255 lines
8.5 KiB
Python
255 lines
8.5 KiB
Python
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"
|