The AM system implements a defense-in-depth security architecture with multiple layers of protection.
Internal services are NOT exposed to the internet.
Internet
│
├─→ API Gateway (8000) ✅ EXPOSED
├─→ User Management (8010) ✅ EXPOSED (registration only)
└─→ Auth Tokens (8001) ✅ EXPOSED (token operations)
Internal Docker Network:
├─→ Python Service (8002) ⛔ NOT EXPOSED
├─→ Java Service (8003) ⛔ NOT EXPOSED
└─→ PostgreSQL (5432) ⛔ NOT EXPOSED
Benefits:
- 57% reduction in attack surface (3 vs 7 exposed services)
- Direct attacks on internal services impossible
- Network-level defense before application security
All requests require valid JWT tokens.
Client Request
↓
[User JWT Token]
↓
API Gateway validates
↓
[Generates Service JWT]
↓
Internal Service validates
↓
Process Request
Token Types:
-
User JWT - Client authentication
- Issued by Auth Tokens service
- Contains: user_id, email, roles, permissions
- Lifetime: Configurable (default: 1 hour)
- Used: Client → API Gateway
-
Service JWT - Service authentication
- Generated by API Gateway
- Contains: service_id, permissions, target_service
- Lifetime: Short-lived (5 minutes)
- Used: API Gateway → Internal Services
Prevents abuse and DDoS attacks.
- Limit: 100 requests per 60 seconds
- Scope: Per IP address
- Response: 429 Too Many Requests
- Headers:
X-RateLimit-Limit: 100X-RateLimit-Remaining: 95X-RateLimit-Reset: 1703001234
Configuration (docker-compose.yml):
environment:
- RATE_LIMIT_REQUESTS=100
- RATE_LIMIT_WINDOW=60Users have roles that determine permissions.
Default Roles:
user- Standard user accessadmin- Administrative accessservice- Service-to-service access
Example:
@router.get("/documents/all")
async def get_all_documents(current_user: CurrentUser = Depends(get_current_user)):
if "admin" not in current_user.roles:
raise HTTPException(status_code=403, detail="Admin access required")
# ... proceed with admin-only operationAll requests are logged for security analysis.
Logged information:
- Timestamp
- Client IP address
- User ID (if authenticated)
- HTTP method and path
- Request body (sanitized)
- Response status
- Processing time
- Rate limit status
Log Location: logs/ directory
Client → User Management Service
POST /api/v1/users/register
Body: {email, password, full_name}
Response:
- User created with "pending_activation" status
- Password hashed with bcrypt
- Returns user_id
Admin/System → User Management Service
POST /api/v1/users/{user_id}/activate
Response:
- User status changed to "active"
- User can now login
Client → Auth Tokens Service
POST /api/v1/auth/login
Body: {email, password}
Process:
1. Validate credentials
2. Check user is active
3. Generate JWT token
Response:
- access_token (JWT)
- token_type: "bearer"
- user_id
Client → API Gateway
GET /api/v1/documents
Header: Authorization: Bearer <user_jwt>
Process:
1. Rate limiter checks request count
2. Validate user JWT with auth service
3. Generate service JWT
4. Call internal service with service JWT
5. Return response to client
Internal Service:
1. Validate service JWT
2. Execute business logic
3. Return data
- Minimum 8 characters
- At least 1 uppercase letter
- At least 1 lowercase letter
- At least 1 number
- At least 1 special character
Implementation: app/database/models.py in user-management service
- Tokens stored only in memory (not localStorage for web apps)
- Short expiration times (1 hour for user, 5 min for service)
- Refresh token rotation (implement in production)
- Token revocation on logout
- HTTPS only in production (TLS 1.3)
- CORS configured for specific origins
- Content-Type validation
- Request size limits
- SQL injection prevention (parameterized queries)
- XSS prevention (input sanitization)
Never commit secrets to git!
.env.docker example:
# JWT Secrets (CHANGE IN PRODUCTION!)
JWT_SECRET=jwt-super-secret-signing-key-change-in-production
INTERNAL_JWT_SECRET=internal-service-super-secret-key
# Database
DATABASE_URL=postgresql://user:password@postgres:5432/am_db
# Service URLs
AUTH_SERVICE_URL=http://auth-tokens:8001
USER_MANAGEMENT_URL=http://am-user-management:8000Protection: SQLAlchemy ORM with parameterized queries
# ✅ SAFE
user = session.query(User).filter(User.email == email).first()
# ❌ UNSAFE (don't do this)
query = f"SELECT * FROM users WHERE email = '{email}'"Protection: Input validation and sanitization
from pydantic import EmailStr, validator
class UserCreate(BaseModel):
email: EmailStr
full_name: str
@validator('full_name')
def sanitize_name(cls, v):
# Remove HTML tags, scripts, etc.
return sanitize_input(v)Protection:
- JWT tokens (not cookies)
- SameSite cookie attribute if using cookies
- CORS origin validation
Protection:
- Rate limiting (100 req/60s)
- Connection limits
- Request timeout (30 seconds)
- Load balancer in production
Protection:
- Rate limiting on login endpoint
- Account lockout after N failed attempts
- CAPTCHA after failures (implement in production)
Protection:
- Short expiration times
- Token revocation list
- Refresh token rotation
- Secure storage (memory only, no localStorage)
Protection:
- HTTPS/TLS in production
- Certificate pinning (mobile apps)
- HSTS headers
API Gateway should add these headers:
# In production, add security headers:
@app.middleware("http")
async def add_security_headers(request, call_next):
response = await call_next(request)
response.headers["X-Content-Type-Options"] = "nosniff"
response.headers["X-Frame-Options"] = "DENY"
response.headers["X-XSS-Protection"] = "1; mode=block"
response.headers["Strict-Transport-Security"] = "max-age=31536000"
response.headers["Content-Security-Policy"] = "default-src 'self'"
return response- ✅ A01:2021 - Broken Access Control → RBAC implemented
- ✅ A02:2021 - Cryptographic Failures → JWT, bcrypt hashing
- ✅ A03:2021 - Injection → Parameterized queries
- ✅ A04:2021 - Insecure Design → Defense in depth
- ✅ A05:2021 - Security Misconfiguration → Environment variables
- ✅ A06:2021 - Vulnerable Components → Regular updates
- ✅ A07:2021 - Authentication Failures → JWT + bcrypt
- ✅ A08:2021 - Software/Data Integrity → Code signing
- ✅ A09:2021 - Security Logging → Audit logs
- ✅ A10:2021 - Server-Side Request Forgery → Input validation
- User data encryption at rest
- Right to be forgotten (user deletion)
- Data minimization (collect only needed data)
- Audit logs for data access
- Consent management
- Environment variables for secrets
- Input validation on all endpoints
- Parameterized database queries
- Password hashing (bcrypt)
- JWT token validation
- Rate limiting configured
- Audit logging enabled
- HTTPS/TLS configured
- Security headers enabled
- Rate limiting tuned for traffic
- Secrets rotated regularly
- Monitoring and alerting
- Intrusion detection system (IDS)
- Web application firewall (WAF)
- Regular security audits
- Penetration testing
- Backup and disaster recovery
-
Immediate Actions:
- Rotate all JWT secrets
- Revoke all active tokens
- Lock affected accounts
- Enable debug logging
-
Investigation:
- Review audit logs
- Identify attack vector
- Assess data exposure
- Document timeline
-
Remediation:
- Fix vulnerability
- Deploy patch
- Notify affected users
- Update security documentation
-
Post-Incident:
- Conduct post-mortem
- Update security procedures
- Implement additional monitoring
- Train team on lessons learned
- Failed login attempts
- Rate limit violations
- 401/403 error rates
- Token validation failures
- Unusual traffic patterns
- Geographic anomalies
- Spike in failed authentications
- Rate limit threshold exceeded
- Service health degradation
- Unusual service-to-service calls
- Database connection errors
Security is everyone's responsibility! 🔐
Report security issues immediately to the security team.