From a77850dc8348d62dcd69d33b0a27cdb0fbc9d9c5 Mon Sep 17 00:00:00 2001 From: Rehan Khan Date: Sat, 22 Aug 2026 17:21:05 +0500 Subject: [PATCH] feat: connect collector to forensic chain --- .github/workflows/ci.yml | 1 + collector/README.md | 15 ++- collector/pyproject.toml | 1 + collector/tests/test_server.py | 149 ++++++++++++++++++++++- collector/vestrix_collector/forensics.py | 45 +++++-- docs/DPIA.md | 5 +- docs/threat-model.md | 23 ++-- forensics/README.md | 6 +- 8 files changed, 218 insertions(+), 27 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index bc5bd3b..002b74d 100644 --- a/.github/workflows/ci.yml +++ b/.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 diff --git a/collector/README.md b/collector/README.md index 48258f8..e0d05af 100644 --- a/collector/README.md +++ b/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. diff --git a/collector/pyproject.toml b/collector/pyproject.toml index f4d20b4..eaa4cb4 100644 --- a/collector/pyproject.toml +++ b/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] diff --git a/collector/tests/test_server.py b/collector/tests/test_server.py index b63ee2a..5e28de4 100644 --- a/collector/tests/test_server.py +++ b/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: diff --git a/collector/vestrix_collector/forensics.py b/collector/vestrix_collector/forensics.py index 1c24077..ac8cdcf 100644 --- a/collector/vestrix_collector/forensics.py +++ b/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), + ) diff --git a/docs/DPIA.md b/docs/DPIA.md index 6520fef..18dec65 100644 --- a/docs/DPIA.md +++ b/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 diff --git a/docs/threat-model.md b/docs/threat-model.md index 4c85883..9fc7035 100644 --- a/docs/threat-model.md +++ b/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. diff --git a/forensics/README.md b/forensics/README.md index b956ee5..561a4e1 100644 --- a/forensics/README.md +++ b/forensics/README.md @@ -3,5 +3,7 @@ Hash-chained event logger and OpenTimestamps anchoring interface. Status: hash-chain logging, Ed25519 signing and verification, and concurrency-safe -append are implemented and tested. The production OpenTimestamps backend remains -a placeholder. +append are implemented and tested. The collector's default adapter now appends each +accepted ingest event using the version-1 collector-ingest profile and fails closed +when the append cannot complete. Collector rejections remain operational logs, not +signed-chain records. The production OpenTimestamps backend remains a placeholder.