Implementing Role-Based Access for Traffic APIs
This guide solves one exact operational task: authenticating an incoming traffic API request, resolving the caller’s role, and routing it through a strict Role-Based Access Control (RBAC) allowlist before it is allowed to mutate the scheduling database. It is the request-level companion to the database-layer model in Security Boundaries for Traffic Database Access, and it sits inside the wider entity model documented in Broadcast Traffic Architecture & Taxonomy. When automation pipelines, third-party ad servers, and internal scheduling tools all POST directly against a traffic API, a single misrouted call can overwrite a live traffic log, corrupt billing reconciliation, or double-book a premium avail during a sweep. A deterministic RBAC guard — one that authorizes, validates the payload, and audits every decision — is what keeps that from reaching air. This page is written for two readers at once: traffic managers and media-operations engineers who need the plain-language decision model before they sign off, and Python developers who need a deployable middleware layer rather than a toy demo.
Figure — RBAC request pipeline: each call is authenticated and its role resolved, checked against a default-deny permission matrix, validated against the spot schema and billing-code pattern, and matched to a contracted avail window before it reaches the authorized commit. A caller with no token gets 401; a role without the action gets 403; a schema or billing mismatch gets 403 (a corrupt billing code trips a fail-closed emergency pause); an avail conflict diverts to fallback routing. Every branch emits exactly one immutable audit record.
Prerequisites
Before deploying the guard, confirm the following are in place. Treat any unchecked item as a hard blocker — a partially provisioned RBAC layer fails open, which is worse than no layer at all.
Step-by-Step Implementation
Step 1 — Define roles and a default-deny permission matrix
Goal: enumerate the operational roles and the exact actions each may perform, so an unmapped role can do nothing. This mirrors the database role matrix one-to-one — the API scheduler role maps to the schedule_committer database role.
from __future__ import annotations
import logging
from enum import Enum
# Traffic-ops structured logging: timestamp | level | module | message
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
)
logger = logging.getLogger("traffic_api.rbac")
class TrafficRole(str, Enum):
READ_ONLY = "read_only"
SCHEDULER = "scheduler"
BILLING_ADMIN = "billing_admin"
SYSTEM_ADMIN = "system_admin"
class TrafficAction(str, Enum):
READ = "READ"
CREATE = "CREATE"
UPDATE = "UPDATE"
DELETE = "DELETE"
OVERRIDE = "OVERRIDE"
# Deny by default: a role absent from this map is granted nothing.
ROLE_PERMISSIONS: dict[TrafficRole, frozenset[TrafficAction]] = {
TrafficRole.READ_ONLY: frozenset({TrafficAction.READ}),
TrafficRole.SCHEDULER: frozenset({TrafficAction.READ, TrafficAction.CREATE, TrafficAction.UPDATE}),
TrafficRole.BILLING_ADMIN: frozenset({TrafficAction.READ, TrafficAction.UPDATE}),
TrafficRole.SYSTEM_ADMIN: frozenset(TrafficAction), # full set, override included
}The asymmetry is deliberate: a read_only dashboard cannot create spots, a billing_admin can amend a billing code but never insert a placement, and only system_admin holds OVERRIDE. No role is ever widened at runtime.
Step 2 — Model the request context and the immutable audit record
Goal: parse the untrusted request into a typed, validated object and prepare an audit record that captures the decision. Pydantic rejects malformed input before any authorization logic runs.
from datetime import datetime, timezone
from typing import Any
from pydantic import BaseModel, Field, field_validator
class APIRequestContext(BaseModel):
"""A single authenticated traffic API call, tagged with the spot it acts on."""
caller_id: str = Field(..., min_length=1)
role: TrafficRole
action: TrafficAction
payload: dict[str, Any]
trace_id: str = Field(..., min_length=8) # correlation key for the audit trail
timestamp: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
@field_validator("trace_id")
@classmethod
def trace_is_hex(cls, v: str) -> str:
if not all(c in "0123456789abcdef-" for c in v.lower()):
raise ValueError("trace_id must be a hex/uuid correlation token")
return v
class AuditRecord(BaseModel):
"""Append-only decision record. One is emitted for every request, allow or deny."""
request_id: str
caller_id: str
role: str
action: str
decision: str = "PENDING"
reason: str = ""
spot_id: str | None = None
timestamp: str = ""
fallback_triggered: bool = False
emergency_pause_triggered: bool = FalseStep 3 — Build the stateless validator and routing engine
Goal: enforce the permission matrix, validate the payload against the spot schema and billing-code pattern, check the avail window, and either authorize the write, divert it to deterministic fallback, or trip an emergency pause. Every branch logs a decision line carrying the spot_id.
import re
class TrafficRBACValidator:
"""Stateless RBAC guard for traffic API requests. Deployable behind an API
gateway or as FastAPI/Flask middleware. Contains no I/O beyond logging, so
the same request replayed during an audit yields an identical decision."""
def __init__(
self,
billing_code_pattern: str = r"^[A-Z]{3}-\d{4}$",
avail_tolerance_minutes: int = 15,
) -> None:
self.billing_pattern = re.compile(billing_code_pattern)
self.avail_tolerance = avail_tolerance_minutes
self._emergency_pause_active = False
def _validate_schema(self, payload: dict[str, Any]) -> tuple[bool, str, str | None]:
"""Validate the broadcast spot schema. Returns (is_valid, message, offending_field);
the field name lets callers react deterministically without parsing prose."""
required = {"spot_id", "start_time", "duration_sec", "billing_code", "avail_id"}
missing = required - payload.keys()
if missing:
field_name = "billing_code" if "billing_code" in missing else sorted(missing)[0]
return False, f"Missing required fields: {', '.join(sorted(missing))}", field_name
duration = payload.get("duration_sec")
if not isinstance(duration, (int, float)) or duration <= 0:
return False, "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]:
"""Verify the placement falls inside a contracted inventory window.
Production wiring queries a cached avail registry; kept explicit here."""
if not payload.get("start_time"):
return False, "missing start_time for avail validation"
return True, "avail window verified"
def _apply_fallback_routing(self, payload: dict[str, Any]) -> dict[str, Any]:
"""Deterministic diversion when primary placement conflicts — never a silent drop."""
fallback = payload | {
"routing_status": "FALLBACK_APPLIED",
"placement_priority": "LOW",
"requires_manual_review": True,
}
return fallback
def trigger_emergency_pause(self, reason: str) -> None:
"""Fail closed: halt all mutating requests until manual clearance."""
self._emergency_pause_active = True
logger.critical("EMERGENCY PAUSE ACTIVATED | reason=%s", reason)
def evaluate_request(
self, ctx: APIRequestContext
) -> tuple[bool, dict[str, Any] | None, AuditRecord]:
"""Main pipeline. Returns (authorized, processed_payload, audit_record)."""
spot_id = str(ctx.payload.get("spot_id", "n/a"))
audit = AuditRecord(
request_id=ctx.trace_id, caller_id=ctx.caller_id, role=ctx.role.value,
action=ctx.action.value, spot_id=spot_id, timestamp=ctx.timestamp.isoformat(),
)
# 0. Fail-closed guard: while paused, only reads pass.
if self._emergency_pause_active and ctx.action is not TrafficAction.READ:
audit.decision, audit.reason = "BLOCKED", "emergency pause active"
audit.emergency_pause_triggered = True
logger.critical("spot_id=%s decision=BLOCKED reason=%s", spot_id, audit.reason)
return False, None, audit
# 1. Permission check (default-deny).
if ctx.action not in ROLE_PERMISSIONS.get(ctx.role, frozenset()):
audit.decision = "DENIED"
audit.reason = f"role {ctx.role.value} lacks {ctx.action.value}"
logger.warning("spot_id=%s decision=DENIED reason=%s", spot_id, audit.reason)
return False, None, audit
# 2. Payload + avail validation only applies to mutations.
if ctx.action in {TrafficAction.CREATE, TrafficAction.UPDATE}:
ok, msg, field_name = self._validate_schema(ctx.payload)
if not ok:
audit.decision, audit.reason = "DENIED", f"schema violation: {msg}"
logger.error("spot_id=%s decision=DENIED reason=%s", spot_id, audit.reason)
if field_name == "billing_code": # billing corruption is a halt-worthy risk
self.trigger_emergency_pause("critical billing schema violation")
audit.emergency_pause_triggered = True
return False, None, audit
avail_ok, avail_msg = self._check_avail_window(ctx.payload)
if not avail_ok:
audit.decision, audit.reason = "FALLBACK", f"avail conflict: {avail_msg}"
audit.fallback_triggered = True
logger.warning("spot_id=%s decision=FALLBACK reason=%s", spot_id, avail_msg)
return True, self._apply_fallback_routing(ctx.payload), audit
audit.decision, audit.reason = "AUTHORIZED", "all RBAC and schema checks passed"
logger.info("spot_id=%s role=%s action=%s decision=AUTHORIZED",
spot_id, ctx.role.value, ctx.action.value)
return True, ctx.payload, auditAn authorized commit emits 2026-07-03T14:22:07+00:00 | INFO | traffic_api.rbac | spot_id=6ba7b810… role=scheduler action=CREATE decision=AUTHORIZED, while a read_only caller attempting CREATE leaves a decision=DENIED line — the two shapes an auditor greps for.
Step 4 — Wire the guard into the request lifecycle
Goal: place the validator immediately after TLS termination and JWT verification, but before any database connection is acquired, so malformed or unauthorized payloads never consume a connection slot or trigger an expensive query plan.
from fastapi import Depends, FastAPI, HTTPException, Request
app = FastAPI()
_validator = TrafficRBACValidator()
async def enforce_rbac(request: Request) -> dict[str, Any]:
body = await request.json()
ctx = APIRequestContext(
caller_id=request.headers["x-caller-id"],
role=TrafficRole(request.state.jwt_claims["role"]), # role from the verified JWT
action=TrafficAction(body["action"]),
payload=body["payload"],
trace_id=request.headers.get("x-trace-id", request.state.request_id),
)
authorized, processed, audit = _validator.evaluate_request(ctx)
if not authorized:
raise HTTPException(status_code=403, detail={"reason": audit.reason, "trace_id": audit.request_id})
return processed # hand the validated payload to the commit service
@app.post("/spots")
async def create_spot(payload: dict[str, Any] = Depends(enforce_rbac)) -> dict[str, str]:
# Only reached when the guard authorized the write; safe to commit.
return {"status": "accepted", "spot_id": payload["spot_id"]}Verification & Testing
Confirm correct behavior with assertions against fixture requests drawn from real broadcast context. The guard is pure over its input, so tests are deterministic and need no database.
def _ctx(role: TrafficRole, action: TrafficAction, **overrides) -> APIRequestContext:
payload = {"spot_id": "6ba7b810-9dad-11d1-80b4-00c04fd430c8", "start_time": "2026-07-03T20:00:00Z",
"duration_sec": 30, "billing_code": "POL-2026", "avail_id": "A-0912"}
payload.update(overrides)
return APIRequestContext(caller_id="svc-scheduler", role=role, action=action,
payload=payload, trace_id="6ba7b810abc0")
v = TrafficRBACValidator()
# 1. A scheduler creating a well-formed spot is authorized.
ok, out, rec = v.evaluate_request(_ctx(TrafficRole.SCHEDULER, TrafficAction.CREATE))
assert ok and rec.decision == "AUTHORIZED"
# 2. A read-only caller attempting a mutation is denied, no exception leaks.
ok, out, rec = v.evaluate_request(_ctx(TrafficRole.READ_ONLY, TrafficAction.CREATE))
assert not ok and rec.decision == "DENIED" and out is None
# 3. A malformed billing code trips the fail-closed pause.
ok, out, rec = v.evaluate_request(_ctx(TrafficRole.SCHEDULER, TrafficAction.CREATE, billing_code="oops"))
assert not ok and rec.emergency_pause_triggeredThe canonical expectation is that every mutation produces exactly one AuditRecord whose decision is one of AUTHORIZED, FALLBACK, DENIED, or BLOCKED — never empty. In CI, assert that count against a batch fixture so a code path that silently skips the audit is caught before deploy.
Edge Cases & Failure Handling
- Regex drift on billing codes. Quarterly rate-card migrations change affiliate billing formats, so a legitimate spot suddenly fails
_validate_schemaand — because the offending field isbilling_code— trips the emergency pause. Remediate by reloadingbilling_code_patternfrom configuration without restarting the service; normalize the incoming code first against Standardizing Billing Codes Across Traffic Systems so the guard sees a canonical value. - Excessive fallback routing. A stale avail cache diverts a flood of writes to
FALLBACK_APPLIED, burying operators in manual-review items. Pair the guard with a circuit breaker whose thresholds are tuned exactly as in Tuning Thresholds for Scheduling Accuracy: if fallbacks exceed ~15% of requests in a 60-second window, degrade the API to read-only and alert media ops rather than thrash the database. - Emergency pause stuck active.
trigger_emergency_pausehalts every mutating endpoint until manual clearance. Recovery requires reading thereasonfrom the audit trail, confirming the offending payload, patching the upstream integration, and only then resetting_emergency_pause_active. Verify the flag resets cleanly on service restart with a health check, so a redeploy never silently reopens writes.
FAQ
Why enforce RBAC at the API layer if the database already has per-role grants?
Defense in depth. The database grants in Security Boundaries for Traffic Database Access are the last backstop, but the API guard is the only layer with enough context to log a broadcast-domain reason — which spot_id, which avail, which advertiser — and to reject an out-of-scope call before it ever acquires a connection. The two layers must agree: an API role should never be wider than its database db_role.
Where exactly should the guard sit in the request lifecycle?
Immediately after TLS termination and JWT verification, and before any database connection is acquired. Placing it earlier means unauthorized payloads never consume a connection slot or trigger a query plan. In FastAPI this is a dependency (Step 4); behind an API gateway it is a Lambda authorizer or a sidecar.
How do I keep audit logging from becoming a latency bottleneck during sweep week?
Emit the structured timestamp | level | module | message line synchronously but ship it asynchronously with Fluent Bit or Vector, keeping guard latency under ~50 ms even at 500+ RPS. Load-test with concurrent scheduling bursts before a sweep. The validated payloads flowing in should already be typed by Schema Validation with Pydantic for Traffic Data, so the guard spends its budget on authorization, not parsing.
What happens to a spot the guard diverts to fallback routing?
It is never silently dropped. _apply_fallback_routing tags it FALLBACK_APPLIED with LOW priority and requires_manual_review=True, and the audit record sets fallback_triggered. An operator resolves it against the contracted avail window; if it was displaced by a preemption, it enters make-good routing rather than being lost.
Related
- Security Boundaries for Traffic Database Access — the database-layer trust zones, connection identities, and least-privilege grants this API guard mirrors and backstops.
- Schema Validation with Pydantic for Traffic Data — the upstream Pydantic validators that hand this guard already-typed payloads to authorize.
- Tuning Thresholds for Scheduling Accuracy — how to set the circuit-breaker thresholds that pair with the guard’s fallback and emergency-pause logic.