Security: Configuration & Examples

from svc_infra.security.lockout import get_lockout_status, record_attempt

# Compute ip_hash from request.client.host
status = await get_lockout_status(session, user_id=None, ip_hash=ip_hash)
if status.locked:
		raise HTTPException(429, headers={"Retry-After": ..})

user = await user_manager.user_db.get_by_email(email)
if not user:
		await record_attempt(session, user_id=None, ip_hash=ip_hash, success=False)
		raise HTTPException(400, "LOGIN_BAD_CREDENTIALS")

GET /sessions/me
Permission: security.session.list

POST /sessions/{session_id}/revoke
Permission: security.session.revoke

DELETE /sessions/{session_id}
Permission: security.session.revoke

DELETE /sessions/me/all
Permission: security.session.revoke

from svc_infra.security.signed_cookies import sign_cookie, verify_cookie

sig = sign_cookie({"sub": "user-123"}, secret="k1", exp_seconds=3600)
payload = verify_cookie(sig, secret="k1", old_secrets=["k0"])  # returns dict

default-src 'self';
script-src 'self' 'unsafe-inline' https://cdn.jsdelivr.net;
style-src 'self' 'unsafe-inline' https://cdn.jsdelivr.net;
img-src 'self' data: https:;
connect-src 'self';
font-src 'self' https://cdn.jsdelivr.net;
frame-ancestors 'none';
base-uri 'self';
form-action 'self'

from fastapi import FastAPI

from svc_infra.security.add import add_security

app = FastAPI()

add_security(
    app,
    cors_origins=["https://app.example.com"],
    headers_overrides={"Content-Security-Policy": "default-src 'self'; script-src 'self'"},  # Stricter CSP
    install_session_middleware=True,  # adds Starlette's SessionMiddleware
)

from svc_infra.api.fastapi.setup import setup_service_api
from svc_infra.security.add import add_security

app = setup_service_api(...)

add_security(
    app,
    headers_overrides={"Strict-Transport-Security": "max-age=63072000; includeSubDomains"},
    enable_hsts_preload=False,
)

from svc_infra.security.lockout import LockoutConfig, get_lockout_status, record_attempt
from svc_infra.security.hibp import check_password_breach

# 1. Rate limiting by IP
lockout_cfg = LockoutConfig(
    threshold=5,              # 5 failed attempts
    window_minutes=15,        # Within 15 minutes
    base_cooldown_seconds=30, # Start at 30s lockout
    max_cooldown_seconds=3600 # Max 1 hour lockout
)

@router.post("/login")
async def login(request: Request, creds: LoginRequest, session = Depends(get_session)):
    ip_hash = hashlib.sha256(request.client.host.encode()).hexdigest()[:16]

    # Check lockout
    status = await get_lockout_status(session, user_id=None, ip_hash=ip_hash, cfg=lockout_cfg)
    if status.locked:
        raise HTTPException(
            429,
            "Too many failed attempts",
            headers={"Retry-After": str(status.retry_after_seconds)}
        )

    user = await authenticate(session, creds.email, creds.password)
    if not user:
        await record_attempt(session, user_id=None, ip_hash=ip_hash, success=False)
        await session.commit()
        raise HTTPException(401, "Invalid credentials")

    # 2. Check if password was in a breach
    is_breached = await check_password_breach(creds.password)
    if is_breached:
        # Force password change on next login
        await flag_password_change_required(session, user.id)

    await record_attempt(session, user_id=user.id, ip_hash=ip_hash, success=True)
    return create_tokens(user)

# 1. Secure cookie configuration (automatic in svc-infra)
response.set_cookie(
    key="session",
    value=token,
    httponly=True,    # No JavaScript access
    secure=True,      # HTTPS only
    samesite="lax",   # CSRF protection
    max_age=3600,     # Short TTL
)

# 2. Session binding to fingerprint
def create_session_token(user_id: str, request: Request) -> str:
    fingerprint = hashlib.sha256(
        f"{request.headers.get('user-agent', '')}"
        f"{request.client.host}".encode()
    ).hexdigest()[:16]

    return jwt.encode({
        "sub": user_id,
        "fp": fingerprint,
        "iat": datetime.utcnow(),
    }, SECRET)

# 3. Validate fingerprint on each request
def validate_session(token: str, request: Request) -> bool:
    payload = jwt.decode(token, SECRET)
    expected_fp = hashlib.sha256(
        f"{request.headers.get('user-agent', '')}"
        f"{request.client.host}".encode()
    ).hexdigest()[:16]
    return payload.get("fp") == expected_fp

#  VULNERABLE: String concatenation
query = f"SELECT * FROM users WHERE email = '{user_input}'"

#  SAFE: Parameterized query (SQLAlchemy)
stmt = select(User).where(User.email == user_input)
result = await session.execute(stmt)

#  SAFE: Using text() with parameters
from sqlalchemy import text
stmt = text("SELECT * FROM users WHERE email = :email")
result = await session.execute(stmt, {"email": user_input})

# 1. SameSite cookies (automatic)
# Cookie: session=...; SameSite=Lax

# 2. CSRF tokens for state-changing operations
from svc_infra.security.csrf import generate_csrf_token, validate_csrf_token

@router.get("/profile")
async def get_profile(request: Request):
    csrf_token = generate_csrf_token(request)
    return {"data": ..., "csrf_token": csrf_token}

@router.post("/profile")
async def update_profile(request: Request, body: ProfileUpdate):
    csrf_token = request.headers.get("X-CSRF-Token")
    if not validate_csrf_token(request, csrf_token):
        raise HTTPException(403, "Invalid CSRF token")
    # Process update

# 3. Custom header requirement
# Attacker cannot set custom headers in cross-origin requests
X-Requested-With: XMLHttpRequest

# 1. Short token TTL
ACCESS_TOKEN_TTL = 900  # 15 minutes

# 2. Refresh token rotation with revocation
async def refresh_tokens(refresh_token: str, session) -> TokenPair:
    payload = jwt.decode(refresh_token, SECRET)

    # Check if token was already used (rotation)
    if await is_token_revoked(session, payload["jti"]):
        # Token reuse detected - revoke all user's tokens
        await revoke_all_tokens_for_user(session, payload["sub"])
        raise HTTPException(401, "Token reuse detected - all sessions revoked")

    # Revoke old refresh token
    await revoke_token(session, payload["jti"])

    # Issue new token pair
    return create_token_pair(payload["sub"])

# 3. One-time use tokens for sensitive operations
async def generate_password_reset_token(user_id: str, session) -> str:
    token = secrets.token_urlsafe(32)
    await store_token(session, user_id, token, ttl=3600, one_time=True)
    return token

from svc_infra.security.permissions import RequirePermission, RequireRoles

# 1. Defense in depth - both role AND permission checks
@router.get(
    "/admin/users",
    dependencies=[
        RequireRoles("admin"),           # Role gate
        RequirePermission("user.read"),  # Permission gate
    ]
)
async def list_users():
    ...

# 2. ABAC for resource ownership
from svc_infra.security.permissions import RequireABAC, owns_resource

@router.delete(
    "/documents/{doc_id}",
    dependencies=[RequireABAC(
        permission="document.delete",
        predicate=owns_resource("owner_id"),
        resource_getter=get_document,
    )]
)
async def delete_document(doc_id: str):
    ...

# 3. Tenant isolation at data layer
from svc_infra.tenancy import TenantSqlService

class UserService(TenantSqlService[User]):
    # All queries automatically scoped to current tenant
    pass

# Discover API endpoints
curl -s https://api.example.com/openapi.json | jq '.paths | keys'

# Enumerate users (should fail with 404, not 401)
for email in admin@example.com test@example.com; do
  curl -s -o /dev/null -w "%{http_code}" \
    "https://api.example.com/v1/auth/forgot-password" \
    -d "{\"email\": \"$email\"}"
done

# Test lockout
for i in {1..10}; do
  curl -X POST https://api.example.com/v1/auth/login \
    -H "Content-Type: application/json" \
    -d '{"email":"test@example.com","password":"wrong"}'
done
# Expected: 429 after threshold

# Test JWT signature validation
# Modify token and replay
TOKEN="eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJ1c2VyIn0.SIGNATURE"
MODIFIED=$(echo $TOKEN | sed 's/user/admin/')
curl -H "Authorization: Bearer $MODIFIED" https://api.example.com/v1/profile
# Expected: 401

# IDOR - Access other user's resources
curl -H "Authorization: Bearer $USER_TOKEN" \
  https://api.example.com/v1/users/OTHER_USER_ID/documents
# Expected: 403 or 404 (not 200)

# Privilege escalation
curl -X POST -H "Authorization: Bearer $USER_TOKEN" \
  https://api.example.com/admin/users
# Expected: 403

# Tenant isolation
curl -H "Authorization: Bearer $TENANT_A_TOKEN" \
  https://api.example.com/v1/resources?tenant_id=TENANT_B
# Expected: 403 or only TENANT_A resources

# SQL injection
curl "https://api.example.com/v1/search?q=test'%20OR%201=1--"
# Expected: 400 (validation error)

# Command injection
curl -X POST https://api.example.com/v1/export \
  -d '{"format": "; rm -rf /"}'
# Expected: 400

# Path traversal
curl "https://api.example.com/v1/files/..%2F..%2Fetc%2Fpasswd"
# Expected: 400 or 404

# Example OWASP ZAP configuration
zap:
  target: https://api.example.com
  authentication:
    type: http
    loginUrl: /v1/auth/login
    loginRequestData: '{"email":"test@example.com","password":"test123"}'
    tokenName: access_token
  rules:
    - id: 10020  # X-Frame-Options
    - id: 10021  # X-Content-Type-Options
    - id: 10038  # Content Security Policy
    - id: 40012  # Cross Site Scripting (Reflected)
    - id: 40014  # Cross Site Scripting (Persistent)
    - id: 40018  # SQL Injection

## Finding: [SEVERITY] - Title

### Description
Brief description of the vulnerability.

### Affected Endpoint
`POST /v1/endpoint`

### Steps to Reproduce
1. Step one
2. Step two
3. Observe result

### Evidence

### Impact
What an attacker could do.

### Remediation
Recommended fix.

### References
- OWASP reference
- CWE number

# Required configuration
AUTH_JWT__TTL = 900                    # Short token TTL
AUTH_MFA__REQUIRED_ROLES = "admin"     # MFA for admins
AUTH_PASSWORD_MIN_LENGTH = 12
AUTH_PASSWORD_BREACH_CHECK = True

# Required code
from svc_infra.security.audit import append_audit_event

# Log all security-relevant events
await append_audit_event(
    session,
    actor_id=user.id,
    event_type="user.login.success",
    resource_ref=f"user:{user.id}",
    metadata={"ip": request.client.host, "user_agent": request.headers.get("user-agent")},
)

# Required configuration
AUTH_JWT__TTL = 300                    # 5 minute tokens for PHI access
SESSION_IDLE_TIMEOUT = 900             # 15 minute idle timeout
AUDIT_RETENTION_DAYS = 2190            # 6 year retention

# Required: Log access to PHI
await append_audit_event(
    session,
    actor_id=user.id,
    event_type="phi.access",
    resource_ref=f"patient:{patient_id}",
    metadata={"fields_accessed": fields, "reason": "treatment"},
)