feat(ics-protocol): finding-source cross-stack alignment (Phase 2 / Rule digitalandrew#25 Shape-1)

Rule #52 instance digitalandrew#3 / Phase 2: extends the finding-source cross-stack
alignment surface to cover the 5 IcsProtocolFamily Literal values per
W2-β §SC5-NEW-ICS-S2-η mitigation — Rule digitalandrew#25 single-slice exception digitalandrew#2
atomic commit because test_finding_source_alignment.py enforces pairwise
agreement across DB CHECK + Pydantic Literal + frontend Union + frontend
FINDING_SOURCE_CONFIG, and splitting leaves the alignment test RED
between commits.

5 NEW source values (one per IcsProtocolFamily Literal entry):
- ics_modbus_tcp_detected — exercised by Session 1 modbus_tcp.yaml
- ics_modbus_rtu_detected — forward-prepared for serial-protocol YAMLs
- ics_dnp3_detected — Phase 5 dnp3.yaml will exercise
- ics_s7comm_detected — Phase 5 s7comm.yaml will exercise
- ics_unknown_ics_detected — Phase 4 plugin escape-hatch envelope

Cross-stack surfaces (all in this commit per Rule #48 5-part shape):
- DB CHECK extension: alembic 2cba3d4e5f6a (chains from 1c52a4b5c6d7)
- Pydantic Literal: app/schemas/finding.py::IcsProtocolFindingSource
- Frontend union: frontend/src/types/index.ts::FindingSource
- Frontend config: frontend/src/constants/statusConfig.ts entries
  (5 new entries; cyan/sky/indigo gradient per Scout D operator-UX)

Walker emit (app/services/ics_protocol_walker.py):
- New _ALLOWED_ICS_FINDING_SOURCES frozenset derived from
  IcsProtocolFindingSource via typing.get_args (single source of truth)
- INNER runner emits one Finding row per (firmware, protocol_family)
  tuple at severity=info — closed-allowlist guarded so unmapped
  families skip emit instead of raising 422 at FindingService.create
- Result aggregate adds `findings_emitted_count` field;
  _empty_result_aggregate updated for shape consistency

Rule #46 paired META-CANARIES (test_ics_protocol_walker.py):
- test_walker_ics_finding_source_allowlist_matches_pydantic_literal:
  walker allowlist EXACTLY equals Pydantic Literal (no drift either way)
- test_walker_ics_finding_source_naming_convention: every Literal entry
  follows ics_<family>_detected — protects per-family lookup from
  silent rename drift

Verified:
- alembic upgrade 1c52a4b5c6d7 -> 2cba3d4e5f6a applied; psql confirmed
  5 new ics_* sources in ck_findings_source CHECK
- pytest tests/test_ics_protocol_walker.py: 11 passed (was 9)
- pytest tests/test_finding_source_alignment.py on host: 3/3 passed
  (DB CHECK + frontend Union + frontend Config pairwise alignment)
- Rule digitalandrew#24 canary + npx tsc -b --force frontend typecheck: exit 0
- broader container pytest sweep (jsonb_normalizers + walker +
  reapers + auth-gate + sbom_status_alignment): 808 passed

Phase 3 (next): MCP tool category (backend/app/ai/tools/ics_protocol.py)
with 4 tools — trigger_ics_protocol_walk + list_ics_protocols +
lookup_ics_protocol_across_firmwares (Rule #44 mandatory) +
describe_ics_protocol_anomalies. Includes W2-β §SC5-NEW-ICS-S2-3
+ §SC5-NEW-ICS-S2-γ + §SC5-NEW-ICS-S2-ε + §SC5-NEW-ICS-S2-ι mitigations
(status filter + supply_chain_signal curated-only + project_id scope +
schema_version legacy-rows filter) per Wave-1 C + W2-β cross-feature.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: eastmadc

backend/alembic/versions/2cba3d4e5f6a_extend_findings_source_ics_protocol.py

@@ -0,0 +1,169 @@
+r"""extend findings.source CHECK with ICS protocol catalog sources (Rule #52 instance #3 Phase 2)
+
+Revision ID: 2cba3d4e5f6a
+Revises: 1c52a4b5c6d7
+Create Date: 2026-05-22 10:00:00.000000
+
+Rule #25 Shape-1 cross-stack alignment commit. Mirrors the bare_metal
+``fd6e7f8a9b0c`` precedent — extends ``ck_findings_source`` with 5 NEW
+finding sources matching the closed ``IcsProtocolFamily`` Literal in
+``app/schemas/ics_protocol.py``. Per W2-β §SC5-NEW-ICS-S2-η mitigation,
+ALL family values get a corresponding finding source — adding a new
+protocol YAML (Phase 5: DNP3, S7Comm; future: opc_ua, ethercat) without
+a matching source value would silently reject finding emits at runtime.
+
+Cross-stack contract surfaces (Rule #48 5-part):
+  - DB CHECK: this migration extends ``ck_findings_source``
+  - Pydantic Literal: ``IcsProtocolFindingSource`` in app/schemas/finding.py
+  - Frontend mirror: ``FindingSource`` union in types/index.ts +
+    ``FINDING_SOURCE_CONFIG`` keys in statusConfig.ts
+  - Alignment test: ``test_finding_source_alignment`` auto-discovers
+    this migration via ``_latest_source_check_migration_path()`` walking
+    the alembic chain from HEAD; pulls ``_NEW_SOURCE_VALUES`` tuple as
+    the authoritative DB-side allowlist
+
+NEW SOURCES:
+
+- ``ics_modbus_tcp_detected`` — emitted when the walker matches at
+  least one binary against a Modbus/TCP manifest (Session 1's
+  ``_system/modbus_tcp.yaml`` is the production reference; Phase 5
+  will not extend with sibling Modbus/RTU YAMLs).
+- ``ics_modbus_rtu_detected`` — Modbus RTU over serial (RS-485);
+  forward-prepared for future serial-protocol YAML extensions.
+- ``ics_dnp3_detected`` — emitted when the walker matches at least one
+  binary against a DNP3 manifest (Phase 5 ``_system/dnp3.yaml`` will
+  exercise this source).
+- ``ics_s7comm_detected`` — emitted when the walker matches at least
+  one binary against a Siemens S7Comm/S7Comm-Plus manifest (Phase 5
+  ``_system/s7comm.yaml`` will exercise this source).
+- ``ics_unknown_ics_detected`` — emitted when a plugin (Phase 4) flags
+  an ICS-like stack the closed-grammar catalog cannot identify;
+  forward-prepared for the W2-β §SC5-NEW-ICS-7 hot-reload mitigation
+  envelope (plugin escape hatch).
+
+Live audit at migration-authoring time (2026-05-22):
+
+    SELECT source, COUNT(*) FROM findings GROUP BY source ORDER BY 2 DESC;
+    -- 0 rows for any ics_* source (Phase 2 first consumer).
+
+Zero existing rows ⇒ "extend CHECK" path; no defensive backfill needed.
+
+Mirrors ``fd6e7f8a9b0c_extend_findings_source_bare_metal.py``. Same
+drop-and-recreate shape with extended ``_NEW_SOURCE_VALUES`` tuple.
+Tuple is exposed under that exact name so
+``test_finding_source_alignment._load_db_source_values`` can import +
+assert agreement.
+
+Chains from ICS Phase 1.A DDL ``1c52a4b5c6d7``.
+"""
+from __future__ import annotations
+
+from alembic import op
+
+
+revision: str = "2cba3d4e5f6a"
+down_revision: str | None = "1c52a4b5c6d7"
+branch_labels: str | None = None
+depends_on: str | None = None
+
+
+_NEW_SOURCE_VALUES: tuple[str, ...] = (
+    # ── canonical core ─────────────────────────────────────────────────
+    "manual", "security_audit", "yara_scan", "attack_surface", "sbom_scan",
+    "hardware_firmware_graph", "apk-manifest-scan", "apk-bytecode-scan",
+    "apk-mobsfscan", "cwe_checker", "uefi_scan", "clamav_scan", "vt_scan",
+    "abusech_scan", "fuzzing", "unpack_audit", "security_review", "ai_discovered",
+    # ── windows family ────────────────────────────────────────────────
+    "windows_authenticode", "windows_dbx_revoked",
+    "windows_registry_persistence", "windows_inf", "windows_driver_imports",
+    "windows_r2r_stomp", "windows_il_capa",
+    "windows_sysmon_proc_create", "windows_logon_success", "windows_logon_failure",
+    "windows_amcache_install", "windows_prefetch_execution",
+    "windows_srum_network_activity", "windows_srum_application_runtime",
+    "windows_powershell_script_block", "windows_scheduled_task_persistence",
+    "windows_lnk_abnormal_target",
+    "windows_mft_ads_hidden_content", "windows_mft_timestomping",
+    "windows_byovd_driver",
+    "windows_bcd_suspicious_path", "windows_bcd_testsigning_enabled",
+    "windows_wmi_persistence",
+    "windows_esp_unsigned", "windows_esp_dbx_revoked",
+    "windows_mbr_bootkit", "windows_vbr_anomaly",
+    "windows_sdb_inject_dll", "windows_sdb_redirect_exe", "windows_sdb_custom_shim",
+    "windows_etl_kernel_proc_after_clear", "windows_etl_provider_disabled",
+    "windows_etl_unusual_provider", "windows_etl_non_microsoft_in_diagtrack",
+    "windows_efs_orphaned_drf", "windows_efs_unusual_recovery_agent",
+    "windows_efs_domain_admin_in_ddf", "windows_efs_large_drf",
+    "windows_appcompat_suspicious_path", "windows_appcompat_temp_execution",
+    "windows_appcompat_recent_baseline",
+    "windows_dpapi_orphaned_masterkey", "windows_dpapi_admin_creator_sid",
+    "windows_dpapi_large_masterkey",
+    "windows_usnjrnl_file_deletion", "windows_usnjrnl_temp_create_delete_pair",
+    "windows_usnjrnl_renamed_executable",
+    # ── linux family ──────────────────────────────────────────────────
+    "linux_journald_priority_critical", "linux_journald_oom_killer",
+    "linux_journald_suspicious_unit", "linux_journald_log_clear",
+    "linux_journald_selinux_denied",
+    "linux_systemd_suspicious_path", "linux_systemd_obfuscated_exec",
+    "linux_systemd_socket_unusual_port", "linux_systemd_root_minimal_deps",
+    "linux_systemd_enabled_outside_standard",
+    "linux_container_privileged_mode", "linux_container_dangerous_capability",
+    "linux_container_unsafe_host_mount", "linux_container_unconfined_security",
+    "linux_container_unknown_registry_image",
+    "linux_bash_history_clear", "linux_cron_suspicious_command",
+    "linux_ld_preload_hijack",
+    # ── Rule #52 instance #1 — bare-metal MCU/DSP family ─────────────
+    "c28x_unsecure_csm",
+    "c28x_csm_perma_lock",
+    "bare_metal_chip_unknown_with_hints",
+    "bare_metal_encrypted_region_skipped",
+    # ── Rule #52 instance #3 — ICS protocol catalog family (THIS) ────
+    # All 5 IcsProtocolFamily Literal values get an ics_*_detected
+    # source per W2-β §SC5-NEW-ICS-S2-η mitigation — forward-prepares
+    # future protocol YAMLs (opc_ua, ethercat) without requiring a
+    # second cross-stack alignment commit each time.
+    "ics_modbus_tcp_detected",
+    "ics_modbus_rtu_detected",
+    "ics_dnp3_detected",
+    "ics_s7comm_detected",
+    "ics_unknown_ics_detected",
+)
+
+
+_REMOVED_IN_THIS_REVISION: tuple[str, ...] = (
+    "ics_modbus_tcp_detected",
+    "ics_modbus_rtu_detected",
+    "ics_dnp3_detected",
+    "ics_s7comm_detected",
+    "ics_unknown_ics_detected",
+)
+
+
+def _in_list_sql(column: str, values: tuple[str, ...]) -> str:
+    quoted = ", ".join(f"'{v}'" for v in values)
+    return f"{column} IN ({quoted})"
+
+
+def upgrade() -> None:
+    op.drop_constraint("ck_findings_source", "findings", type_="check")
+    op.create_check_constraint(
+        "ck_findings_source",
+        "findings",
+        _in_list_sql("source", _NEW_SOURCE_VALUES),
+    )
+
+
+def downgrade() -> None:
+    op.execute(
+        "UPDATE findings SET source = 'manual' "
+        "WHERE source IN ("
+        + ", ".join(f"'{v}'" for v in _REMOVED_IN_THIS_REVISION)
+        + ")"
+    )
+
+    op.drop_constraint("ck_findings_source", "findings", type_="check")
+    prior = tuple(v for v in _NEW_SOURCE_VALUES if v not in _REMOVED_IN_THIS_REVISION)
+    op.create_check_constraint(
+        "ck_findings_source",
+        "findings",
+        _in_list_sql("source", prior),
+    )

backend/app/schemas/finding.py

@@ -575,6 +575,36 @@
 ]
 
 
+# CLAUDE.md Rule #52 instance #3 — ICS protocol catalog walker emit sources.
+# Sibling of BareMetalFindingSource. Walker
+# (``backend/app/services/ics_protocol_walker.py``) emits one finding row per
+# ``(firmware, protocol_family)`` tuple at severity=info when the closed-
+# grammar resolver matches at least one binary against a manifest declaring
+# the family. The 5 family values mirror the IcsProtocolFamily Literal in
+# ``app/schemas/ics_protocol.py`` per W2-β §SC5-NEW-ICS-S2-η mitigation —
+# every protocol_family the catalog can output MUST have a corresponding
+# finding source declared here (catalog loader validates at YAML-load time
+# in a future Phase 4 commit; missing source rejects the manifest with
+# WARN, not silent skip).
+#
+# - ``ics_modbus_tcp_detected`` — Session 1's ``_system/modbus_tcp.yaml``
+#   exercises this source.
+# - ``ics_modbus_rtu_detected`` — forward-prepared for future Modbus-RTU
+#   serial YAML extension.
+# - ``ics_dnp3_detected`` — Phase 5 ``_system/dnp3.yaml`` exercises this.
+# - ``ics_s7comm_detected`` — Phase 5 ``_system/s7comm.yaml`` exercises this.
+# - ``ics_unknown_ics_detected`` — Phase 4 plugin escape hatch (W2-β
+#   §SC5-NEW-ICS-7 hot-reload mitigation envelope); reserved for plugins
+#   that detect an ICS stack but cannot identify the family.
+IcsProtocolFindingSource = Literal[
+    "ics_modbus_tcp_detected",
+    "ics_modbus_rtu_detected",
+    "ics_dnp3_detected",
+    "ics_s7comm_detected",
+    "ics_unknown_ics_detected",
+]
+
+
 class Severity(str, Enum):
     critical = "critical"
     high = "high"

backend/app/services/ics_protocol_walker.py

@@ -62,11 +62,18 @@
 import traceback
 import uuid
 from collections import Counter
+from typing import get_args
 
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.database import async_session_factory
 from app.models.firmware import Firmware
+from app.schemas.finding import (
+    FindingCreate,
+    IcsProtocolFindingSource,
+    Severity,
+)
+from app.services.finding_service import FindingService
 from app.services.firmware_paths import get_detection_roots
 from app.services.ics_protocol_catalog import (
     IcsResolverContext,
@@ -77,6 +84,19 @@
     _stamp_firmware_ics_protocol_walk_result,
 )
 
+
+# Closed allowlist of ICS protocol-family finding sources (Rule #25 Shape-1
+# alignment surface — DB CHECK ↔ Pydantic Literal ↔ frontend mirror). The
+# walker emits a Finding per (firmware, protocol_family) tuple where the
+# source `ics_{protocol_family}_detected` is a declared member of
+# IcsProtocolFindingSource. Per W2-β §SC5-NEW-ICS-S2-η mitigation: if a
+# future protocol_family value lands in IcsProtocolFamily Literal without
+# a corresponding finding source declared here + in the DB CHECK, the
+# walker SKIPS emit (no silent FastAPI 422 on FindingService.create).
+_ALLOWED_ICS_FINDING_SOURCES: frozenset[str] = frozenset(
+    get_args(IcsProtocolFindingSource)
+)
+
 logger = logging.getLogger(__name__)
 
 
@@ -176,6 +196,7 @@ def _empty_result_aggregate(
         "protocol_family_counts": {},
         "manifest_ids_seen": [],
         "manifest_sources_seen": [],
+        "findings_emitted_count": 0,
         "errors": errors,
     }
 
@@ -294,6 +315,47 @@ async def _do_ics_protocol_walk(
                 "matches": match_dicts,
             })
 
+    # Emit findings — one per (firmware, protocol_family). Severity is
+    # always INFO (an ICS protocol detected on a PLC is informational
+    # context, not a vulnerability). Gated by closed allowlist to
+    # respect the Rule #25 Shape-1 cross-stack alignment contract per
+    # W2-β §SC5-NEW-ICS-S2-η — protocol families without a declared
+    # finding source skip emit silently.
+    findings_emitted = 0
+    finding_emit_errors: list[str] = []
+    if family_counts:
+        findings_service = FindingService(db)
+        for protocol_family, count in family_counts.items():
+            source = f"ics_{protocol_family}_detected"
+            if source not in _ALLOWED_ICS_FINDING_SOURCES:
+                finding_emit_errors.append(
+                    f"protocol_family={protocol_family!r} has no declared "
+                    f"finding source — manifest output declares a family "
+                    f"the IcsProtocolFindingSource Literal does not list"
+                )
+                continue
+            try:
+                await findings_service.create(
+                    project_id=firmware.project_id,
+                    data=FindingCreate(
+                        title=f"ICS protocol detected: {protocol_family}",
+                        severity=Severity.info,
+                        firmware_id=firmware_id,
+                        source=source,
+                        description=(
+                            f"Walker detected {count} binary match(es) for "
+                            f"ICS protocol family {protocol_family}. See "
+                            f"ics_protocol_walk_result JSONB for per-binary "
+                            f"match detail + manifest_ids_seen."
+                        ),
+                    ),
+                )
+                findings_emitted += 1
+            except Exception as exc:  # noqa: BLE001 — emit-boundary
+                finding_emit_errors.append(
+                    f"finding emit failed for {protocol_family}: {exc}"
+                )
+
     # W2-β §SC5-NEW-ICS-S2-β: read snapshot at EXIT; flag drift.
     exit_snapshot = catalog.get_snapshot()
     snapshot_id_at_exit = exit_snapshot.snapshot_id
@@ -316,7 +378,8 @@ async def _do_ics_protocol_walk(
         "protocol_family_counts": dict(family_counts),
         "manifest_ids_seen": manifest_ids,
         "manifest_sources_seen": manifest_sources,
-        "errors": errors,
+        "findings_emitted_count": findings_emitted,
+        "errors": errors + finding_emit_errors,
     }
 
 

backend/tests/test_ics_protocol_walker.py

@@ -465,3 +465,58 @@ def test_ics_protocol_walk_status_literal_matches_db_check_values():
         f"the DB CHECK ck_firmware_ics_protocol_walk_status value set "
         f"{db_check_values!r}. Rule #33 .c contract."
     )
+
+
+# ───────────────────────────────────────────────────────────────────────
+# Phase 2: walker finding-emit Rule #25 Shape-1 alignment META-CANARIES.
+# ───────────────────────────────────────────────────────────────────────
+
+
+def test_walker_ics_finding_source_allowlist_matches_pydantic_literal():
+    """W2-β §SC5-NEW-ICS-S2-η META-CANARY — the walker's
+    ``_ALLOWED_ICS_FINDING_SOURCES`` frozenset MUST EXACTLY equal the
+    Pydantic ``IcsProtocolFindingSource`` Literal value set.
+
+    Drift means either: (a) the walker can emit a source the DB CHECK
+    rejects (Pydantic 422 at FindingService.create), or (b) the
+    Literal admits a value the walker would never emit (dead schema).
+    """
+    from typing import get_args
+
+    from app.schemas.finding import IcsProtocolFindingSource
+    from app.services.ics_protocol_walker import _ALLOWED_ICS_FINDING_SOURCES
+
+    literal_values = frozenset(get_args(IcsProtocolFindingSource))
+    assert _ALLOWED_ICS_FINDING_SOURCES == literal_values, (
+        f"_ALLOWED_ICS_FINDING_SOURCES {_ALLOWED_ICS_FINDING_SOURCES!r} "
+        f"MUST equal IcsProtocolFindingSource {literal_values!r}. "
+        f"Drift means the walker either emits a source the DB rejects "
+        f"or admits dead schema. Rule #25 Shape-1 cross-stack alignment."
+    )
+
+
+def test_walker_ics_finding_source_naming_convention():
+    """Every entry in the IcsProtocolFindingSource Literal MUST follow
+    the ``ics_<protocol_family>_detected`` naming convention so the
+    walker's per-family lookup
+    (``f"ics_{family}_detected"``) hits all declared values.
+
+    Without this canary, a future commit could rename a source value
+    (e.g. ``ics_modbus_tcp_detected`` -> ``modbus_detected``) and the
+    walker would silently stop emitting for that family.
+    """
+    from typing import get_args
+
+    from app.schemas.finding import IcsProtocolFindingSource
+
+    for source in get_args(IcsProtocolFindingSource):
+        assert source.startswith("ics_") and source.endswith("_detected"), (
+            f"source {source!r} does NOT match the "
+            f"`ics_<family>_detected` convention; the walker's per-family "
+            f"emit lookup `f'ics_{{family}}_detected'` will not hit it."
+        )
+        family = source[len("ics_") : -len("_detected")]
+        assert family, (
+            f"source {source!r} has empty protocol family between "
+            f"`ics_` and `_detected`"
+        )

frontend/src/constants/statusConfig.ts

@@ -170,6 +170,11 @@ export const FINDING_SOURCE_CONFIG: Record<FindingSource, FindingSourceConfigEnt
   c28x_csm_perma_lock: { icon: KeyRound, label: 'C28x CSM Perma-Lock', className: 'border-amber-600/50 text-amber-700 dark:text-amber-300' },
   bare_metal_chip_unknown_with_hints: { icon: Info, label: 'Bare-Metal Chip Unknown', className: 'border-blue-500/50 text-blue-600 dark:text-blue-400' },
   bare_metal_encrypted_region_skipped: { icon: Info, label: 'Encrypted Region Skipped', className: 'border-slate-500/50 text-slate-600 dark:text-slate-400' },
+  ics_modbus_tcp_detected: { icon: Network, label: 'ICS: Modbus/TCP', className: 'border-cyan-500/50 text-cyan-600 dark:text-cyan-400' },
+  ics_modbus_rtu_detected: { icon: Network, label: 'ICS: Modbus RTU', className: 'border-cyan-500/50 text-cyan-700 dark:text-cyan-300' },
+  ics_dnp3_detected: { icon: Network, label: 'ICS: DNP3', className: 'border-sky-500/50 text-sky-600 dark:text-sky-400' },
+  ics_s7comm_detected: { icon: Network, label: 'ICS: S7Comm', className: 'border-indigo-500/50 text-indigo-600 dark:text-indigo-400' },
+  ics_unknown_ics_detected: { icon: Info, label: 'ICS: Unknown Stack', className: 'border-slate-500/50 text-slate-600 dark:text-slate-400' },
 }
 
 // ── Finding confidence ──