Deepen fedramplens 4x: tests, demos, hardening + bug fixes

.github/labeler.yml

@@ -1,4 +1,15 @@
-docs: ['**/*.md']
-tests: ['tests/**']
-ci: ['.github/**']
-ports: ['ports/**']
+docs:
+  - changed-files:
+      - any-glob-to-any-file: '**/*.md'
+tests:
+  - changed-files:
+      - any-glob-to-any-file: 'tests/**'
+ci:
+  - changed-files:
+      - any-glob-to-any-file: '.github/**'
+ports:
+  - changed-files:
+      - any-glob-to-any-file: 'ports/**'
+demos:
+  - changed-files:
+      - any-glob-to-any-file: 'demos/**'

.gitignore

@@ -8,3 +8,4 @@ build/
 !demos/**
 
 media/walkthrough.mp4
+demos/__pycache__/

VERSION

@@ -1 +1 @@
-0.4.9
+0.5.0

demos/06_ao_risk_dashboard.py

@@ -0,0 +1,50 @@
+"""Scenario 6 - authorizing-official risk dashboard.
+
+Audience: Authorizing Officials (AOs) / risk executives.
+
+An AO signs the ATO and owns the residual risk. This scenario ranks the
+portfolio by POA&M risk score and blocking-finding count, so the AO sees --
+at a glance -- where the risk concentrates and which systems need a
+conditional ATO or a remediation deadline before signature.
+"""
+from _common import load, rule, bullet
+from fedramplens.core import analyze_boundary
+
+PORTFOLIO = ["clean_low", "basic", "overdue_poam", "high_ready",
+             "boundary_creep", "multi_external"]
+
+
+def main() -> None:
+    rule("AO RISK DASHBOARD  -  residual-risk ranking for signature")
+
+    rows = []
+    for key in PORTFOLIO:
+        s = analyze_boundary(load(key))
+        counts = s["finding_counts"]
+        blocking = counts.get("high", 0) + counts.get("critical", 0)
+        rows.append((s, blocking))
+
+    # Highest risk first: blocking findings, then POA&M risk score.
+    rows.sort(key=lambda r: (r[1], r[0]["poam_risk_score"]), reverse=True)
+
+    print("\n  rank  system                         risk  blk  overdue  ready")
+    print("  " + "-" * 66)
+    for i, (s, blocking) in enumerate(rows, 1):
+        print(f"  {i:>4}  {s['system_name'][:28]:28}  "
+              f"{s['poam_risk_score']:>4}  {blocking:>3}  "
+              f"{len(s['poam_overdue']):>7}  "
+              f"{'YES' if s['authorization_ready'] else 'NO':>5}")
+
+    rule("SIGNATURE GUIDANCE")
+    ready = [s for s, _ in rows if s["authorization_ready"]]
+    blocked = [s for s, _ in rows if not s["authorization_ready"]]
+    bullet(f"Clear to sign now      : {len(ready)} system(s)")
+    bullet(f"Needs remediation first: {len(blocked)} system(s)")
+    top = rows[0][0]
+    bullet(f"Highest residual risk  : {top['system_name']} "
+           f"(risk score {top['poam_risk_score']})")
+    print("\nThe AO uses this to prioritize conditional ATOs and set deadlines.")
+
+
+if __name__ == "__main__":
+    main()

demos/07_ci_gate_sarif_upload.py

@@ -0,0 +1,49 @@
+"""Scenario 7 - CI/CD gate that fails the build on blocking findings.
+
+Audience: DevSecOps / platform CI owners.
+
+Wire fedramplens into a pipeline: analyze the boundary, emit SARIF for the
+code-scanning dashboard, and fail the job (non-zero) when the package is not
+authorization-ready. This scenario mimics that gate exactly -- it computes the
+process exit code the CLI would return and shows the SARIF artifact the step
+would upload -- without actually calling sys.exit, so it stays runnable.
+"""
+import json
+
+from _common import load, rule, bullet
+from fedramplens.core import analyze_boundary, to_sarif
+
+
+def _gate(key):
+    s = analyze_boundary(load(key))
+    sarif = to_sarif(s)
+    # This is the exact rule the CLI applies: exit 1 unless ready.
+    exit_code = 0 if s["authorization_ready"] else 1
+    return s, sarif, exit_code
+
+
+def main() -> None:
+    rule("CI GATE  -  fail the build on blocking FedRAMP findings")
+
+    for key in ("clean_low", "boundary_creep"):
+        s, sarif, code = _gate(key)
+        errors = sum(1 for r in sarif["runs"][0]["results"]
+                     if r["level"] == "error")
+        print(f"\n  step: fedramplens analyze {key}")
+        bullet(f"authorization-ready : {s['authorization_ready']}")
+        bullet(f"SARIF error results : {errors} (uploaded to code-scanning)")
+        bullet(f"process exit code   : {code} "
+               f"({'build FAILS' if code else 'build passes'})")
+
+    rule("SARIF ARTIFACT (what the upload step ships)")
+    _, sarif, _ = _gate("boundary_creep")
+    blob = json.dumps(sarif)
+    print(f"\n  sarif log: {len(blob)} bytes, "
+          f"{len(sarif['runs'][0]['results'])} results, valid JSON")
+    assert json.loads(blob)["version"] == "2.1.0"
+    bullet("round-trips cleanly -> ready for actions/upload-sarif")
+    print("\nDrop this into a workflow step to enforce ATO readiness pre-merge.")
+
+
+if __name__ == "__main__":
+    main()

demos/08_control_coverage_report.py

@@ -0,0 +1,47 @@
+"""Scenario 8 - control-coverage report vs the FedRAMP baseline.
+
+Audience: control owners / compliance analysts.
+
+Coverage against the applicable baseline is the headline metric in a readiness
+review. This scenario reports implemented-vs-baseline counts per impact level,
+lists the distinct controls a system implements with their official NIST
+800-53 rev5 titles (resolved offline), and shows how coverage differs across
+low/moderate/high baselines for the same control set.
+"""
+from _common import load, rule, bullet, use_offline_feed_cache
+from fedramplens.core import analyze_boundary, BASELINE_CONTROL_COUNTS
+from fedramplens import controls
+
+
+def main() -> None:
+    rule("CONTROL COVERAGE  -  implemented vs FedRAMP baseline")
+    use_offline_feed_cache()
+
+    b = load("basic")
+    s = analyze_boundary(b, resolve_titles=True, offline=True)
+    print(f"\nSystem: {s['system_name']} ({s['system_id']}), "
+          f"{s['impact'].upper()} impact")
+    bullet(f"controls implemented : {s['controls_implemented']}")
+    bullet(f"baseline controls    : {s['baseline_controls']}")
+    bullet(f"coverage             : {s['coverage_pct']}% of baseline")
+
+    # Distinct implemented controls with their real titles.
+    implemented = sorted({c for comp in b.components
+                          for c in comp.get("controls", [])})
+    rule("IMPLEMENTED CONTROLS (with NIST 800-53 rev5 titles)")
+    for cid in implemented:
+        title = controls.control_title(cid, offline=True) or "(title unresolved)"
+        bullet(f"{cid:8} {title}")
+
+    # Same control set, different baselines -> different coverage.
+    rule("COVERAGE ACROSS BASELINES (same implemented set)")
+    n = s["controls_implemented"]
+    for impact, total in BASELINE_CONTROL_COUNTS.items():
+        pct = round(100.0 * n / total, 1)
+        bullet(f"{impact:9} baseline = {total:3} controls -> {pct}% covered")
+
+    print("\nThis is the coverage snapshot a control owner brings to a gap review.")
+
+
+if __name__ == "__main__":
+    main()

demos/09_poam_tracker.py

@@ -0,0 +1,44 @@
+"""Scenario 9 - POA&M tracker: overdue, risk, and closure planning.
+
+Audience: ISSOs running POA&M remediation.
+
+The POA&M is a living backlog. This scenario drives the analyzer over a system
+with an overdue backlog and a system with a malformed date, then renders the
+POA&M the way an ISSO tracks it: open vs closed, overdue items to escalate,
+the weighted risk score, and any data-quality problems (bad dates) that would
+bounce a package back from the PMO.
+"""
+from _common import load, rule, bullet
+from fedramplens.core import analyze_boundary, generate_poam
+
+
+def _report(key):
+    b = load(key)
+    s = analyze_boundary(b)
+    poam = generate_poam(b)["plan-of-action-and-milestones"]
+    print(f"\n  {s['system_name']} ({s['system_id']})")
+    bullet(f"open items    : {s['poam_open']}")
+    bullet(f"overdue       : {', '.join(s['poam_overdue']) or 'none'}")
+    bullet(f"risk score    : {s['poam_risk_score']}")
+    bad = [f for f in s["findings"] if f["type"] == "bad_poam_date"]
+    bullet(f"bad dates     : {len(bad)}")
+    for f in bad:
+        bullet(f"  -> {f['detail']}")
+    print("    OSCAL POA&M items:")
+    for it in poam["poam-items"]:
+        props = {p["name"]: p["value"] for p in it["props"]}
+        print(f"      - {it['title']:7} [{props['severity']:8}/"
+              f"{props['status']:10}] due={props['scheduled-completion'] or '-'}")
+
+
+def main() -> None:
+    rule("POA&M TRACKER  -  overdue, risk, and closure planning")
+    for key in ("overdue_poam", "bad_poam_date"):
+        _report(key)
+    rule("ESCALATION")
+    bullet("Overdue items go to the AO with a revised milestone date.")
+    bullet("Bad-date findings are fixed before the package returns to the PMO.")
+
+
+if __name__ == "__main__":
+    main()

demos/10_dependency_inventory.py

@@ -0,0 +1,47 @@
+"""Scenario 10 - external-dependency inventory (leveraged / interconnections).
+
+Audience: ISSOs / architects documenting the boundary.
+
+FedRAMP requires every external service and interconnection to be inventoried.
+This scenario walks a system with several external dependencies, lists each
+one outside the authorization boundary, and shows which data flows cross to
+them and whether those crossings are encrypted -- the raw material for the
+CIS/interconnection tables in an SSP.
+"""
+from _common import load, rule, bullet
+from fedramplens.core import analyze_boundary
+
+
+def main() -> None:
+    rule("EXTERNAL DEPENDENCY INVENTORY  -  what leaves the boundary")
+
+    b = load("multi_external")
+    s = analyze_boundary(b)
+    ext = {c["id"]: c for c in b.components if c.get("zone") == "external"}
+
+    print(f"\nSystem: {b.system_name} ({b.system_id}), {b.impact.upper()} impact")
+    print(f"External dependencies: {len(ext)}")
+    for cid, c in ext.items():
+        bullet(f"{c.get('name', cid)} ({cid}) [type={c.get('type', '?')}]")
+
+    rule("CROSSING FLOWS (source -> external dependency)")
+    for fl in b.flows:
+        if fl["to"] in ext or fl["from"] in ext:
+            enc = "encrypted" if fl.get("encrypted") else "UNENCRYPTED"
+            bullet(f"{fl['from']} -> {fl['to']}  [{fl.get('data', '?')}] {enc}")
+
+    unenc = [f for f in s["findings"]
+             if f["type"] == "unencrypted_boundary_crossing"]
+    rule("SC-8 GAPS ON EXTERNAL CROSSINGS")
+    if unenc:
+        for f in unenc:
+            bullet(f"({f['severity']}) {f['detail']}")
+    else:
+        bullet("none -- every external crossing is encrypted")
+
+    print(f"\nInventory feeds the SSP interconnection table; "
+          f"{len(unenc)} crossing(s) need encryption before ATO.")
+
+
+if __name__ == "__main__":
+    main()

demos/11_boundary_hygiene_lint.py

@@ -0,0 +1,51 @@
+"""Scenario 11 - boundary hygiene lint (dangling flows + orphans).
+
+Audience: engineers maintaining the boundary-as-code file.
+
+Before a boundary definition reaches an assessor it should be internally
+consistent: no flow should reference a component that doesn't exist, and no
+in-boundary component should be stranded with zero data flows. This scenario
+runs those structural lints over the typo and orphan fixtures and prints a
+tidy pass/fail lint report an engineer fixes in the JSON.
+"""
+from _common import load, rule, bullet
+from fedramplens.core import analyze_boundary
+
+LINTS = {
+    "dangling_flow": "flow references an undefined component",
+    "orphan_component": "in-boundary component has no data flows",
+    "bad_poam_date": "POA&M scheduled date is not ISO-8601",
+}
+
+
+def _lint(key):
+    s = analyze_boundary(load(key))
+    hits = {}
+    for f in s["findings"]:
+        if f["type"] in LINTS:
+            hits.setdefault(f["type"], []).append(f["detail"])
+    return s, hits
+
+
+def main() -> None:
+    rule("BOUNDARY HYGIENE LINT  -  structural consistency checks")
+
+    for key in ("dangling_flow", "orphan", "clean_low"):
+        s, hits = _lint(key)
+        status = "FAIL" if hits else "PASS"
+        print(f"\n  [{status}] {s['system_name']} ({s['system_id']})")
+        if not hits:
+            bullet("clean -- no structural issues")
+        for lint, details in hits.items():
+            bullet(f"{lint}: {LINTS[lint]} ({len(details)} hit(s))")
+            for d in details:
+                print(f"        - {d}")
+
+    rule("LINT RULES CHECKED")
+    for lint, desc in LINTS.items():
+        bullet(f"{lint:22} {desc}")
+    print("\nRun this as a pre-commit hook on the boundary-as-code file.")
+
+
+if __name__ == "__main__":
+    main()

demos/12_high_baseline_walkthrough.py

@@ -0,0 +1,56 @@
+"""Scenario 12 - high-baseline package walkthrough.
+
+Audience: teams pursuing a FedRAMP High ATO.
+
+High-impact systems are held to the 410-control High baseline. This scenario
+takes a high-impact, SIEM-integrated boundary that is authorization-ready,
+walks its posture end to end (coverage against the High baseline, boundary
+crossings, POA&M, findings), and then emits its OSCAL SSP header so the team
+sees the machine-readable package it would submit.
+"""
+from _common import load, rule, bullet
+from fedramplens.core import analyze_boundary, generate_ssp
+
+
+def main() -> None:
+    rule("HIGH BASELINE WALKTHROUGH  -  a ready High-impact package")
+
+    b = load("high_ready")
+    s = analyze_boundary(b)
+
+    print(f"\nSystem: {s['system_name']} ({s['system_id']})")
+    bullet(f"impact              : {s['impact'].upper()} "
+           f"(baseline {s['baseline_controls']} controls)")
+    bullet(f"coverage            : {s['coverage_pct']}% "
+           f"({s['controls_implemented']} controls)")
+    bullet(f"components in bound. : {s['components_in_boundary']}")
+    bullet(f"external deps        : {len(s['external_dependencies'])}")
+    bullet(f"data flows           : {s['flows']}")
+    bullet(f"POA&M open / overdue : {s['poam_open']} / {len(s['poam_overdue'])}")
+    bullet(f"authorization-ready  : {s['authorization_ready']}")
+
+    counts = s["finding_counts"]
+    rule("FINDINGS")
+    if s["findings"]:
+        for f in s["findings"]:
+            bullet(f"({f['severity']}) {f['type']}: {f['detail']}")
+    else:
+        bullet("none -- no structural or crossing issues")
+    print(f"\n  severity counts: "
+          f"{', '.join(f'{k}={v}' for k, v in sorted(counts.items())) or 'none'}")
+
+    rule("OSCAL SSP HEADER (submission artifact)")
+    ssp = generate_ssp(b)["system-security-plan"]
+    meta = ssp["metadata"]
+    bullet(f"title          : {meta['title']}")
+    bullet(f"oscal-version  : {meta['oscal-version']}")
+    bullet(f"import-profile : {ssp['import-profile']['href']}")
+    bullet(f"components      : "
+           f"{len(ssp['system-implementation']['components'])}")
+    bullet(f"implemented reqs: "
+           f"{len(ssp['control-implementation']['implemented-requirements'])}")
+    print("\nA ready High package: coverage documented, crossings encrypted.")
+
+
+if __name__ == "__main__":
+    main()