Add runnable audience-targeted demo scenarios + architecture/demos docs

Five narrated demos (ISSO/ISSM, SOC, sysadmin/DevSecOps, auditor, edge/air-gap)
drive the real comint_osquery/cognis_mil API over bundled offline fixtures and
the committed feed cache; each exits 0 and is covered by tests. Adds
docs/ARCHITECTURE.md (mermaid pipeline + fleet + data-model) and docs/DEMOS.md
(audience table), a Demos section + diagram in README, and a pytest module that
executes every scenario. Fix pre-existing demo-dir test to skip __pycache__.

Author: cognis-digital

README.md

@@ -114,7 +114,40 @@ pip install -e ../../shared
 pip install -e .
 ```
 
-## Demos — real-use-case library
+## Demos
+
+Five runnable, narrated scenarios in [`demos/`](demos/) drive the **real**
+`comint-osquery` / `cognis_mil` API over the bundled offline fixtures — no
+network, no fabricated output, each exits 0. Each targets a different audience.
+Full write-up: [`docs/DEMOS.md`](docs/DEMOS.md).
+
+```bash
+PYTHONUTF8=1 python demos/run_all.py            # all five, end to end
+PYTHONUTF8=1 python demos/03_sysadmin_fleet.py  # or just one
+```
+
+| # | Scenario | Audience | What it shows |
+|---|----------|----------|---------------|
+| 1 | `01_isso_assessment.py` | **ISSO / ISSM** | Scan → composite risk + full RMF crosswalk → OSCAL 1.1.2 SAR for eMASS |
+| 2 | `02_soc_detection.py` | **SOC / endpoint** | Emit the scheduled osquery STIG pack; map every query to its ATT&CK technique |
+| 3 | `03_sysadmin_fleet.py` | **Sysadmins / DevSecOps** | Fleet correlation (systemic vs isolated) + golden-baseline drift |
+| 4 | `04_auditor_poam.py` | **Auditors / assessors** | eMASS POA&M workbook + tamper-evident hash-chained audit trail |
+| 5 | `05_airgap_enrichment.py` | **Edge / air-gap** | Resolve official NIST titles + ATT&CK→CTID countermeasures, fully offline |
+
+How a scan flows end to end (full diagram in [`docs/ARCHITECTURE.md`](docs/ARCHITECTURE.md)):
+
+```mermaid
+flowchart LR
+    osq[osquery agent<br/>scheduled pack] -->|JSON| scan[scan / scan_fleet]
+    stig[(STIG_PACK<br/>NIST+STIG+CCI+ATT&CK)] --> scan
+    scan --> findings[Findings + risk score]
+    findings --> exp[6 exporters<br/>console/json/sarif/md/oscal/csv]
+    findings --> corr[fleet correlation<br/>+ POA&M]
+    findings --> enrich[offline feed enrich<br/>OSCAL 800-53 + ATT&CK<->NIST]
+    findings --> audit[(hash-chained<br/>audit log)]
+```
+
+### Fixture library
 
 Each `demos/<NN-name>/` holds osquery snapshot JSON in the tool's real input
 shape (`{query_name: [failing rows…]}`) plus a `SCENARIO.md` that explains where

demos/01_isso_assessment.py

@@ -0,0 +1,55 @@
+"""Scenario 1 - ISSO / ISSM: turn host telemetry into an RMF assessment.
+
+The ISSO's question is never "is this one setting wrong?" — it is "what is my
+control posture, what's the composite risk, and can I hand an AO an OSCAL
+artifact?" This demo scans an un-hardened host fixture with the REAL
+``comint_osquery.core.scan`` and shows the full RMF crosswalk every finding
+already carries (NIST 800-53 control + DISA STIG rule + CCI + ATT&CK technique),
+then renders OSCAL 1.1.2 Assessment Results — the package an ISSO actually
+uploads to eMASS.
+
+Offline: reads the bundled ``demos/01-failing-host`` osquery snapshot fixture.
+"""
+import json
+
+from _common import fixture, rule, section
+from comint_osquery.core import scan
+from cognis_mil import to_console, to_oscal_skeleton
+
+
+def main() -> None:
+    rule("ISSO / ISSM  -  host telemetry -> RMF control posture -> OSCAL")
+
+    target = fixture("01-failing-host")
+    print("\nScanning an un-hardened Ubuntu host snapshot (osquery JSON results).")
+    print(f"  target: demos/01-failing-host/\n")
+
+    result = scan(target)
+
+    print(f"Composite risk: {result.composite_score}/100  ({result.risk_level})")
+    print(f"Findings: {result.total_findings()}  over {result.items_scanned} snapshot file(s)\n")
+
+    section("Each finding carries its full RMF crosswalk")
+    print(f"  {'CAT':<10} {'NIST':<10} {'DISA STIG':<11} {'CCI':<13} {'ATT&CK':<11} Title")
+    for f in result.findings:
+        print(f"  {f.severity.value.upper():<10} {f.nist_800_53:<10} {f.disa_stig:<11} "
+              f"{(f.cci or '-'):<13} {(f.mitre_attack or '-'):<11} {f.title}")
+
+    section("OSCAL 1.1.2 Assessment Results (the eMASS-ingestible artifact)")
+    oscal = json.loads(to_oscal_skeleton(result))
+    ar = oscal["assessment-results"]
+    res0 = ar["results"][0]
+    print(f"  oscal-version : {ar['metadata']['oscal-version']}")
+    print(f"  classification: {ar['metadata']['props'][0]['value']}")
+    print(f"  observations  : {len(res0['observations'])}")
+    print(f"  findings      : {len(res0['findings'])} (each target status = not-satisfied)")
+    f0 = res0["findings"][0]
+    print(f"  e.g. finding  : '{f0['title']}' -> control {f0['target']['target-id']} "
+          f"[{f0['target']['status']['state']}]")
+    print("\n  UUIDs are deterministic (uuid5), so re-scans diff cleanly in the RMF package.")
+
+    print("\nThe ISSO now has a scored posture and an OSCAL package to attach to the SAR.")
+
+
+if __name__ == "__main__":
+    main()

demos/02_soc_detection.py

@@ -0,0 +1,55 @@
+"""Scenario 2 - SOC / endpoint detection engineering: ship the osquery pack.
+
+A SOC doesn't want a one-off scan — it wants the *scheduled* query pack loaded
+into the osquery fleet agent so the weak configs surface continuously, and it
+wants every query mapped to the MITRE ATT&CK technique the weak config enables,
+so the detection lands in the right place on the ATT&CK matrix.
+
+This demo emits the REAL osquery YAML pack with ``core.emit_query_pack`` and
+prints the ATT&CK coverage the ``STIG_PACK`` provides — straight from the
+shipped pack metadata, no fabricated techniques.
+
+Offline: no network, no files written (pack is returned as a string).
+"""
+from collections import defaultdict
+
+from _common import rule, section
+from comint_osquery.core import STIG_PACK, emit_query_pack
+
+
+def main() -> None:
+    rule("SOC / ENDPOINT  -  schedule the pack, map every query to ATT&CK")
+
+    section("osquery scheduled-query pack (load into the fleet agent)")
+    pack = emit_query_pack()
+    # show the header + the first scheduled query so the shape is concrete
+    shown = 0
+    for line in pack.splitlines():
+        print("  " + line)
+        if line.strip().startswith("snapshot:"):
+            shown += 1
+        if shown >= 2:
+            print("  ...")
+            break
+    print(f"\n  {len(STIG_PACK)} STIG-aligned queries, interval=3600s, snapshot mode.")
+    print("  Load with: osqueryi --config_path=stig_pack.yaml")
+
+    section("ATT&CK coverage  (technique <- the weak config that enables it)")
+    by_tech = defaultdict(list)
+    for name, cfg in STIG_PACK.items():
+        by_tech[cfg.get("attack", "")].append((name, cfg))
+    for tech in sorted(by_tech):
+        rows = by_tech[tech]
+        cfg0 = rows[0][1]
+        print(f"  {tech:<12} {cfg0['severity'].value.upper():<10} "
+              f"({len(rows)} query/queries)")
+        for name, cfg in rows:
+            print(f"      - {name}  [{cfg['nist']} / {cfg['stig']}]")
+
+    techniques = sorted({c['attack'] for c in STIG_PACK.values() if c.get('attack')})
+    print(f"\n  {len(techniques)} distinct ATT&CK techniques covered: {', '.join(techniques)}")
+    print("\n  Failing rows from these queries become detections the SOC can pivot on.")
+
+
+if __name__ == "__main__":
+    main()

demos/03_sysadmin_fleet.py

@@ -0,0 +1,59 @@
+"""Scenario 3 - sysadmins / DevSecOps: systemic vs isolated, and baseline drift.
+
+One host failing a control is a remediation ticket. The *same* control failing
+on every host is a broken golden image / GPO / Ansible role — a very different,
+much higher-leverage fix. This demo drives the REAL ``comint_osquery.fleet``
+engine over three edge hosts built from one image: it correlates failures into
+``systemic`` / ``widespread`` / ``isolated`` scope, picks the cleanest host as a
+golden baseline, and reports per-host drift.
+
+Offline: reads the bundled ``demos/12-fleet-systemic`` per-host snapshots.
+"""
+from _common import fixture, rule, section
+from comint_osquery import fleet as fl
+
+
+def main() -> None:
+    rule("SYSADMIN / DEVSECOPS  -  fleet correlation + baseline drift")
+
+    target = fixture("12-fleet-systemic")
+    print("\nScanning 3 edge hosts built from one golden image.")
+    print("  target: demos/12-fleet-systemic/  (host-edge01..03.json)\n")
+
+    hosts = fl.scan_fleet(target)
+    for h in hosts:
+        state = "clean" if h.ok else f"failing: {', '.join(sorted(h.failing))}"
+        print(f"  {h.host:<10} {state}")
+
+    section("Correlation: where is the blast radius?")
+    summ = fl.fleet_summary(hosts)
+    print(f"  hosts scanned : {summ['hosts_scanned']}  "
+          f"(clean={len(summ['hosts_clean'])}, failing={len(summ['hosts_failing'])})")
+    corr = summ["correlation"]
+    if summ["systemic_findings"]:
+        print("\n  SYSTEMIC (every host -> fix the IMAGE, not the host):")
+        for q in summ["systemic_findings"]:
+            d = corr[q]
+            print(f"    [SYS] {d['nist']:<8} {d['title']}  ({d['count']}/{summ['hosts_scanned']} hosts)")
+    if summ["isolated_findings"]:
+        print("\n  ISOLATED (single host -> per-host ticket):")
+        for q in summ["isolated_findings"]:
+            d = corr[q]
+            print(f"    [ISO] {d['nist']:<8} {d['title']}  -> {', '.join(d['hosts'])}")
+
+    section("Baseline drift against the cleanest host")
+    base = fl.pick_baseline(hosts)
+    drift = fl.baseline_drift(hosts, base)
+    print(f"  auto-selected baseline: {drift['baseline']}  "
+          f"(its own failing set: {drift['in_baseline'] or 'none'})")
+    if not drift["drifted_hosts"]:
+        print("  no drift detected.")
+    for host in drift["drifted_hosts"]:
+        regs = drift["drift"][host]["regression_controls"]
+        print(f"    [DRIFT-] {host}: regressions beyond baseline -> {', '.join(regs)}")
+
+    print("\n  The systemic FIPS failure says: re-bake the image. The rest are tickets.")
+
+
+if __name__ == "__main__":
+    main()

demos/04_auditor_poam.py

@@ -0,0 +1,63 @@
+"""Scenario 4 - auditors / assessors: a POA&M workbook + a provable audit trail.
+
+The artifact an auditor lives by is the Plan of Action & Milestones (POA&M): one
+row per (weakness, asset), with a CAT level, the security check (STIG/CCI), and a
+severity-driven scheduled-completion date. This demo builds the REAL eMASS-style
+POA&M with ``comint_osquery.fleet.poam_items`` and shows the columns, then uses
+the shipped hash-chained ``AuditLog`` to make the assessment itself
+tamper-evident — and proves the chain catches an edit.
+
+Offline: reads the bundled ``demos/12-fleet-systemic`` snapshots; the audit log
+is written to a throwaway temp file.
+"""
+import csv
+import io
+import tempfile
+from pathlib import Path
+
+from _common import fixture, rule, section
+from comint_osquery import fleet as fl
+from cognis_mil import AuditLog
+
+
+def main() -> None:
+    rule("AUDITOR / ASSESSOR  -  eMASS POA&M + tamper-evident audit trail")
+
+    target = fixture("12-fleet-systemic")
+    hosts = fl.scan_fleet(target)
+
+    section("POA&M workbook (one row per failing control, per host)")
+    # assessed_at fixed so the scheduled-completion dates are deterministic.
+    items = fl.poam_items(hosts, office="J6 / Cyber", assessed_at=0)
+    print(f"  {len(items)} POA&M item(s). eMASS columns: {len(fl.POAM_COLUMNS)}")
+    print(f"\n  {'Item ID':<13} {'CAT':<7} {'Control':<9} {'Sched. Complete':<16} Check")
+    for it in items:
+        print(f"  {it['POA&M Item ID']:<13} {it['Raw Severity']:<7} "
+              f"{it['Security Control Number (NC/NA)']:<9} "
+              f"{it['Scheduled Completion Date']:<16} {it['Security Checks']}")
+
+    # Render the real CSV and confirm it round-trips (eMASS import sanity).
+    csv_text = fl.poam_to_csv(items)
+    rows = list(csv.DictReader(io.StringIO(csv_text)))
+    print(f"\n  CSV renders {len(rows)} data row(s), RFC 4180 quoted, header = eMASS columns.")
+
+    section("Tamper-evident audit trail (hash-chained, local-only)")
+    log_path = Path(tempfile.mkdtemp(prefix="comint_audit_")) / "audit.log"
+    log = AuditLog(log_path)
+    log.append({"actor": "isso", "action": "scan", "target": "demos/12-fleet-systemic"})
+    for it in items:
+        log.append({"actor": "isso", "action": "poam_item", "id": it["POA&M Item ID"]})
+    ok, msg = log.verify()
+    print(f"  appended {len(items) + 1} entries -> verify(): intact={ok}  ({msg})")
+
+    # Tamper: rewrite one line's body directly, bypassing append().
+    lines = log_path.read_text().splitlines()
+    lines[1] = lines[1].replace("poam_item", "poam_item_HACKED")
+    log_path.write_text("\n".join(lines) + "\n")
+    ok2, msg2 = log.verify()
+    print(f"  after editing one row directly: intact={ok2}  ({msg2})")
+    print("\n  The chain catches the edit -> the assessment record is provable, not asserted.")
+
+
+if __name__ == "__main__":
+    main()

demos/05_airgap_enrichment.py

@@ -0,0 +1,44 @@
+"""Scenario 5 - edge / air-gap operators: enrich findings fully offline.
+
+On disconnected gear you still want a finding's bare control id resolved to its
+*official* NIST 800-53 title, and its single STIG-mapped ATT&CK technique
+expanded into the full CTID-recommended countermeasure control set
+(defense-in-depth for the RMF package). This demo runs the REAL
+``comint_osquery.feeds`` enrichment with ``offline=True`` against the committed
+fixture feed cache — exactly the sneakernet workflow: the cache was carried
+across the air gap, and nothing here touches the network.
+
+Offline: ``_common`` points ``COGNIS_FEEDS_CACHE`` at ``tests/fixtures/feeds-cache``
+(trimmed OSCAL 800-53 rev5 catalog + CTID ATT&CK<->800-53 crosswalk).
+"""
+from _common import fixture, rule, section
+from comint_osquery.core import scan
+from comint_osquery import feeds
+
+
+def main() -> None:
+    rule("EDGE / AIR-GAP  -  offline feed enrichment (official titles + countermeasures)")
+
+    print("\n  Feed cache served offline (no network) — the sneakernet posture.")
+
+    result = scan(fixture("01-failing-host"))
+    print(f"  scanned demos/01-failing-host/ -> {result.total_findings()} finding(s)\n")
+
+    section("Enrich each finding from the authoritative feeds")
+    summary = feeds.enrich_result(result, offline=True)
+    for fid, info in summary.items():
+        title = info["control_title"] or "(unresolved)"
+        cms = info["attack_countermeasures"]
+        print(f"  {fid}")
+        print(f"      NIST {info['nist_800_53']:<8} -> {title}")
+        if info["mitre_attack"]:
+            head = ", ".join(cms[:6]) + (f"  (+{len(cms) - 6} more)" if len(cms) > 6 else "")
+            print(f"      ATT&CK {info['mitre_attack']:<10} -> {len(cms)} countermeasure control(s): {head}")
+
+    resolved = sum(1 for i in summary.values() if i["control_title"])
+    print(f"\n  {resolved}/{len(summary)} control titles resolved from the OSCAL catalog, offline.")
+    print("  Every finding now carries its official NIST title and a defense-in-depth set.")
+
+
+if __name__ == "__main__":
+    main()

demos/_common.py

@@ -0,0 +1,48 @@
+"""Shared helpers for the runnable demo scenarios.
+
+The ``demos/`` directory bundles two kinds of artifact:
+
+  * ``NN-name/`` directories — osquery snapshot JSON fixtures (the tool's real
+    input shape) plus a ``SCENARIO.md``. These are the offline data the demos
+    feed into the real API.
+  * ``NN_name.py`` scripts (this family) — runnable, narrated scenarios that
+    drive the **real** ``comint_osquery`` / ``cognis_mil`` API over those
+    fixtures. No network, no fabricated functions, exit 0.
+
+Every scenario is self-contained: it rebuilds nothing on disk it can't throw
+away, and points the data-feed engine at the committed fixture cache so feed
+enrichment works fully offline.
+"""
+from __future__ import annotations
+
+import os
+import sys
+
+# allow `python demos/NN_name.py` from anywhere
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+DEMOS_DIR = os.path.join(REPO_ROOT, "demos")
+
+# The committed, trimmed offline feed cache (OSCAL 800-53 rev5 catalog + the
+# CTID ATT&CK<->800-53 crosswalk). Pointing COGNIS_FEEDS_CACHE here makes every
+# `offline=True` feed read serve from disk — never the network.
+FIXTURE_CACHE = os.path.join(REPO_ROOT, "tests", "fixtures", "feeds-cache")
+os.environ.setdefault("COGNIS_FEEDS_CACHE", FIXTURE_CACHE)
+
+
+def fixture(name: str) -> str:
+    """Absolute path to a bundled demo fixture directory, e.g. ``01-failing-host``."""
+    return os.path.join(DEMOS_DIR, name)
+
+
+def rule(title: str) -> None:
+    print("\n" + "=" * 72)
+    print(f"  {title}")
+    print("=" * 72)
+
+
+def section(title: str) -> None:
+    print("\n" + "-" * 72)
+    print(f"  {title}")
+    print("-" * 72)