frankbria
diff --git a/‎codeframe/auth/api_key_router.py‎
Lines changed: 238 additions & 0 deletions b/‎codeframe/auth/api_key_router.py‎
Lines changed: 238 additions & 0 deletions
diff --git a/‎codeframe/auth/api_keys.py‎
Lines changed: 156 additions & 0 deletions b/‎codeframe/auth/api_keys.py‎
Lines changed: 156 additions & 0 deletions
@@ -0,0 +1,238 @@
+"""API key management endpoints.
+
+Provides endpoints for creating, listing, and revoking API keys.
+API key creation requires JWT authentication (not API key auth) to prevent
+privilege escalation attacks.
+
+Uses core/api_key_service.py for business logic (shared with CLI).
+"""
+
+import logging
+import os
+from datetime import datetime, timezone
+from typing import List, Optional
+
+from fastapi import APIRouter, Depends, HTTPException, Request, status
+from pydantic import BaseModel, field_validator
+
+from codeframe.auth.dependencies import get_current_user, require_auth
+from codeframe.auth.models import User
+from codeframe.auth.api_keys import (
+    validate_scopes,
+    SCOPE_READ,
+    SCOPE_WRITE,
+)
+from codeframe.core.api_key_service import ApiKeyService
+from codeframe.persistence.database import Database
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(prefix="/api/auth/api-keys", tags=["auth", "api-keys"])
+
+
+# =============================================================================
+# Pydantic Models
+# =============================================================================
+
+
+class CreateApiKeyRequest(BaseModel):
+    """Request body for creating an API key."""
+
+    name: str
+    scopes: List[str] = [SCOPE_READ, SCOPE_WRITE]
+    expires_at: Optional[datetime] = None
+
+    @field_validator("scopes")
+    @classmethod
+    def validate_scopes_field(cls, v: List[str]) -> List[str]:
+        if not validate_scopes(v):
+            raise ValueError(f"Invalid scopes: {v}. Valid scopes are: read, write, admin")
+        # Deduplicate while preserving order
+        seen = set()
+        deduped = [s for s in v if not (s in seen or seen.add(s))]
+        if not deduped:
+            raise ValueError("At least one scope is required")
+        return deduped
+
+    @field_validator("expires_at")
+    @classmethod
+    def validate_expires_at(cls, v: Optional[datetime]) -> Optional[datetime]:
+        if v is not None:
+            now = datetime.now(timezone.utc)
+            # Handle naive datetime by assuming UTC
+            if v.tzinfo is None:
+                v = v.replace(tzinfo=timezone.utc)
+            if v <= now:
+                raise ValueError("expires_at must be in the future")
+        return v
+
+
+class CreateApiKeyResponse(BaseModel):
+    """Response body for API key creation."""
+
+    key: str  # Full key - shown only once
+    id: str
+    prefix: str
+    created_at: str
+
+
+class ApiKeyInfoResponse(BaseModel):
+    """API key information (without sensitive data)."""
+
+    id: str
+    name: str
+    prefix: str
+    scopes: List[str]
+    created_at: str
+    last_used_at: Optional[str]
+    expires_at: Optional[str]
+    is_active: bool
+
+
+class RevokeApiKeyResponse(BaseModel):
+    """Response body for API key revocation."""
+
+    id: str
+    revoked: bool
+
+
+# =============================================================================
+# Database Helper
+# =============================================================================
+
+
+def get_db(request: Request) -> Database:
+    """Get database instance from app state (singleton managed by lifespan handler).
+
+    Uses the app-scoped database to avoid per-request connection leaks.
+    Falls back to DATABASE_PATH env var if app.state.db not available.
+    """
+    # Prefer app-scoped singleton (set by lifespan handler in server.py)
+    db = getattr(request.app.state, "db", None)
+    if db is not None:
+        return db
+
+    # Fallback for tests or standalone usage
+    db = getattr(request.state, "db", None)
+    if db is None:
+        logger.warning("No db in app.state, creating fallback connection")
+        db_path = os.getenv(
+            "DATABASE_PATH",
+            os.path.join(os.getcwd(), ".codeframe", "state.db")
+        )
+        db = Database(db_path)
+        db.initialize()
+        request.state.db = db
+    return db
+
+
+def get_api_key_service(request: Request) -> ApiKeyService:
+    """Get API key service instance."""
+    db = get_db(request)
+    return ApiKeyService(db)
+
+
+# =============================================================================
+# Endpoints
+# =============================================================================
+
+
+@router.post("", status_code=status.HTTP_201_CREATED, response_model=CreateApiKeyResponse)
+async def create_api_key(
+    request: Request,
+    body: CreateApiKeyRequest,
+    current_user: User = Depends(get_current_user),  # JWT required, not API key
+):
+    """Create a new API key.
+
+    **Important**: This endpoint requires JWT authentication. API keys cannot
+    be used to create new API keys (prevents privilege escalation).
+
+    The full API key is returned only once. Store it securely - it cannot
+    be retrieved again.
+
+    Args:
+        body: API key configuration (name, scopes, optional expiration)
+
+    Returns:
+        Created API key details including the full key (shown once)
+    """
+    service = get_api_key_service(request)
+
+    result = service.create_api_key(
+        user_id=current_user.id,
+        name=body.name,
+        scopes=body.scopes,
+        expires_at=body.expires_at,
+    )
+
+    return CreateApiKeyResponse(
+        key=result.key,
+        id=result.id,
+        prefix=result.prefix,
+        created_at=result.created_at,
+    )
+
+
+@router.get("", response_model=List[ApiKeyInfoResponse])
+async def list_api_keys(
+    request: Request,
+    auth: dict = Depends(require_auth),  # Either JWT or API key
+):
+    """List all API keys for the authenticated user.
+
+    Does not expose the key hash or full key. Shows prefix for identification.
+
+    Returns:
+        List of API key information
+    """
+    service = get_api_key_service(request)
+
+    keys = service.list_api_keys(user_id=auth["user_id"])
+
+    return [
+        ApiKeyInfoResponse(
+            id=k.id,
+            name=k.name,
+            prefix=k.prefix,
+            scopes=k.scopes,
+            created_at=k.created_at,
+            last_used_at=k.last_used_at,
+            expires_at=k.expires_at,
+            is_active=k.is_active,
+        )
+        for k in keys
+    ]
+
+
+@router.delete("/{key_id}", response_model=RevokeApiKeyResponse)
+async def revoke_api_key(
+    request: Request,
+    key_id: str,
+    auth: dict = Depends(require_auth),  # Either JWT or API key
+):
+    """Revoke an API key.
+
+    The key will be marked as inactive and can no longer be used for
+    authentication.
+
+    Args:
+        key_id: The API key ID to revoke
+
+    Returns:
+        Confirmation of revocation
+
+    Raises:
+        404: If key not found or not owned by user
+    """
+    service = get_api_key_service(request)
+
+    success = service.revoke_api_key(key_id, user_id=auth["user_id"])
+
+    if not success:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="API key not found",
+        )
+
+    return RevokeApiKeyResponse(id=key_id, revoked=True)
@@ -0,0 +1,156 @@
+"""API key generation, verification, and utility functions.
+
+Provides utilities for creating and validating API keys for server authentication.
+API keys follow the format: cf_{environment}_{32_hex_chars}
+
+Security considerations:
+- Keys are hashed using SHA256 (industry standard for high-entropy API keys)
+- API keys have 128 bits of entropy from secrets.token_hex(16)
+- SHA256 is appropriate here (vs bcrypt for passwords) because API keys
+  are already high-entropy random strings, not human-chosen passwords
+- Only the prefix is stored for efficient lookup
+- Full key is shown only once during creation
+
+This approach matches GitHub, Stripe, and other API key implementations.
+"""
+
+import hashlib
+import hmac
+import secrets
+import logging
+from typing import Tuple
+
+logger = logging.getLogger(__name__)
+
+# Scope constants for API key permissions
+SCOPE_READ = "read"
+SCOPE_WRITE = "write"
+SCOPE_ADMIN = "admin"
+
+# All valid scopes
+VALID_SCOPES = frozenset({SCOPE_READ, SCOPE_WRITE, SCOPE_ADMIN})
+
+# Key format constants
+# PREFIX_LENGTH=12 captures "cf_live_" (8) + 4 random chars for efficient DB lookup
+# Example: "cf_live_1abc" from "cf_live_1abc2b2f78d93788b9fa00e383832be3"
+KEY_PREFIX_LENGTH = 12
+KEY_RANDOM_BYTES = 16   # 16 bytes = 32 hex characters
+
+
+def generate_api_key(environment: str = "live") -> Tuple[str, str, str]:
+    """Generate a new API key with SHA256 hash.
+
+    Args:
+        environment: Environment identifier ('live' or 'test')
+
+    Returns:
+        Tuple of (full_key, key_hash, prefix):
+        - full_key: The complete API key to give to the user (shown once)
+        - key_hash: SHA256 hash of the key for storage (bcrypt-like format)
+        - prefix: First 12 characters for efficient lookup
+
+    Example:
+        >>> key, hash, prefix = generate_api_key()
+        >>> key
+        'cf_live_a1b2c3d4e5f6789012345678abcdef01'
+        >>> prefix
+        'cf_live_a1b2'
+    """
+    # Generate 16 random bytes = 32 hex characters (128 bits entropy)
+    random_part = secrets.token_hex(KEY_RANDOM_BYTES)
+
+    # Construct the full key: cf_{environment}_{32_hex}
+    full_key = f"cf_{environment}_{random_part}"
+
+    # Hash using SHA256 - appropriate for high-entropy API keys
+    # Format: $sha256${hex_digest} to distinguish from bcrypt hashes
+    digest = hashlib.sha256(full_key.encode()).hexdigest()
+    key_hash = f"$sha256${digest}"
+
+    # Extract prefix for database lookup optimization
+    prefix = full_key[:KEY_PREFIX_LENGTH]
+
+    logger.debug(f"Generated API key with prefix {prefix}")
+
+    return full_key, key_hash, prefix
+
+
+def verify_api_key(key: str, key_hash: str) -> bool:
+    """Verify an API key against its stored hash.
+
+    Supports both SHA256 (preferred for API keys) and bcrypt hashes.
+
+    Args:
+        key: The full API key provided by the user
+        key_hash: The stored hash (SHA256 or bcrypt format)
+
+    Returns:
+        True if the key matches the hash, False otherwise
+
+    Note:
+        Uses constant-time comparison to prevent timing attacks.
+        Returns False (not exception) for invalid inputs.
+    """
+    if not key or not key_hash:
+        return False
+
+    try:
+        if key_hash.startswith("$sha256$"):
+            # SHA256 hash verification with constant-time comparison
+            expected_digest = key_hash[8:]  # Remove "$sha256$" prefix
+            actual_digest = hashlib.sha256(key.encode()).hexdigest()
+            return hmac.compare_digest(expected_digest, actual_digest)
+
+        elif key_hash.startswith("$2"):
+            # Bcrypt hash (legacy or if we switch back)
+            # Import here to avoid startup issues with bcrypt compatibility
+            from passlib.context import CryptContext
+            pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
+            return pwd_context.verify(key, key_hash)
+
+        else:
+            # Unknown hash format
+            logger.warning(f"Unknown hash format: {key_hash[:10]}...")
+            return False
+
+    except Exception as e:
+        # Handle malformed hashes gracefully
+        logger.debug(f"API key verification failed: {e}")
+        return False
+
+
+def extract_prefix(key: str) -> str:
+    """Extract the prefix from a full API key for database lookup.
+
+    The prefix is the first 12 characters (cf_{env}_xxxx) which provides
+    enough entropy for efficient indexed lookup while avoiding full key
+    comparison in the database.
+
+    Args:
+        key: The full API key
+
+    Returns:
+        The 12-character prefix
+
+    Raises:
+        ValueError: If key is shorter than 12 characters
+    """
+    if len(key) < KEY_PREFIX_LENGTH:
+        raise ValueError(f"API key too short: expected at least {KEY_PREFIX_LENGTH} chars")
+
+    return key[:KEY_PREFIX_LENGTH]
+
+
+def validate_scopes(scopes: list[str]) -> bool:
+    """Validate that all scopes in the list are valid.
+
+    Args:
+        scopes: List of scope strings to validate
+
+    Returns:
+        True if all scopes are valid and list is non-empty, False otherwise
+    """
+    if not scopes:
+        return False
+
+    return all(scope in VALID_SCOPES for scope in scopes)