API Security Engineer

# VULNERABLE: No authorization check on object access
@app.route('/api/v1/accounts/<account_id> output_file')
@login_required
def get_transactions(account_id):
    transactions = db.query(Transaction).filter_by(account_id=account_id).all()
    return jsonify([t.to_dict() for t in transactions])

# SECURE: Object-level authorization check
@app.route('/api/v1/accounts/<account_id> output_file')
@login_required
def get_transactions(account_id):
    account = db.query(Account).get(account_id)
    if account is None:
        abort(404)  # Don't reveal resource existence
    if not current_user.can_access(account):
        abort(403)
    transactions = db.query(Transaction).filter_by(account_id=account_id).all()
    return jsonify([t.to_dict() for t in transactions])

# API2 - Rate-limited authentication
@app.route('/api/v1/auth/login', methods=['POST'])
@limiter.limit("5 per minute")
def login():
    email = request.json.get('email', '')
    if is_account_locked(email):
        return jsonify({"error": "Account temporarily locked"}), 429
    user = authenticate(email, request.json.get('password', ''))
    if not user:
        record_failed_attempt(email)
        return jsonify({"error": "Invalid credentials"}), 401
    return jsonify({"token": generate_token(user)})

# API3 - Prevent mass assignment with allowlists
@app.route('/api/v1/users', methods=['POST'])
def create_user():
    ALLOWED_FIELDS = {'name', 'email', 'phone'}
    data = {k: v for k, v in request.json.items() if k in ALLOWED_FIELDS}
    user = User(**data)
    user.role = 'user'  # Always default
    db.save(user)
    return jsonify({"id": user.id, "name": user.name, "email": user.email})

# API4 - Enforced pagination
@app.route('/api/v1/products')
def list_products():
    MAX_LIMIT = 100
    limit = min(request.args.get('limit', 20, type=int), MAX_LIMIT)
    offset = max(request.args.get('offset', 0, type=int), 0)
    products = db.query(Product).limit(limit).offset(offset).all()
    total = db.query(Product).count()
    return jsonify({
        "data": [p.to_dict() for p in products],
        "pagination": {"limit": limit, "offset": offset, "total": total}
    })

import jwt, uuid
from datetime import datetime, timedelta, timezone

class JWTManager:
    def __init__(self, secret_key, issuer, audience):
        self.secret_key = secret_key
        self.issuer = issuer
        self.audience = audience

    def create_token(self, user_id, roles, expiry_minutes=15):
        now = datetime.now(timezone.utc)
        return jwt.encode({
            "sub": user_id, "roles": roles,
            "iss": self.issuer, "aud": self.audience,
            "iat": now, "exp": now + timedelta(minutes=expiry_minutes),
            "jti": str(uuid.uuid4()),
        }, self.secret_key, algorithm="HS256")

    def verify_token(self, token):
        payload = jwt.decode(
            token, self.secret_key,
            algorithms=["HS256"],  # Explicit allowlist
            issuer=self.issuer, audience=self.audience,
            options={"require": ["exp", "iss", "aud", "sub", "jti"]}
        )
        if self.is_revoked(payload["jti"]):
            raise jwt.InvalidTokenError("Token revoked")
        return payload

Creation:
  [ ] Strong algorithm (RS256 distributed, HS256 single service)
  [ ] Short expiration (15 min access, longer refresh)
  [ ] Include sub, iss, aud, exp, iat, jti
  [ ] Never store sensitive data in payload
  [ ] Sign with 256+ bit secret

Validation:
  [ ] Always verify signature
  [ ] Validate algorithm against allowlist (prevent "none" attack)
  [ ] Check exp, iss, aud claims
  [ ] Check revocation status

Transport:
  [ ] HTTPS only
  [ ] httpOnly secure SameSite cookies (browsers)
  [ ] Authorization: Bearer header (API clients)
  [ ] Never in localStorage (XSS vulnerable)

import secrets, hashlib

class APIKeyManager:
    def create_key(self, owner_id, key_type="live", scopes=None):
        prefix = {"live": "sk_live_", "test": "sk_test_"}[key_type]
        raw_key = prefix + secrets.token_urlsafe(32)
        key_hash = hashlib.sha256(raw_key.encode()).hexdigest()
        db.save_api_key({
            "key_hash": key_hash,
            "prefix": raw_key[:12],
            "owner_id": owner_id,
            "scopes": scopes or ["read"],
            "is_active": True
        })
        return {"api_key": raw_key, "warning": "Store securely. Cannot be retrieved again."}

    def validate_key(self, raw_key):
        key_hash = hashlib.sha256(raw_key.encode()).hexdigest()
        record = db.get_by_hash(key_hash)
        if not record or not record["is_active"]:
            raise AuthError("Invalid or revoked API key")
        return record

[ ] Keys stored as SHA-256 hashes (never plaintext)
[ ] Displayed to user exactly once at creation
[ ] In env variables or vault (never in code)
[ ] Excluded from logs (masked in middleware)
[ ] Per-key rate limits enforced
[ ] Maximum age policy (90-365 days)
[ ] Automated expiration warnings
[ ] Rotation with grace period
[ ] Unused key detection (revoke after 90+ days idle)

global:              10000 req/sec (DDoS protection)
per_ip:              60 req/min (abuse prevention)
per_key:
  free:              100 req/hour
  basic:             1000 req/hour
  pro:               10000 req/hour
per_endpoint:
  /auth/login:       5 req/min per IP
  /auth/reset:       3 req/hour per IP
  /search:           30 req/min per key
  /export:           10 req/hour per key

RateLimit-Limit: 100
RateLimit-Remaining: 42
RateLimit-Reset: 1625000000
Retry-After: 30          # Only on 429 responses

from pydantic import BaseModel, Field, validator
import re

class OrderCreateRequest(BaseModel):
    product_id: str = Field(
        ..., min_length=36, max_length=36,
        pattern=r'^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$'
    )
    quantity: int = Field(..., ge=1, le=1000)
    shipping_address: str = Field(..., min_length=10, max_length=500)
    notes: str | None = Field(None, max_length=2000)

    @validator('shipping_address')
    def sanitize(cls, v):
        return re.sub(r'[<>{}]', '', v).strip()

security:
  tls:
    min_version: "TLSv1.2"
    hsts: {enabled: true, max_age: 31536000, include_subdomains: true}
  request_limits:
    max_body_size: "10MB"
    max_header_size: "8KB"
    max_uri_length: 2048
    request_timeout: "30s"
  response_headers:
    X-Content-Type-Options: "nosniff"
    X-Frame-Options: "DENY"
    Cache-Control: "no-store"
  cors:
    allowed_origins: ["[reference URL]"]
    allowed_methods: ["GET", "POST", "PUT", "DELETE", "PATCH"]
    allow_credentials: true

Authentication & Authorization:
  [ ] All endpoints require authentication
  [ ] Object-level authorization on every resource endpoint
  [ ] Function-level authorization (admin routes restricted)
  [ ] No mass assignment vulnerabilities

Input & Output:
  [ ] Strict schema validation (type, length, format)
  [ ] Explicit output serialization (no raw object dumps)
  [ ] Pagination enforced with maximum
  [ ] Error responses don't leak internals

Rate Limiting:
  [ ] Global rate limit configured
  [ ] Per-client limits enforced
  [ ] Auth endpoints strictly limited
  [ ] 429 responses include Retry-After

Transport:
  [ ] TLS 1.2+ enforced
  [ ] HSTS header set
  [ ] No sensitive data in URLs

Monitoring:
  [ ] All API calls logged
  [ ] Auth failures alerted
  [ ] Rate limit violations tracked
  [ ] API inventory maintained (no shadow APIs)

# OWASP ZAP API scan
docker run -t zaproxy/zap-stable zap-api-scan.py \
    -t [reference URL] -f openapi -r report.html

# BOLA test pattern:
#   1. Auth as User A, create resource, note ID
#   2. Auth as User B
#   3. Access User A's resource with User B's token
#   4. Expect: 404 (not 200)