from __future__ import annotations import io import hashlib import json import re import urllib.error from pathlib import Path import pytest from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from cryptography.hazmat.primitives import serialization import yovision_brain.app.__main__ as cli import yovision_brain.integration.event_export.runtime as export_runtime from yovision_brain.integration.event_export import ( EventDeliveryError, HTTPSMachineIdentitySender, ) from yovision_brain.integration.machine_identity import ( KeyRecord, Registry, ReplayStore, Signer, TransportPolicy, Verifier, bearer_token, ) FIXTURE = Path(__file__).parents[1] / "fixtures" / "events" / "area.json" def _policy() -> TransportPolicy: return TransportPolicy( tls_min_version="1.2", verify_certificate=True, verify_hostname=True, connect_timeout_ms=1_000, response_header_timeout_ms=1_000, request_timeout_ms=2_000, max_request_bytes=64 * 1024, ) class _Response: status = 202 def __init__(self, body: bytes, request_id: str) -> None: self.body = body self.headers = {"X-Request-ID": request_id} self.closed = False def read(self, amount: int = -1) -> bytes: return self.body[:amount] if amount >= 0 else self.body def close(self) -> None: self.closed = True class _VerifyingOpener: def __init__(self, verifier: Verifier) -> None: self.verifier = verifier self.requests = [] def open(self, request, timeout: float) -> _Response: # noqa: ANN001 self.requests.append((request, timeout)) body = request.data self.verifier.verify( bearer_token(request.get_header("Authorization")), "yovision-sense", "events:ingest", request.method, "/v1/events", body, ) event = json.loads(body) return _Response(json.dumps({ "producer_id": event["producer_id"], "source_event_id": event["source_event_id"], "payload_sha256": hashlib.sha256(body).hexdigest(), "disposition": "accepted", }).encode(), request.get_header("X-request-id")) def test_sender_binds_exact_path_body_and_safe_request_id() -> None: private_key = Ed25519PrivateKey.generate() signer = Signer("yv:brain:school-a", "brain-key-0001", private_key, clock=lambda: 100) registry = Registry( [ KeyRecord( principal="yv:brain:school-a", key_id="brain-key-0001", public_key=private_key.public_key(), audience="yovision-sense", scopes=frozenset({"events:ingest"}), ) ] ) opener = _VerifyingOpener(Verifier(registry, ReplayStore(), clock=lambda: 100)) sender = HTTPSMachineIdentitySender( "https://sense.example:8443", signer, _policy(), opener=opener ) result = sender.send( b'{"producer_id":"brain-school-a","schema_version":"yovision.event/v1",' b'"source_event_id":"evt-1"}' ) request, timeout = opener.requests[0] assert request.full_url == "https://sense.example:8443/v1/events" assert request.method == "POST" assert request.get_header("Content-type") == "application/json" assert re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._:-]{15,127}", result.request_id) assert request.get_header("X-request-id") == result.request_id assert (result.disposition, timeout) == ("accepted", 2.0) def test_sender_rejects_plaintext_and_marks_conflict_terminal() -> None: key = Ed25519PrivateKey.generate() signer = Signer("yv:brain:school-a", "brain-key-0001", key) with pytest.raises(ValueError, match="HTTPS origin"): HTTPSMachineIdentitySender("http://sense.example", signer, _policy()) class ConflictOpener: def open(self, request, timeout: float): # noqa: ANN001, ARG002 raise urllib.error.HTTPError( request.full_url, 409, "Conflict", {}, io.BytesIO(b'{"code":"event_identity_conflict"}'), ) sender = HTTPSMachineIdentitySender( "https://sense.example", signer, _policy(), opener=ConflictOpener() ) with pytest.raises(EventDeliveryError) as caught: sender.send(b"{}") assert (caught.value.code, caught.value.terminal) == ( "event_identity_conflict", True, ) def test_disabled_connector_keeps_existing_json_lines_output(tmp_path: Path) -> None: raw = json.loads(FIXTURE.read_text(encoding="utf-8")) raw["event_export"] = {"enabled": False} config = tmp_path / "brain.json" config.write_text(json.dumps(raw), encoding="utf-8") output = tmp_path / "events.jsonl" assert cli.main(["--config", str(config), "--output", str(output)]) == 0 assert json.loads(output.read_text(encoding="utf-8"))["schema"] == ( "brain.internal.event-candidate/v1" ) def test_delivery_failure_is_nonzero_and_not_silently_reported_as_success( monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], tmp_path: Path, ) -> None: class FailingSink: def write(self, candidate) -> None: # noqa: ANN001, ARG002 raise EventDeliveryError("event_delivery_unavailable", terminal=False) monkeypatch.setattr(cli, "build_event_export_sink", lambda raw, base_dir: FailingSink()) unused_output = tmp_path / "disabled-json-lines-target" unused_output.write_text("must remain unchanged", encoding="utf-8") result = cli.main( ["--config", str(FIXTURE), "--output", str(unused_output)] ) assert result == 3 error = json.loads(capsys.readouterr().err.splitlines()[0]) assert error == {"status": "error", "message": "event_delivery_unavailable"} assert unused_output.read_text(encoding="utf-8") == "must remain unchanged" def test_inline_private_key_material_is_rejected_without_echo( tmp_path: Path, capsys: pytest.CaptureFixture[str] ) -> None: raw = json.loads(FIXTURE.read_text(encoding="utf-8")) marker = "INLINE-PRIVATE-MATERIAL-MUST-NOT-LEAK" raw["event_export"] = { "enabled": True, "endpoint": "https://sense.example", "producer_id": "brain-school-a", "site_ref": "site-school-a", "severity": "high", "region_refs": {}, "crossing_directions": {}, "machine_identity": { "principal": "yv:brain:school-a", "key_id": "brain-key-0001", "private_key_path": "external.pem", "private_key": marker, }, "transport": {}, } config = tmp_path / "brain.json" config.write_text(json.dumps(raw), encoding="utf-8") assert cli.main(["--config", str(config)]) == 3 assert marker not in capsys.readouterr().err def test_enabled_config_loads_machine_key_only_from_external_path( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: key_path = tmp_path / "brain-machine.pem" key_path.write_bytes( Ed25519PrivateKey.generate().private_bytes( serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8, serialization.NoEncryption(), ) ) captured = {} class Sender: def __init__(self, endpoint, signer, policy) -> None: # noqa: ANN001 captured.update(endpoint=endpoint, signer=signer, policy=policy) monkeypatch.setattr(export_runtime, "HTTPSMachineIdentitySender", Sender) sink = export_runtime.build_event_export_sink( { "enabled": True, "endpoint": "https://sense.example", "producer_id": "brain-school-a", "site_ref": "site-school-a", "severity": "high", "region_refs": {}, "crossing_directions": {}, "machine_identity": { "principal": "yv:brain:school-a", "key_id": "brain-key-0001", "private_key_path": key_path.name, }, "transport": { "tls_min_version": "1.2", "verify_certificate": True, "verify_hostname": True, "connect_timeout_ms": 1_000, "response_header_timeout_ms": 1_000, "request_timeout_ms": 2_000, "max_request_bytes": 64 * 1024, }, }, base_dir=tmp_path, ) assert sink is not None assert captured["endpoint"] == "https://sense.example"