feat(ics-protocol): plugin infrastructure + bundled string_scanner + W2-β §SC5-NEW-ICS-S2-α HARDENED freeze (Phase 4)

Rule #52 instance #3 / Phase 4: closes Session 1 W2-β §SC5-NEW-ICS-7
(hot-reload × plugin module-level shadow) — the scariest unmitigated
attack identified by Session 1 deep-research. Ships with W2-β
§SC5-NEW-ICS-S2-α HARDENED shape (Wave-2 β cross-feature critique
explicit recommendation): MappingProxyType wrap + closure-capture gate
+ freeze sentinel + namespace-disjointness collision check.

Resolver plugin infrastructure (app/services/ics_protocol_catalog/
resolver.py):
- IcsProtocolMatcherProto Protocol — typed plugin contract;
  cost_class + protocol_families + detect(blob_head, path, size,
  context) signature.
- Private _PLUGIN_REGISTRY mutable dict; PRIVATE — never re-exported.
- Public PLUGIN_REGISTRY = MappingProxyType(_PLUGIN_REGISTRY) —
  read-only view. Direct ``PLUGIN_REGISTRY[x] = y`` writes raise
  TypeError. CLOSES §SC5-NEW-ICS-S2-α: hostile bundled plugin
  ``from ... import PLUGIN_REGISTRY; PLUGIN_REGISTRY["x"] = matcher``
  attack now fails at the proxy layer (NOT just the freeze check).
- register_matcher(name, matcher) — three gates: freeze sentinel,
  namespace-disjointness collision, closure-capture rejection.
- _closure_capture_check (W2-β §SC5-NEW-ICS-S2-ζ): rejects matchers
  whose detect() callable captures AsyncSession/Settings/ContextVar/
  ToolContext via closure (request-scope leak prevention).
- freeze_plugin_registry() flips sentinel; _unfreeze_plugin_registry_for_tests
  for test isolation only.

Bundled string_scanner plugin (NEW
app/services/ics_protocol_catalog/plugins/string_scanner.py):
- Stateless StringScannerPlugin — closure-clean (no captured state).
- Coarse byte-needle signatures for modbus_tcp / dnp3 / s7comm —
  detects library-symbol patterns the closed-grammar
  string_in_binary_constraint cannot express compactly.

Plugin discovery (NEW app/services/ics_protocol_catalog/plugins/__init__.py):
- register_default_plugins(freeze=True) — static imports of bundled
  plugins; no importlib against operator-controlled name strings
  (defense against §SC5-NEW-ICS-S2-α dynamic-import attack vector).

Lifespan wire (app/main.py):
- register_default_plugins(freeze=True) called AFTER LOLDrivers probe
  + BEFORE yield. Order matters — the freeze gate is the security
  boundary; calling after yield exposes a startup-to-first-request
  window. Defensive try/except so plugin registration failure does
  NOT block startup — walker degrades gracefully to closed-grammar
  YAML detection only.

Rule #46 paired META-CANARIES (test_ics_protocol_plugin.py, 13 tests):
- test_plugin_registry_is_mappingproxytype_read_only: type assertion
- test_mappingproxytype_gate_actually_blocks_direct_dict_write:
  paired canary — synthesizes hostile direct write; confirms TypeError
- test_register_matcher_pre_freeze_succeeds: paired canary (gate
  doesn't degenerate to always-reject)
- test_register_matcher_post_freeze_raises: §SC5-NEW-ICS-S2-α
  primary mitigation; RuntimeError on post-freeze write
- test_freeze_does_not_block_existing_lookups: defensive — reads
  still work post-freeze
- test_register_matcher_rejects_async_session_closure_capture:
  §SC5-NEW-ICS-S2-ζ — synthesizes hostile matcher with AsyncSession
  in __closure__; confirms ValueError
- test_register_matcher_accepts_stateless_matcher: paired canary
- test_register_matcher_rejects_protocol_family_collision: namespace
  disjointness (analog of file_format A7)
- test_register_default_plugins_registers_string_scanner: smoke
- test_register_default_plugins_freezes_when_requested: lifespan
  contract
- test_string_scanner_plugin_detect_returns_hits_on_modbus_signature:
  plugin functional smoke
- test_string_scanner_plugin_returns_none_on_clean_binary: paired
- test_main_py_lifespan_imports_and_calls_register_default_plugins:
  Rule #46 META-CANARY — confirms main.py call shape + BEFORE-yield
  textual ordering

Phase 4 test sweep: 13 passed in 0.48s; broader Phase 1-4 + baseline
sweep: 952 passed.

DEFERRED per W2-β recommendation (queued in Phase 6 postmortem):
- AST pre-import scan of plugin module source (complex; covered
  partially by closure-capture gate + MappingProxyType wrap)
- Rule #21 backfill to file_format_catalog (same MappingProxyType +
  closure-capture hardening) — file_format precedent uses bare dict
  shape; backfill is a Rule-of-Two sweep but ICS-S2 scope deferred
  it; documented in Phase 6 postmortem as follow-up.

Phase 5 (next): DNP3 + S7Comm production YAML manifests as Rule #25
single-slice commits (one per protocol per Rule #25).

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

Author: eastmadc

backend/app/main.py

@@ -350,6 +350,26 @@ async def _reap_one_config(config: WalkerReaperConfig) -> None:
             "LOLDrivers bundle probe failed unexpectedly", exc_info=True,
         )
 
+    # CLAUDE.md Rule #52 instance #3 / Phase 4 — ICS protocol catalog
+    # plugin registration + freeze (W2-β §SC5-NEW-ICS-S2-α HARDENED).
+    # AFTER any catalog mtime-cache warmup AND BEFORE the FastAPI yield
+    # so the freeze sentinel is locked before any incoming request can
+    # reach a register_matcher() call. Closes the
+    # Session 1 W2-β §SC5-NEW-ICS-7 hot-reload × plugin attack vector.
+    try:
+        from app.services.ics_protocol_catalog.plugins import (
+            register_default_plugins,
+        )
+
+        register_default_plugins(freeze=True)
+    except Exception:
+        import logging
+        logging.getLogger(__name__).exception(
+            "ics_protocol_catalog: bundled plugin registration failed at "
+            "lifespan startup — walker will operate WITHOUT plugin "
+            "matchers; closed-grammar YAML detection still works",
+        )
+
     yield
 
     # Shutdown Redis

backend/app/services/ics_protocol_catalog/__init__.py

@@ -32,8 +32,14 @@
 from app.services.ics_protocol_catalog.resolver import (
     _SIGNAL_COST_CLASS,
     _SOURCE_PRECEDENCE,
+    PLUGIN_REGISTRY,
     SIGNAL_EVALUATORS,
+    IcsProtocolMatcherProto,
     IcsResolverContext,
+    _unfreeze_plugin_registry_for_tests,
+    freeze_plugin_registry,
+    is_plugin_registry_frozen,
+    register_matcher,
     resolve_all,
 )
 from app.services.ics_protocol_catalog.snapshot import (
@@ -49,13 +55,19 @@ def get_default_snapshot() -> IcsProtocolCatalogSnapshot:
 __all__ = [
     "IcsProtocolCatalog",
     "IcsProtocolCatalogSnapshot",
+    "IcsProtocolMatcherProto",
     "IcsResolverContext",
+    "PLUGIN_REGISTRY",
     "SIGNAL_EVALUATORS",
     "_SIGNAL_COST_CLASS",
     "_SOURCE_PRECEDENCE",
+    "_unfreeze_plugin_registry_for_tests",
     "derive_vendor_authority",
+    "freeze_plugin_registry",
     "get_default_catalog",
     "get_default_snapshot",
+    "is_plugin_registry_frozen",
+    "register_matcher",
     "reset_default_catalog",
     "resolve_all",
 ]

backend/app/services/ics_protocol_catalog/plugins/__init__.py

@@ -0,0 +1,52 @@
+"""Bundled plugin discovery + registration for the ICS protocol catalog.
+
+CLAUDE.md Rule #52 instance #3 / Phase 4. Wired into ``main.py:lifespan``
+AFTER catalog mtime-cache warmup AND BEFORE the FastAPI ``yield`` so
+the freeze sentinel is set before any incoming request can reach a
+``register_matcher`` call.
+
+Discovery shape (W2-β §SC5-NEW-ICS-S2-α partial mitigation): bundled
+plugins are static imports below — NOT dynamic via ``importlib`` against
+operator-controlled name strings. The catalog loader's YAML
+``plugin.name`` reference (future cross-feature gate I21) MUST resolve
+against the frozen registry; unknown names reject the manifest at
+YAML-load time.
+
+Hot-reload of YAML at runtime is allowed (operators tune detection
+without restart); plugin registration at runtime is NOT — closes the
+§SC5-NEW-ICS-7 hot-reload × plugin attack vector that Session 1 W2-β
+identified as the scariest unmitigated case.
+"""
+from __future__ import annotations
+
+import logging
+
+from app.services.ics_protocol_catalog.plugins.string_scanner import (
+    StringScannerPlugin,
+)
+from app.services.ics_protocol_catalog.resolver import (
+    freeze_plugin_registry,
+    register_matcher,
+)
+
+logger = logging.getLogger(__name__)
+
+
+def register_default_plugins(*, freeze: bool = True) -> None:
+    """Register every bundled plugin, then optionally freeze the registry.
+
+    Wired from ``main.py:lifespan`` with ``freeze=True``. Tests can call
+    with ``freeze=False`` to register-and-test without locking the
+    sentinel for sibling tests.
+    """
+    register_matcher("string_scanner", StringScannerPlugin())
+    if freeze:
+        freeze_plugin_registry()
+        logger.info(
+            "ics_protocol_catalog: plugin registry frozen post-startup "
+            "(W2-β §SC5-NEW-ICS-S2-α HARDENED — MappingProxyType view + "
+            "freeze sentinel; bundled plugins: string_scanner)"
+        )
+
+
+__all__ = ["register_default_plugins"]

backend/app/services/ics_protocol_catalog/plugins/string_scanner.py

@@ -0,0 +1,92 @@
+"""Bundled string-scanner plugin (CLAUDE.md Rule #52 instance #3, Phase 4).
+
+Detects ICS protocol stacks by scanning the binary's head bytes for
+library-symbol byte signatures that closed-grammar
+``string_in_binary_constraint`` evaluators cannot express compactly.
+The plugin is STATELESS — accepts head + path + size + context, returns
+a (protocol_family, evidence) tuple per hit.
+
+**Stateless contract — Rule #45 + W2-β §SC5-NEW-ICS-S2-ζ.** The plugin
+does NOT capture session state, settings, or context-var values in its
+closure. The ``register_matcher`` closure-capture gate would reject
+this plugin at registration if it did; the gate's META-CANARY proves
+the rejection fires.
+
+**Bundled byte signatures.** Each entry maps a protocol_family to a
+list of byte-sequence needles. Hits in ANY needle imply detection
+(disjunctive). For the v0 release we ship coarse signatures for the
+3 SHIPPED protocols (modbus_tcp, dnp3, s7comm) — future plugin updates
+can extend.
+"""
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from app.services.ics_protocol_catalog.resolver import IcsResolverContext
+
+
+# Per-protocol byte-needle library. Operators can extend this by
+# authoring YAML manifests with ``string_in_binary`` signal kinds
+# (closed-grammar path); this plugin covers cases where the byte
+# sequence is too varied for a single YAML constraint.
+_BYTE_NEEDLES: dict[str, list[bytes]] = {
+    "modbus_tcp": [
+        b"libmodbus",
+        b"modbus_tcp_listen",
+        b"modbus.so",
+        b"mbap_header",
+    ],
+    "dnp3": [
+        b"dnp3-bin",
+        b"dnp3_outstation",
+        b"opendnp3",
+        b"libdnp3",
+    ],
+    "s7comm": [
+        b"libs7comm",
+        b"s7_open_connection",
+        b"snap7",
+        b"libsnap7",
+    ],
+}
+
+
+class StringScannerPlugin:
+    """Stateless string-scanning matcher. Closure carries no session
+    state — confirmed by the register_matcher closure-capture gate.
+    """
+
+    # IcsProtocolMatcherProto contract — accessed by namespace-collision
+    # check + cost-class ranking.
+    cost_class: int = 3  # expensive — substring scan over head window
+    protocol_families: frozenset[str] = frozenset(_BYTE_NEEDLES.keys())
+
+    def detect(
+        self,
+        blob_head: bytes,
+        path: str,
+        size: int,
+        context: "IcsResolverContext | None",
+    ) -> list[dict] | None:
+        """Return a list of ``{protocol_family, evidence}`` dicts or None.
+
+        Stateless: reads only its arguments + the module-level
+        ``_BYTE_NEEDLES`` constant.
+        """
+        hits: list[dict] = []
+        for protocol_family, needles in _BYTE_NEEDLES.items():
+            for needle in needles:
+                if needle in blob_head:
+                    hits.append({
+                        "protocol_family": protocol_family,
+                        "evidence": (
+                            f"byte-needle {needle.decode('latin-1')!r} "
+                            f"in head"
+                        ),
+                    })
+                    break  # one needle hit per family is enough
+        return hits or None
+
+
+__all__ = ["StringScannerPlugin"]