feat: extract the assessment into a shared, headless surface (4.1.0)

README.md

@@ -394,17 +394,28 @@ to suppress it everywhere.
 
 **Package layout** (`caia/`)
 
-| Module | Responsibility |
-| ------ | -------------- |
-| `cli.py` | single entry point; refuses removed options, then dispatches `--help`/`--version`/`--update`/`--uninstall`/`--config` and the run |
-| `analyzer.py` | run flow + interactive prompts (working-folder inventory + `output/`) |
-| `inventory.py` | read the inventory workbook into per-firewall records |
-| `psirt.py` | the only module that talks to Cisco's advisory API |
-| `matching.py` | pure verdict logic: release normalisation, model→platform mapping, verdicts |
-| `html_report.py` | render the self-contained HTML report |
-| `config.py` | per-user configuration (paths, precedence, `--config`) |
-| `version.py` | installed version, GitHub release discovery, uv-based update |
-| `ui.py` | terminal colors + prompts (no dependencies) |
+| Module | Responsibility | Shared surface |
+| ------ | -------------- | -------------- |
+| `assessment.py` | the assessment itself — `analyze()`, with no user interface attached | ✅ |
+| `matching.py` | pure verdict logic: release normalisation, model→platform mapping, verdicts | ✅ |
+| `psirt.py` | the only module that talks to Cisco's advisory API | ✅ |
+| `inventory.py` | read the inventory workbook into per-firewall records | ✅ |
+| `cli.py` | single entry point; refuses removed options, then dispatches `--help`/`--version`/`--update`/`--uninstall`/`--config` and the run | |
+| `analyzer.py` | terminal run flow + interactive prompts (working-folder inventory + `output/`) | |
+| `html_report.py` | render the self-contained HTML report | |
+| `config.py` | per-user configuration (paths, precedence, `--config`) | |
+| `version.py` | installed version, GitHub release discovery, uv-based update | |
+| `ui.py` | terminal colors + prompts (no dependencies) | |
+
+**Shared surface.** The four modules marked above are the ones a consumer other than this
+CLI is expected to import; the rest are the terminal program built on top of them. The
+entry point is `assessment.analyze(inventory, client)` — it takes a list of per-device
+dicts (whatever produced them) plus a `psirt.Client` the caller owns, prints nothing,
+writes nothing, and never exits the process. `analyzer.run()` is one caller of it; the
+terminal output you see is supplied by the `on_plan` and `on_progress` callbacks it passes
+in. The split exists so that a second consumer cannot drift from this tool's verdicts:
+the rules that read Cisco's data live in one place, and anything that reimplements them
+reimplements the safety property in Constitution IV.
 
 The version is single-sourced from the committed `VERSION` file into the package metadata and
 reported at runtime via `importlib.metadata`. Devices sharing a product family, hardware

VERSION

@@ -1 +1 @@
-4.0.0
+4.1.0

caia/analyzer.py

@@ -7,9 +7,13 @@
 PSIRT advisory API — the same data behind Cisco's public Software Checker — and the
 answer is reported directly. Writes one timestamped, self-contained HTML report.
 
-This module holds the run flow; the command-line entry point is `caia.cli`. Inputs are
-read from, and the report is written to, the current working folder (an `output/`
-subfolder), so the tool works from any folder once installed.
+This module holds the *terminal* run flow: credential resolution, inventory discovery and
+its prompts, progress output, and writing the report into the current working folder (an
+`output/` subfolder), so the tool works from any folder once installed. The command-line
+entry point is `caia.cli`.
+
+The assessment itself lives in `caia.assessment` and is shared with any other consumer of
+this package, so a verdict cannot differ between the terminal and anything else.
 """
 
 from __future__ import annotations
@@ -18,7 +22,7 @@
 import sys
 from pathlib import Path
 
-from caia import config, html_report, matching, psirt
+from caia import assessment, config, html_report, matching, psirt
 from caia import inventory as inv
 from caia import ui
 
@@ -138,36 +142,6 @@ def _resolve_cisco_credentials():
     return client_id, client_secret
 
 
-def _families_in(inventory):
-    """Distinct lower-cased product families present in the inventory."""
-    seen = []
-    for device in inventory:
-        family = str(device.get("family") or "").strip().lower()
-        if family and family not in seen:
-            seen.append(family)
-    return seen
-
-
-def _load_catalogues(client, families):
-    """Fetch the platform and release catalogues Cisco publishes, per family.
-
-    Only families Cisco can answer about are fetched; anything else is resolved locally to
-    Needs review, so an inventory containing switches does not fail the run.
-    """
-    catalogues = matching.Catalogues()
-    for family in families:
-        if not matching.family_supported(family):
-            continue
-        catalogues.add_platforms(family, client.get_platforms(family))
-        catalogues.add_releases(family, client.get_os_data(family))
-    return catalogues
-
-
-def _describe_check(check):
-    os_type, alias, release = check
-    return f"{os_type}/{alias or '-'}/{release}"
-
-
 def run(args):
     """Assess the whole inventory against Cisco's published advisory data.
 
@@ -196,25 +170,29 @@ def run(args):
     client = psirt.Client(client_id, client_secret)
     ui.plain()
     ui.system("Fetching Cisco platform and release catalogues ...")
+
+    def report_plan(check_count, unresolved_count):
+        ui.ok(f"{len(inventory)} firewalls collapse to {ui.bold(str(check_count))} distinct "
+              f"checks.")
+        if unresolved_count:
+            ui.info(f"{unresolved_count} firewall(s) need review before Cisco can be asked.")
+
+    def report_progress(index, total, check):
+        ui.plain(ui.dim(f"  [{index}/{total}] {assessment.describe_check(check)}"))
+
+    # Only the catalogue fetch can raise here; a failure once they are in hand comes back
+    # as `partial_cause` so the answers already obtained survive (assessment.analyze).
     try:
-        catalogues = _load_catalogues(client, _families_in(inventory))
+        devices, partial_cause = assessment.analyze(
+            inventory, client, on_plan=report_plan, on_progress=report_progress)
     except psirt.PsirtAuthError as e:
         die(f"Cisco rejected the configured credentials: {e}",
             hint="Re-check them with 'caia --config'.")
     except psirt.PsirtUnreachableError as e:
         die(f"Could not reach Cisco: {e}",
             hint="Check your connection or proxy and try again.")
 
-    checks, unresolved = matching.build_checks(inventory, catalogues)
-    ui.ok(f"{len(inventory)} firewalls collapse to {ui.bold(str(len(checks)))} distinct "
-          f"checks.")
-    if unresolved:
-        ui.info(f"{len(unresolved)} firewall(s) need review before Cisco can be asked.")
-
-    results, partial_cause = _run_checks(client, checks)
-
     # 4. Verdicts.
-    devices = _assemble_devices(inventory, catalogues, checks, unresolved, results)
     summary = _build_summary(inv_path, devices, partial_cause)
     _print_verdict_tally(summary)
 
@@ -230,57 +208,6 @@ def run(args):
     return 0
 
 
-def _run_checks(client, checks):
-    """Ask Cisco once per check. Returns ({check: (outcome, advisories)}, partial_cause).
-
-    A failure part-way through stops further requests but keeps every answer already
-    obtained: the remaining devices become Needs review and the report is flagged partial
-    rather than the run being thrown away (FR-016).
-    """
-    results = {}
-    total = len(checks)
-    for index, check in enumerate(checks, 1):
-        ui.plain(ui.dim(f"  [{index}/{total}] {_describe_check(check)}"))
-        os_type, alias, release = check
-        try:
-            outcome, advisories = client.get_advisories(os_type, release, alias)
-        except psirt.PsirtAuthError as e:
-            return results, f"Cisco rejected the access token part-way through: {e}"
-        except psirt.PsirtUnreachableError as e:
-            return results, f"Cisco became unreachable part-way through: {e}"
-        results[check] = (outcome, advisories)
-    return results, None
-
-
-def _assemble_devices(inventory, catalogues, checks, unresolved, results):
-    """Attach a verdict, reason and advisories to every inventory row."""
-    reasons = {id(device): reason for device, reason in unresolved}
-    by_device = {}
-    for check, members in checks.items():
-        outcome, advisories = results.get(check, (None, None))
-        for device in members:
-            by_device[id(device)] = (check, outcome, advisories)
-
-    devices = []
-    for device in inventory:
-        entry = dict(device)
-        reason = reasons.get(id(device))
-        if reason is not None:
-            entry["verdict"] = matching.NEEDS_REVIEW
-            entry["reason"] = reason
-            entry["advisories"] = []
-        else:
-            check, outcome, advisories = by_device.get(id(device), (None, None, None))
-            verdict, reason = matching.verdict_from_outcome(
-                check, catalogues, outcome, advisories,
-                raw_release=device.get("version"))
-            entry["verdict"] = verdict
-            entry["reason"] = reason
-            entry["advisories"] = list(advisories or []) if verdict == matching.IMPACTED else []
-        devices.append(entry)
-    return devices
-
-
 def _build_summary(inv_path, devices, partial_cause):
     counts = {matching.IMPACTED: 0, matching.NOT_IMPACTED: 0, matching.NEEDS_REVIEW: 0}
     for device in devices:

caia/assessment.py

@@ -0,0 +1,172 @@
+"""The assessment itself, with no user interface attached.
+
+`analyze()` is the whole check: fetch Cisco's catalogues, collapse the inventory into the
+distinct questions Cisco needs to be asked, ask them, and attach a verdict to every device.
+It prints nothing, writes no file, never exits the process, and never consults the working
+folder — so the same code path serves the terminal and any other consumer.
+
+This module and `matching`, `psirt` and `inventory` are the package's **shared surface**:
+the modules a consumer other than the CLI is expected to import. `analyzer`, `cli`, `ui`,
+`config`, `version` and `html_report` are the terminal program built on top of them, and
+are not part of that surface.
+
+That split exists so a second consumer cannot drift from the CLI's verdicts. Cisco's
+published data is the authority (Constitution IV) and the rules that read it live here
+exactly once; anything that reimplements them is reimplementing the safety property.
+"""
+
+from __future__ import annotations
+
+from caia import matching, psirt
+
+
+# --------------------------------------------------------------------------- #
+# Inspecting an inventory
+# --------------------------------------------------------------------------- #
+def families_in(inventory):
+    """Distinct lower-cased product families present in the inventory.
+
+    Order-preserving, so a caller fetching catalogues per family does so predictably.
+    """
+    seen = []
+    for device in inventory:
+        family = str(device.get("family") or "").strip().lower()
+        if family and family not in seen:
+            seen.append(family)
+    return seen
+
+
+def describe_check(check):
+    """A short label for a check, e.g. `ftd/ISA3000/7.4.2.3`. For progress reporting."""
+    os_type, alias, release = check
+    return f"{os_type}/{alias or '-'}/{release}"
+
+
+# --------------------------------------------------------------------------- #
+# Catalogues
+# --------------------------------------------------------------------------- #
+def load_catalogues(client, families):
+    """Fetch the platform and release catalogues Cisco publishes, per family.
+
+    Only families Cisco can answer about are fetched; anything else is resolved locally to
+    Needs review, so an inventory containing switches does not fail the run.
+
+    These catalogues are identical for every caller and change rarely, so a long-running
+    consumer may fetch them once and pass the result to `analyze(catalogues=...)` instead
+    of paying for them on every assessment.
+    """
+    catalogues = matching.Catalogues()
+    for family in families:
+        if not matching.family_supported(family):
+            continue
+        catalogues.add_platforms(family, client.get_platforms(family))
+        catalogues.add_releases(family, client.get_os_data(family))
+    return catalogues
+
+
+# --------------------------------------------------------------------------- #
+# The assessment
+# --------------------------------------------------------------------------- #
+def analyze(inventory, client, catalogues=None, on_plan=None, on_progress=None):
+    """Assess an inventory against Cisco's published advisory data.
+
+    Args:
+      inventory: list of per-device dicts, as produced by `inventory.load_inventory` —
+        each carrying at least `name`, `model`, `family` and `version`. Any other keys are
+        carried through onto the corresponding result entry untouched, so a caller's own
+        columns survive into a report.
+      client: a `psirt.Client`, or anything exposing its `get_platforms`, `get_os_data`
+        and `get_advisories` methods. The caller owns its credentials and its pacing.
+      catalogues: an optional pre-built `matching.Catalogues`. When omitted, catalogues are
+        fetched from Cisco for the families present in `inventory`.
+      on_plan: optional callable(check_count, unresolved_count), called once after
+        deduplication and before the first advisory request. `check_count` is exactly how
+        many requests the run will make, so a caller can estimate how long it will take.
+      on_progress: optional callable(index, total, check), called before each advisory
+        request. `index` is 1-based; pass `check` to `describe_check` for a label.
+
+    Returns:
+      (devices, partial_cause).
+
+      `devices` holds one dict per inventory row, in inventory order — a copy of the input
+      row plus `verdict`, `reason`, and `advisories` (populated only for an Impacted
+      verdict). Every row appears exactly once; none is ever dropped.
+
+      `partial_cause` is None for a complete run, or a sentence explaining why it stopped
+      early. A run that stops early keeps every answer already obtained and leaves the
+      devices it never reached as Needs review (FR-016) — it is never discarded.
+
+    Raises:
+      psirt.PsirtAuthError, psirt.PsirtUnreachableError: when the catalogues cannot be
+        fetched. Without them nothing can be confirmed, and an inventory of Needs review
+        would read as a data-quality problem rather than the access failure it is — so
+        this fails loudly instead. A failure *after* the catalogues are in hand is
+        reported through `partial_cause`, never raised.
+    """
+    if catalogues is None:
+        catalogues = load_catalogues(client, families_in(inventory))
+
+    checks, unresolved = matching.build_checks(inventory, catalogues)
+    if on_plan is not None:
+        on_plan(len(checks), len(unresolved))
+
+    results, partial_cause = _run_checks(client, checks, on_progress)
+    devices = _assemble_devices(inventory, catalogues, checks, unresolved, results)
+    return devices, partial_cause
+
+
+def _run_checks(client, checks, on_progress=None):
+    """Ask Cisco once per check. Returns ({check: (outcome, advisories)}, partial_cause).
+
+    A failure part-way through stops further requests but keeps every answer already
+    obtained: the remaining devices become Needs review and the run is reported partial
+    rather than being thrown away (FR-016).
+    """
+    results = {}
+    total = len(checks)
+    for index, check in enumerate(checks, 1):
+        if on_progress is not None:
+            on_progress(index, total, check)
+        os_type, alias, release = check
+        try:
+            outcome, advisories = client.get_advisories(os_type, release, alias)
+        except psirt.PsirtAuthError as e:
+            return results, f"Cisco rejected the access token part-way through: {e}"
+        except psirt.PsirtUnreachableError as e:
+            return results, f"Cisco became unreachable part-way through: {e}"
+        results[check] = (outcome, advisories)
+    return results, None
+
+
+def _assemble_devices(inventory, catalogues, checks, unresolved, results):
+    """Attach a verdict, reason and advisories to every inventory row.
+
+    Devices are tracked by object identity, which is why `analyze` owns the whole span
+    from `build_checks` to here: the same dicts must flow through both. Callers only ever
+    hand in a list and receive a list back, so they cannot break that.
+    """
+    reasons = {id(device): reason for device, reason in unresolved}
+    by_device = {}
+    for check, members in checks.items():
+        outcome, advisories = results.get(check, (None, None))
+        for device in members:
+            by_device[id(device)] = (check, outcome, advisories)
+
+    devices = []
+    for device in inventory:
+        entry = dict(device)
+        reason = reasons.get(id(device))
+        if reason is not None:
+            entry["verdict"] = matching.NEEDS_REVIEW
+            entry["reason"] = reason
+            entry["advisories"] = []
+        else:
+            check, outcome, advisories = by_device.get(id(device), (None, None, None))
+            verdict, reason = matching.verdict_from_outcome(
+                check, catalogues, outcome, advisories,
+                raw_release=device.get("version"))
+            entry["verdict"] = verdict
+            entry["reason"] = reason
+            entry["advisories"] = list(advisories or []) if verdict == matching.IMPACTED else []
+        devices.append(entry)
+    return devices

caia/inventory.py

@@ -9,12 +9,13 @@
 
 from __future__ import annotations
 
-import sys
-
 try:
     import openpyxl
-except ImportError:  # pragma: no cover
-    sys.exit("openpyxl is required. Reinstall the tool with uv.")
+except ImportError as exc:  # pragma: no cover - only a broken install reaches this
+    # Raised, not exited: this module is imported by consumers other than the command-line
+    # program, and a library must never terminate its host process on import.
+    raise ImportError("openpyxl is required to read an inventory workbook. "
+                      "Reinstall the tool with uv.") from exc
 
 
 class InventoryError(Exception):