diff --git a/replicant/core/models.py b/replicant/core/models.py index b820464..d5e7e86 100644 --- a/replicant/core/models.py +++ b/replicant/core/models.py @@ -222,6 +222,19 @@ class RunManifest(BaseModel): # ones that raise, so a short manifest and a failed one have to be # distinguishable: without this a run that died after two events looks # exactly like a run that was only ever meant to emit two. + # Which vendor dialect rendered these events. ScenarioManifest has always + # recorded it; RunManifest did not, so a manifest could not answer "which + # profile produced this?" without re-reading the settings that made it. + vendor: str = "" + #: The --duration the operator asked for, verbatim, or None for the default. + duration: str | None = None + #: The events-per-second ceiling actually in force for this run. + rate: int | None = None + #: What the socket really did: sends, bytes, errors, oversize. None when the + #: run had no collector. `event_count` counts events rendered; this counts + #: datagrams handed to the kernel, and the two differing is the interesting + #: case rather than an inconsistency. + send_stats: dict[str, int] | None = None status: RunStatus = "done" #: Bounded description of the failure, or None. Type and message only, never #: a traceback: this is an operator record, not a debugger. diff --git a/replicant/core/orchestrator.py b/replicant/core/orchestrator.py index 38f0301..03b2a40 100644 --- a/replicant/core/orchestrator.py +++ b/replicant/core/orchestrator.py @@ -231,6 +231,9 @@ def __init__( self.engine = engine or ScenarioEngine() self.profile = profile or build_profile(self.settings) self._stop = threading.Event() + #: What the last run's socket actually did, or None when it had no + #: collector. Set by _emit on every exit path and read into the manifest. + self.last_send_stats: dict[str, int] | None = None # -- vendor identity ------------------------------------------------------- @@ -357,6 +360,7 @@ def run( file_path = request.to_file self.reset() + self.last_send_stats = None plan = self.build_plan(request) target, transport = self._describe_target(request, send) @@ -412,6 +416,10 @@ def run( warmup_note=warmup, pace=pace, speed=request.speed, + vendor=self.settings.vendor, + duration=request.duration, + rate=eps_cap, + send_stats=self.last_send_stats, status=_run_status(failure, stopped), error=describe_error(failure) if failure is not None else None, ) @@ -641,6 +649,17 @@ def _emit( if sink is not None: sink.close() if emitter is not None: + # Captured before close, and on every exit path, so a run that + # failed part-way still records what its socket actually did. + # `event_count` counts events rendered; this counts datagrams the + # kernel accepted, and the two differing is the interesting case. + # getattr, not emitter.stats: the emitter is an injection point + # and several test doubles implement only send/close. Requiring a + # stats attribute would make this record cost every future fake a + # field it does not otherwise need. + stats = getattr(emitter, "stats", None) + if stats is not None: + self.last_send_stats = stats.as_dict() emitter.close() return count, stopped diff --git a/tests/test_run_stream_fanout.py b/tests/test_run_stream_fanout.py index 8a9089e..7881edb 100644 --- a/tests/test_run_stream_fanout.py +++ b/tests/test_run_stream_fanout.py @@ -198,3 +198,88 @@ def test_the_replay_is_bounded(self) -> None: handle.publish({"type": "line", "data": str(i)}) assert len(handle.history) == MAX_HISTORY_ITEMS + + +class TestManifestCompleteness: + """The manifest should answer questions about its own run without help. + + Review finding #1: ScenarioManifest recorded vendor and duration, RunManifest + did not, and neither recorded the rate in force or what the socket actually + did. A manifest that cannot say which profile rendered it, or how many + datagrams left, is a weaker audit record than safety rule 5 implies. + """ + + def test_a_run_records_its_vendor_duration_and_rate(self, tmp_path: Path) -> None: + orch = Orchestrator(CATALOG, Settings(manifest_dir=str(tmp_path))) + result = orch.run( + RunRequest( + technique_id="REP-001", + intensity="low", + duration="90s", + seed=1337, + no_send=True, + ) + ) + + assert result.manifest.vendor == "fortigate" + assert result.manifest.duration == "90s" + assert result.manifest.rate is not None and result.manifest.rate > 0 + + def test_a_sending_run_records_what_the_socket_did(self, tmp_path: Path) -> None: + import socket as _socket + + listener = _socket.socket(_socket.AF_INET, _socket.SOCK_DGRAM) + listener.bind(("127.0.0.1", 0)) + port = int(listener.getsockname()[1]) + try: + orch = Orchestrator(CATALOG, Settings(manifest_dir=str(tmp_path))) + result = orch.run( + RunRequest( + technique_id="REP-001", + intensity="low", + duration="60s", + no_send=False, + pace="burst", + collector=CollectorProfile( + name="t", host="127.0.0.1", port=port, transport="udp" + ), + ) + ) + finally: + listener.close() + + stats = result.manifest.send_stats + assert stats is not None + assert stats["sends"] == result.event_count + assert stats["errors"] == 0 + + def test_a_run_with_no_collector_records_no_send_stats(self, tmp_path: Path) -> None: + """None is the honest answer, not a row of zeroes that reads like a send.""" + orch = Orchestrator(CATALOG, Settings(manifest_dir=str(tmp_path))) + result = orch.run(RunRequest(technique_id="REP-001", intensity="low", no_send=True)) + + assert result.manifest.send_stats is None + + def test_an_older_manifest_without_the_new_fields_still_loads(self) -> None: + """Every new field is defaulted, as pace and speed were before them.""" + from replicant.core.models import RunManifest + + manifest = RunManifest( + replicant_version="0.1.0", + technique_id="REP-001", + technique_name="x", + ndr_uc="UC-001", + intensity="low", + seed=1, + params={}, + entities={}, + target="none", + transport="none", + event_count=1, + started_at="t", + ended_at="t", + anchor_epoch=0, + ) + + assert manifest.vendor == "" + assert manifest.send_stats is None