Keyarc API Security

KeyArc API Security Patterns

Overview

Security patterns for KeyArc's zero-knowledge FastAPI backend services. Covers JWT authentication, encrypted payload validation, audit logging, and OWASP top 10 prevention.

Service Architecture Context

KeyArc uses a multi-service architecture:

Key security boundaries:

• Gateway handles JWT validation - private services trust user context headers from Gateway
• Auth Service is the only service that validates authHash (never passwords)
• RBAC checks happen in Account Service and Key Service using shared RBAC module
• Audit logging is a shared module used by all services

When to Use

✅ Apply this skill when:

• Creating new API endpoints in any service
• Implementing authentication/authorization
• Handling secrets or encrypted data
• Adding audit logging
• Reviewing security of existing endpoints
• Implementing rate limiting
• Handling errors that might leak data

Core Security Principles

1. Validate encrypted payloads - Never accept plaintext secrets
2. JWT token authentication - Validate using authHash, not passwords
3. Audit everything - Log all secret access (not values)
4. Fail securely - Errors don't leak sensitive data
5. Rate limit auth - Prevent brute force attacks
6. RBAC enforcement - Check permissions at API level

Endpoint Security Checklist

Before merging any new endpoint:

Auth Service endpoints:

• Rate limiting applied (5/minute for login)
• authHash validated (never password)
• Audit log entry created
• Error responses don't leak data

Gateway:

• JWT token validated
• User context headers added for downstream
• Invalid tokens rejected with 401

Account/Key Service endpoints (private):

• User context from Gateway headers (use Depends(get_current_user_from_gateway))
• RBAC permissions checked for team operations (use shared RBAC module)
• Encrypted payloads validated - no plaintext secrets (Key Service)
• Audit log entry created
• Error responses don't leak secrets
• Input validation with Pydantic
• SQL injection prevented (use ORM)

JWT Authentication Pattern

# dependencies.py
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from jose import JWTError, jwt
from sqlalchemy.ext.asyncio import AsyncSession

security = HTTPBearer()

async def get_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(security),
    db: AsyncSession = Depends(get_db)
) -> User:
    """Validate JWT token and return current user."""
    token = credentials.credentials

    try:
        payload = jwt.decode(
            token,
            SECRET_KEY,
            algorithms=[ALGORITHM]
        )
        user_id: int = payload.get("sub")
        if user_id is None:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Invalid authentication credentials"
            )
    except JWTError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid authentication credentials"
        )

    # Get user from database
    result = await db.execute(select(User).where(User.id == user_id))
    user = result.scalar_one_or_none()

    if user is None:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="User not found"
        )

    return user

Creating Tokens (Auth Service Only)

# src/services/auth/app/utils/security.py
from datetime import datetime, timedelta
from jose import jwt

SECRET_KEY = os.getenv("JWT_SECRET")  # From environment
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 120  # 2 hours

def create_access_token(data: dict) -> str:
    """Create JWT access token. Auth Service only."""
    to_encode = data.copy()
    expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire})

    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

Gateway User Context

The Gateway validates JWTs and passes user context to private services via headers:

# Gateway adds these headers after JWT validation:
# X-User-ID: <user_id>
# X-User-Email: <email>

# Private services (Account, Key) read user from headers:
# src/services/account/app/dependencies.py or src/services/keys/app/dependencies.py

from fastapi import Header, HTTPException

async def get_current_user_from_gateway(
    x_user_id: str = Header(..., alias="X-User-ID"),
    x_user_email: str = Header(None, alias="X-User-Email"),
) -> dict:
    """Get user context from Gateway headers. For private services only."""
    if not x_user_id:
        raise HTTPException(401, "Missing user context")

    return {
        "id": int(x_user_id),
        "email": x_user_email
    }

Important: Private services should only be accessible via Gateway (Fly.io private networking). They trust Gateway headers because external traffic cannot reach them directly.

Secure Endpoint Example

@router.post("/secrets", response_model=SecretResponse, status_code=201)
async def create_secret(
    secret_data: SecretCreate,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db)
):
    """Create a new encrypted secret."""

    # Validate encrypted payload format
    if not is_valid_ciphertext(secret_data.encrypted_value):
        raise HTTPException(400, "Invalid encrypted format")

    # Create secret (store ciphertext only)
    secret = Secret(
        user_id=current_user.id,
        name=secret_data.name,
        encrypted_value=secret_data.encrypted_value,
        folder_id=secret_data.folder_id,
        tags=secret_data.tags
    )

    db.add(secret)
    await db.commit()
    await db.refresh(secret)

    # Audit log (NO secret value!)
    await create_audit_log(
        db=db,
        user_id=current_user.id,
        action="secret.create",
        resource_type="secret",
        resource_id=secret.id,
        ip_address=request.client.host  # From request context
    )

    return secret

Pydantic Schemas for Security

from pydantic import BaseModel, EmailStr, Field, validator

class SecretCreate(BaseModel):
    """Schema for creating a secret."""
    name: str = Field(min_length=1, max_length=255)
    encrypted_value: str = Field(min_length=1)
    folder_id: int | None = None
    tags: list[str] = Field(default_factory=list)

    @validator('encrypted_value')
    def validate_encrypted(cls, v):
        """Ensure value is actually encrypted (base64)."""
        try:
            import base64
            base64.b64decode(v)
            return v
        except Exception:
            raise ValueError('encrypted_value must be base64-encoded ciphertext')

class LoginRequest(BaseModel):
    """Login request with auth hash (NOT password)."""
    email: EmailStr
    auth_hash: str = Field(min_length=1)

    # NO password field! Only authHash!

Audit Logging

async def create_audit_log(
    db: AsyncSession,
    user_id: int,
    action: str,
    resource_type: str,
    resource_id: int,
    ip_address: str,
    metadata: dict | None = None
) -> AuditLog:
    """Create audit log entry."""
    log = AuditLog(
        user_id=user_id,
        action=action,  # 'secret.view', 'secret.create', 'secret.delete', etc.
        resource_type=resource_type,
        resource_id=resource_id,
        ip_address=ip_address,
        metadata=metadata,  # Never include secret values!
        timestamp=datetime.utcnow()
    )

    db.add(log)
    await db.flush()  # Don't commit yet (part of larger transaction)

    return log

# Usage in endpoint:
await create_audit_log(
    db=db,
    user_id=current_user.id,
    action="secret.view",
    resource_type="secret",
    resource_id=secret_id,
    ip_address=request.client.host
)

RBAC Permission Checking

async def check_team_permission(
    user_id: int,
    team_id: int,
    required_role: str,  # 'owner', 'admin', 'member'
    db: AsyncSession
) -> TeamMembership:
    """Check if user has required role in team."""
    result = await db.execute(
        select(TeamMembership).where(
            TeamMembership.user_id == user_id,
            TeamMembership.team_id == team_id
        )
    )
    membership = result.scalar_one_or_none()

    if not membership:
        raise HTTPException(403, "Not a team member")

    role_hierarchy = {'member': 0, 'admin': 1, 'owner': 2}
    if role_hierarchy.get(membership.role, -1) < role_hierarchy.get(required_role, 999):
        raise HTTPException(403, "Insufficient permissions")

    return membership

Rate Limiting

from slowapi import Limiter
from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address)

@router.post("/auth/login")
@limiter.limit("5/minute")  # 5 attempts per minute per IP
async def login(
    request: Request,
    login_data: LoginRequest,
    db: AsyncSession = Depends(get_db)
):
    # ... login logic

Error Handling (Secure)

# ❌ BAD: Leaks sensitive data
@router.get("/secrets/{secret_id}")
async def get_secret(secret_id: int):
    secret = await db.get(Secret, secret_id)
    if not secret:
        raise HTTPException(404, f"Secret {secret.encrypted_value} not found")  # LEAK!

# ✅ GOOD: Generic errors
@router.get("/secrets/{secret_id}")
async def get_secret(
    secret_id: int,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db)
):
    secret = await db.get(Secret, secret_id)

    if not secret:
        raise HTTPException(404, "Secret not found")

    # Check ownership
    if secret.user_id != current_user.id:
        # Same error (don't reveal if secret exists)
        raise HTTPException(404, "Secret not found")

    # Audit log
    await create_audit_log(
        db=db,
        user_id=current_user.id,
        action="secret.view",
        resource_type="secret",
        resource_id=secret_id,
        ip_address=request.client.host
    )

    await db.commit()

    return secret

OWASP Top 10 Quick Reference

See owasp-checklist.md for detailed prevention patterns.

See Also

• jwt-patterns.md - Detailed JWT implementation
• audit-logging.md - Comprehensive audit logging requirements
• owasp-checklist.md - Detailed OWASP top 10 prevention

Key Principles

1. Validate encrypted payloads: Never accept plaintext secrets
2. Auth with JWT: Token-based, stateless authentication
3. Audit everything: Log access, not values
4. RBAC at API layer: Enforce permissions in endpoints
5. Rate limit sensitive endpoints: Prevent brute force
6. Secure error handling: Generic errors, no leaks
7. Input validation: Always use Pydantic schemas

Security is not optional - apply these patterns to every endpoint.