Implementing Role-Based Access for Flight Ops Data

Flight operations and crew scheduling platforms process highly regulated data streams: crew qualification matrices, duty time limitation calculations, medical certificate statuses, and published flight manifests. Under FAA Part 119, EASA ORO.GEN.205, and ICAO Annex 6, strict data segregation is not an IT convenience but a continuous regulatory obligation. The exact task this guide solves is narrow: given a daily RBAC audit export from a scheduling platform, build a deterministic Python pipeline that detects unauthorized privilege escalation — a dispatcher or scheduler holding an administrative override longer than policy allows — before it corrupts operational integrity or surfaces as a regulatory finding.

This work sits inside the broader System Security & Access Boundaries topic within the Core Architecture & Regulatory Mapping domain. It assumes access events already reach you through a normalized flight data ingestion pipeline, and that role and resource names follow the canonical crew duty time taxonomy so a positioning tool is never mistaken for a manifest editor. The result is a nightly job that turns thousands of raw access events into a short, signed list of provable violations.

Prerequisites

Before building the pipeline, confirm the following are in place:

Defining boundaries and regulatory alignment

Establishing precise access boundaries means mapping operational functions to strict, non-overlapping data entitlements. A flight dispatcher requires read and write access to flight planning modules and NOTAM ingestion tools, but must be explicitly restricted from viewing pilot medical records or fatigue risk management system outputs. A crew scheduler requires read-only query access to qualification databases and seniority lists, but must be prevented from modifying published manifests or altering automated duty-time calculations. Codifying these entitlements as data — rather than as scattered configuration in your identity provider — is what lets the pipeline below validate them deterministically and lets an inspector replay any verdict months later.

Manual RBAC review cannot scale across platforms that generate thousands of access events per day. The regulatory edge case that matters most is duration-based: a scheduler or dispatcher granted an administrative override for more than three consecutive hours without documented supervisor approval. The pipeline calculates that duration by subtracting the grant timestamp from the revoke timestamp, flattens any roles inherited during irregular operations, and flags any session that exceeds its approved ceiling.

Step 1 — Encode the entitlement matrix as versioned data

Keep the approved role definitions in their own typed structure so a policy change is a reviewable diff, not a code edit buried in the validator. Each rule pins the role’s allowed resources, its maximum override window in hours, and whether that override requires supervisor approval.

from dataclasses import dataclass
from typing import List


@dataclass(frozen=True)
class EntitlementRule:
    role: str
    allowed_resources: List[str]
    max_override_hours: float = 3.0
    requires_approval: bool = True


ENTITLEMENT_MATRIX = {
    "ADMIN_OVERRIDE": EntitlementRule(
        "ADMIN_OVERRIDE", ["*"],
        max_override_hours=3.0, requires_approval=True,
    ),
    "DISPATCHER_BASE": EntitlementRule(
        "DISPATCHER_BASE", ["flight_plan", "weather_data"],
        max_override_hours=12.0, requires_approval=False,
    ),
    "SCHEDULER_BASE": EntitlementRule(
        "SCHEDULER_BASE", ["qualification_db", "seniority_list"],
        max_override_hours=12.0, requires_approval=False,
    ),
}

Verify: ENTITLEMENT_MATRIX["ADMIN_OVERRIDE"].requires_approval is True and its max_override_hours == 3.0 — the elevated role is the one gated by the three-hour ceiling, while the base roles carry the routine shift length.

Step 2 — Parse the audit stream and normalize to UTC

Read the export line by line so a multi-gigabyte daily log never has to fit in memory. Every timestamp is parsed into a timezone-aware value; malformed records are logged and skipped rather than aborting the run.

import json
import logging
from datetime import datetime
from pathlib import Path
from typing import Iterator, Optional
from dataclasses import dataclass

logging.basicConfig(
    level=logging.INFO,
    format='{"timestamp": "%(asctime)s", "level": "%(levelname)s", '
           '"module": "%(module)s", "message": "%(message)s"}',
    datefmt="%Y-%m-%dT%H:%M:%SZ",
)
logger = logging.getLogger("rbac_compliance_engine")


@dataclass
class AuditEvent:
    timestamp: datetime
    user_id: str
    role_assigned: str
    resource_accessed: str
    action_type: str
    session_id: Optional[str] = None


def parse_audit_stream(file_path: Path) -> Iterator[AuditEvent]:
    """Memory-efficient line-by-line JSON parser with UTC normalization."""
    with open(file_path, "r", encoding="utf-8") as handle:
        for line_num, line in enumerate(handle, 1):
            line = line.strip()
            if not line or line.startswith("#"):
                continue
            try:
                raw = json.loads(line)
                ts = datetime.fromisoformat(
                    raw["timestamp"].replace("Z", "+00:00")
                )
                if ts.tzinfo is None:
                    raise ValueError("naive timestamp rejected")
                yield AuditEvent(
                    timestamp=ts,
                    user_id=str(raw["user_id"]),
                    role_assigned=str(raw["role_assigned"]),
                    resource_accessed=str(raw["resource_accessed"]),
                    action_type=str(raw["action_type"]),
                    session_id=raw.get("session_id"),
                )
            except (json.JSONDecodeError, KeyError, ValueError) as exc:
                logger.warning(
                    "Skipping malformed record at line %s: %s", line_num, exc
                )

Verify: feeding a two-line log where one record omits user_id yields exactly one AuditEvent and emits one WARNING for the malformed line — the run continues instead of crashing.

Step 3 — Flatten inherited roles for irregular operations

Many scheduling platforms assign a base role that silently inherits elevated permissions during irregular operations or a system outage. A duration check that only inspects the literally assigned role misses these. Flatten each role to the full set it effectively grants before applying any ceiling.

_INHERITANCE_MAP = {
    "SCHEDULER_BASE": ["SCHEDULER_BASE", "SCHEDULER_READ_ONLY"],
    "DISPATCHER_BASE": ["DISPATCHER_BASE", "FLIGHT_PLANNING_RW"],
    "ADMIN_OVERRIDE": ["ADMIN_OVERRIDE", "DISPATCHER_BASE", "SCHEDULER_BASE"],
}


def flatten_inherited_roles(base_role: str) -> List[str]:
    """Resolve nested role inheritance; in production, query the IdP graph."""
    return _INHERITANCE_MAP.get(base_role, [base_role])

Verify: flatten_inherited_roles("ADMIN_OVERRIDE") returns the override role plus the dispatcher and scheduler bases it subsumes, so a session opened under ADMIN_OVERRIDE is measured against the strictest approving rule in that set.

Step 4 — Pair grants with revokes and flag over-limit sessions

Sessions open on a GRANT or ASSIGN and close on a REVOKE or EXPIRE. The duration is the delta between the two events. Any session whose effective roles include an approval-gated rule, and whose duration exceeds that rule’s ceiling, is a violation — unless a SUPERVISOR_APPROVAL event covers it.

from datetime import timedelta
from typing import Dict, List


class RBACComplianceValidator:
    def __init__(self, matrix: Dict[str, EntitlementRule]):
        self.matrix = matrix
        self.active_sessions: Dict[str, AuditEvent] = {}
        self.approved_sessions: set[str] = set()
        self.violations: List[Dict] = []

    def _session_key(self, event: AuditEvent) -> str:
        sid = event.session_id or event.timestamp.isoformat()
        return f"{event.user_id}:{sid}"

    def validate_event(self, event: AuditEvent) -> None:
        key = self._session_key(event)
        if event.action_type in ("GRANT", "ASSIGN"):
            self.active_sessions[key] = event
        elif event.action_type == "SUPERVISOR_APPROVAL":
            self.approved_sessions.add(key)
        elif event.action_type in ("REVOKE", "EXPIRE"):
            start = self.active_sessions.pop(key, None)
            if start is not None:
                self._check_duration(key, start, event.timestamp - start.timestamp)

    def _check_duration(self, key: str, start: AuditEvent,
                        duration: timedelta) -> None:
        for role in flatten_inherited_roles(start.role_assigned):
            rule = self.matrix.get(role)
            if not rule or not rule.requires_approval:
                continue
            if duration <= timedelta(hours=rule.max_override_hours):
                continue
            if key in self.approved_sessions:
                continue
            self.violations.append({
                "type": "UNAUTHORIZED_OVERRIDE",
                "user_id": start.user_id,
                "role": role,
                "duration_hours": round(duration.total_seconds() / 3600, 2),
                "threshold_hours": rule.max_override_hours,
                "session_start": start.timestamp.isoformat(),
                "resource": start.resource_accessed,
            })
            logger.error("Compliance violation: %s", self.violations[-1])

Verify: an ADMIN_OVERRIDE session opened at 01:00Z and revoked at 05:30Z (4.5 h, no approval event) appends one UNAUTHORIZED_OVERRIDE with duration_hours 4.5 against a threshold_hours of 3.0; adding a SUPERVISOR_APPROVAL event for the same session key suppresses it.

Step 5 — Run the pipeline and route violations to the SIEM

The driver streams the export, catches sessions that never emitted a revoke (still open past their ceiling at run time), and returns a structured findings list for the SIEM or ticketing sink.

from datetime import timezone


def run_audit(matrix: Dict[str, EntitlementRule],
              audit_log_path: Path) -> List[Dict]:
    validator = RBACComplianceValidator(matrix)
    logger.info("Starting RBAC audit for %s", audit_log_path)
    for event in parse_audit_stream(audit_log_path):
        validator.validate_event(event)

    # Sessions still open at run time are measured against elapsed time.
    now = datetime.now(timezone.utc)
    for key, start in list(validator.active_sessions.items()):
        validator._check_duration(key, start, now - start.timestamp)

    logger.info("Audit complete. %s violations identified.",
                len(validator.violations))
    return validator.violations


if __name__ == "__main__":
    findings = run_audit(ENTITLEMENT_MATRIX, Path("daily_rbac_audit.json"))
    # POST findings to the SIEM / incident-response API here.

Schedule this immediately after the nightly audit export lands, via cron or an orchestrator, and route each UNAUTHORIZED_OVERRIDE to an automated response playbook. The pairing logic is summarised below.

RBAC drift detection pipeline The daily RBAC audit export is streamed line by line and each timestamp is normalized to UTC. Every event is branched on its action type. A GRANT or ASSIGN opens a session that is held in memory until a matching close event arrives. A REVOKE or EXPIRE computes the session duration as revoke minus grant, flattens the assigned role into the full inherited set, then tests the duration against the role's maximum override window. A session that exceeds its window with no supervisor approval is flagged as an UNAUTHORIZED_OVERRIDE and routed to the SIEM; a session within its window is compliant and needs no action. RBAC audit JSON streamed line by line Parse & normalize every timestamp → UTC action_type Compute duration revoke − grant Open session held until close event Flatten roles full inherited set duration > max override? Flag violation UNAUTHORIZED_OVERRIDE Route to SIEM incident playbook Compliant no action REVOKE / EXPIRE GRANT / ASSIGN yes no
RBAC drift pipeline: sessions open on a GRANT and close on a REVOKE, and any inherited role whose held duration exceeds its override window is flagged as an UNAUTHORIZED_OVERRIDE and routed to the SIEM, while compliant sessions take no action.

Verification queries and assertions

After a run, assert the pipeline behaves deterministically on a known fixture:

# A 4.5-hour unapproved admin override must produce exactly one violation.
findings = run_audit(ENTITLEMENT_MATRIX, Path("fixture_override.json"))
assert len(findings) == 1
assert findings[0]["type"] == "UNAUTHORIZED_OVERRIDE"
assert findings[0]["duration_hours"] == 4.5

Where access events are also mirrored into the append-only audit_event table used across the access boundary, the same over-limit sessions can be surfaced directly in PostgreSQL, which is useful for spot-checks that do not run the Python job:

-- Admin overrides held longer than 3 hours without a supervisor approval.
SELECT g.principal_id,
       g.occurred_at            AS granted_at,
       r.occurred_at            AS revoked_at,
       EXTRACT(EPOCH FROM r.occurred_at - g.occurred_at) / 3600.0
           AS override_hours
FROM audit_event g
JOIN audit_event r
  ON r.principal_id = g.principal_id
 AND r.decision = 'revoke'
 AND r.occurred_at > g.occurred_at
WHERE g.decision = 'admit'
  AND g.reason = 'ADMIN_OVERRIDE'
  AND r.occurred_at - g.occurred_at > INTERVAL '3 hours'
  AND NOT EXISTS (
        SELECT 1 FROM audit_event a
        WHERE a.principal_id = g.principal_id
          AND a.reason = 'SUPERVISOR_APPROVAL'
          AND a.occurred_at BETWEEN g.occurred_at AND r.occurred_at
  );
-- Expected: zero rows on a clean day.

Failure modes and troubleshooting

Frequently Asked Questions

How do I stop a legitimate supervisor-approved override from being flagged?

Emit a SUPERVISOR_APPROVAL event carrying the same session key while the override is active. The validator records that key in approved_sessions and skips the duration check for it. Because approval is itself an audit event, the exemption is provable on replay — an inspector sees both the override and the approval that authorised it, rather than an unexplained gap.

How deep should role inheritance be flattened during irregular operations?

Flatten to every role that carries an independent approval-gated ceiling, not just the immediate parent. During irregular operations an ADMIN_OVERRIDE typically subsumes both dispatcher and scheduler bases, and the correct verdict is the strictest one in that set. Resolve the graph from your identity provider rather than a hardcoded map in production, and cache it per run so a mid-run permission change cannot produce two different verdicts for the same session.

The audit export arrives in local station time — how do I handle it?

Convert at the boundary before evaluation. Duration is computed as an absolute delta, so both endpoints must reference the same instant; mixing a spring-forward local grant with a UTC revoke over-counts or under-counts by the offset. Keep the originating zone as metadata for display, but let datetime.fromisoformat produce a timezone-aware UTC value that every comparison uses.

Which regulation actually mandates this segregation?

All three overlap. 14 CFR Part 119 requires an operator to demonstrate who was authorised to alter operational records and when; EASA ORO.GEN.205 governs the contracted activities and data handoffs that make role segregation necessary; and ICAO Annex 6 sets the operational-control expectations the other two implement. The pipeline satisfies all three by producing a deterministic, replayable record of every override and its authorisation, which is the evidence an inspector requests.

Should the entitlement matrix live in code or in the database?

Store it as versioned data keyed by an effective date, exactly as the parent system security and access boundaries schema does for scopes and contracts. Hardcoding limits makes every policy change a deployment and destroys the audit trail; an effective-dated table lets the validator evaluate each historical session against the rules in force when it occurred, and lets a permission change be reviewed as a diff and rolled back cleanly.

Back to System Security & Access Boundaries.