Implementing Role-Based Access for Traffic APIs

import logging
import enum
import json
import re
from typing import Dict, List, Optional, Any, Tuple, Set
from dataclasses import dataclass, field, asdict
from datetime import datetime, timezone
import hashlib

# ---------------------------------------------------------------------------
# Production Logging Configuration (Compliance-Ready Audit Trail)
# ---------------------------------------------------------------------------
class JSONAuditFormatter(logging.Formatter):
    """Formats log records into structured JSON for SIEM ingestion and compliance audits."""
    def format(self, record: logging.LogRecord) -> str:
        log_entry = {
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
            "module": record.module,
            "function": record.funcName,
            "trace_id": getattr(record, "trace_id", None)
        }
        if hasattr(record, "audit_payload"):
            log_entry["audit"] = record.audit_payload
        return json.dumps(log_entry, default=str)

logger = logging.getLogger("traffic_rbac_middleware")
logger.setLevel(logging.INFO)
_handler = logging.StreamHandler()
_handler.setFormatter(JSONAuditFormatter())
logger.addHandler(_handler)

# ---------------------------------------------------------------------------
# Domain Models & Enums
# ---------------------------------------------------------------------------
class TrafficRole(enum.Enum):
    READ_ONLY = "read_only"
    SCHEDULER = "scheduler"
    BILLING_ADMIN = "billing_admin"
    SYSTEM_ADMIN = "system_admin"

class TrafficAction(enum.Enum):
    READ = "READ"
    CREATE = "CREATE"
    UPDATE = "UPDATE"
    DELETE = "DELETE"
    OVERRIDE = "OVERRIDE"

ROLE_PERMISSIONS: Dict[TrafficRole, Set[TrafficAction]] = {
    TrafficRole.READ_ONLY: {TrafficAction.READ},
    TrafficRole.SCHEDULER: {TrafficAction.READ, TrafficAction.CREATE, TrafficAction.UPDATE},
    TrafficRole.BILLING_ADMIN: {TrafficAction.READ, TrafficAction.UPDATE},
    TrafficRole.SYSTEM_ADMIN: {TrafficAction.READ, TrafficAction.CREATE, TrafficAction.UPDATE, TrafficAction.DELETE, TrafficAction.OVERRIDE}
}

@dataclass
class APIRequestContext:
    caller_id: str
    role: TrafficRole
    action: TrafficAction
    payload: Dict[str, Any]
    trace_id: str
    timestamp: datetime = field(default_factory=lambda: datetime.now(timezone.utc))

@dataclass
class AuditRecord:
    request_id: str
    caller_id: str
    role: str
    action: str
    decision: str
    reason: str
    timestamp: str
    fallback_triggered: bool = False
    emergency_pause_triggered: bool = False

# ---------------------------------------------------------------------------
# Validation & Routing Engine
# ---------------------------------------------------------------------------
class TrafficRBACValidator:
    """
    Stateless middleware for validating, authorizing, and routing traffic API requests.
    Designed for integration with FastAPI/Flask or API Gateway Lambda functions.
    """
    def __init__(self, billing_code_pattern: str = r"^[A-Z]{3}-\d{4}$", 
                 avail_window_tolerance_minutes: int = 15):
        self.billing_pattern = re.compile(billing_code_pattern)
        self.avail_tolerance = avail_window_tolerance_minutes
        self._emergency_pause_active = False

    def _log_audit(self, record: AuditRecord, level: int = logging.INFO) -> None:
        extra = {"trace_id": record.request_id, "audit_payload": asdict(record)}
        logger.log(level, f"RBAC_DECISION: {record.decision} | {record.reason}", extra=extra)

    def _validate_schema(self, payload: Dict[str, Any]) -> Tuple[bool, str, Optional[str]]:
        """Validates broadcast spot schema requirements.

        Returns a (is_valid, message, offending_field) tuple. The third element
        names the failing field so callers can react deterministically without
        parsing the human-readable message.
        """
        required_fields = {"spot_id", "start_time", "duration_sec", "billing_code", "avail_id"}
        missing = required_fields - set(payload.keys())
        if missing:
            # Surface the billing field explicitly when it is among the missing keys.
            field_name = "billing_code" if "billing_code" in missing else sorted(missing)[0]
            return False, f"Missing required fields: {', '.join(sorted(missing))}", field_name

        if not isinstance(payload.get("duration_sec"), (int, float)) or payload["duration_sec"] <= 0:
            return False, "Invalid duration_sec: must be positive numeric", "duration_sec"

        if not self.billing_pattern.match(str(payload.get("billing_code"))):
            return False, "Billing code format mismatch", "billing_code"

        return True, "Schema valid", None

    def _check_avail_window(self, payload: Dict[str, Any]) -> Tuple[bool, str]:
        """Verifies placement aligns with contracted inventory windows."""
        # In production, this queries a cached avail registry or Redis layer
        # Simplified for middleware demonstration
        start = payload.get("start_time")
        if not start:
            return False, "Missing start_time for avail validation"
        # Placeholder logic: real implementation would cross-reference contracted windows
        return True, "Avail window verified"

    def _apply_fallback_routing(self, payload: Dict[str, Any]) -> Dict[str, Any]:
        """Deterministic fallback when primary placement fails validation or conflicts."""
        fallback = payload.copy()
        fallback["routing_status"] = "FALLBACK_APPLIED"
        fallback["placement_priority"] = "LOW"
        fallback["requires_manual_review"] = True
        logger.info("Fallback routing logic executed for payload", extra={"trace_id": "N/A"})
        return fallback

    def trigger_emergency_pause(self, reason: str) -> None:
        """Halts all mutating traffic API requests until manual clearance."""
        self._emergency_pause_active = True
        logger.critical(f"EMERGENCY PAUSE ACTIVATED: {reason}")

    def evaluate_request(self, ctx: APIRequestContext) -> Tuple[bool, Optional[Dict[str, Any]], AuditRecord]:
        """
        Main evaluation pipeline. Returns (authorized, processed_payload, audit_record).
        """
        audit = AuditRecord(
            request_id=ctx.trace_id,
            caller_id=ctx.caller_id,
            role=ctx.role.value,
            action=ctx.action.value,
            decision="PENDING",
            reason="",
            timestamp=ctx.timestamp.isoformat()
        )

        if self._emergency_pause_active and ctx.action != TrafficAction.READ:
            audit.decision = "BLOCKED"
            audit.reason = "Emergency pause protocol active"
            audit.emergency_pause_triggered = True
            self._log_audit(audit, logging.CRITICAL)
            return False, None, audit

        allowed_actions = ROLE_PERMISSIONS.get(ctx.role, set())
        if ctx.action not in allowed_actions:
            audit.decision = "DENIED"
            audit.reason = f"Role {ctx.role.value} lacks permission for {ctx.action.value}"
            self._log_audit(audit, logging.WARNING)
            return False, None, audit

        if ctx.action in {TrafficAction.CREATE, TrafficAction.UPDATE}:
            schema_valid, schema_msg, offending_field = self._validate_schema(ctx.payload)
            if not schema_valid:
                audit.decision = "DENIED"
                audit.reason = f"Schema violation: {schema_msg}"
                self._log_audit(audit, logging.ERROR)
                # Auto-pause on critical billing schema corruption risk
                if offending_field == "billing_code":
                    self.trigger_emergency_pause("Critical billing schema violation detected")
                    audit.emergency_pause_triggered = True
                return False, None, audit

            avail_valid, avail_msg = self._check_avail_window(ctx.payload)
            if not avail_valid:
                audit.decision = "FALLBACK"
                audit.reason = f"Avail conflict: {avail_msg}"
                audit.fallback_triggered = True
                processed = self._apply_fallback_routing(ctx.payload)
                self._log_audit(audit, logging.WARNING)
                return True, processed, audit

        audit.decision = "AUTHORIZED"
        audit.reason = "All RBAC and schema checks passed"
        self._log_audit(audit, logging.INFO)
        return True, ctx.payload, audit

from fastapi import Depends, HTTPException, Request

def rbac_middleware(request: Request, validator: TrafficRBACValidator = Depends(lambda: TrafficRBACValidator())):
    # Extract claims, build context, evaluate, and route
    pass