Role-Based Access Control for Waste Operations Dashboards

This page shows how to build the role-to-scope registry and least-privilege enforcement layer that stops a dispatcher, compliance officer, route engineer, or municipal auditor from touching any dashboard action their role was never granted — the exact failure that turns a shared routing console into an unaccountable control surface.

The broader perimeter — credential verification, tenant isolation, Hours-of-Service-linked revocation, and the hash-chained audit ledger — is defined in the parent Security & Access Boundaries reference. That reference resolves a request into a verified Principal and then defers the role definitions to this page: here we build the closed registry that maps each operator role to the granular scopes it may exercise, wire it into a FastAPI dependency, and serialize every decision into a compliance record an auditor can replay.

Environment & Data Prerequisites

This targets Python 3.10+ and models roles and scopes as closed enums so an unrecognized capability cannot be represented. FastAPI provides the enforcement point, and PyJWT verifies the operator’s token. Pin the versions below before proceeding — they match the pins used across the Security & Access Boundaries reference so the two layers share one JWT stack:

python -m pip install \
  fastapi==0.111.0 \
  "uvicorn[standard]==0.30.1" \
  "pyjwt[crypto]==2.8.0" \
  pydantic==2.7.1

The input to this layer is a signed operator JWT. The claims it must carry, and their expected ranges, are:

A token whose role is not one of the four recognized values must be rejected at parse time, never defaulted to a low-privilege role — a silent default is how a typo in an identity-provider mapping becomes an unlogged privilege grant.

Step-by-Step Implementation

Step 1 — Define the role-to-scope registry

Permission boundaries are declared statically at import time to eliminate runtime resolution overhead and to make the entire grant surface reviewable in one place. Each role maps to a frozenset of granular scope strings aligned with municipal fleet management and hazardous-material handling duties.

from enum import Enum


class OpsRole(str, Enum):
    DISPATCHER = "dispatcher"
    COMPLIANCE = "compliance_officer"
    ROUTE_ENGINEER = "route_engineer"
    MUNICIPAL_AUDITOR = "municipal_auditor"


class Scope(str, Enum):
    ROUTE_DISPATCH = "route:dispatch"
    TELEMETRY_READ = "telemetry:read"
    MANIFEST_EDIT = "manifest:edit"
    MANIFEST_VIEW = "manifest:view"
    COMPLIANCE_AUDIT = "compliance:audit"
    HAZMAT_VIEW = "hazmat:view"
    HOS_OVERRIDE = "hos:override"
    GRAPH_MODIFY = "graph:modify"
    WEIGHT_ADJUST = "weight:adjust"
    FALLBACK_TRIGGER = "fallback:trigger"
    REPORT_READ = "report:read"


SCOPE_REGISTRY: dict[OpsRole, frozenset[Scope]] = {
    OpsRole.DISPATCHER: frozenset(
        {Scope.ROUTE_DISPATCH, Scope.TELEMETRY_READ, Scope.MANIFEST_EDIT}
    ),
    OpsRole.COMPLIANCE: frozenset(
        {Scope.COMPLIANCE_AUDIT, Scope.HAZMAT_VIEW, Scope.HOS_OVERRIDE}
    ),
    OpsRole.ROUTE_ENGINEER: frozenset(
        {Scope.GRAPH_MODIFY, Scope.WEIGHT_ADJUST, Scope.FALLBACK_TRIGGER}
    ),
    OpsRole.MUNICIPAL_AUDITOR: frozenset(
        {Scope.REPORT_READ, Scope.TELEMETRY_READ, Scope.MANIFEST_VIEW}
    ),
}

The auditor role is deliberately read-only: it can pull reports, watch telemetry, and view manifests, but holds none of the mutating scopes (graph:modify, manifest:edit, route:dispatch). This is the least-privilege boundary — a role gets exactly the scopes its job requires and no more, so a compromised auditor session cannot reshape a route.

Step 2 — Verify the token and resolve scopes

Verification decodes the JWT, enforces the signature and expiry, and resolves the operator’s role to its scope set. Restrict algorithms explicitly to block the algorithm-confusion attack where a client re-signs a token with the public key as an HMAC secret. Resolution is memoized with an LRU cache so a burst of dashboard requests from one operator does not re-parse and re-map on every call — the registry is immutable, so the cached frozenset is always safe to return.

from functools import lru_cache

import jwt  # PyJWT >= 2.0
from fastapi import Depends, HTTPException
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials

security = HTTPBearer()

# In production, load from a secrets manager, not source.
JWT_SECRET = "your-256-bit-secret"
JWT_ALGORITHM = "HS256"


@lru_cache(maxsize=4096)
def scopes_for_role(role: str) -> frozenset[Scope]:
    """Map a role string to its granted scopes. Raises ValueError for an
    unrecognized role so an unknown role can never resolve to a default grant."""
    return SCOPE_REGISTRY[OpsRole(role)]


def resolve_principal(
    credentials: HTTPAuthorizationCredentials = Depends(security),
) -> dict:
    """Decode and validate the operator JWT, then attach the resolved scopes."""
    try:
        payload = jwt.decode(
            credentials.credentials,
            JWT_SECRET,
            algorithms=[JWT_ALGORITHM],   # never accept "none" or a client-chosen alg
            options={"require": ["exp", "sub", "role", "tenant"]},
        )
        scopes = scopes_for_role(payload["role"])
    except jwt.ExpiredSignatureError:
        raise HTTPException(status_code=401, detail="Token expired")
    except (jwt.InvalidTokenError, ValueError, KeyError) as exc:
        raise HTTPException(status_code=401, detail=f"Token invalid: {exc}")

    return {
        "sub": payload["sub"],
        "tenant_id": payload["tenant"],
        "role": payload["role"],
        "scopes": scopes,
    }

Step 3 — Bind a required scope to each dashboard endpoint

The registry is only as good as the point that enforces it. A dependency factory produces a FastAPI dependency parameterized by the Scope a given endpoint requires; it runs before the route handler, so an under-privileged request is rejected with a 403 before any business logic executes. This is the same wrap-the-operation discipline the parent reference applies around the OR-Tools Implementation solver, applied here to dashboard routes.

from collections.abc import Callable

from fastapi import FastAPI

app = FastAPI()


def require_scope(scope: Scope) -> Callable[[dict], dict]:
    """Return a dependency that admits a request only if the resolved
    principal holds `scope`; otherwise it raises 403 before the handler runs."""
    def dependency(principal: dict = Depends(resolve_principal)) -> dict:
        if scope not in principal["scopes"]:
            raise HTTPException(
                status_code=403,
                detail=f"Missing scope: {scope.value}",
            )
        return principal
    return dependency


@app.post("/dashboard/dispatch")
def dispatch_route(
    plan: dict,
    principal: dict = Depends(require_scope(Scope.ROUTE_DISPATCH)),
) -> dict:
    return {"committed": True, "tenant_id": principal["tenant_id"]}


@app.get("/dashboard/audit-report")
def audit_report(
    principal: dict = Depends(require_scope(Scope.REPORT_READ)),
) -> dict:
    return {"report": "generated", "requested_by": principal["sub"]}

A dispatcher calling /dashboard/dispatch passes; the same dispatcher calling /dashboard/audit-report is rejected 403 because report:read is not in the dispatcher grant. No endpoint reads scopes ad hoc — every protected route names the one scope it needs, which keeps the authorization model auditable as a single list rather than scattered if checks.

Step 4 — Serialize every decision into a structured audit record

Every grant and denial is written as a structured JSON record so a compliance officer can reconstruct who reached which action and when. The JSONFormatter below is the custom formatter used site-wide — the standard logging module ships no JSON formatter, so structured audit output requires one.

import json
import logging


class JSONFormatter(logging.Formatter):
    """Custom JSON formatter — the standard logging module has no built-in JSONFormatter."""
    def format(self, record: logging.LogRecord) -> str:
        base = logging.LogRecord("", 0, "", 0, "", (), None).__dict__
        log_obj = {
            "timestamp": self.formatTime(record),
            "level": record.levelname,
            "message": record.getMessage(),
        }
        log_obj.update({k: v for k, v in record.__dict__.items() if k not in base})
        return json.dumps(log_obj, sort_keys=True)


_handler = logging.StreamHandler()
_handler.setFormatter(JSONFormatter())
audit_logger = logging.getLogger("waste_ops.rbac")
audit_logger.addHandler(_handler)
audit_logger.setLevel(logging.INFO)


def log_decision(principal: dict, action: str, scope: Scope, granted: bool) -> None:
    audit_logger.info(
        "rbac_decision",
        extra={
            "event": "rbac.decision",
            "decision": "granted" if granted else "denied",
            "subject": principal["sub"],
            "tenant_id": principal["tenant_id"],
            "role": principal["role"],
            "action": action,
            "required_scope": scope.value,
        },
    )

Compliance Output

A denied attempt by a municipal auditor to reach the dispatch endpoint produces exactly this record:

{
  "action": "POST /dashboard/dispatch",
  "decision": "denied",
  "event": "rbac.decision",
  "level": "INFO",
  "message": "rbac_decision",
  "required_scope": "route:dispatch",
  "role": "municipal_auditor",
  "subject": "aud-204",
  "tenant_id": "city-sac",
  "timestamp": "2026-07-02 08:00:30,124"
}

Each field carries a regulatory purpose:

• subject and role identify the operator and the capability set they were acting under, satisfying the accountable-identity requirement an auditor needs to attribute every access.
• tenant_id records the municipal fleet boundary so cross-jurisdiction access attempts are filterable in one query — the isolation control the parent reference maps to the NIST SP 800-53 Access Control (AC) family.
• required_scope and decision show precisely which capability was demanded and whether it was granted, which is what lets the hos:override and compliance:audit grants be tied back to duty-time and hazmat authority under 49 CFR Subchapter B.
• action names the protected operation, so a burst of denied records on one subject becomes an alertable signal of a misconfigured role grant or a probed credential.

These records are retained alongside the tamper-evident chain described in the Security & Access Boundaries reference, and their event keys line up with the manifest fields defined in the JSON Schema for Municipal Disposal Tracking so a single audit query spans access decisions and disposal records.

Verification

Confirm both the granted and denied paths with FastAPI’s TestClient. The test issues a real signed token for each role and asserts the status code the enforcement layer returns.

import jwt
import pytest
from fastapi.testclient import TestClient

client = TestClient(app)


def _token(role: str, sub: str, tenant: str = "city-sac") -> str:
    return jwt.encode(
        {"sub": sub, "role": role, "tenant": tenant, "exp": 9_999_999_999},
        JWT_SECRET,
        algorithm=JWT_ALGORITHM,
    )


def test_dispatcher_can_dispatch():
    headers = {"Authorization": f"Bearer {_token('dispatcher', 'drv-118')}"}
    resp = client.post("/dashboard/dispatch", json={"stops": [1, 2]}, headers=headers)
    assert resp.status_code == 200
    assert resp.json()["committed"] is True


def test_auditor_cannot_dispatch():
    headers = {"Authorization": f"Bearer {_token('municipal_auditor', 'aud-204')}"}
    resp = client.post("/dashboard/dispatch", json={"stops": [1]}, headers=headers)
    assert resp.status_code == 403
    assert resp.json()["detail"] == "Missing scope: route:dispatch"


def test_unknown_role_is_rejected():
    headers = {"Authorization": f"Bearer {_token('mayor', 'x-1')}"}
    resp = client.get("/dashboard/audit-report", headers=headers)
    assert resp.status_code == 401   # ValueError from OpsRole("mayor") -> 401

The third test is the important one: an unrecognized role resolves to a ValueError inside scopes_for_role, which the dependency maps to 401 rather than silently granting an empty scope set that would leak through some future endpoint with a permissive default.

Common Errors

jwt.exceptions.InvalidAlgorithmError: The specified alg value is not allowed — the token was signed with an algorithm not in your algorithms list, or you left algorithms unset. Never omit it: an unset or wildcard algorithm list is exactly the algorithm-confusion hole. Set algorithms=[JWT_ALGORITHM] and re-sign test tokens with the matching algorithm.

fastapi.exceptions.FastAPIError: Invalid args for response field / dependency not enforced — calling Depends(require_scope) instead of Depends(require_scope(Scope.ROUTE_DISPATCH)). require_scope is a factory; you must invoke it so FastAPI receives the inner dependency callable. If you pass the factory itself, FastAPI injects the scope argument as a query parameter and the check never runs.

KeyError: <OpsRole.X> raised from SCOPE_REGISTRY[OpsRole(role)] — a role was added to the OpsRole enum but not to SCOPE_REGISTRY. Because the registry is the single source of grants, every enum member must have an entry; add the missing role with an explicit (possibly empty) frozenset and cover it with a test so the gap fails in CI, not production.

Up: Security & Access Boundaries

Related

• Security & Access Boundaries — the full perimeter: credential verification, tenant isolation, HOS-linked revocation, and the hash-chained audit ledger this registry feeds.
• DOT/FMCSA Rule Mapping — how the hos:override and duty-time scopes trace back to federal Hours-of-Service rules.
• JSON Schema for Municipal Disposal Tracking — the manifest record whose fields the audit event keys align with.
• OR-Tools Implementation — the solver the dispatch scope ultimately authorizes access to.
• Fallback Routing Logic — deterministic degradation when an identity or routing dependency fails.