Setting Up Role-Based Access Control for Municipal Billing APIs

The parent guide, Security Boundaries & Role-Based Access — part of Municipal Utility Billing Architecture & Rate Taxonomy — establishes why a least-privilege role matrix exists and how a decision is audited. This page resolves the concrete wiring problem underneath it: how a billing API stops a caller who holds a perfectly valid token from reaching an endpoint their role was never meant to touch. The failure mode it prevents is the scattered guard — one endpoint checks a role inline, another forgets, a third trusts a client-supplied X-Role header — so a rate_engineer token quietly reaches the collections routes, or a read-only analyst pulls a response body that still carries assistance_eligibility fields. The fix is a single source of truth: one endpoint-to-scope registry, one token verifier bound to the municipal identity provider (IdP), one deny-by-default dependency mounted on every route, and response models that strip protected fields before serialization.

# Minimal prerequisites

• Python 3.11+ — for StrEnum, the X | None union syntax, and datetime.UTC.
• FastAPI (fastapi>=0.110) — the guard is a dependency; the examples use FastAPI’s HTTPBearer security utility and Depends.
• PyJWT (pyjwt[crypto]>=2.8) — tokens are asymmetric (RS256) and verified against the IdP’s published JWKS, never a shared secret.
• Pydantic v2 (pydantic>=2.6) — response models use v2 model_config and computed serialization to drop protected fields per role.

Data and access assumptions: callers authenticate against the municipal IdP and arrive with a signed bearer token whose claims include a role, an aud, an iss, and a jurisdiction_scope. The role names and the least-privilege matrix are the same BillingRole / ROLE_MATRIX structures defined in the parent guide — this page consumes that matrix rather than redefining it. Payloads reaching these endpoints are assumed already conformant per upstream schema validation & data quality checks, so RBAC governs who may act, not whether the body is well-formed.

# Annotated implementation

The registry is the load-bearing idea: every route the service exposes is declared once, next to the exact scope it requires. Nothing is guarded by a decorator buried in a router module, and no endpoint can ship without a scope because a startup check (below, in verification) refuses to boot if one is missing.

from __future__ import annotations

from enum import StrEnum


class BillingRole(StrEnum):
    METER_VALIDATOR = "meter_validator"
    RATE_ENGINEER = "rate_engineer"
    COLLECTIONS_AGENT = "collections_agent"
    ASSISTANCE_CASEWORKER = "assistance_caseworker"
    FINANCE_ADMIN = "finance_admin"


# Central endpoint-to-scope registry, keyed by (HTTP method, route template).
# This is the single source of truth an auditor reads to answer "who can call what?".
ENDPOINT_SCOPES: dict[tuple[str, str], str] = {
    ("GET",  "/api/telemetry/{account_id}/reads"):     "telemetry:read",
    ("POST", "/api/validation/{account_id}/flags"):    "validation:write",
    ("GET",  "/api/tariffs/schedules"):                "tariffs:read",
    ("POST", "/api/tariffs/schedules"):                "tariffs:write",
    ("PUT",  "/api/tariffs/block-rate/thresholds"):    "tariffs:write",
    ("GET",  "/api/arrears/{account_id}"):             "arrears:read",
    ("POST", "/api/payments/{account_id}/route"):      "payment_routing:execute",
    ("GET",  "/api/assistance/{account_id}"):          "assistance:read",
    ("POST", "/api/assistance/{account_id}/verify"):   "eligibility:verify",
    ("POST", "/api/ledger/sync"):                      "ledger:sync",
    ("GET",  "/api/audit/decisions"):                  "audit:read",
    ("POST", "/api/adjustments/{adjustment_id}/approve"): "adjustments:approve",
}

Verification of the token happens once, in a helper that trusts nothing the client asserts about itself. The signature is checked against the IdP’s JWKS, and iss/aud are pinned so a token minted for a different service cannot be replayed here.

import jwt
from jwt import PyJWKClient

IDP_JWKS_URL = "https://idp.internal.example.gov/.well-known/jwks.json"
IDP_ISSUER = "https://idp.internal.example.gov/"
IDP_AUDIENCE = "municipal-billing-api"

# cache_keys avoids a JWKS round-trip on every request; the client refreshes on rotation.
_jwk_client = PyJWKClient(IDP_JWKS_URL, cache_keys=True)


def decode_verified_claims(token: str) -> dict:
    """Verify signature, issuer, audience, and expiry. Raise on anything unexpected."""
    signing_key = _jwk_client.get_signing_key_from_jwt(token)
    return jwt.decode(
        token,
        signing_key.key,
        algorithms=["RS256"],          # pin the algorithm; never accept "none" or HS256
        audience=IDP_AUDIENCE,
        issuer=IDP_ISSUER,
        options={"require": ["exp", "iss", "aud", "role"]},
    )

The guard dependency ties the two together. It reads the matched route template from the request (not the raw path, so /api/arrears/ACCT-000123 resolves to the /api/arrears/{account_id} rule), looks up the required scope, resolves the caller’s role from the verified claim, and consults the least-privilege matrix. Every allow and every deny is chained into the tamper-evident trail the parent guide defines, so the decision is reconstructable later.

from fastapi import Depends, HTTPException, Request
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer

# ROLE_MATRIX and record_decision are imported from the shared security module
# defined in "Security Boundaries & Role-Based Access".
from billing.security import ROLE_MATRIX, record_decision

security_scheme = HTTPBearer()


async def rbac_guard(
    request: Request,
    credentials: HTTPAuthorizationCredentials = Depends(security_scheme),
) -> dict:
    # 1. Verify the token cryptographically before trusting any claim.
    try:
        claims = decode_verified_claims(credentials.credentials)
    except jwt.ExpiredSignatureError:
        raise HTTPException(status_code=401, detail="token expired; request a fresh assertion")
    except jwt.InvalidTokenError:
        raise HTTPException(status_code=401, detail="invalid token signature or claims")

    # 2. Resolve the role from the verified claim, never a client header.
    try:
        role = BillingRole(claims["role"])
    except (KeyError, ValueError):
        raise HTTPException(status_code=401, detail="missing or unknown role assignment")

    # 3. Look up the required scope by the MATCHED route template, not the raw URL.
    route = request.scope.get("route")
    key = (request.method, route.path if route else request.url.path)
    required_scope = ENDPOINT_SCOPES.get(key)
    if required_scope is None:
        # An unregistered endpoint is a deployment error, not a public one — fail closed.
        record_decision(role, key[1], "UNREGISTERED", "DENIED_NO_SCOPE", request)
        raise HTTPException(status_code=403, detail="endpoint has no scope registration")

    # 4. Deny by default: the scope must be explicitly granted to this role.
    permissions = ROLE_MATRIX[role]
    if required_scope not in permissions.allowed_scopes:
        record_decision(role, key[1], required_scope, "DENIED_SCOPE", request)
        raise HTTPException(status_code=403, detail=f"scope {required_scope!r} not permitted")

    record_decision(role, key[1], required_scope, "ALLOW", request)
    # Hand sanitized context to the route; jurisdiction feeds the ABAC overlay downstream.
    request.state.role = role
    request.state.jurisdiction_scope = claims.get("jurisdiction_scope")
    return claims

Mounting the guard once, at the application level, is what closes the “one endpoint forgot to check” gap: it applies to every route, so a new endpoint is protected the moment it is registered.

from fastapi import FastAPI

app = FastAPI(dependencies=[Depends(rbac_guard)])  # global: no route escapes the guard

Even with the route gated, the response body can still over-share. A GET /api/assistance/{account_id} is legitimately reachable by an assistance_caseworker, but the serialized record must never leak protected fields to any other context. Field-level redaction lives in the response model, keyed off the resolved role, so the Assistance Program Eligibility Taxonomy flags and raw PII are dropped before they cross the wire.

from pydantic import BaseModel

_PROTECTED_FIELDS = {"assistance_eligibility", "subsidy_tier", "ssn_last4"}


class AccountView(BaseModel):
    account_id: str
    customer_class: str
    balance_cents: int
    assistance_eligibility: str | None = None
    subsidy_tier: str | None = None
    ssn_last4: str | None = None

    def for_role(self, role: BillingRole) -> dict:
        data = self.model_dump()
        # Only a certified caseworker may see eligibility + limited PII; strip for everyone else.
        if role is not BillingRole.ASSISTANCE_CASEWORKER:
            for field in _PROTECTED_FIELDS:
                data.pop(field, None)
        return data

# Edge cases and billing gotchas

• Raw path vs matched route template. Keying the registry on request.url.path breaks the moment a path carries an identifier — /api/arrears/ACCT-000123 will not match /api/arrears/{account_id}. Resolve the scope from request.scope["route"].path, the template FastAPI already matched, so parameterized routes register once and every account resolves correctly.
• Unregistered endpoint fails open. If a route is mounted but omitted from ENDPOINT_SCOPES, a naive guard that only checks known scopes would let it through unguarded. This implementation fails closed — an unregistered endpoint returns 403 and logs DENIED_NO_SCOPE, turning a missed registration into a loud deploy-time failure instead of a silent hole.
• Algorithm-confusion and alg: none. A token verifier that does not pin algorithms=["RS256"] can be tricked into accepting an unsigned token or an HMAC forged with the public key as the secret. Pin the algorithm, require exp/iss/aud, and reject anything else — the same discipline the hash-chained audit trail in Data Governance & Privacy Compliance depends on to stay defensible.
• Duplicate batch posts to /api/ledger/sync. A retried reconciliation run can double-post. Require an Idempotency-Key header on the ledger sync endpoint, store the first result against that key, and return 409 Conflict with the original transaction ID on a duplicate — RBAC decides who may sync, idempotency decides that syncing twice is not two events.
• Stale role after a demotion. A token minted before a job change still carries the old role until it expires. Keep token TTLs short (15–30 minutes) and resolve the live ROLE_MATRIX on every request rather than caching a permission set for the token’s whole life, so a revoked privilege takes effect at the next call.

# Verification snippet

Two checks matter before shipping: that every mounted route has a scope registration (no route can escape the registry), and that the guard denies a scope the role does not hold. Run the registration check at startup so a missing entry fails the boot, not a production request.

def assert_every_route_registered(app: FastAPI) -> None:
    """Boot-time guard: every non-docs route must appear in ENDPOINT_SCOPES."""
    for route in app.routes:
        methods = getattr(route, "methods", set()) - {"HEAD", "OPTIONS"}
        path = getattr(route, "path", "")
        if path.startswith(("/docs", "/openapi", "/redoc")):
            continue
        for method in methods:
            assert (method, path) in ENDPOINT_SCOPES, f"unregistered endpoint: {method} {path}"


def test_analyst_cannot_write_tariffs() -> None:
    # A meter validator holds telemetry/validation scopes, never tariffs:write.
    perms = ROLE_MATRIX[BillingRole.METER_VALIDATOR]
    assert "tariffs:write" not in perms.allowed_scopes


def test_protected_fields_stripped_for_non_caseworker() -> None:
    view = AccountView(
        account_id="ACCT-000123", customer_class="residential", balance_cents=4200,
        assistance_eligibility="LIHEAP", subsidy_tier="T2", ssn_last4="6789",
    )
    finance = view.for_role(BillingRole.FINANCE_ADMIN)
    assert "assistance_eligibility" not in finance and "ssn_last4" not in finance
    caseworker = view.for_role(BillingRole.ASSISTANCE_CASEWORKER)
    assert caseworker["assistance_eligibility"] == "LIHEAP"

Run assert_every_route_registered(app) in a startup hook and the field-redaction test against a fixture that carries every protected field populated — that combination catches both the forgotten-scope regression and an accidental leak of eligibility data in one pass.

Figure: One request, end to end — cryptographic verification and deny-by-default authorization gate the route (401 for identity failures, 403 for authorization failures), then role-aware redaction strips protected fields before the response, and every decision is appended to the audit trail.

# Frequently Asked Questions

# Related Topics

• Security Boundaries & Role-Based Access — the parent guide defining the role matrix, ABAC overlay, and audit chain this API wiring consumes.
• Data Governance & Privacy Compliance — the retention and metadata-only rules the redaction and audit logging satisfy.
• Assistance Program Eligibility Taxonomy — the protected eligibility data that caseworker scopes and field redaction gate.
• Step-Rate vs Block-Rate Structure Design — the tariff structures that only a tariffs:write scope may modify through these endpoints.
• Schema Validation & Data Quality Checks — the upstream contract that lets these endpoints assume well-formed payloads.

Up one level: Security Boundaries & Role-Based Access · Parent guide: Municipal Utility Billing Architecture & Rate Taxonomy · Return to the utilitybilling.org home.