Connect collector to signed forensic chain

.github/workflows/ci.yml

@@ -24,6 +24,7 @@ jobs:
           pip install ruff==0.15.21 pytest==9.1.1
           pip install -r collector/requirements.txt
           pip install -r forensics/requirements.txt
+          pip install -e ./forensics
           pip install -e ./collector
           pip install -e ./soc-integration
 

collector/README.md

@@ -3,5 +3,16 @@
 mTLS ingest service that receives CSI data from sensor nodes.
 
 Status: collector-side mTLS ingest, certificate-CN allow-listing, payload validation,
-and in-memory anti-replay checks are implemented and tested. ESP32 client
-provisioning and automated certificate enrollment/rotation are not implemented.
+in-memory anti-replay checks, and the signed forensic handoff are implemented and
+tested. Configure `VESTRIX_FORENSICS_PRIVATE_KEY` with an unencrypted Ed25519 PKCS#8
+private-key path and, optionally, `VESTRIX_FORENSICS_STORE` with the JSONL path.
+Accepted events are acknowledged only after a durable signed append; append failure
+rejects the event. Rejected ingest attempts remain structured collector decision
+logs and are not appended to the signed chain.
+
+The version-1 collector-ingest profile records `csi_window_sha256` as
+`raw_csi_hash`, uses the SHA-256 of empty content when features are unavailable,
+identifies the adapter in the model fields, records `collector_event_accepted` with
+confidence `1.0` (transport acceptance, not ML confidence), and stores the collector
+schema version and sequence number in `top_shap`. ESP32 client provisioning and
+automated certificate enrollment/rotation are not implemented.

collector/pyproject.toml

@@ -11,6 +11,7 @@ requires-python = ">=3.11"
 license = { text = "Apache-2.0" }
 dependencies = [
     "PyYAML>=6.0.1,<7",
+    "vestrix-forensics==0.1.0",
 ]
 
 [project.optional-dependencies]

collector/tests/test_server.py

@@ -11,6 +11,9 @@
 
 import pytest
 from conftest import CertificateBundle, Identity
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
+from forensics import verify_chain
+from forensics.keys import generate_signing_key, save_private_key
 
 from vestrix_collector.config import CollectorConfig, ServerConfig, TLSConfig
 from vestrix_collector.models import CSIEvent
@@ -69,7 +72,7 @@ def _config(tmp_path: Path, certificates: CertificateBundle) -> CollectorConfig:
 
 async def _with_server(
     config: CollectorConfig,
-    event_logger: Callable[[CSIEvent], None],
+    event_logger: Callable[[CSIEvent], None] | None,
     operation: Callable[[CollectorServer], Any],
 ) -> Any:
     server = CollectorServer(config, event_logger=event_logger)
@@ -80,6 +83,18 @@ async def _with_server(
         await server.close()
 
 
+def _configure_forensics(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> tuple[Path, Ed25519PublicKey]:
+    store = tmp_path / "chain.jsonl"
+    private_key = tmp_path / "forensics-private.pem"
+    signer = generate_signing_key()
+    save_private_key(signer, private_key)
+    monkeypatch.setenv("VESTRIX_FORENSICS_STORE", str(store))
+    monkeypatch.setenv("VESTRIX_FORENSICS_PRIVATE_KEY", str(private_key))
+    return store, signer.public_key()
+
+
 def test_valid_certificate_and_enrolled_node_is_accepted(
     tmp_path: Path, certificates: CertificateBundle
 ) -> None:
@@ -105,6 +120,138 @@ async def scenario(server: CollectorServer) -> dict[str, str]:
     assert received == [_payload()]
 
 
+def test_default_adapter_appends_verifiable_accepted_event(
+    tmp_path: Path,
+    certificates: CertificateBundle,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    store, public_key = _configure_forensics(tmp_path, monkeypatch)
+
+    async def scenario(server: CollectorServer) -> dict[str, str]:
+        return await asyncio.to_thread(
+            _send_event,
+            server.port,
+            certificates.ca_cert,
+            certificates.enrolled_node,
+            _payload(),
+        )
+
+    response = asyncio.run(
+        _with_server(_config(tmp_path, certificates), None, scenario)
+    )
+    record = json.loads(store.read_text(encoding="utf-8"))
+
+    assert response["status"] == "accepted"
+    assert verify_chain(store, public_key).records_verified == 1
+    assert record["raw_csi_hash"] == _payload()["csi_window_sha256"]
+    assert record["class"] == "collector_event_accepted"
+    assert record["top_shap"]["collector_sequence_number"] == 1
+
+
+def test_default_adapter_chains_two_accepted_events(
+    tmp_path: Path,
+    certificates: CertificateBundle,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    store, public_key = _configure_forensics(tmp_path, monkeypatch)
+
+    async def scenario(server: CollectorServer) -> list[dict[str, str]]:
+        return [
+            await asyncio.to_thread(
+                _send_event,
+                server.port,
+                certificates.ca_cert,
+                certificates.enrolled_node,
+                _payload(sequence_number),
+            )
+            for sequence_number in (1, 2)
+        ]
+
+    responses = asyncio.run(
+        _with_server(_config(tmp_path, certificates), None, scenario)
+    )
+    records = [
+        json.loads(line)
+        for line in store.read_text(encoding="utf-8").splitlines()
+    ]
+
+    assert [response["status"] for response in responses] == ["accepted", "accepted"]
+    assert verify_chain(store, public_key).records_verified == 2
+    assert records[1]["prev_hash"] == records[0]["record_hash"]
+
+
+def test_rejected_replay_is_not_appended_to_forensic_chain(
+    tmp_path: Path,
+    certificates: CertificateBundle,
+    monkeypatch: pytest.MonkeyPatch,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    store, public_key = _configure_forensics(tmp_path, monkeypatch)
+    caplog.set_level(logging.INFO, logger="vestrix_collector.decisions")
+
+    async def scenario(server: CollectorServer) -> list[dict[str, str]]:
+        return [
+            await asyncio.to_thread(
+                _send_event,
+                server.port,
+                certificates.ca_cert,
+                certificates.enrolled_node,
+                _payload(),
+            )
+            for _ in range(2)
+        ]
+
+    responses = asyncio.run(
+        _with_server(_config(tmp_path, certificates), None, scenario)
+    )
+
+    assert responses[0]["status"] == "accepted"
+    assert responses[1] == {
+        "status": "rejected",
+        "reason": "replayed_or_out_of_order_sequence",
+    }
+    assert verify_chain(store, public_key).records_verified == 1
+    assert any(
+        getattr(record, "decision_fields", {}).get("reason")
+        == "replayed_or_out_of_order_sequence"
+        for record in caplog.records
+    )
+
+
+def test_forensic_append_failure_rejects_event(
+    tmp_path: Path,
+    certificates: CertificateBundle,
+    monkeypatch: pytest.MonkeyPatch,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    store, _ = _configure_forensics(tmp_path, monkeypatch)
+    caplog.set_level(logging.INFO, logger="vestrix_collector.decisions")
+    blocked_parent = tmp_path / "blocked"
+    blocked_parent.write_text("not a directory", encoding="utf-8")
+    monkeypatch.setenv("VESTRIX_FORENSICS_STORE", str(blocked_parent / store.name))
+
+    async def scenario(server: CollectorServer) -> dict[str, str]:
+        return await asyncio.to_thread(
+            _send_event,
+            server.port,
+            certificates.ca_cert,
+            certificates.enrolled_node,
+            _payload(),
+        )
+
+    response = asyncio.run(
+        _with_server(_config(tmp_path, certificates), None, scenario)
+    )
+
+    assert response == {"status": "rejected", "reason": "forensics_handoff_failed"}
+    assert not (blocked_parent / store.name).exists()
+    assert any(
+        getattr(record, "decision_fields", {}).get("reason")
+        == "forensics_handoff_failed"
+        for record in caplog.records
+    )
+
+
 def test_valid_certificate_with_unenrolled_cn_is_rejected(
     tmp_path: Path, certificates: CertificateBundle
 ) -> None:

collector/vestrix_collector/forensics.py

@@ -1,21 +1,40 @@
-"""Temporary boundary to the separately developed forensic event logger."""
+"""Collector adapter for the signed forensic event logger."""
 
 from __future__ import annotations
 
-from .models import CSIEvent
+import hashlib
+import os
 
+from forensics import log_event as append_forensic_event
+from forensics.keys import load_private_key
 
-def log_event(event: CSIEvent) -> None:
-    """Hand an authenticated raw event to the forensic logging subsystem.
+from .models import CSIEvent
 
-    Integration contract for ``forensics/`` (must retain this exact signature):
+SIGNING_KEY_PATH_ENV = "VESTRIX_FORENSICS_PRIVATE_KEY"
+_EMPTY_CONTENT_HASH = hashlib.sha256(b"").hexdigest()
+_PROFILE_HASH = hashlib.sha256(b"vestrix-collector-accepted-event-v1").hexdigest()
 
-        def log_event(event: vestrix_collector.models.CSIEvent) -> None
 
-    The eventual implementation must raise an exception when durable handoff fails.
-    It owns hash chaining and external timestamp anchoring. The collector must never
-    sign or claim to verify ``timestamp_utc`` itself.
-    """
-    # TODO(forensics integration): replace this no-op with the separately built
-    # hash-chain logger. Hashing/signing deliberately does not belong here.
-    del event
+def log_event(event: CSIEvent) -> None:
+    """Durably append one authenticated collector event to the forensic chain."""
+    key_path = os.environ.get(SIGNING_KEY_PATH_ENV)
+    if not key_path:
+        raise RuntimeError(f"{SIGNING_KEY_PATH_ENV} is not configured")
+
+    append_forensic_event(
+        {
+            "ts_utc": event["timestamp_utc"],
+            "node_id": event["node_id"],
+            "raw_csi_hash": event["csi_window_sha256"],
+            "features_hash": _EMPTY_CONTENT_HASH,
+            "model_id": "vestrix-collector-adapter-v1",
+            "model_config_hash": _PROFILE_HASH,
+            "class": "collector_event_accepted",
+            "confidence": 1.0,
+            "top_shap": {
+                "collector_schema_version": event["schema_version"],
+                "collector_sequence_number": event["sequence_number"],
+            },
+        },
+        load_private_key(key_path),
+    )

docs/DPIA.md

@@ -173,7 +173,7 @@ treated as a statement that every stage is implemented or enabled.
 | 1. Sensor | Capture CSI and optionally buffer or pre-filter it. | ESP32 firmware and sensor-side controls are not yet implemented. | Sensor memory, flash, or another operator-selected store. | Device administrators and anyone with physical or administrative access. |
 | 2. Collector | Authenticate the node and accept an event containing `node_id`, timestamp, sequence number, and `csi_window_sha256`. | Server-side mTLS, allow-listing, identity binding, validation, and in-process replay checking are implemented. The current collector schema receives a CSI-window hash, not the raw window. The firmware client is not implemented. | Collector operational logs and any configured handoff destination. | Collector administrators and service accounts. |
 | 3. Processing / ML | Clean signals, derive features, classify activity, and produce confidence and SHAP contributions. | The architecture and schemas define this stage; the production signal pipeline and trained model are not implemented. | Operator-selected feature, model, training, and raw-window stores. | ML operators, security engineers, and service accounts selected by the operator. |
-| 4. Forensic log | Append timestamp, node, raw/feature hashes, model metadata, class, confidence, SHAP snapshot, chain links, and signature to canonical JSONL. | Logger and independent Rust verifier are implemented. The default collector-to-forensics adapter remains a no-op. | Local or operator-selected append-only JSONL store, backups, and any exported chain tips or timestamp proofs. | Evidence custodians, incident responders, legal teams, and system administrators as authorized by the operator. |
+| 4. Forensic log | Append timestamp, node, raw/feature hashes, model metadata, class, confidence, explanation or transport metadata, chain links, and signature to canonical JSONL. | Logger and independent Rust verifier are implemented. The default adapter durably signs and appends each accepted collector event using the documented collector-ingest profile. Rejected attempts remain in collector operational logs rather than the signed chain; a forensic append failure causes the collector to reject the event. | Local or operator-selected append-only JSONL store, backups, and any exported chain tips or timestamp proofs. | Evidence custodians, incident responders, legal teams, and system administrators as authorized by the operator. |
 | 5. SOC / SIEM | Map a canonical alert to Wazuh JSON and OCSF Detection Finding, including site/zone and optional PACS correlation fields. | Mapper and tested Wazuh rules are implemented; no end-to-end dispatcher or operator-notification guarantee exists. | Wazuh, SIEM, ticketing, notification, archive, and backup systems. | SOC analysts, incident responders, administrators, service providers, and any downstream recipients configured by the operator. |
 
 The controller must create a deployment-specific diagram showing every actual
@@ -375,7 +375,8 @@ As of this template version, Vestrix does **not** provide:
 - built-in user RBAC, access reviews, or audit logs for reads and exports;
 - encryption at rest or deployment-wide key management;
 - automatic certificate revocation, rotation, or durable replay state;
-- an end-to-end dispatcher connecting collector, ML, forensic, and SOC stages;
+- an end-to-end dispatcher connecting the collector through ML and SOC stages
+  (the direct accepted-event collector-to-forensics handoff is implemented);
 - an implemented production model, SHAP generator, or real-world accuracy and
   generalization results;
 - a visible sensor-active indicator, opt-out mechanism, or sensing-boundary

docs/threat-model.md

@@ -173,6 +173,7 @@ integrity and documented handling part of a defensible forensic process
 |---|---|---|
 | `IMPLEMENTED` | Each canonical JSONL record contains a sequence number, the previous record hash, a SHA-256 `record_hash`, and an Ed25519 signature over the canonical record bytes. | Modification, insertion, reordering, or forgery without the signing key causes hash, linkage, sequence, canonicalization, or signature verification to fail. |
 | `IMPLEMENTED` | The Python logger validates the existing chain before append and serializes appends under a shared file lock. | Reduces accidental or concurrent chain corruption at the supported append interface. It does not make the underlying file write-once. |
+| `IMPLEMENTED` | After authentication, payload validation, identity binding, and anti-replay checks, the default collector adapter maps each accepted event to the version-1 collector-ingest profile and durably appends it with the configured Ed25519 key. The collector rejects the event if append fails. | Prevents the collector from acknowledging accepted evidence that was not signed and durably appended. |
 | `IMPLEMENTED` | The independent Rust `vestrix-verify` CLI shares no collector or Python forensics code and validates UTF-8/JSON form, the exact schema, canonical bytes, sequence continuity, hash linkage, SHA-256 hashes, and Ed25519 signatures. | Provides an independent, read-only check of the records that are present. |
 | `IMPLEMENTED` | The Python OpenTimestamps interface can snapshot a chain tip and write a backend receipt; the Rust verifier can inspect a limited offline proof subset and intentionally refuses to report full Bitcoin-anchor success. | Provides fail-closed scaffolding and partial proof binding without overstating timestamp assurance. |
 | `PLANNED` | A pinned production OpenTimestamps submission client and complete verification against an independently trusted Bitcoin best chain. | Intended to bind selected chain tips to an external time source and make later history replacement or truncation detectable relative to a retained checkpoint. |
@@ -204,9 +205,16 @@ integrity and documented handling part of a defensible forensic process
 - Signing-key custody, filesystem authorization, backup retention, export of trusted
   chain tips, and evidence-chain-of-custody procedures remain operator
   responsibilities.
-- The collector-to-forensics adapter is currently a no-op boundary. The logger is
-  implemented as a separate component, but accepted collector events are not yet
-  connected to it by the default collector configuration.
+- Only accepted events enter the signed forensic chain. TLS, enrollment, malformed
+  payload, identity-mismatch, and replay rejections remain structured collector
+  operational logs; they are not signed chain records and therefore do not have the
+  chain's tamper-evidence properties.
+- The collector-ingest profile is a transport-acceptance record, not an ML result.
+  It maps `csi_window_sha256` to `raw_csi_hash`, uses the empty-content SHA-256 for
+  unavailable features, identifies the adapter in the model fields, records the
+  acceptance decision with confidence `1.0`, and preserves collector schema and
+  sequence metadata in `top_shap`. The sensor-supplied timestamp is recorded but is
+  not independently verified by the collector.
 
 ## 4. Alert suppression — attacker attempts to suppress, delay, or interfere with alert delivery
 
@@ -227,7 +235,8 @@ misinterpreted as an absence of intrusion rather than a failed or suppressed sen
 |---|---|---|
 | `IMPLEMENTED` | The canonical alert mapper produces validated Wazuh JSON and OCSF Detection Finding events, and the Wazuh decoder/rules generate defined alerts for supported intrusion and `sensor_tamper` events. | Provides deterministic downstream interpretation for an event that reaches the integration. It does not guarantee transport or operator notification. |
 | `IMPLEMENTED` | The Wazuh integration includes a correlation rule for a Vestrix intrusion following a supported authentication anomaly. | Can raise the severity of a received alert when corroborating telemetry is present. It does not detect a missing Vestrix event. |
-| `PLANNED` | Connect authenticated collector handoff to the signed logger and canonical SOC event path. | Intended to preserve accepted evidence and feed both SOC mappings from the same validated event. The current collector adapter is a no-op, so this path is not end to end. |
+| `IMPLEMENTED` | Connect authenticated collector handoff to the signed logger. | Preserves each accepted collector event in the hash-chained, Ed25519-signed store and fails closed if the append cannot complete. |
+| `PLANNED` | Connect the processing/ML output to the canonical SOC event path. | Intended to feed both SOC mappings from the same validated classified event. This path is not end to end. |
 | `PLANNED` | Sensor heartbeat/liveness records and missed-heartbeat alerts. | Intended to make loss of expected telemetry observable instead of treating silence as normal operation. |
 | `PLANNED` | Correlation of a sensor-tamper or liveness failure with missing or delayed alert traffic. | Intended to identify combined physical interference and alert-path impairment. No such correlation rule is currently shipped. |
 
@@ -248,9 +257,9 @@ misinterpreted as an absence of intrusion rather than a failed or suppressed sen
   implemented. Silence is not currently a reliable security signal.
 - The Wazuh and OCSF components map and classify events; they do not deliver them to
   an operator or prove that an operator received them.
-- No in-repository dispatcher currently connects collector ingest to the signed
-  logger, canonical alert mapper, and Wazuh/OCSF destinations as one operational
-  path.
+- No in-repository dispatcher currently connects collector ingest through ML to the
+  canonical alert mapper and Wazuh/OCSF destinations as one operational path. The
+  direct accepted-event collector-to-signed-logger path is implemented.
 - mTLS provides peer authentication and transport integrity, not availability. It
   does not prevent connection exhaustion, network outage, deliberate packet loss,
   or loss of sensor power.