Security Boundaries & Role-Based Access Control for Municipal Utility Billing

In a billing system, the most expensive mistakes are made by someone who simply had more access than their role required. This guide sits under Municipal Utility Billing Architecture & Rate Taxonomy and addresses one specific production failure: a workflow that grants a role broader reach than its job demands, so an operator, a service account, or a stray script can modify a tariff, read protected customer data, or post a ledger adjustment it was never meant to touch. When automating rate structures, arrears routing, and Public Utility Commission (PUC) synchronization, the boundary between roles is not an IT compliance checkbox — it is the operational guardrail that keeps unauthorized rate modifications, PII exposure, and unreviewed financial writes out of the pipeline.

Role-based access control (RBAC) is the mechanism that makes that boundary deterministic. Instead of trusting that a caller will only do what they are supposed to, every request is mapped to a role, every role is mapped to an explicit set of permitted scopes and forbidden modules, and every allow/deny decision is written to a tamper-evident log. The remedy for over-broad access is engineering discipline: declarative role definitions validated at load time, least-privilege enforcement at both the application and data layers, attribute overlays for jurisdictional scoping, time-bound elevation for batch jobs, and an immutable audit trail a finance team can replay end to end.

# Prerequisites

This workflow targets a modern, reproducible Python stack. Pin the following before implementing anything below:

• Python 3.11+ — required for StrEnum, X | None union syntax, and datetime.UTC, all used throughout.
• Pydantic v2 (pydantic>=2.6) — the role matrix and audit records are v2 models; the frozen=True config and model_validator semantics below assume v2.
• FastAPI (fastapi>=0.110) — scope and module enforcement is implemented as a dependency; any ASGI framework with a request/dependency model works, but the examples use FastAPI security utilities.
• zoneinfo (not pytz) — just-in-time elevation windows and audit timestamps are resolved in the utility’s local civil time, so every expiry comparison must be timezone-aware.
• hashlib from the standard library — audit records are chained with SHA-256; no third-party crypto is required for the tamper-evidence layer.

Data and access assumptions: callers authenticate against the municipal identity provider (IdP) and arrive carrying a signed token whose claims include a role and, for the ABAC overlay, a set of jurisdiction attributes. Role assignments themselves are configuration, not billing execution — write access to the role matrix must be isolated from the code that runs a billing cycle. Upstream account records should already have passed schema validation & data quality checks so that this stage governs who may act, not whether payloads are well-formed.

# Architecture Overview

Access control here is a staged decision, not a single gate. A request enters carrying a role claim; the role is resolved against a least-privilege matrix; the requested scope is checked; the target module is checked against the role’s forbidden set; attribute rules are evaluated for jurisdictional fit; and — allow or deny — an audit record is chained before the request is ever routed to the rate engine, arrears processor, or ledger writer. Each stage has an explicit deny path so an over-scoped call is diverted and logged rather than silently honored.

Figure: Least-privilege separation of duties — each role maps only to the permissions its workflow requires.

The steps below implement the decision path in order: define the matrix, enforce it per request, chain the audit record, overlay attribute rules, then grant time-bound elevation for batch work.

Figure: Deny-by-default decision path — a request must clear the scope, module, and jurisdiction gates in order; any failure returns a specific 403, and every outcome is chained into the audit trail.

# Step-by-Step Implementation

# Step 1 — Define roles and a least-privilege permission matrix

The foundation of a secure billing boundary is a declarative role matrix that is validated once at startup, not scattered across if checks in every endpoint. Each role names exactly the scopes it may exercise and the modules it may never touch. Aligning these to functional workflows is what enforces separation of duties: meter validation gets read-only telemetry and write access to validation flags but is walled off from tariff configuration; a rate engineer may edit tariff tables but cannot reach the collections or assistance modules; an assistance caseworker may read eligibility and limited PII but cannot alter the Step-Rate vs Block-Rate Structure Design that governs consumption pricing.

from enum import StrEnum

from pydantic import BaseModel, ConfigDict, model_validator


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


class RolePermissions(BaseModel):
    model_config = ConfigDict(frozen=True)  # the matrix is immutable at runtime
    allowed_scopes: frozenset[str]
    restricted_modules: frozenset[str]

    @model_validator(mode="after")
    def scopes_and_bans_disjoint(self) -> "RolePermissions":
        # A role must not be granted a scope on a module it is also banned from —
        # that contradiction is the classic source of an accidental privilege leak.
        scoped_modules = {s.split(":", 1)[0] for s in self.allowed_scopes}
        overlap = scoped_modules & self.restricted_modules
        if overlap:
            raise ValueError(f"role grants and bans the same module(s): {sorted(overlap)}")
        return self


ROLE_MATRIX: dict[BillingRole, RolePermissions] = {
    BillingRole.METER_VALIDATOR: RolePermissions(
        allowed_scopes=frozenset({"telemetry:read", "validation:write"}),
        restricted_modules=frozenset({"tariffs", "collections", "assistance"}),
    ),
    BillingRole.RATE_ENGINEER: RolePermissions(
        allowed_scopes=frozenset({"tariffs:read", "tariffs:write", "rate_engine:configure"}),
        restricted_modules=frozenset({"collections", "assistance", "ledger"}),
    ),
    BillingRole.COLLECTIONS_AGENT: RolePermissions(
        allowed_scopes=frozenset({"arrears:read", "payment_routing:execute", "customer:contact"}),
        restricted_modules=frozenset({"tariffs", "rate_engine", "assistance"}),
    ),
    BillingRole.ASSISTANCE_CASEWORKER: RolePermissions(
        allowed_scopes=frozenset({"assistance:read", "eligibility:verify", "customer:pii_read"}),
        restricted_modules=frozenset({"tariffs", "rate_engine", "collections"}),
    ),
    BillingRole.FINANCE_ADMIN: RolePermissions(
        allowed_scopes=frozenset({"ledger:sync", "audit:read", "adjustments:approve"}),
        restricted_modules=frozenset(),  # broad read, but still scope-gated on writes
    ),
}

Because the matrix is a frozen, self-validating structure, a contradictory grant (a scope on a banned module) fails at load time rather than becoming a silent hole in production. The full endpoint-by-endpoint mapping is covered in Setting Up RBAC for Municipal Billing APIs.

# Step 2 — Enforce scope and module isolation per request

With the matrix defined, every billing API request must be intercepted before it reaches business logic. A FastAPI dependency resolves the caller’s role from the verified IdP token, checks the required scope against the role’s allowed set, and checks the target module against its restricted set. Two independent checks matter: a scope check answers “may this role perform this action?” and a module check answers “may this role touch this subsystem at all?” — the second catches path-traversal and endpoint-guessing that a scope check alone would miss.

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

security_scheme = HTTPBearer()


def _module_from_path(path: str) -> str:
    # "/api/tariffs/schedules/42" -> "tariffs". Segment 0 is the "api" prefix.
    parts = [p for p in path.split("/") if p]
    return parts[1] if len(parts) > 1 else "root"


def require_scope(required_scope: str):
    """Dependency factory: guards an endpoint with one scope + module isolation."""

    async def _guard(
        request: Request,
        credentials: HTTPAuthorizationCredentials = Depends(security_scheme),
    ) -> RolePermissions:
        # In production: decode the JWT and verify its signature against the
        # municipal IdP's JWKS before trusting any claim. The role comes from a
        # verified claim, never from a client-supplied header.
        role_claim = decode_verified_role(credentials.credentials)
        try:
            role = BillingRole(role_claim)
        except ValueError:
            raise HTTPException(status_code=401, detail="Invalid or missing role assignment")

        permissions = ROLE_MATRIX[role]
        target_module = _module_from_path(request.url.path)

        if required_scope not in permissions.allowed_scopes:
            record_decision(role, request.url.path, required_scope, "DENIED_SCOPE", request)
            raise HTTPException(status_code=403, detail=f"scope {required_scope!r} not permitted")

        if target_module in permissions.restricted_modules:
            record_decision(role, request.url.path, required_scope, "DENIED_MODULE", request)
            raise HTTPException(status_code=403, detail=f"module {target_module!r} restricted")

        record_decision(role, request.url.path, required_scope, "ALLOW", request)
        return permissions

    return _guard


# Usage: only a rate engineer with tariffs:write may reach this endpoint.
# @app.post("/api/tariffs/schedules", dependencies=[Depends(require_scope("tariffs:write"))])

The dependency denies by default: any role, scope, or module not explicitly permitted results in a 403 and a logged decision. This is the same discipline the arrears path relies on when Fallback Routing for Missing Rate Data diverts an incomplete record — the request is diverted and recorded, never silently honored.

# Step 3 — Chain every decision into a tamper-evident audit trail

An access decision is only defensible if a finance team or PUC auditor can reconstruct it. Every allow and every deny emits an audit record, and records are chained by hash so any later alteration is detectable. The chain captures the role, endpoint, requested scope, the decision, and the previous record’s digest — never the payload or any raw PII.

import hashlib
import json
from datetime import UTC, datetime

from fastapi import Request
from pydantic import BaseModel

_AUDIT_HEAD = "GENESIS"  # replace with the persisted head hash in production


class AuditRecord(BaseModel):
    timestamp: str
    role: str
    endpoint: str
    requested_scope: str
    decision: str
    prev_hash: str
    record_hash: str = ""


def _digest(payload: dict) -> str:
    # Canonical JSON (sorted keys) so the digest is reproducible across runs.
    encoded = json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8")
    return hashlib.sha256(encoded).hexdigest()


def record_decision(role, endpoint: str, scope: str, decision: str, request: Request) -> AuditRecord:
    global _AUDIT_HEAD
    body = {
        "timestamp": datetime.now(UTC).isoformat(),
        "role": str(role),
        "endpoint": endpoint,
        "requested_scope": scope,
        "decision": decision,
        "prev_hash": _AUDIT_HEAD,
    }
    body["record_hash"] = _digest(body)
    _AUDIT_HEAD = body["record_hash"]
    return AuditRecord(**body)

Chaining by digest means altering any historical decision changes its hash and therefore every hash after it — the same append-only guarantee mandated by Data Governance & Privacy Compliance for the snapshots that satisfy municipal retention rules.

# Step 4 — Overlay attribute rules for jurisdictional scoping

Multi-jurisdictional billing introduces a dimension RBAC alone cannot express: two callers may share the role of rate engineer yet be authorized for different tax authorities. Layering attribute-based access control (ABAC) on top of RBAC lets the engine evaluate geographic tags, tax authority IDs, and rate-schedule versions before permitting a ledger-affecting change, so a cross-jurisdictional adjustment can never bypass the local PUC approval path.

from dataclasses import dataclass


@dataclass(frozen=True)
class RequestAttributes:
    role_jurisdictions: frozenset[str]  # authorities this operator may act for
    target_jurisdiction: str            # authority the request would modify
    schedule_version: str
    approved_versions: frozenset[str]   # versions cleared by PUC workflow


def evaluate_attributes(attrs: RequestAttributes) -> tuple[bool, str]:
    if attrs.target_jurisdiction not in attrs.role_jurisdictions:
        return False, "DENIED_ATTR_JURISDICTION"
    if attrs.schedule_version not in attrs.approved_versions:
        # Editing an unapproved schedule version must route through PUC approval.
        return False, "DENIED_ATTR_UNAPPROVED_VERSION"
    return True, "ALLOW"

The attribute check runs after the RBAC gate in Step 2, not instead of it: RBAC decides whether the role may edit tariffs at all, ABAC decides whether this tariff, in this jurisdiction, at this version. Both must pass. Assistance-related requests carry an additional overlay so that eligibility data governed by the Assistance Program Eligibility Taxonomy is reachable only by certified caseworkers acting within their jurisdiction.

# Step 5 — Grant time-bound just-in-time elevation for batch work

Batch reconciliation and ledger synchronization need write access that no interactive human role should hold permanently. The pattern is just-in-time (JIT) elevation: a service account requests a narrowly scoped, time-boxed grant, performs the run, and the grant expires automatically once ledger parity is confirmed. The elevation window is evaluated in the utility’s civil time so a grant issued near a DST boundary expires exactly when intended.

from datetime import datetime, timedelta
from zoneinfo import ZoneInfo

UTILITY_TZ = ZoneInfo("America/Denver")


@dataclass(frozen=True)
class ElevationGrant:
    principal: str
    scope: str
    issued_at: datetime
    expires_at: datetime


def issue_grant(principal: str, scope: str, ttl_minutes: int) -> ElevationGrant:
    now = datetime.now(UTILITY_TZ)
    return ElevationGrant(
        principal=principal,
        scope=scope,
        issued_at=now,
        expires_at=now + timedelta(minutes=ttl_minutes),
    )


def grant_is_active(grant: ElevationGrant, scope: str) -> bool:
    # Timezone-aware comparison; a grant is valid only for its exact scope and window.
    return grant.scope == scope and grant.issued_at <= datetime.now(UTILITY_TZ) < grant.expires_at

Every elevation is itself logged through record_decision, so the audit chain shows precisely when a batch account held ledger:sync, for how long, and against which reconciliation run.

# Edge-Case Handling

Municipal access control produces failure modes generic RBAC tutorials never cover. Handle each deliberately:

• Role drift and orphaned accounts. Staff change jobs; a former rate engineer who becomes a collections lead must not retain tariff write access. Recertify assignments on a fixed cadence and revoke on transfer, or drift accumulates into a standing over-privilege.
• Break-glass access during an outage. When a billing cycle is failing and the on-call engineer needs emergency reach, issue a short JIT grant that is more heavily logged, not an unrestricted admin key. A break-glass event must produce a louder audit record, never a quieter one.
• Retroactive PUC orders requiring a historical write. A commission order effective a prior month may demand a re-price. Do not hand a human a broad ledger write — issue a scoped, time-boxed grant tied to the specific adjustment and let the ABAC version check confirm the order is approved.
• JIT window straddling a DST transition. Because grants are compared in zoneinfo civil time, a window opened just before the spring-forward hour still expires after its true elapsed duration; naive UTC-offset math would expire it an hour early or late.
• Shared service accounts. Two batch jobs sharing one principal make an audit trail ambiguous. Give each job a distinct principal so record_decision attributes every write to a single, identifiable actor.
• Token replay after role change. A still-valid token minted before a demotion could carry a stale role claim. Keep token lifetimes short and check the live matrix on every request rather than caching the permission set for the token’s whole life.

# Verification and Audit Trail

A boundary is only trustworthy if you can prove it held. Recompute the chain from genesis and compare each stored digest; any in-place edit of a past decision breaks the recomputation at the altered record and every record after it.

def verify_audit_chain(records: list[dict]) -> bool:
    prev = "GENESIS"
    for rec in records:
        body = {
            "timestamp": rec["timestamp"],
            "role": rec["role"],
            "endpoint": rec["endpoint"],
            "requested_scope": rec["requested_scope"],
            "decision": rec["decision"],
            "prev_hash": prev,
        }
        if _digest(body) != rec["record_hash"]:
            return False
        prev = rec["record_hash"]
    return True

Beyond chain integrity, verify the matrix itself against expectation before a production run. A property test asserts that no interactive role can reach the ledger writer and that every deny path actually denies:

def test_no_interactive_ledger_write() -> None:
    # Only finance_admin may hold ledger scopes, and only via a JIT grant for writes.
    for role, perms in ROLE_MATRIX.items():
        writes_ledger = any(s.startswith("ledger:") and s.endswith(":write") for s in perms.allowed_scopes)
        assert not writes_ledger, f"{role} holds a standing ledger write scope"


def test_denied_scope_is_denied() -> None:
    perms = ROLE_MATRIX[BillingRole.METER_VALIDATOR]
    assert "tariffs:write" not in perms.allowed_scopes
    assert "tariffs" in perms.restricted_modules

Operationally, mandate quarterly role recertification, automated permission-drift detection that diffs the live matrix against its version-controlled source, and cryptographic signing of every rate-table modification so a tariff change is traceable to an approved operator. These controls, replayed against historical decision logs, are what let a municipality satisfy both internal audit and external penetration testing without reconstructing intent from memory. The same audit records feed batch reconciliation, where access decisions are cross-checked against ledger postings before final invoice generation.

# Troubleshooting

# FAQ

# Related Topics

• Municipal Utility Billing Architecture & Rate Taxonomy — the parent guide; where access boundaries sit in the wider billing stack.
• Setting Up RBAC for Municipal Billing APIs — the endpoint-by-endpoint configuration deep dive.
• Data Governance & Privacy Compliance — retention, redaction, and the PII rules the audit chain must honor.
• Assistance Program Eligibility Taxonomy — the protected eligibility data that caseworker scopes gate.
• Fallback Routing for Missing Rate Data — who may approve fallback thresholds and manual overrides.
• Customer Class & Service Tier Mapping — the tier thresholds whose write access these boundaries isolate.

Up: Municipal Utility Billing Architecture & Rate Taxonomy