From bc8d311b5505427a0a8cbb154ead9418f5c0f418 Mon Sep 17 00:00:00 2001 From: Cognis Digital <215970675+cognis-digital@users.noreply.github.com> Date: Tue, 30 Jun 2026 22:49:06 -0400 Subject: [PATCH 1/2] Deepen fedramplens: 4x tests, 4x demos, hardening + bug fixes Tests (53 -> 242 test functions, all green): - test_core_validation.py: exhaustive boundary/component/flow/POA&M validation + malformed-file error paths + direct-dataclass hardening - test_analysis.py: SC-8 crossing detection (both directions, tokens, missing key), dangling flows, orphans, POA&M overdue/risk/date, coverage math, authorization gate, summary shape - test_oscal.py: SSP + POA&M output shape, deterministic uuids, control-id normalization, DOT structure/escaping - test_sarif_edge.py: envelope, rule de-dup, level mapping + rule-default escalation, preserved properties, empty/robust findings - test_cli.py: every subcommand, all formats, exit codes, error paths - test_datafeeds.py: catalog, cache/age, offline get contract, snapshot round-trip, control normalization/fallback, graceful degrade Demos (5 -> 20 runnable scenarios, each exits 0): - Added 06-20 covering AO risk dashboard, CI gate, coverage report, POA&M tracker, dependency inventory, hygiene lint, High-baseline walkthrough, JSON pipeline, air-gap snapshot transfer, enrichment degrade contract, SARIF rule catalogue, diagram export, SSP deep inspect, error-handling showcase, and a full-package pipeline capstone - run_all.py + docs/DEMOS.md updated Hardening + real bug fixes (public API unchanged): - analyze_boundary raised an uncaught KeyError when a Boundary was built directly via the public dataclass with (a) an out-of-range/uppercased impact or (b) a POA&M severity outside the weight table. Now normalizes impact and raises a clear BoundaryError, and tolerates unknown severity as moderate. - POA&M scheduled-date validation now runs on every item (open or closed), so a malformed date is surfaced as bad_poam_date regardless of status. - _build_boundary rejects non-dict components/flows/POA&M items and a non-list controls field with precise errors. - to_sarif escalates a rule's default level to the most severe finding of its type and tolerates findings missing type/severity/detail. - CLI catches late BoundaryError from analyze/ssp/poam (exit 2). --- .gitignore | 1 + VERSION | 2 +- demos/06_ao_risk_dashboard.py | 50 ++++ demos/07_ci_gate_sarif_upload.py | 49 ++++ demos/08_control_coverage_report.py | 47 ++++ demos/09_poam_tracker.py | 44 ++++ demos/10_dependency_inventory.py | 47 ++++ demos/11_boundary_hygiene_lint.py | 51 ++++ demos/12_high_baseline_walkthrough.py | 56 ++++ demos/13_json_pipeline_integration.py | 56 ++++ demos/14_airgap_snapshot_transfer.py | 58 +++++ demos/15_enrichment_graceful_degrade.py | 61 +++++ demos/16_sarif_rule_catalogue.py | 46 ++++ demos/17_diagram_export_formats.py | 48 ++++ demos/18_oscal_ssp_deep_inspect.py | 65 +++++ demos/19_error_handling_showcase.py | 88 +++++++ demos/20_full_package_pipeline.py | 61 +++++ demos/run_all.py | 15 ++ docs/DEMOS.md | 19 +- fedramplens/cli.py | 62 +++-- fedramplens/core.py | 78 ++++-- tests/test_analysis.py | 333 ++++++++++++++++++++++++ tests/test_cli.py | 207 +++++++++++++++ tests/test_core_validation.py | 280 ++++++++++++++++++++ tests/test_datafeeds.py | 178 +++++++++++++ tests/test_demos.py | 94 +++++++ tests/test_oscal.py | 215 +++++++++++++++ tests/test_sarif_edge.py | 191 ++++++++++++++ 28 files changed, 2450 insertions(+), 52 deletions(-) create mode 100644 demos/06_ao_risk_dashboard.py create mode 100644 demos/07_ci_gate_sarif_upload.py create mode 100644 demos/08_control_coverage_report.py create mode 100644 demos/09_poam_tracker.py create mode 100644 demos/10_dependency_inventory.py create mode 100644 demos/11_boundary_hygiene_lint.py create mode 100644 demos/12_high_baseline_walkthrough.py create mode 100644 demos/13_json_pipeline_integration.py create mode 100644 demos/14_airgap_snapshot_transfer.py create mode 100644 demos/15_enrichment_graceful_degrade.py create mode 100644 demos/16_sarif_rule_catalogue.py create mode 100644 demos/17_diagram_export_formats.py create mode 100644 demos/18_oscal_ssp_deep_inspect.py create mode 100644 demos/19_error_handling_showcase.py create mode 100644 demos/20_full_package_pipeline.py create mode 100644 tests/test_analysis.py create mode 100644 tests/test_cli.py create mode 100644 tests/test_core_validation.py create mode 100644 tests/test_datafeeds.py create mode 100644 tests/test_oscal.py create mode 100644 tests/test_sarif_edge.py diff --git a/.gitignore b/.gitignore index 6a942db..18320d5 100644 --- a/.gitignore +++ b/.gitignore @@ -8,3 +8,4 @@ build/ !demos/** media/walkthrough.mp4 +demos/__pycache__/ diff --git a/VERSION b/VERSION index f51cebc..79a2734 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -0.4.9 +0.5.0 \ No newline at end of file diff --git a/demos/06_ao_risk_dashboard.py b/demos/06_ao_risk_dashboard.py new file mode 100644 index 0000000..ec2560c --- /dev/null +++ b/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() diff --git a/demos/07_ci_gate_sarif_upload.py b/demos/07_ci_gate_sarif_upload.py new file mode 100644 index 0000000..22554ef --- /dev/null +++ b/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() diff --git a/demos/08_control_coverage_report.py b/demos/08_control_coverage_report.py new file mode 100644 index 0000000..5c20e55 --- /dev/null +++ b/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() diff --git a/demos/09_poam_tracker.py b/demos/09_poam_tracker.py new file mode 100644 index 0000000..2add180 --- /dev/null +++ b/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() diff --git a/demos/10_dependency_inventory.py b/demos/10_dependency_inventory.py new file mode 100644 index 0000000..95c3b9d --- /dev/null +++ b/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() diff --git a/demos/11_boundary_hygiene_lint.py b/demos/11_boundary_hygiene_lint.py new file mode 100644 index 0000000..b95c8ed --- /dev/null +++ b/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() diff --git a/demos/12_high_baseline_walkthrough.py b/demos/12_high_baseline_walkthrough.py new file mode 100644 index 0000000..b300d0f --- /dev/null +++ b/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() diff --git a/demos/13_json_pipeline_integration.py b/demos/13_json_pipeline_integration.py new file mode 100644 index 0000000..269bbba --- /dev/null +++ b/demos/13_json_pipeline_integration.py @@ -0,0 +1,56 @@ +"""Scenario 13 - JSON output as a pipeline integration point. + +Audience: tooling / integration engineers. + +fedramplens's --format json is a stable machine contract. This scenario runs +the CLI in-process, captures the JSON, and shows how downstream tooling would +consume it: pull the top-level posture keys, filter findings by severity, and +confirm the exit code matches the authorization-ready verdict. No shelling out +-- it calls the real CLI main() and parses its stdout. +""" +import contextlib +import io +import json + +from _common import fixture_path, rule, bullet +from fedramplens.cli import main as cli_main + + +def _analyze_json(path): + buf = io.StringIO() + with contextlib.redirect_stdout(buf): + code = cli_main(["--format", "json", "analyze", path]) + return code, json.loads(buf.getvalue()) + + +def main() -> None: + rule("JSON PIPELINE - consume fedramplens output downstream") + + path = fixture_path("boundary_creep") + code, data = _analyze_json(path) + + print(f"\nCLI exit code : {code} " + f"(0=ready, 1=blocking findings)") + print("Top-level posture keys a pipeline reads:") + for key in ("system_id", "impact", "coverage_pct", "poam_risk_score", + "authorization_ready"): + bullet(f"{key:20} = {data[key]}") + + # Contract check: exit code agrees with the ready flag. + assert code == (0 if data["authorization_ready"] else 1) + bullet("exit code matches authorization_ready (contract holds)") + + rule("FILTER FINDINGS BY SEVERITY (downstream triage)") + for sev in ("critical", "high", "moderate", "low"): + hits = [f for f in data["findings"] if f["severity"] == sev] + if hits: + print(f" {sev.upper()} ({len(hits)}):") + for f in hits: + ctl = f.get("control", "-") + bullet(f"[{ctl}] {f['type']}: {f['detail']}") + + print("\nThe JSON is the integration seam -- stable keys, no scraping text.") + + +if __name__ == "__main__": + main() diff --git a/demos/14_airgap_snapshot_transfer.py b/demos/14_airgap_snapshot_transfer.py new file mode 100644 index 0000000..2a0770e --- /dev/null +++ b/demos/14_airgap_snapshot_transfer.py @@ -0,0 +1,58 @@ +"""Scenario 14 - air-gap snapshot transfer of the OSCAL catalog. + +Audience: teams operating disconnected / classified enclaves. + +An air-gapped enclave has no path to NIST's OSCAL content. This scenario shows +the sneakernet workflow: export the cached OSCAL 800-53 catalog to a portable +tarball, import it into a *fresh, empty* cache (simulating the enclave), and +then resolve control titles there with offline=True -- proving enrichment +works with zero network once the snapshot has crossed the gap. +""" +import os +import tempfile + +from _common import rule, bullet, use_offline_feed_cache +from fedramplens import controls, datafeeds + +FEED = "oscal-800-53-rev5-catalog" + + +def main() -> None: + rule("AIR-GAP SNAPSHOT - sneakernet the OSCAL catalog across the gap") + + # Source side: the connected cache holding the catalog. + use_offline_feed_cache() + src = os.environ["COGNIS_FEEDS_CACHE"] + print(f"\nSource cache (connected side): {src}") + bullet(f"catalog cached: {datafeeds.cached_age_hours(FEED) is not None}") + + # Export to a portable archive. + tmp = tempfile.mkdtemp(prefix="fedramplens_airgap_") + archive = os.path.join(tmp, "oscal-feeds.tar.gz") + n = datafeeds.snapshot_export(archive) + bullet(f"exported {n} feed(s) -> {os.path.basename(archive)} " + f"({os.path.getsize(archive)} bytes)") + + # Enclave side: a fresh empty cache with no network. + rule("INSIDE THE ENCLAVE - fresh empty cache, no network") + enclave = tempfile.mkdtemp(prefix="fedramplens_enclave_") + os.environ["COGNIS_FEEDS_CACHE"] = enclave + controls.reset_cache() + print(f"\nEnclave cache: {enclave}") + print(f"Before import, control_title('SC-8', offline) = " + f"{controls.control_title('SC-8', offline=True)!r} (nothing cached)") + + imported = datafeeds.snapshot_import(archive) + controls.reset_cache() + bullet(f"imported {imported} feed(s) from the tarball") + title = controls.control_title("SC-8", offline=True) + bullet(f"after import, control_title('SC-8', offline) = {title!r}") + assert title == "Transmission Confidentiality and Integrity" + + # restore the working cache for any later demo in the same process + use_offline_feed_cache() + print("\nOSCAL enrichment now works fully offline inside the enclave.") + + +if __name__ == "__main__": + main() diff --git a/demos/15_enrichment_graceful_degrade.py b/demos/15_enrichment_graceful_degrade.py new file mode 100644 index 0000000..22f7d90 --- /dev/null +++ b/demos/15_enrichment_graceful_degrade.py @@ -0,0 +1,61 @@ +"""Scenario 15 - enrichment contract: enriched vs graceful-degrade. + +Audience: integrators relying on control-title enrichment. + +Enrichment must never turn a working analysis into a crash. This scenario runs +the SAME boundary two ways -- against the bundled OSCAL cache (titles resolve) +and against an empty cache (titles unresolved) -- and shows the analysis +summary is structurally identical either way, only the control_titles map and +feed_available flag differ. That is the degrade contract downstream code leans +on. +""" +import os +import tempfile + +from _common import load, rule, bullet, use_offline_feed_cache +from fedramplens import controls +from fedramplens.core import analyze_boundary + + +def _summary(key): + return analyze_boundary(load(key), resolve_titles=True, offline=True) + + +def main() -> None: + rule("ENRICHMENT CONTRACT - enriched vs graceful degrade") + + # Enriched path. + use_offline_feed_cache() + enriched = _summary("oscal_enrichment") + print("\nWith the bundled OSCAL cache:") + bullet(f"feed_available : {enriched['feed_available']}") + bullet(f"titles resolved : {len(enriched['control_titles'])}") + bullet(f"findings : {len(enriched['findings'])}") + + # Degraded path: empty cache. + empty = tempfile.mkdtemp(prefix="fedramplens_degrade_") + os.environ["COGNIS_FEEDS_CACHE"] = empty + controls.reset_cache() + degraded = _summary("oscal_enrichment") + print("\nWith an empty cache (no network):") + bullet(f"feed_available : {degraded['feed_available']}") + bullet(f"titles resolved : {len(degraded['control_titles'])}") + bullet(f"findings : {len(degraded['findings'])}") + + rule("CONTRACT CHECK") + # Same structural analysis both ways. + same_findings = len(enriched["findings"]) == len(degraded["findings"]) + same_coverage = enriched["coverage_pct"] == degraded["coverage_pct"] + same_ready = enriched["authorization_ready"] == degraded["authorization_ready"] + bullet(f"identical finding count : {same_findings}") + bullet(f"identical coverage : {same_coverage}") + bullet(f"identical ready verdict : {same_ready}") + assert same_findings and same_coverage and same_ready + bullet("only control_titles + feed_available differ -> contract holds") + + use_offline_feed_cache() + print("\nEnrichment is purely additive: it never changes the verdict.") + + +if __name__ == "__main__": + main() diff --git a/demos/16_sarif_rule_catalogue.py b/demos/16_sarif_rule_catalogue.py new file mode 100644 index 0000000..f96e8e4 --- /dev/null +++ b/demos/16_sarif_rule_catalogue.py @@ -0,0 +1,46 @@ +"""Scenario 16 - the SARIF rule catalogue explained. + +Audience: security-tooling engineers integrating the SARIF output. + +Every finding type fedramplens can emit becomes a SARIF rule with a stable id, +a remediation description, and a default level. This scenario aggregates +findings across the whole demo corpus, then prints the resulting rule +catalogue -- id, PascalCase name, default level, and the fix guidance -- so an +integrator knows exactly what rules their code-scanning dashboard will show. +""" +from _common import load, rule, bullet, FIXTURES +from fedramplens.core import analyze_boundary, to_sarif + + +def main() -> None: + rule("SARIF RULE CATALOGUE - every finding type as a scanning rule") + + # Merge findings from all fixtures so each rule type appears at least once. + merged = [] + for key in FIXTURES: + merged.extend(analyze_boundary(load(key))["findings"]) + sarif = to_sarif({"system_id": "CORPUS", "system_name": "Demo Corpus", + "findings": merged}) + rules = sarif["runs"][0]["tool"]["driver"]["rules"] + + print(f"\n{len(rules)} distinct rule(s) across the demo corpus " + f"({len(merged)} findings total):\n") + for r in rules: + lvl = r["defaultConfiguration"]["level"] + print(f" - id={r['id']}") + print(f" name : {r['name']}") + print(f" level : {lvl}") + print(f" fix : {r['fullDescription']['text']}") + + rule("LEVEL DISTRIBUTION ACROSS RESULTS") + levels = {} + for res in sarif["runs"][0]["results"]: + levels[res["level"]] = levels.get(res["level"], 0) + 1 + for lvl in ("error", "warning", "note"): + bullet(f"{lvl:8} = {levels.get(lvl, 0)} result(s)") + + print("\nThese are the exact rules your code-scanning dashboard renders.") + + +if __name__ == "__main__": + main() diff --git a/demos/17_diagram_export_formats.py b/demos/17_diagram_export_formats.py new file mode 100644 index 0000000..674ce89 --- /dev/null +++ b/demos/17_diagram_export_formats.py @@ -0,0 +1,48 @@ +"""Scenario 17 - boundary diagram in multiple export formats. + +Audience: architects producing SSP diagrams. + +The authorization-boundary diagram is a required SSP artifact. This scenario +renders the same boundary as both Graphviz DOT (pipe to `dot -Tsvg` for the +SSP) and a Mermaid flowchart (renders inline in GitHub/GitLab), and highlights +how each format visually distinguishes encrypted vs unencrypted crossings and +in-boundary vs external components. +""" +from _common import load, rule, bullet, boundary_to_mermaid +from fedramplens.core import generate_dot + + +def main() -> None: + rule("DIAGRAM EXPORT - DOT + Mermaid for the SSP boundary figure") + + b = load("basic") + dot = generate_dot(b) + mer = boundary_to_mermaid(b) + + in_b = [c["id"] for c in b.components if c.get("zone") != "external"] + ext = [c["id"] for c in b.components if c.get("zone") == "external"] + print(f"\nSystem: {b.system_name} ({b.system_id})") + bullet(f"in-boundary : {', '.join(in_b)}") + bullet(f"external : {', '.join(ext)}") + + rule("GRAPHVIZ DOT (pipe to: dot -Tsvg -o boundary.svg)") + print() + for line in dot.splitlines()[:14]: + print(" " + line) + print(f" ... ({len(dot.splitlines())} lines total)") + bullet(f"unencrypted crossings drawn red: {'color=red' in dot}") + bullet(f"cluster subgraph for boundary : {'cluster_boundary' in dot}") + + rule("MERMAID (paste into a GitHub issue/README to render)") + print("\n```mermaid") + print(mer) + print("```") + bullet(f"dotted UNENCRYPTED edges present: {'-.->' in mer}") + bullet(f"authorization-boundary subgraph : " + f"{'Authorization Boundary' in mer}") + + print("\nSame boundary, two renderers -- one for the SSP, one for the repo.") + + +if __name__ == "__main__": + main() diff --git a/demos/18_oscal_ssp_deep_inspect.py b/demos/18_oscal_ssp_deep_inspect.py new file mode 100644 index 0000000..b01acf7 --- /dev/null +++ b/demos/18_oscal_ssp_deep_inspect.py @@ -0,0 +1,65 @@ +"""Scenario 18 - deep inspection of the generated OSCAL SSP. + +Audience: OSCAL toolers / package validators. + +Beyond "it produced JSON", this scenario validates the internal integrity of +the generated SSP: every implemented-requirement's by-components link resolves +to a declared component uuid, control ids are lower-case OSCAL form, uuids are +deterministic across runs, and enrichment (offline) annotates requirements +with their NIST 800-53 rev5 title. It's the check an OSCAL pipeline runs before +accepting the package. +""" +from _common import load, rule, bullet, use_offline_feed_cache +from fedramplens.core import generate_ssp + + +def main() -> None: + rule("OSCAL SSP DEEP INSPECT - internal-integrity validation") + use_offline_feed_cache() + + b = load("basic") + ssp = generate_ssp(b, resolve_titles=True, offline=True) + root = ssp["system-security-plan"] + comps = root["system-implementation"]["components"] + reqs = root["control-implementation"]["implemented-requirements"] + comp_uuids = {c["uuid"] for c in comps} + + print(f"\nSSP: {root['metadata']['title']}") + bullet(f"components : {len(comps)}") + bullet(f"implemented-requirements : {len(reqs)}") + + rule("INTEGRITY CHECKS") + # 1. Every by-component link resolves. + dangling = [r for r in reqs + if r["by-components"][0]["component-uuid"] not in comp_uuids] + bullet(f"by-component links all resolve : {not dangling}") + assert not dangling + + # 2. Control ids are lower-case OSCAL form. + bad_ids = [r["control-id"] for r in reqs + if r["control-id"] != r["control-id"].lower()] + bullet(f"control-ids are OSCAL lowercase: {not bad_ids}") + assert not bad_ids + + # 3. Deterministic uuids across a second generation. + ssp2 = generate_ssp(b)["system-security-plan"] + bullet(f"top-level uuid deterministic : {root['uuid'] == ssp2['uuid']}") + assert root["uuid"] == ssp2["uuid"] + + # 4. Enrichment attached titles. + labelled = [r for r in reqs if any( + p.get("class") == "nist-800-53-rev5-title" + for p in r.get("props", []))] + bullet(f"requirements annotated w/ title: {len(labelled)}/{len(reqs)}") + + rule("ANNOTATED REQUIREMENTS") + for r in labelled: + label = next(p["value"] for p in r["props"] + if p.get("class") == "nist-800-53-rev5-title") + print(f" - {r['control-id']:8} -> {label}") + + print("\nThe SSP passes structural validation an OSCAL pipeline would run.") + + +if __name__ == "__main__": + main() diff --git a/demos/19_error_handling_showcase.py b/demos/19_error_handling_showcase.py new file mode 100644 index 0000000..9ec45ac --- /dev/null +++ b/demos/19_error_handling_showcase.py @@ -0,0 +1,88 @@ +"""Scenario 19 - error-handling showcase (clean failures, not crashes). + +Audience: anyone feeding fedramplens untrusted / hand-edited boundary files. + +Good tooling fails loudly and clearly. This scenario deliberately feeds the +loader a series of malformed boundary definitions -- missing fields, a bad +impact, a duplicate component id, a flow with no endpoints, a wrong-typed +controls field, and an out-of-range impact discovered only at analysis time -- +and shows that each one raises a precise BoundaryError instead of a traceback. +""" +import json +import os +import tempfile + +from _common import rule, bullet +from fedramplens.core import ( + Boundary, BoundaryError, analyze_boundary, load_boundary, +) + +BAD_CASES = [ + ("missing impact", + {"system_name": "x", "system_id": "y", + "components": [{"id": "a", "zone": "boundary"}]}), + ("invalid impact 'medium'", + {"system_name": "x", "system_id": "y", "impact": "medium", + "components": [{"id": "a", "zone": "boundary"}]}), + ("no components", + {"system_name": "x", "system_id": "y", "impact": "low", + "components": []}), + ("duplicate component id", + {"system_name": "x", "system_id": "y", "impact": "low", + "components": [{"id": "a"}, {"id": "a"}]}), + ("flow missing endpoints", + {"system_name": "x", "system_id": "y", "impact": "low", + "components": [{"id": "a", "zone": "boundary"}], + "flows": [{"from": "a"}]}), + ("controls not a list", + {"system_name": "x", "system_id": "y", "impact": "low", + "components": [{"id": "a", "zone": "boundary", "controls": "AC-2"}]}), + ("bad POA&M severity", + {"system_name": "x", "system_id": "y", "impact": "low", + "components": [{"id": "a", "zone": "boundary"}], + "poam": [{"id": "P1", "severity": "showstopper"}]}), +] + + +def main() -> None: + rule("ERROR HANDLING - malformed boundaries fail cleanly") + + print("\nEach malformed input raises a precise BoundaryError:\n") + for label, raw in BAD_CASES: + fh = tempfile.NamedTemporaryFile("w", suffix=".json", delete=False) + json.dump(raw, fh) + fh.close() + try: + load_boundary(fh.name) + bullet(f"[{label}] -> NO ERROR (unexpected!)") + except BoundaryError as exc: + bullet(f"[{label}] -> BoundaryError: {exc}") + finally: + os.unlink(fh.name) + + rule("LATE VALIDATION - direct dataclass, caught at analysis time") + # A Boundary built directly can carry an out-of-range impact; analyze + # surfaces it as a BoundaryError rather than a raw KeyError. + b = Boundary("x", "y", "medium", [{"id": "a", "zone": "boundary"}], [], []) + try: + analyze_boundary(b) + bullet("analyze -> NO ERROR (unexpected!)") + except BoundaryError as exc: + bullet(f"analyze(out-of-range impact) -> BoundaryError: {exc}") + + rule("MALFORMED JSON") + fh = tempfile.NamedTemporaryFile("w", suffix=".json", delete=False) + fh.write("{ not: valid, json ]") + fh.close() + try: + load_boundary(fh.name) + except json.JSONDecodeError as exc: + bullet(f"load_boundary(bad json) -> JSONDecodeError: {exc.msg}") + finally: + os.unlink(fh.name) + + print("\nEvery bad input produced a clear, typed error -- no tracebacks.") + + +if __name__ == "__main__": + main() diff --git a/demos/20_full_package_pipeline.py b/demos/20_full_package_pipeline.py new file mode 100644 index 0000000..1f17d03 --- /dev/null +++ b/demos/20_full_package_pipeline.py @@ -0,0 +1,61 @@ +"""Scenario 20 - end-to-end package pipeline for one system. + +Audience: package leads running the whole flow. + +This capstone runs a single boundary through every stage of the fedramplens +pipeline in order: analyze -> SARIF -> boundary diagram -> OSCAL SSP -> OSCAL +POA&M, with offline control-title enrichment throughout. It's the full journey +from a boundary-as-code file to a submission-ready, machine-readable package, +narrated stage by stage. +""" +from _common import load, rule, bullet, use_offline_feed_cache +from fedramplens.core import ( + analyze_boundary, generate_dot, generate_poam, generate_ssp, to_sarif, +) + + +def main() -> None: + rule("FULL PACKAGE PIPELINE - boundary-as-code to OSCAL package") + use_offline_feed_cache() + + b = load("basic") + print(f"\nInput: {b.system_name} ({b.system_id}), {b.impact.upper()} impact\n") + + rule("STAGE 1 - analyze (enriched, offline)") + s = analyze_boundary(b, resolve_titles=True, offline=True) + bullet(f"coverage : {s['coverage_pct']}% of baseline") + bullet(f"findings : {len(s['findings'])}") + bullet(f"control titles : {len(s['control_titles'])} resolved") + bullet(f"authorization-ready : {s['authorization_ready']}") + + rule("STAGE 2 - SARIF export") + sarif = to_sarif(s) + bullet(f"schema : {sarif['version']}") + bullet(f"rules : {len(sarif['runs'][0]['tool']['driver']['rules'])}") + bullet(f"results : {len(sarif['runs'][0]['results'])}") + + rule("STAGE 3 - boundary diagram (DOT)") + dot = generate_dot(b) + bullet(f"DOT lines: {len(dot.splitlines())}") + bullet(f"boundary cluster present: {'cluster_boundary' in dot}") + + rule("STAGE 4 - OSCAL SSP") + ssp = generate_ssp(b, resolve_titles=True, offline=True)["system-security-plan"] + bullet(f"oscal-version : {ssp['metadata']['oscal-version']}") + bullet(f"components : " + f"{len(ssp['system-implementation']['components'])}") + bullet(f"implemented req: " + f"{len(ssp['control-implementation']['implemented-requirements'])}") + + rule("STAGE 5 - OSCAL POA&M") + poam = generate_poam(b)["plan-of-action-and-milestones"] + bullet(f"oscal-version : {poam['metadata']['oscal-version']}") + bullet(f"poam-items : {len(poam['poam-items'])}") + + rule("PIPELINE COMPLETE") + bullet("analyze -> SARIF -> diagram -> SSP -> POA&M, all offline") + print("\nOne boundary file produced the full machine-readable package.") + + +if __name__ == "__main__": + main() diff --git a/demos/run_all.py b/demos/run_all.py index dca304d..555f84e 100644 --- a/demos/run_all.py +++ b/demos/run_all.py @@ -18,6 +18,21 @@ "03_platform_engineer_boundary_map", "04_isso_oscal_packages", "05_offline_control_enrichment", + "06_ao_risk_dashboard", + "07_ci_gate_sarif_upload", + "08_control_coverage_report", + "09_poam_tracker", + "10_dependency_inventory", + "11_boundary_hygiene_lint", + "12_high_baseline_walkthrough", + "13_json_pipeline_integration", + "14_airgap_snapshot_transfer", + "15_enrichment_graceful_degrade", + "16_sarif_rule_catalogue", + "17_diagram_export_formats", + "18_oscal_ssp_deep_inspect", + "19_error_handling_showcase", + "20_full_package_pipeline", ] diff --git a/docs/DEMOS.md b/docs/DEMOS.md index 3bb1a10..835c4ad 100644 --- a/docs/DEMOS.md +++ b/docs/DEMOS.md @@ -1,13 +1,13 @@ # Demos -Five runnable scenarios live in [`../demos/`](../demos/), each written for a +Twenty runnable scenarios live in [`../demos/`](../demos/), each written for a different audience. Every scenario loads a **bundled** boundary fixture (the `demos/NN-*/boundary.json` packages) and drives the real `fedramplens` API — no fabricated data, no network. They print narrated output and exit 0, so they double as smoke tests (`tests/test_demos.py` covers the same code paths). ```bash -# all five, end to end +# all twenty, end to end PYTHONUTF8=1 python demos/run_all.py # or just one @@ -25,6 +25,21 @@ PYTHONUTF8=1 python demos/01_pm_authorization_readiness.py | 3 | [`03_platform_engineer_boundary_map.py`](../demos/03_platform_engineer_boundary_map.py) | **Cloud platform engineers** | Boundary map as Mermaid + Graphviz DOT; external dependencies and unencrypted boundary crossings (SC-8) to fix | `analyze_boundary`, `generate_dot` | | 4 | [`04_isso_oscal_packages.py`](../demos/04_isso_oscal_packages.py) | **ISSOs / package authors** | Generate the OSCAL SSP + POA&M, inspect metadata, component inventory, implemented-requirements, POA&M items | `generate_ssp`, `generate_poam` | | 5 | [`05_offline_control_enrichment.py`](../demos/05_offline_control_enrichment.py) | **ISSOs / assessors in an air-gap** | Resolve official NIST 800-53 rev5 control titles from the bundled OSCAL cache `offline=True`; graceful degrade on empty cache | `analyze_boundary`, `generate_ssp`, `controls.control_title` | +| 6 | [`06_ao_risk_dashboard.py`](../demos/06_ao_risk_dashboard.py) | **Authorizing Officials / risk executives** | Portfolio ranked by POA&M risk score and blocking-finding count; signature guidance for conditional ATOs | `analyze_boundary` | +| 7 | [`07_ci_gate_sarif_upload.py`](../demos/07_ci_gate_sarif_upload.py) | **DevSecOps / CI owners** | CI gate that fails the build on blocking findings and ships the SARIF artifact to code-scanning | `analyze_boundary`, `to_sarif` | +| 8 | [`08_control_coverage_report.py`](../demos/08_control_coverage_report.py) | **Control owners / compliance analysts** | Implemented-vs-baseline coverage, distinct controls with NIST titles, coverage across low/moderate/high baselines | `analyze_boundary`, `controls.control_title` | +| 9 | [`09_poam_tracker.py`](../demos/09_poam_tracker.py) | **ISSOs running remediation** | POA&M as a living backlog: open/closed, overdue, weighted risk, and bad-date data-quality findings | `analyze_boundary`, `generate_poam` | +| 10 | [`10_dependency_inventory.py`](../demos/10_dependency_inventory.py) | **ISSOs / architects** | External-dependency + interconnection inventory and which crossings are encrypted (SC-8) | `analyze_boundary` | +| 11 | [`11_boundary_hygiene_lint.py`](../demos/11_boundary_hygiene_lint.py) | **Engineers maintaining boundary-as-code** | Structural lint (dangling flows, orphan components, bad dates) as a pre-commit pass/fail report | `analyze_boundary` | +| 12 | [`12_high_baseline_walkthrough.py`](../demos/12_high_baseline_walkthrough.py) | **Teams pursuing FedRAMP High** | End-to-end posture of a ready High-impact package against the 410-control baseline + OSCAL SSP header | `analyze_boundary`, `generate_ssp` | +| 13 | [`13_json_pipeline_integration.py`](../demos/13_json_pipeline_integration.py) | **Tooling / integration engineers** | Consume `--format json` downstream: read posture keys, filter findings by severity, verify the exit-code contract | `cli.main` (`--format json`) | +| 14 | [`14_airgap_snapshot_transfer.py`](../demos/14_airgap_snapshot_transfer.py) | **Disconnected / classified enclaves** | Sneakernet the OSCAL catalog: export a tarball, import into a fresh empty cache, resolve titles offline | `datafeeds.snapshot_export/import`, `controls.control_title` | +| 15 | [`15_enrichment_graceful_degrade.py`](../demos/15_enrichment_graceful_degrade.py) | **Integrators using enrichment** | Same boundary enriched vs empty-cache: proves the analysis verdict is identical, only titles differ | `analyze_boundary` | +| 16 | [`16_sarif_rule_catalogue.py`](../demos/16_sarif_rule_catalogue.py) | **Security-tooling engineers** | Aggregate findings across the corpus and print the full SARIF rule catalogue (id, level, fix guidance) | `analyze_boundary`, `to_sarif` | +| 17 | [`17_diagram_export_formats.py`](../demos/17_diagram_export_formats.py) | **Architects producing SSP diagrams** | The same boundary as Graphviz DOT and Mermaid, highlighting encrypted vs unencrypted crossings | `generate_dot`, `boundary_to_mermaid` | +| 18 | [`18_oscal_ssp_deep_inspect.py`](../demos/18_oscal_ssp_deep_inspect.py) | **OSCAL toolers / validators** | Internal-integrity validation of the SSP: resolving links, lowercase control ids, deterministic uuids, title props | `generate_ssp` | +| 19 | [`19_error_handling_showcase.py`](../demos/19_error_handling_showcase.py) | **Anyone feeding untrusted files** | Malformed boundaries fail with precise `BoundaryError`s instead of tracebacks, incl. late validation | `load_boundary`, `analyze_boundary` | +| 20 | [`20_full_package_pipeline.py`](../demos/20_full_package_pipeline.py) | **Package leads** | Capstone: one boundary through analyze → SARIF → diagram → SSP → POA&M, enriched and offline | all of the above | ## Notes diff --git a/fedramplens/cli.py b/fedramplens/cli.py index 9b00f3f..ff2d134 100644 --- a/fedramplens/cli.py +++ b/fedramplens/cli.py @@ -191,33 +191,41 @@ def main(argv: Optional[List[str]] = None) -> int: print(f"error: invalid boundary: {exc}", file=sys.stderr) return 2 - if args.command == "analyze": - summary = analyze_boundary( - b, resolve_titles=args.enrich, offline=args.offline - ) - if args.format == "json": - print(json.dumps(summary, indent=2)) - elif args.format == "sarif": - print(json.dumps(to_sarif(summary), indent=2)) - else: - _print_analysis_table(summary) - # Non-zero exit if not authorization-ready. - return 0 if summary["authorization_ready"] else 1 - - if args.command == "diagram": - print(generate_dot(b)) - return 0 - - if args.command == "ssp": - print(json.dumps( - generate_ssp(b, resolve_titles=args.enrich, offline=args.offline), - indent=2, - )) - return 0 - - if args.command == "poam": - print(json.dumps(generate_poam(b), indent=2)) - return 0 + try: + if args.command == "analyze": + summary = analyze_boundary( + b, resolve_titles=args.enrich, offline=args.offline + ) + if args.format == "json": + print(json.dumps(summary, indent=2)) + elif args.format == "sarif": + print(json.dumps(to_sarif(summary), indent=2)) + else: + _print_analysis_table(summary) + # Non-zero exit if not authorization-ready. + return 0 if summary["authorization_ready"] else 1 + + if args.command == "diagram": + print(generate_dot(b)) + return 0 + + if args.command == "ssp": + print(json.dumps( + generate_ssp( + b, resolve_titles=args.enrich, offline=args.offline + ), + indent=2, + )) + return 0 + + if args.command == "poam": + print(json.dumps(generate_poam(b), indent=2)) + return 0 + except BoundaryError as exc: + # A Boundary can still fail validation late (e.g. an out-of-range + # impact only checked at analysis time). Surface it as a clean error. + print(f"error: invalid boundary: {exc}", file=sys.stderr) + return 2 parser.error(f"unknown command: {args.command}") return 2 diff --git a/fedramplens/core.py b/fedramplens/core.py index 1050da7..ca6baa9 100644 --- a/fedramplens/core.py +++ b/fedramplens/core.py @@ -112,6 +112,8 @@ def _build_boundary(raw: Dict[str, Any]) -> Boundary: raise BoundaryError("at least one component is required") seen = set() for c in components: + if not isinstance(c, dict): + raise BoundaryError("each component must be a JSON object") cid = c.get("id") if not cid: raise BoundaryError("component missing 'id'") @@ -123,11 +125,18 @@ def _build_boundary(raw: Dict[str, Any]) -> Boundary: raise BoundaryError( f"component {cid}: zone must be one of {VALID_ZONES}" ) + ctls = c.get("controls", []) + if not isinstance(ctls, list): + raise BoundaryError( + f"component {cid}: 'controls' must be a list" + ) flows = raw.get("flows", []) if not isinstance(flows, list): raise BoundaryError("'flows' must be a list") for f in flows: + if not isinstance(f, dict): + raise BoundaryError("each flow must be a JSON object") if not f.get("from") or not f.get("to"): raise BoundaryError("each flow needs 'from' and 'to'") @@ -135,6 +144,8 @@ def _build_boundary(raw: Dict[str, Any]) -> Boundary: if not isinstance(poam, list): raise BoundaryError("'poam' must be a list") for p in poam: + if not isinstance(p, dict): + raise BoundaryError("each POA&M item must be a JSON object") sev = str(p.get("severity", "moderate")).lower() if sev not in VALID_SEVERITIES: raise BoundaryError( @@ -209,9 +220,18 @@ def analyze_boundary( # 4. Control coverage estimate vs FedRAMP baseline. implemented = set() for c in b.components: - for ctl in c.get("controls", []): + for ctl in c.get("controls", []) or []: implemented.add(_normalize_control(ctl)) - baseline = BASELINE_CONTROL_COUNTS[b.impact] + # Normalize impact defensively: a Boundary may be constructed directly via + # the public dataclass (bypassing load_boundary/_build_boundary), so accept + # any case and reject an out-of-range impact with a clear BoundaryError + # rather than leaking a raw KeyError from the baseline table. + impact = str(b.impact).strip().lower() + if impact not in BASELINE_CONTROL_COUNTS: + raise BoundaryError( + f"impact must be one of {VALID_IMPACTS}, got {b.impact!r}" + ) + baseline = BASELINE_CONTROL_COUNTS[impact] coverage_pct = round(100.0 * len(implemented) / baseline, 1) # 5. POA&M risk roll-up + overdue detection. @@ -221,20 +241,27 @@ def analyze_boundary( overdue = [] open_items = 0 for p in b.poam: + # Validate the scheduled date on EVERY item (open or closed): a + # malformed date is a data-quality problem regardless of status. + sched = p.get("scheduled") + parsed_date = None + if sched: + try: + parsed_date = datetime.date.fromisoformat(str(sched)) + except (ValueError, TypeError): + findings.append({ + "severity": "low", + "type": "bad_poam_date", + "detail": f"POA&M {p.get('id')}: invalid date {sched!r}", + }) if str(p.get("status", "open")).lower() == "open": open_items += 1 - risk_score += sev_weight[str(p.get("severity", "moderate")).lower()] - sched = p.get("scheduled") - if sched: - try: - if datetime.date.fromisoformat(sched) < today: - overdue.append(p.get("id")) - except ValueError: - findings.append({ - "severity": "low", - "type": "bad_poam_date", - "detail": f"POA&M {p.get('id')}: invalid date {sched!r}", - }) + sev = str(p.get("severity", "moderate")).lower() + # Tolerate an unknown severity (e.g. from a directly-built + # Boundary): weight it as moderate rather than raising KeyError. + risk_score += sev_weight.get(sev, sev_weight["moderate"]) + if parsed_date is not None and parsed_date < today: + overdue.append(p.get("id")) for oid in overdue: findings.append({ "severity": "high", @@ -271,7 +298,7 @@ def analyze_boundary( summary = { "system_name": b.system_name, "system_id": b.system_id, - "impact": b.impact, + "impact": impact, "baseline_controls": baseline, "controls_implemented": len(implemented), "coverage_pct": coverage_pct, @@ -320,11 +347,15 @@ def to_sarif(summary: Dict[str, Any]) -> Dict[str, Any]: "bad_poam_date": "Use ISO-8601 (YYYY-MM-DD) for POA&M scheduled-completion dates.", } + # Rank severities so a rule's default level reflects the most severe + # finding of its type, not merely the first one encountered. + level_rank = {"note": 0, "warning": 1, "error": 2} rules: List[Dict[str, Any]] = [] rule_index: Dict[str, int] = {} results: List[Dict[str, Any]] = [] for f in summary.get("findings", []): - ftype = f["type"] + ftype = f.get("type", "finding") + level = level_map.get(f.get("severity"), "warning") if ftype not in rule_index: rule_index[ftype] = len(rules) rules.append({ @@ -336,18 +367,21 @@ def to_sarif(summary: Dict[str, Any]) -> Dict[str, Any]: }, "helpUri": "https://github.com/cognis-digital/fedramplens#findings", - "defaultConfiguration": { - "level": level_map.get(f["severity"], "warning") - }, + "defaultConfiguration": {"level": level}, }) - props = {"fedramp-severity": f["severity"], "finding-type": ftype} + else: + # Escalate the rule default if this occurrence is more severe. + cfg = rules[rule_index[ftype]]["defaultConfiguration"] + if level_rank[level] > level_rank[cfg["level"]]: + cfg["level"] = level + props = {"fedramp-severity": f.get("severity"), "finding-type": ftype} if f.get("control"): props["nist-control"] = f["control"] results.append({ "ruleId": ftype, "ruleIndex": rule_index[ftype], - "level": level_map.get(f["severity"], "warning"), - "message": {"text": f["detail"]}, + "level": level, + "message": {"text": f.get("detail", "")}, "properties": props, "locations": [{ "logicalLocations": [{ diff --git a/tests/test_analysis.py b/tests/test_analysis.py new file mode 100644 index 0000000..5914806 --- /dev/null +++ b/tests/test_analysis.py @@ -0,0 +1,333 @@ +"""Deep tests for analyze_boundary: SC-8 crossing detection, orphans, +dangling flows, POA&M overdue/risk/date handling, coverage math, and the +authorization-ready gate. Stdlib only, no network. +""" +import datetime +import os +import sys +import unittest + +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) + +from fedramplens.core import ( # noqa: E402 + BASELINE_CONTROL_COUNTS, + Boundary, + _build_boundary, + _is_external_token, + _normalize_control, + analyze_boundary, +) + + +def _b(**kw): + raw = { + "system_name": "S", "system_id": "ID", "impact": "moderate", + "components": [{"id": "a", "zone": "boundary", "controls": []}], + "flows": [], "poam": [], + } + raw.update(kw) + return _build_boundary(raw) + + +def _types(summary): + return [f["type"] for f in summary["findings"]] + + +class TestCrossingDetection(unittest.TestCase): + def _cross(self, encrypted): + # internal component 'a' <-> external component 'ext' + comps = [ + {"id": "a", "zone": "internal", "controls": []}, + {"id": "ext", "zone": "external", "controls": []}, + ] + flows = [{"from": "a", "to": "ext", "data": "d", "encrypted": encrypted}] + return analyze_boundary(_b(components=comps, flows=flows)) + + def test_unencrypted_crossing_flagged(self): + s = self._cross(False) + self.assertIn("unencrypted_boundary_crossing", _types(s)) + + def test_encrypted_crossing_not_flagged(self): + s = self._cross(True) + self.assertNotIn("unencrypted_boundary_crossing", _types(s)) + + def test_crossing_carries_sc8_control(self): + s = self._cross(False) + f = next(f for f in s["findings"] + if f["type"] == "unencrypted_boundary_crossing") + self.assertEqual(f["control"], "SC-8") + self.assertEqual(f["severity"], "high") + + def test_internal_to_internal_never_crosses(self): + comps = [ + {"id": "a", "zone": "internal", "controls": []}, + {"id": "b", "zone": "boundary", "controls": []}, + ] + flows = [{"from": "a", "to": "b", "encrypted": False}] + s = analyze_boundary(_b(components=comps, flows=flows)) + self.assertNotIn("unencrypted_boundary_crossing", _types(s)) + + def test_internet_token_to_boundary_is_a_crossing(self): + # "internet" is external-by-definition; unencrypted -> crossing + comps = [{"id": "web", "zone": "boundary", "controls": []}] + flows = [{"from": "internet", "to": "web", "encrypted": False}] + s = analyze_boundary(_b(components=comps, flows=flows)) + self.assertIn("unencrypted_boundary_crossing", _types(s)) + + def test_internet_token_encrypted_no_crossing(self): + comps = [{"id": "web", "zone": "boundary", "controls": []}] + flows = [{"from": "internet", "to": "web", "encrypted": True}] + s = analyze_boundary(_b(components=comps, flows=flows)) + self.assertNotIn("unencrypted_boundary_crossing", _types(s)) + + def test_missing_encrypted_key_treated_as_unencrypted(self): + comps = [ + {"id": "a", "zone": "internal", "controls": []}, + {"id": "ext", "zone": "external", "controls": []}, + ] + flows = [{"from": "a", "to": "ext"}] # no 'encrypted' key + s = analyze_boundary(_b(components=comps, flows=flows)) + self.assertIn("unencrypted_boundary_crossing", _types(s)) + + def test_reverse_direction_crossing(self): + comps = [ + {"id": "a", "zone": "internal", "controls": []}, + {"id": "ext", "zone": "external", "controls": []}, + ] + flows = [{"from": "ext", "to": "a", "encrypted": False}] + s = analyze_boundary(_b(components=comps, flows=flows)) + self.assertIn("unencrypted_boundary_crossing", _types(s)) + + +class TestDanglingFlows(unittest.TestCase): + def test_unknown_target_flagged(self): + comps = [{"id": "a", "zone": "boundary", "controls": []}] + flows = [{"from": "a", "to": "ghost", "encrypted": True}] + s = analyze_boundary(_b(components=comps, flows=flows)) + self.assertIn("dangling_flow", _types(s)) + + def test_unknown_source_flagged(self): + comps = [{"id": "a", "zone": "boundary", "controls": []}] + flows = [{"from": "ghost", "to": "a", "encrypted": True}] + s = analyze_boundary(_b(components=comps, flows=flows)) + self.assertIn("dangling_flow", _types(s)) + + def test_external_token_source_not_dangling(self): + comps = [{"id": "a", "zone": "boundary", "controls": []}] + flows = [{"from": "internet", "to": "a", "encrypted": True}] + s = analyze_boundary(_b(components=comps, flows=flows)) + self.assertNotIn("dangling_flow", _types(s)) + + def test_both_endpoints_unknown_yields_two_findings(self): + comps = [{"id": "a", "zone": "boundary", "controls": []}] + flows = [{"from": "x", "to": "y", "encrypted": True}] + s = analyze_boundary(_b(components=comps, flows=flows)) + self.assertEqual(_types(s).count("dangling_flow"), 2) + + +class TestOrphanComponents(unittest.TestCase): + def test_orphan_in_boundary_flagged(self): + comps = [ + {"id": "a", "zone": "boundary", "controls": []}, + {"id": "lonely", "zone": "internal", "controls": []}, + ] + flows = [{"from": "internet", "to": "a", "encrypted": True}] + s = analyze_boundary(_b(components=comps, flows=flows)) + detail = " ".join(f["detail"] for f in s["findings"] + if f["type"] == "orphan_component") + self.assertIn("lonely", detail) + + def test_external_component_never_orphan(self): + comps = [ + {"id": "a", "zone": "boundary", "controls": []}, + {"id": "ext", "zone": "external", "controls": []}, + ] + flows = [{"from": "internet", "to": "a", "encrypted": True}] + s = analyze_boundary(_b(components=comps, flows=flows)) + # 'ext' is untouched but external, so it must NOT be an orphan + detail = " ".join(f["detail"] for f in s["findings"] + if f["type"] == "orphan_component") + self.assertNotIn("ext", detail) + + def test_touched_component_not_orphan(self): + comps = [{"id": "a", "zone": "boundary", "controls": []}] + flows = [{"from": "internet", "to": "a", "encrypted": True}] + s = analyze_boundary(_b(components=comps, flows=flows)) + self.assertNotIn("orphan_component", _types(s)) + + def test_orphan_is_low_severity(self): + comps = [{"id": "a", "zone": "boundary", "controls": []}] + s = analyze_boundary(_b(components=comps, flows=[])) + f = next(f for f in s["findings"] if f["type"] == "orphan_component") + self.assertEqual(f["severity"], "low") + + +class TestPoamAnalysis(unittest.TestCase): + def test_open_count(self): + poam = [ + {"id": "P1", "severity": "high", "status": "open"}, + {"id": "P2", "severity": "low", "status": "completed"}, + {"id": "P3", "severity": "moderate", "status": "open"}, + ] + s = analyze_boundary(_b(poam=poam)) + self.assertEqual(s["poam_open"], 2) + + def test_risk_score_sums_open_only(self): + poam = [ + {"id": "P1", "severity": "critical", "status": "open"}, # 8 + {"id": "P2", "severity": "high", "status": "open"}, # 4 + {"id": "P3", "severity": "high", "status": "completed"}, # 0 + ] + s = analyze_boundary(_b(poam=poam)) + self.assertEqual(s["poam_risk_score"], 12) + + def test_overdue_detected(self): + poam = [{"id": "P1", "severity": "high", "status": "open", + "scheduled": "2000-01-01"}] + s = analyze_boundary(_b(poam=poam)) + self.assertIn("P1", s["poam_overdue"]) + self.assertIn("overdue_poam", _types(s)) + + def test_future_date_not_overdue(self): + future = (datetime.date.today() + datetime.timedelta(days=365)).isoformat() + poam = [{"id": "P1", "severity": "high", "status": "open", + "scheduled": future}] + s = analyze_boundary(_b(poam=poam)) + self.assertEqual(s["poam_overdue"], []) + + def test_closed_overdue_item_not_flagged(self): + poam = [{"id": "P1", "severity": "high", "status": "completed", + "scheduled": "2000-01-01"}] + s = analyze_boundary(_b(poam=poam)) + self.assertEqual(s["poam_overdue"], []) + + def test_bad_date_flagged_for_open_item(self): + poam = [{"id": "P1", "severity": "high", "status": "open", + "scheduled": "31/12/2026"}] + s = analyze_boundary(_b(poam=poam)) + self.assertIn("bad_poam_date", _types(s)) + + def test_bad_date_flagged_even_for_closed_item(self): + # Hardened behavior: a malformed date is a data-quality issue on ANY item. + poam = [{"id": "P1", "severity": "high", "status": "completed", + "scheduled": "not-a-date"}] + s = analyze_boundary(_b(poam=poam)) + self.assertIn("bad_poam_date", _types(s)) + + def test_no_scheduled_date_is_fine(self): + poam = [{"id": "P1", "severity": "high", "status": "open"}] + s = analyze_boundary(_b(poam=poam)) + self.assertNotIn("bad_poam_date", _types(s)) + self.assertEqual(s["poam_overdue"], []) + + def test_empty_poam(self): + s = analyze_boundary(_b(poam=[])) + self.assertEqual(s["poam_open"], 0) + self.assertEqual(s["poam_risk_score"], 0) + self.assertEqual(s["poam_overdue"], []) + + +class TestCoverageMath(unittest.TestCase): + def test_baseline_counts(self): + self.assertEqual(BASELINE_CONTROL_COUNTS["low"], 156) + self.assertEqual(BASELINE_CONTROL_COUNTS["moderate"], 323) + self.assertEqual(BASELINE_CONTROL_COUNTS["high"], 410) + + def test_coverage_percentage(self): + comps = [{"id": "a", "zone": "boundary", + "controls": ["AC-2", "SC-7", "SC-8"]}] + s = analyze_boundary(_b(components=comps, impact="moderate")) + # 3 / 323 * 100 = 0.9 + self.assertEqual(s["controls_implemented"], 3) + self.assertAlmostEqual(s["coverage_pct"], round(300 / 323, 1)) + + def test_duplicate_controls_counted_once(self): + comps = [ + {"id": "a", "zone": "boundary", "controls": ["AC-2", "ac-2"]}, + {"id": "b", "zone": "internal", "controls": ["AC-2"]}, + ] + flows = [{"from": "a", "to": "b", "encrypted": True}] + s = analyze_boundary(_b(components=comps, flows=flows)) + self.assertEqual(s["controls_implemented"], 1) + + def test_zero_controls_zero_coverage(self): + s = analyze_boundary(_b()) + self.assertEqual(s["coverage_pct"], 0.0) + + +class TestAuthorizationGate(unittest.TestCase): + def test_high_finding_blocks(self): + comps = [ + {"id": "a", "zone": "internal", "controls": []}, + {"id": "ext", "zone": "external", "controls": []}, + ] + flows = [{"from": "a", "to": "ext", "encrypted": False}] + s = analyze_boundary(_b(components=comps, flows=flows)) + self.assertFalse(s["authorization_ready"]) + + def test_only_low_findings_still_ready(self): + # a lone orphan (low) should not block authorization + s = analyze_boundary(_b()) + self.assertEqual(_types(s), ["orphan_component"]) + self.assertTrue(s["authorization_ready"]) + + def test_clean_system_is_ready(self): + comps = [{"id": "a", "zone": "boundary", "controls": ["AC-2"]}] + flows = [{"from": "internet", "to": "a", "encrypted": True}] + s = analyze_boundary(_b(components=comps, flows=flows)) + self.assertTrue(s["authorization_ready"]) + self.assertEqual(s["findings"], []) + + def test_finding_counts_by_severity(self): + comps = [ + {"id": "a", "zone": "internal", "controls": []}, + {"id": "ext", "zone": "external", "controls": []}, + {"id": "orphan", "zone": "internal", "controls": []}, + ] + flows = [{"from": "a", "to": "ext", "encrypted": False}] + s = analyze_boundary(_b(components=comps, flows=flows)) + counts = s["finding_counts"] + self.assertEqual(counts.get("high"), 1) + self.assertEqual(counts.get("low"), 1) + + +class TestSummaryShape(unittest.TestCase): + def test_summary_keys_present(self): + s = analyze_boundary(_b()) + for key in ( + "system_name", "system_id", "impact", "baseline_controls", + "controls_implemented", "coverage_pct", "components_in_boundary", + "external_dependencies", "flows", "poam_open", "poam_overdue", + "poam_risk_score", "findings", "finding_counts", + "authorization_ready", + ): + self.assertIn(key, s) + + def test_no_enrich_keys_by_default(self): + s = analyze_boundary(_b()) + self.assertNotIn("control_titles", s) + self.assertNotIn("feed_available", s) + + def test_external_dependencies_listed(self): + comps = [ + {"id": "a", "zone": "boundary", "controls": []}, + {"id": "ext1", "zone": "external", "controls": []}, + {"id": "ext2", "zone": "external", "controls": []}, + ] + s = analyze_boundary(_b(components=comps)) + self.assertEqual(set(s["external_dependencies"]), {"ext1", "ext2"}) + + +class TestHelpers(unittest.TestCase): + def test_normalize_control(self): + self.assertEqual(_normalize_control("ac-2"), "AC-2") + self.assertEqual(_normalize_control(" sc-8 "), "SC-8") + + def test_is_external_token(self): + for tok in ("internet", "USER", "External", "public", "saas"): + self.assertTrue(_is_external_token(tok)) + self.assertFalse(_is_external_token("web-tier")) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_cli.py b/tests/test_cli.py new file mode 100644 index 0000000..62ade2b --- /dev/null +++ b/tests/test_cli.py @@ -0,0 +1,207 @@ +"""Tests for the fedramplens command-line interface. + +Covers every subcommand (analyze/diagram/ssp/poam/feeds), all output formats +(table/json/sarif), exit codes (0 ready, 1 findings, 2 usage/error), and the +error paths for missing files, malformed JSON, invalid boundaries, and +late-validation failures surfaced from analyze. Offline; uses the committed +OSCAL fixture cache. Stdlib only. +""" +import contextlib +import io +import json +import os +import sys +import tempfile +import unittest + +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) + +_HERE = os.path.dirname(__file__) +_FIXTURE_CACHE = os.path.join(_HERE, "fixtures", "feedcache") +os.environ["COGNIS_FEEDS_CACHE"] = _FIXTURE_CACHE + +from fedramplens import controls # noqa: E402 +from fedramplens.cli import main # noqa: E402 + +_ROOT = os.path.abspath(os.path.join(_HERE, "..")) +DEMO = os.path.join(_ROOT, "demos", "01-basic", "boundary.json") +CLEAN = os.path.join(_ROOT, "demos", "02-clean-low-saas", "boundary.json") + +VALID = { + "system_name": "CLI Sys", "system_id": "FR-CLI", "impact": "moderate", + "components": [{"id": "web", "name": "Web", "zone": "boundary", + "controls": ["AC-2", "SC-8"]}], + "flows": [{"from": "internet", "to": "web", "encrypted": True}], + "poam": [], +} + + +def _run(argv): + out, err = io.StringIO(), io.StringIO() + with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err): + code = main(argv) + return code, out.getvalue(), err.getvalue() + + +def _write(data): + fh = tempfile.NamedTemporaryFile("w", suffix=".json", delete=False) + if isinstance(data, str): + fh.write(data) + else: + json.dump(data, fh) + fh.close() + return fh.name + + +class TestAnalyzeCommand(unittest.TestCase): + def test_analyze_table_findings_exit_1(self): + code, out, _ = _run(["analyze", DEMO]) + self.assertEqual(code, 1) + self.assertIn("Authorization-ready", out) + + def test_analyze_clean_exit_0(self): + code, out, _ = _run(["analyze", CLEAN]) + self.assertEqual(code, 0) + + def test_analyze_json(self): + code, out, _ = _run(["--format", "json", "analyze", DEMO]) + self.assertEqual(code, 1) + data = json.loads(out) + self.assertEqual(data["system_id"], "FR2026ACME01") + + def test_analyze_sarif(self): + code, out, _ = _run(["--format", "sarif", "analyze", DEMO]) + self.assertEqual(code, 1) + self.assertEqual(json.loads(out)["version"], "2.1.0") + + def test_analyze_own_valid_file_ready(self): + path = _write(VALID) + try: + code, out, _ = _run(["analyze", path]) + self.assertEqual(code, 0) + self.assertIn("YES", out) + finally: + os.unlink(path) + + def test_analyze_table_prints_coverage(self): + code, out, _ = _run(["analyze", DEMO]) + self.assertIn("Controls:", out) + self.assertIn("of baseline", out) + + +class TestErrorPaths(unittest.TestCase): + def test_missing_file_exit_2(self): + code, _, err = _run(["analyze", os.path.join(tempfile.gettempdir(), + "no-such.json")]) + self.assertEqual(code, 2) + self.assertIn("file not found", err) + + def test_malformed_json_exit_2(self): + path = _write("{ bad json") + try: + code, _, err = _run(["analyze", path]) + self.assertEqual(code, 2) + self.assertIn("invalid JSON", err) + finally: + os.unlink(path) + + def test_invalid_boundary_exit_2(self): + path = _write({"system_name": "x", "system_id": "y", + "impact": "medium", "components": [{"id": "a"}]}) + try: + code, _, err = _run(["analyze", path]) + self.assertEqual(code, 2) + self.assertIn("invalid boundary", err) + finally: + os.unlink(path) + + def test_missing_components_exit_2(self): + path = _write({"system_name": "x", "system_id": "y", + "impact": "low", "components": []}) + try: + code, _, err = _run(["analyze", path]) + self.assertEqual(code, 2) + finally: + os.unlink(path) + + +class TestDiagramCommand(unittest.TestCase): + def test_diagram_exit_0(self): + code, out, _ = _run(["diagram", DEMO]) + self.assertEqual(code, 0) + self.assertIn("digraph", out) + + def test_diagram_missing_file_exit_2(self): + code, _, err = _run(["diagram", "/no/such/file.json"]) + self.assertEqual(code, 2) + + +class TestSspPoamCommands(unittest.TestCase): + def test_ssp_exit_0(self): + code, out, _ = _run(["ssp", DEMO]) + self.assertEqual(code, 0) + self.assertIn("system-security-plan", json.loads(out)) + + def test_poam_exit_0(self): + code, out, _ = _run(["poam", DEMO]) + self.assertEqual(code, 0) + self.assertIn("plan-of-action-and-milestones", json.loads(out)) + + def test_ssp_enrich_offline(self): + controls.reset_cache() + code, out, _ = _run(["ssp", DEMO, "--enrich", "--offline"]) + self.assertEqual(code, 0) + # AC-2 title should appear from the fixture catalog + self.assertIn("Account Management", out) + + +class TestFeedsCommand(unittest.TestCase): + def setUp(self): + os.environ["COGNIS_FEEDS_CACHE"] = _FIXTURE_CACHE + controls.reset_cache() + + def test_feeds_list_exit_0(self): + code, out, _ = _run(["feeds", "list"]) + self.assertEqual(code, 0) + self.assertIn("oscal-800-53-rev5-catalog", out) + + def test_feeds_get_offline(self): + code, out, _ = _run(["feeds", "get", "oscal-800-53-rev5-catalog", + "--offline"]) + self.assertEqual(code, 0) + self.assertIn("catalog", out) + + def test_feeds_get_unrelated_exit_2(self): + code, _, err = _run(["feeds", "get", "cisa-kev", "--offline"]) + self.assertEqual(code, 2) + self.assertIn("not a feed", err) + + def test_feeds_get_unknown_exit_2(self): + code, _, err = _run(["feeds", "get", "made-up-feed", "--offline"]) + self.assertEqual(code, 2) + + +class TestVersionAndUsage(unittest.TestCase): + def test_version_flag(self): + with self.assertRaises(SystemExit) as cm: + _run(["--version"]) + self.assertEqual(cm.exception.code, 0) + + def test_no_command_errors(self): + with self.assertRaises(SystemExit) as cm: + _run([]) + self.assertNotEqual(cm.exception.code, 0) + + def test_analyze_enrich_offline_shows_titles(self): + controls.reset_cache() + path = _write(VALID) + try: + code, out, _ = _run(["analyze", path, "--enrich", "--offline"]) + self.assertEqual(code, 0) + self.assertIn("Account Management", out) + finally: + os.unlink(path) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_core_validation.py b/tests/test_core_validation.py new file mode 100644 index 0000000..125a830 --- /dev/null +++ b/tests/test_core_validation.py @@ -0,0 +1,280 @@ +"""Edge-case + error-path tests for boundary loading and validation. + +Exercises _build_boundary / load_boundary hard against malformed input: +missing/empty fields, wrong types, duplicate ids, bad zones/severities, and +the defensive normalization that keeps analyze_boundary from leaking raw +KeyErrors when a Boundary is constructed directly through the public dataclass. +Stdlib only, no network. +""" +import json +import os +import sys +import tempfile +import unittest + +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) + +from fedramplens.core import ( # noqa: E402 + Boundary, + BoundaryError, + _build_boundary, + analyze_boundary, + load_boundary, +) + +BASE = { + "system_name": "Base", + "system_id": "FR-BASE", + "impact": "moderate", + "components": [ + {"id": "a", "name": "A", "zone": "boundary", "controls": ["AC-2"]}, + ], + "flows": [], + "poam": [], +} + + +def _mut(**changes): + raw = json.loads(json.dumps(BASE)) + raw.update(changes) + return raw + + +class TestTopLevelValidation(unittest.TestCase): + def test_not_a_dict_rejected(self): + for bad in ([], "x", 3, None): + with self.assertRaises(BoundaryError): + _build_boundary(bad) + + def test_missing_system_name(self): + raw = _mut() + del raw["system_name"] + with self.assertRaises(BoundaryError): + _build_boundary(raw) + + def test_empty_system_name_rejected(self): + with self.assertRaises(BoundaryError): + _build_boundary(_mut(system_name="")) + + def test_missing_system_id(self): + raw = _mut() + del raw["system_id"] + with self.assertRaises(BoundaryError): + _build_boundary(raw) + + def test_missing_impact(self): + raw = _mut() + del raw["impact"] + with self.assertRaises(BoundaryError): + _build_boundary(raw) + + def test_impact_case_insensitive(self): + b = _build_boundary(_mut(impact="MODERATE")) + self.assertEqual(b.impact, "moderate") + + def test_impact_low_moderate_high_all_valid(self): + for imp in ("low", "moderate", "high"): + b = _build_boundary(_mut(impact=imp)) + self.assertEqual(b.impact, imp) + + def test_impact_invalid_value(self): + for bad in ("medium", "severe", "critical", "unknown", ""): + with self.assertRaises(BoundaryError): + _build_boundary(_mut(impact=bad)) + + def test_error_message_names_missing_field(self): + raw = _mut() + del raw["impact"] + try: + _build_boundary(raw) + self.fail("expected BoundaryError") + except BoundaryError as exc: + self.assertIn("impact", str(exc)) + + +class TestComponentValidation(unittest.TestCase): + def test_components_must_be_present(self): + with self.assertRaises(BoundaryError): + _build_boundary(_mut(components=[])) + + def test_components_must_be_a_list(self): + with self.assertRaises(BoundaryError): + _build_boundary(_mut(components={"id": "a"})) + + def test_component_not_a_dict(self): + with self.assertRaises(BoundaryError): + _build_boundary(_mut(components=["a", "b"])) + + def test_component_missing_id(self): + with self.assertRaises(BoundaryError): + _build_boundary(_mut(components=[{"name": "no id"}])) + + def test_component_empty_id(self): + with self.assertRaises(BoundaryError): + _build_boundary(_mut(components=[{"id": ""}])) + + def test_duplicate_component_id(self): + dup = [{"id": "a", "zone": "boundary"}, {"id": "a", "zone": "internal"}] + with self.assertRaises(BoundaryError): + _build_boundary(_mut(components=dup)) + + def test_bad_zone_rejected(self): + comps = [{"id": "a", "zone": "dmz"}] + with self.assertRaises(BoundaryError): + _build_boundary(_mut(components=comps)) + + def test_all_valid_zones_accepted(self): + for zone in ("boundary", "internal", "external"): + comps = [{"id": "a", "zone": zone, "controls": []}] + b = _build_boundary(_mut(components=comps)) + self.assertEqual(b.components[0]["zone"], zone) + + def test_zone_defaults_to_internal(self): + comps = [{"id": "a", "controls": []}] + b = _build_boundary(_mut(components=comps)) + # in_boundary treats a missing zone as internal (in boundary) + self.assertTrue(b.in_boundary("a")) + + def test_controls_must_be_a_list(self): + comps = [{"id": "a", "zone": "boundary", "controls": "AC-2"}] + with self.assertRaises(BoundaryError): + _build_boundary(_mut(components=comps)) + + def test_controls_optional(self): + comps = [{"id": "a", "zone": "boundary"}] + b = _build_boundary(_mut(components=comps)) + self.assertEqual(b.components[0].get("controls", []), []) + + +class TestFlowValidation(unittest.TestCase): + def test_flows_must_be_a_list(self): + with self.assertRaises(BoundaryError): + _build_boundary(_mut(flows={"from": "a", "to": "b"})) + + def test_flow_not_a_dict(self): + with self.assertRaises(BoundaryError): + _build_boundary(_mut(flows=["a->b"])) + + def test_flow_missing_from(self): + with self.assertRaises(BoundaryError): + _build_boundary(_mut(flows=[{"to": "a"}])) + + def test_flow_missing_to(self): + with self.assertRaises(BoundaryError): + _build_boundary(_mut(flows=[{"from": "a"}])) + + def test_flow_empty_endpoints(self): + with self.assertRaises(BoundaryError): + _build_boundary(_mut(flows=[{"from": "", "to": "a"}])) + + def test_valid_flow_accepted(self): + b = _build_boundary( + _mut(flows=[{"from": "internet", "to": "a", "encrypted": True}]) + ) + self.assertEqual(len(b.flows), 1) + + +class TestPoamValidation(unittest.TestCase): + def test_poam_must_be_a_list(self): + with self.assertRaises(BoundaryError): + _build_boundary(_mut(poam={"id": "P1"})) + + def test_poam_item_not_a_dict(self): + with self.assertRaises(BoundaryError): + _build_boundary(_mut(poam=["P1"])) + + def test_poam_bad_severity_rejected(self): + item = [{"id": "P1", "severity": "showstopper"}] + with self.assertRaises(BoundaryError): + _build_boundary(_mut(poam=item)) + + def test_poam_all_valid_severities(self): + for sev in ("low", "moderate", "high", "critical"): + item = [{"id": "P1", "severity": sev, "status": "open"}] + b = _build_boundary(_mut(poam=item)) + self.assertEqual(len(b.poam), 1) + + def test_poam_severity_case_insensitive(self): + item = [{"id": "P1", "severity": "HIGH", "status": "open"}] + b = _build_boundary(_mut(poam=item)) + self.assertEqual(len(b.poam), 1) + + def test_poam_severity_defaults_to_moderate(self): + item = [{"id": "P1", "status": "open"}] + b = _build_boundary(_mut(poam=item)) + self.assertEqual(len(b.poam), 1) + + +class TestLoadBoundaryFile(unittest.TestCase): + def _write(self, text): + fh = tempfile.NamedTemporaryFile("w", suffix=".json", delete=False) + fh.write(text) + fh.close() + self.addCleanup(os.unlink, fh.name) + return fh.name + + def test_missing_file_raises(self): + with self.assertRaises(FileNotFoundError): + load_boundary(os.path.join(tempfile.gettempdir(), "nope-xyz.json")) + + def test_malformed_json_raises(self): + path = self._write("{ not valid json ]") + with self.assertRaises(json.JSONDecodeError): + load_boundary(path) + + def test_valid_file_loads(self): + path = self._write(json.dumps(BASE)) + b = load_boundary(path) + self.assertEqual(b.system_id, "FR-BASE") + + def test_json_array_top_level_rejected(self): + path = self._write("[]") + with self.assertRaises(BoundaryError): + load_boundary(path) + + +class TestDirectDataclassHardening(unittest.TestCase): + """A Boundary built directly (public API) bypasses _build_boundary; + analyze_boundary must not leak raw KeyErrors on odd values.""" + + def test_uppercase_impact_normalized(self): + b = Boundary("X", "Y", "HIGH", [{"id": "a", "zone": "boundary"}], [], []) + s = analyze_boundary(b) + self.assertEqual(s["impact"], "high") + self.assertEqual(s["baseline_controls"], 410) + + def test_out_of_range_impact_raises_boundaryerror(self): + b = Boundary("X", "Y", "medium", [{"id": "a", "zone": "boundary"}], [], []) + with self.assertRaises(BoundaryError): + analyze_boundary(b) + + def test_unknown_severity_weighted_as_moderate(self): + b = Boundary( + "X", "Y", "low", [{"id": "a", "zone": "boundary"}], [], + [{"id": "P1", "severity": "bogus", "status": "open"}], + ) + s = analyze_boundary(b) + self.assertEqual(s["poam_risk_score"], 2) # moderate weight + + def test_none_controls_do_not_crash(self): + b = Boundary( + "X", "Y", "low", + [{"id": "a", "zone": "boundary", "controls": None}], [], [], + ) + s = analyze_boundary(b) + self.assertEqual(s["controls_implemented"], 0) + + def test_component_helpers(self): + b = Boundary( + "X", "Y", "low", + [{"id": "a", "zone": "boundary"}, {"id": "e", "zone": "external"}], + [], [], + ) + self.assertEqual(b.component_ids(), {"a", "e"}) + self.assertTrue(b.in_boundary("a")) + self.assertFalse(b.in_boundary("e")) + self.assertFalse(b.in_boundary("unknown")) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_datafeeds.py b/tests/test_datafeeds.py new file mode 100644 index 0000000..419c3a9 --- /dev/null +++ b/tests/test_datafeeds.py @@ -0,0 +1,178 @@ +"""Offline tests for the datafeeds layer + controls enrichment edge cases. + +Never touches the network: COGNIS_FEEDS_CACHE points at the committed OSCAL +fixture cache and everything runs offline=True. Covers the catalog loader, +cache paths/age metadata, offline get() error contract, snapshot export/import +round-trip, control-id normalization, enhancement fallback, and graceful +degrade to {} on a missing cache. Stdlib only. +""" +import os +import sys +import tempfile +import unittest + +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) + +_HERE = os.path.dirname(__file__) +_FIXTURE_CACHE = os.path.join(_HERE, "fixtures", "feedcache") +os.environ["COGNIS_FEEDS_CACHE"] = _FIXTURE_CACHE + +from fedramplens import controls, datafeeds # noqa: E402 + +FEED = "oscal-800-53-rev5-catalog" + + +class TestCatalog(unittest.TestCase): + def test_load_catalog_has_feeds(self): + cat = datafeeds.load_catalog() + self.assertIn("feeds", cat) + self.assertTrue(cat["feeds"]) + + def test_oscal_feed_present(self): + ids = {f["id"] for f in datafeeds.list_feeds()} + self.assertIn(FEED, ids) + + def test_oscal_feed_is_oscal_format(self): + feed = next(f for f in datafeeds.list_feeds() if f["id"] == FEED) + self.assertEqual(feed.get("format"), "oscal") + + def test_list_feeds_filter_by_domain(self): + feed = next(f for f in datafeeds.list_feeds() if f["id"] == FEED) + domain = feed.get("domain") + if domain: + filtered = datafeeds.list_feeds(domain=domain) + self.assertTrue(all(f.get("domain") == domain for f in filtered)) + + def test_load_missing_catalog_returns_empty(self): + missing = os.path.join(tempfile.gettempdir(), "no-catalog-xyz.json") + self.assertEqual(datafeeds.load_catalog(missing), {"feeds": []}) + + +class TestCachePaths(unittest.TestCase): + def setUp(self): + os.environ["COGNIS_FEEDS_CACHE"] = _FIXTURE_CACHE + + def test_cache_dir_created(self): + d = datafeeds.cache_dir() + self.assertTrue(os.path.isdir(d)) + + def test_cached_age_for_fixture(self): + age = datafeeds.cached_age_hours(FEED) + self.assertIsNotNone(age) + self.assertGreaterEqual(age, 0) + + def test_cached_age_none_for_unknown(self): + self.assertIsNone(datafeeds.cached_age_hours("never-fetched-feed")) + + +class TestOfflineGet(unittest.TestCase): + def setUp(self): + os.environ["COGNIS_FEEDS_CACHE"] = _FIXTURE_CACHE + + def test_offline_get_returns_dict(self): + data = datafeeds.get(FEED, offline=True) + self.assertIsInstance(data, dict) + self.assertIn("catalog", data) + + def test_offline_get_missing_raises(self): + empty = tempfile.mkdtemp(prefix="ff-empty-") + os.environ["COGNIS_FEEDS_CACHE"] = empty + try: + with self.assertRaises(FileNotFoundError): + datafeeds.get(FEED, offline=True) + finally: + os.environ["COGNIS_FEEDS_CACHE"] = _FIXTURE_CACHE + + +class TestSnapshot(unittest.TestCase): + def test_export_import_roundtrip(self): + os.environ["COGNIS_FEEDS_CACHE"] = _FIXTURE_CACHE + tmp = tempfile.mkdtemp(prefix="ff-snap-") + archive = os.path.join(tmp, "feeds.tar.gz") + n = datafeeds.snapshot_export(archive) + self.assertGreaterEqual(n, 1) + self.assertTrue(os.path.exists(archive)) + + # Import into a fresh empty cache and confirm the feed reads back. + dest = tempfile.mkdtemp(prefix="ff-dest-") + os.environ["COGNIS_FEEDS_CACHE"] = dest + try: + imported = datafeeds.snapshot_import(archive) + self.assertGreaterEqual(imported, 1) + data = datafeeds.get(FEED, offline=True) + self.assertIn("catalog", data) + finally: + os.environ["COGNIS_FEEDS_CACHE"] = _FIXTURE_CACHE + + +class TestControlNormalization(unittest.TestCase): + def setUp(self): + os.environ["COGNIS_FEEDS_CACHE"] = _FIXTURE_CACHE + controls.reset_cache() + + def test_normalize_forms(self): + self.assertEqual(controls._normalize("AC-2(1)"), "ac-2.1") + self.assertEqual(controls._normalize("ac-2.1"), "ac-2.1") + self.assertEqual(controls._normalize(" SC-8 "), "sc-8") + + def test_title_lookup(self): + self.assertEqual( + controls.control_title("SC-8", offline=True), + "Transmission Confidentiality and Integrity", + ) + + def test_title_lookup_case_insensitive(self): + self.assertEqual( + controls.control_title("sc-8", offline=True), + controls.control_title("SC-8", offline=True), + ) + + def test_enhancement_exact_hit(self): + self.assertEqual( + controls.control_title("AC-2(1)", offline=True), + "Automated System Account Management", + ) + + def test_enhancement_falls_back_to_base(self): + self.assertEqual( + controls.control_title("AC-2(99)", offline=True), + "Account Management", + ) + + def test_unknown_returns_none(self): + self.assertIsNone(controls.control_title("ZZ-9", offline=True)) + + def test_empty_control_returns_none(self): + self.assertIsNone(controls.control_title("", offline=True)) + self.assertIsNone(controls.control_title(None, offline=True)) + + def test_enrich_controls_map(self): + m = controls.enrich_controls(["AC-2", "ZZ-9"], offline=True) + self.assertEqual(m["AC-2"], "Account Management") + self.assertIsNone(m["ZZ-9"]) + + +class TestGracefulDegrade(unittest.TestCase): + def test_missing_cache_returns_none(self): + empty = tempfile.mkdtemp(prefix="ff-degrade-") + os.environ["COGNIS_FEEDS_CACHE"] = empty + controls.reset_cache() + try: + self.assertIsNone(controls.control_title("SC-8", offline=True)) + finally: + os.environ["COGNIS_FEEDS_CACHE"] = _FIXTURE_CACHE + controls.reset_cache() + + def test_build_title_map_direct(self): + cat = datafeeds.get(FEED, offline=True) + titles = controls.build_title_map(cat) + self.assertEqual(titles["ac-2"], "Account Management") + self.assertEqual(titles["sc-13"], "Cryptographic Protection") + + def test_build_title_map_handles_empty(self): + self.assertEqual(controls.build_title_map({}), {}) + self.assertEqual(controls.build_title_map({"catalog": {}}), {}) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_demos.py b/tests/test_demos.py index d6847ed..921d199 100644 --- a/tests/test_demos.py +++ b/tests/test_demos.py @@ -24,6 +24,21 @@ "03_platform_engineer_boundary_map", "04_isso_oscal_packages", "05_offline_control_enrichment", + "06_ao_risk_dashboard", + "07_ci_gate_sarif_upload", + "08_control_coverage_report", + "09_poam_tracker", + "10_dependency_inventory", + "11_boundary_hygiene_lint", + "12_high_baseline_walkthrough", + "13_json_pipeline_integration", + "14_airgap_snapshot_transfer", + "15_enrichment_graceful_degrade", + "16_sarif_rule_catalogue", + "17_diagram_export_formats", + "18_oscal_ssp_deep_inspect", + "19_error_handling_showcase", + "20_full_package_pipeline", ] @@ -95,6 +110,85 @@ def test_enrichment_scenario_resolves_titles_offline(self): self.assertIn("Transmission Confidentiality and Integrity", text) self.assertIn("GRACEFUL DEGRADE", text) + def test_ao_dashboard_ranks_portfolio(self): + text = self._run("06_ao_risk_dashboard") + self.assertIn("AO RISK DASHBOARD", text) + self.assertIn("SIGNATURE GUIDANCE", text) + + def test_ci_gate_reports_exit_codes(self): + text = self._run("07_ci_gate_sarif_upload") + self.assertIn("CI GATE", text) + self.assertIn("build FAILS", text) + self.assertIn("SARIF ARTIFACT", text) + + def test_coverage_report_lists_titles(self): + text = self._run("08_control_coverage_report") + self.assertIn("CONTROL COVERAGE", text) + self.assertIn("baseline", text) + self.assertIn("Account Management", text) + + def test_poam_tracker_shows_overdue(self): + text = self._run("09_poam_tracker") + self.assertIn("POA&M TRACKER", text) + self.assertIn("overdue", text) + + def test_dependency_inventory_lists_external(self): + text = self._run("10_dependency_inventory") + self.assertIn("EXTERNAL DEPENDENCY INVENTORY", text) + self.assertIn("SC-8", text) + + def test_hygiene_lint_pass_fail(self): + text = self._run("11_boundary_hygiene_lint") + self.assertIn("BOUNDARY HYGIENE LINT", text) + self.assertIn("PASS", text) + self.assertIn("FAIL", text) + + def test_high_baseline_walkthrough(self): + text = self._run("12_high_baseline_walkthrough") + self.assertIn("HIGH BASELINE WALKTHROUGH", text) + self.assertIn("oscal-version", text) + + def test_json_pipeline_contract(self): + text = self._run("13_json_pipeline_integration") + self.assertIn("JSON PIPELINE", text) + self.assertIn("contract holds", text) + + def test_airgap_snapshot_transfer(self): + text = self._run("14_airgap_snapshot_transfer") + self.assertIn("AIR-GAP SNAPSHOT", text) + self.assertIn("Transmission Confidentiality and Integrity", text) + + def test_enrichment_degrade_contract(self): + text = self._run("15_enrichment_graceful_degrade") + self.assertIn("ENRICHMENT CONTRACT", text) + self.assertIn("contract holds", text) + + def test_sarif_rule_catalogue(self): + text = self._run("16_sarif_rule_catalogue") + self.assertIn("SARIF RULE CATALOGUE", text) + self.assertIn("unencrypted_boundary_crossing", text) + + def test_diagram_export_formats(self): + text = self._run("17_diagram_export_formats") + self.assertIn("DIAGRAM EXPORT", text) + self.assertIn("```mermaid", text) + self.assertIn("digraph boundary", text) + + def test_ssp_deep_inspect(self): + text = self._run("18_oscal_ssp_deep_inspect") + self.assertIn("OSCAL SSP DEEP INSPECT", text) + self.assertIn("INTEGRITY CHECKS", text) + + def test_error_handling_showcase(self): + text = self._run("19_error_handling_showcase") + self.assertIn("ERROR HANDLING", text) + self.assertIn("BoundaryError", text) + + def test_full_package_pipeline(self): + text = self._run("20_full_package_pipeline") + self.assertIn("FULL PACKAGE PIPELINE", text) + self.assertIn("PIPELINE COMPLETE", text) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_oscal.py b/tests/test_oscal.py new file mode 100644 index 0000000..f9aa822 --- /dev/null +++ b/tests/test_oscal.py @@ -0,0 +1,215 @@ +"""Tests for OSCAL SSP + POA&M generation and Graphviz DOT rendering. + +Covers the machine-readable artifact shape (metadata, oscal-version, +components, implemented-requirements, poam-items), deterministic UUIDs, +control-id normalization, and DOT structure/escaping. Stdlib only, no network. +""" +import os +import sys +import unittest + +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) + +from fedramplens.core import ( # noqa: E402 + _build_boundary, + _uuid_like, + generate_dot, + generate_poam, + generate_ssp, +) + + +def _b(**kw): + raw = { + "system_name": "OSCAL Sys", "system_id": "FR-OSCAL", "impact": "high", + "components": [ + {"id": "web", "name": "Web", "type": "service", "zone": "boundary", + "controls": ["AC-2", "SC-8"]}, + {"id": "db", "name": "DB", "type": "database", "zone": "internal", + "controls": ["SC-13"]}, + ], + "flows": [{"from": "web", "to": "db", "encrypted": True}], + "poam": [ + {"id": "V-1", "weakness": "weak cipher", "control": "SC-13", + "severity": "high", "status": "open", "scheduled": "2027-01-01"}, + ], + } + raw.update(kw) + return _build_boundary(raw) + + +class TestSSP(unittest.TestCase): + def setUp(self): + self.ssp = generate_ssp(_b())["system-security-plan"] + + def test_top_level_uuid(self): + self.assertTrue(self.ssp["uuid"]) + + def test_metadata_oscal_version(self): + self.assertEqual(self.ssp["metadata"]["oscal-version"], "1.1.2") + + def test_metadata_title(self): + self.assertIn("System Security Plan", self.ssp["metadata"]["title"]) + + def test_last_modified_is_iso(self): + # datetime.fromisoformat parses it back without raising + import datetime + datetime.datetime.fromisoformat(self.ssp["metadata"]["last-modified"]) + + def test_import_profile_references_baseline(self): + self.assertEqual( + self.ssp["import-profile"]["href"], "#fedramp-high-baseline" + ) + + def test_sensitivity_level(self): + sc = self.ssp["system-characteristics"] + self.assertEqual(sc["security-sensitivity-level"], "high") + self.assertEqual(sc["system-ids"][0]["id"], "FR-OSCAL") + + def test_components_inventory(self): + comps = self.ssp["system-implementation"]["components"] + self.assertEqual(len(comps), 2) + titles = {c["title"] for c in comps} + self.assertEqual(titles, {"Web", "DB"}) + + def test_component_zone_prop(self): + comps = self.ssp["system-implementation"]["components"] + web = next(c for c in comps if c["title"] == "Web") + zone = next(p["value"] for p in web["props"] if p["name"] == "zone") + self.assertEqual(zone, "boundary") + + def test_implemented_requirements(self): + reqs = self.ssp["control-implementation"]["implemented-requirements"] + ids = {r["control-id"] for r in reqs} + self.assertEqual(ids, {"ac-2", "sc-8", "sc-13"}) + + def test_control_ids_are_lowercase_oscal(self): + reqs = self.ssp["control-implementation"]["implemented-requirements"] + for r in reqs: + self.assertEqual(r["control-id"], r["control-id"].lower()) + + def test_by_components_link(self): + reqs = self.ssp["control-implementation"]["implemented-requirements"] + for r in reqs: + self.assertTrue(r["by-components"][0]["component-uuid"]) + self.assertIn("description", r["by-components"][0]) + + def test_component_without_controls_yields_no_reqs(self): + comps = [{"id": "x", "name": "X", "zone": "boundary"}] + flows = [{"from": "internet", "to": "x", "encrypted": True}] + ssp = generate_ssp(_b(components=comps, flows=flows)) + reqs = ssp["system-security-plan"]["control-implementation"][ + "implemented-requirements"] + self.assertEqual(reqs, []) + + def test_ssp_deterministic(self): + a = generate_ssp(_b())["system-security-plan"]["uuid"] + b = generate_ssp(_b())["system-security-plan"]["uuid"] + self.assertEqual(a, b) + + +class TestPoamGen(unittest.TestCase): + def setUp(self): + self.poam = generate_poam(_b())["plan-of-action-and-milestones"] + + def test_metadata(self): + self.assertEqual(self.poam["metadata"]["oscal-version"], "1.1.2") + self.assertIn("POA&M", self.poam["metadata"]["title"]) + + def test_system_id(self): + self.assertEqual(self.poam["system-id"]["id"], "FR-OSCAL") + + def test_item_count(self): + self.assertEqual(len(self.poam["poam-items"]), 1) + + def test_item_props(self): + item = self.poam["poam-items"][0] + props = {p["name"]: p["value"] for p in item["props"]} + self.assertEqual(props["severity"], "high") + self.assertEqual(props["status"], "open") + self.assertEqual(props["control"], "SC-13") + self.assertEqual(props["scheduled-completion"], "2027-01-01") + + def test_item_description_from_weakness(self): + item = self.poam["poam-items"][0] + self.assertEqual(item["description"], "weak cipher") + + def test_empty_poam_yields_no_items(self): + poam = generate_poam(_b(poam=[]))["plan-of-action-and-milestones"] + self.assertEqual(poam["poam-items"], []) + + def test_poam_item_missing_fields_default(self): + raw_poam = [{"id": "V-9", "severity": "low"}] + poam = generate_poam(_b(poam=raw_poam))["plan-of-action-and-milestones"] + item = poam["poam-items"][0] + props = {p["name"]: p["value"] for p in item["props"]} + self.assertEqual(props["status"], "open") # default + self.assertEqual(props["control"], "") # empty default + self.assertEqual(item["description"], "") # no weakness + + +class TestUuid(unittest.TestCase): + def test_uuid_shape(self): + u = _uuid_like("seed") + parts = u.split("-") + self.assertEqual([len(p) for p in parts], [8, 4, 4, 4, 12]) + + def test_uuid_deterministic(self): + self.assertEqual(_uuid_like("abc"), _uuid_like("abc")) + + def test_uuid_distinct_seeds(self): + self.assertNotEqual(_uuid_like("a"), _uuid_like("b")) + + +class TestDot(unittest.TestCase): + def test_digraph_header(self): + dot = generate_dot(_b()) + self.assertIn("digraph boundary", dot) + self.assertIn("cluster_boundary", dot) + + def test_edges_present(self): + dot = generate_dot(_b()) + self.assertIn("->", dot) + + def test_label_has_system_and_impact(self): + dot = generate_dot(_b()) + self.assertIn("FR-OSCAL", dot) + self.assertIn("HIGH", dot) + + def test_unencrypted_flow_is_red(self): + comps = [ + {"id": "a", "zone": "internal", "controls": []}, + {"id": "ext", "zone": "external", "controls": []}, + ] + flows = [{"from": "a", "to": "ext", "data": "x", "encrypted": False}] + dot = generate_dot(_b(components=comps, flows=flows)) + self.assertIn("color=red", dot) + + def test_encrypted_flow_is_black(self): + dot = generate_dot(_b()) + self.assertIn("color=black", dot) + + def test_external_node_synthesized_from_flow_token(self): + comps = [{"id": "web", "zone": "boundary", "controls": []}] + flows = [{"from": "internet", "to": "web", "encrypted": True}] + dot = generate_dot(_b(components=comps, flows=flows)) + self.assertIn("internet", dot) + + def test_quotes_escaped_in_labels(self): + comps = [{"id": "a", "name": 'Web "prod" tier', "zone": "boundary", + "controls": []}] + flows = [{"from": "internet", "to": "a", "encrypted": True}] + dot = generate_dot(_b(components=comps, flows=flows)) + # double quotes are replaced with single quotes to keep DOT valid + self.assertNotIn('"prod"', dot) + self.assertIn("'prod'", dot) + + def test_node_ids_sanitized(self): + comps = [{"id": "web-tier.01", "zone": "boundary", "controls": []}] + flows = [{"from": "internet", "to": "web-tier.01", "encrypted": True}] + dot = generate_dot(_b(components=comps, flows=flows)) + self.assertIn("n_web_tier_01", dot) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_sarif_edge.py b/tests/test_sarif_edge.py new file mode 100644 index 0000000..e0eef31 --- /dev/null +++ b/tests/test_sarif_edge.py @@ -0,0 +1,191 @@ +"""Edge-case tests for the SARIF 2.1.0 exporter. + +Covers the envelope/schema, rule catalogue de-duplication, severity->level +mapping (including the hardened rule-default escalation), preserved control + +fedramp-severity properties, empty runs, and robustness against findings that +are missing optional keys. Stdlib only, no network. +""" +import os +import sys +import unittest + +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) + +from fedramplens.core import ( # noqa: E402 + TOOL_NAME, + _build_boundary, + analyze_boundary, + to_sarif, +) + + +def _b(**kw): + raw = { + "system_name": "Sarif", "system_id": "FR-SARIF", "impact": "moderate", + "components": [{"id": "a", "zone": "boundary", "controls": []}], + "flows": [], "poam": [], + } + raw.update(kw) + return _build_boundary(raw) + + +def _crossing_boundary(): + comps = [ + {"id": "a", "zone": "internal", "controls": []}, + {"id": "ext", "zone": "external", "controls": []}, + ] + flows = [{"from": "a", "to": "ext", "data": "d", "encrypted": False}] + return _b(components=comps, flows=flows) + + +class TestEnvelope(unittest.TestCase): + def test_schema_and_version(self): + s = to_sarif(analyze_boundary(_b())) + self.assertEqual(s["version"], "2.1.0") + self.assertIn("sarif-2.1.0", s["$schema"]) + + def test_single_run(self): + s = to_sarif(analyze_boundary(_b())) + self.assertEqual(len(s["runs"]), 1) + + def test_driver_identity(self): + driver = to_sarif(analyze_boundary(_b()))["runs"][0]["tool"]["driver"] + self.assertEqual(driver["name"], TOOL_NAME) + self.assertTrue(driver["version"]) + self.assertIn("informationUri", driver) + + def test_run_properties(self): + s = to_sarif(analyze_boundary(_crossing_boundary())) + props = s["runs"][0]["properties"] + self.assertEqual(props["system-id"], "FR-SARIF") + self.assertEqual(props["impact"], "moderate") + self.assertIn("coverage-pct", props) + self.assertIn("authorization-ready", props) + + +class TestRuleCatalogue(unittest.TestCase): + def test_rules_deduplicated(self): + # Two dangling flows share one finding-type -> one rule, two results. + comps = [{"id": "a", "zone": "boundary", "controls": []}] + flows = [ + {"from": "a", "to": "ghost1", "encrypted": True}, + {"from": "a", "to": "ghost2", "encrypted": True}, + ] + s = to_sarif(analyze_boundary(_b(components=comps, flows=flows))) + run = s["runs"][0] + rule_ids = [r["id"] for r in run["tool"]["driver"]["rules"]] + self.assertEqual(rule_ids.count("dangling_flow"), 1) + dangs = [r for r in run["results"] if r["ruleId"] == "dangling_flow"] + self.assertEqual(len(dangs), 2) + + def test_rule_index_consistent(self): + s = to_sarif(analyze_boundary(_crossing_boundary())) + run = s["runs"][0] + rules = run["tool"]["driver"]["rules"] + for res in run["results"]: + self.assertEqual(rules[res["ruleIndex"]]["id"], res["ruleId"]) + + def test_rule_has_help_uri_and_descriptions(self): + s = to_sarif(analyze_boundary(_crossing_boundary())) + for r in s["runs"][0]["tool"]["driver"]["rules"]: + self.assertIn("helpUri", r) + self.assertIn("shortDescription", r) + self.assertIn("fullDescription", r) + + def test_rule_name_is_pascal_case(self): + s = to_sarif(analyze_boundary(_crossing_boundary())) + r = next(r for r in s["runs"][0]["tool"]["driver"]["rules"] + if r["id"] == "unencrypted_boundary_crossing") + self.assertEqual(r["name"], "UnencryptedBoundaryCrossing") + + +class TestLevelMapping(unittest.TestCase): + def test_high_maps_to_error(self): + s = to_sarif(analyze_boundary(_crossing_boundary())) + res = next(r for r in s["runs"][0]["results"] + if r["ruleId"] == "unencrypted_boundary_crossing") + self.assertEqual(res["level"], "error") + + def test_low_maps_to_note(self): + # a lone orphan is a low finding -> note + s = to_sarif(analyze_boundary(_b())) + res = next(r for r in s["runs"][0]["results"] + if r["ruleId"] == "orphan_component") + self.assertEqual(res["level"], "note") + + def test_rule_default_escalates_to_most_severe(self): + # Build findings of the same type with differing severity by hand, + # then confirm the rule default reflects the highest (error). + summary = { + "system_id": "X", "system_name": "X", + "findings": [ + {"type": "t", "severity": "low", "detail": "a"}, + {"type": "t", "severity": "high", "detail": "b"}, + ], + } + s = to_sarif(summary) + rule = s["runs"][0]["tool"]["driver"]["rules"][0] + self.assertEqual(rule["defaultConfiguration"]["level"], "error") + + def test_unknown_severity_defaults_to_warning(self): + summary = { + "findings": [{"type": "t", "severity": "weird", "detail": "x"}], + } + s = to_sarif(summary) + self.assertEqual(s["runs"][0]["results"][0]["level"], "warning") + + +class TestProperties(unittest.TestCase): + def test_control_preserved(self): + s = to_sarif(analyze_boundary(_crossing_boundary())) + controls = [r["properties"].get("nist-control") + for r in s["runs"][0]["results"]] + self.assertIn("SC-8", controls) + + def test_fedramp_severity_preserved(self): + s = to_sarif(analyze_boundary(_crossing_boundary())) + sevs = [r["properties"]["fedramp-severity"] + for r in s["runs"][0]["results"]] + self.assertIn("high", sevs) + + def test_result_has_logical_location(self): + s = to_sarif(analyze_boundary(_crossing_boundary())) + loc = s["runs"][0]["results"][0]["locations"][0]["logicalLocations"][0] + self.assertEqual(loc["name"], "FR-SARIF") + self.assertEqual(loc["kind"], "module") + + def test_no_control_property_when_absent(self): + # orphan findings carry no control + s = to_sarif(analyze_boundary(_b())) + res = next(r for r in s["runs"][0]["results"] + if r["ruleId"] == "orphan_component") + self.assertNotIn("nist-control", res["properties"]) + + +class TestEmptyAndRobust(unittest.TestCase): + def test_clean_boundary_empty_results(self): + comps = [{"id": "a", "zone": "boundary", "controls": ["AC-2"]}] + flows = [{"from": "internet", "to": "a", "encrypted": True}] + s = to_sarif(analyze_boundary(_b(components=comps, flows=flows))) + self.assertEqual(s["runs"][0]["results"], []) + self.assertEqual(s["runs"][0]["tool"]["driver"]["rules"], []) + + def test_missing_findings_key(self): + s = to_sarif({"system_id": "X"}) + self.assertEqual(s["runs"][0]["results"], []) + + def test_finding_missing_optional_keys(self): + # No 'severity', no 'detail' — exporter must still produce valid output. + s = to_sarif({"findings": [{"type": "custom"}]}) + res = s["runs"][0]["results"][0] + self.assertEqual(res["ruleId"], "custom") + self.assertEqual(res["level"], "warning") + self.assertEqual(res["message"]["text"], "") + + def test_finding_missing_type_key(self): + s = to_sarif({"findings": [{"severity": "high", "detail": "d"}]}) + self.assertEqual(s["runs"][0]["results"][0]["ruleId"], "finding") + + +if __name__ == "__main__": + unittest.main() From af900e552e5051a0b2ff4ae263d6b372753852e5 Mon Sep 17 00:00:00 2001 From: Cognis Digital <215970675+cognis-digital@users.noreply.github.com> Date: Tue, 30 Jun 2026 23:57:43 -0400 Subject: [PATCH 2/2] Fix labeler.yml to actions/labeler@v5 schema (was v4, failed on every PR) The config used the deprecated v4 flat glob syntax; labeler@v5 requires each label to be an array of config options with changed-files / any-glob-to-any-file. It errored 'found unexpected type for label docs' on every pull_request_target run. Migrated all rules and added a demos label. --- .github/labeler.yml | 19 +++++++++++++++---- 1 file changed, 15 insertions(+), 4 deletions(-) diff --git a/.github/labeler.yml b/.github/labeler.yml index d910b08..d7f5df9 100644 --- a/.github/labeler.yml +++ b/.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/**'