diff --git a/docs/adr/0001-episode-semantics-boundaries.md b/docs/adr/0001-episode-semantics-boundaries.md index b58e893..aca5443 100644 --- a/docs/adr/0001-episode-semantics-boundaries.md +++ b/docs/adr/0001-episode-semantics-boundaries.md @@ -110,6 +110,17 @@ rejects more than 200 fixture events or 1,000 candidates per segment. These limits make research execution bounded; they do not make the algorithm suitable for `Detector::analyze()`. +A fail-closed baseline contract is a separate prerequisite for oracle +materialization. Candidate v1 accepts only the `brute_force` / `source_ip` / +inclusive-window fixture contract, replays the v0.6 best-count selection, and +requires the declared segments, selected window, finding identity, and excluded +event partition to match that replay. Timestamp comparison uses instants and +finding identity uses canonical UTC, so equivalent offsets cannot create a new +episode identity. Rule scalars are type-strict: the subject is a string, threshold +and window_seconds are integers, and the terminal-failure flag is a Boolean. +Coercible strings and numeric Boolean equivalents fail closed. Unsupported rules +or semantically stale baselines are rejected before candidate selection. + A separate oracle slice must bind this core to the committed baseline, validate all candidate-to-episode evidence references, and record the measured fixture outcome before the hypothesis is accepted. The runtime and `loglens.report.v3` @@ -148,3 +159,5 @@ remain unchanged. - [`Detector tests`](../../tests/test_detector.cpp) - [`Candidate selection core`](../../scripts/episode_candidate_core.py) - [`Candidate core tests`](../../tests/test_episode_candidate_core.py) +- [`Baseline contract`](../../scripts/episode_baseline_contract.py) +- [`Baseline contract tests`](../../tests/test_episode_candidate_baseline_contract.py) diff --git a/scripts/episode_baseline_contract.py b/scripts/episode_baseline_contract.py new file mode 100644 index 0000000..1661a81 --- /dev/null +++ b/scripts/episode_baseline_contract.py @@ -0,0 +1,332 @@ +"""Fail-closed binding between a research fixture and its v0.6 baseline.""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime, timezone +from typing import Any, Sequence + +if __package__: + from .episode_candidate_core import ( + CandidateWindow, + activity_segments, + ordered_events, + parse_timestamp, + ) +else: + from episode_candidate_core import ( + CandidateWindow, + activity_segments, + ordered_events, + parse_timestamp, + ) + + +RULE_FIELDS = ( + "rule_id", + "grouping_key", + "subject", + "threshold", + "window_seconds", + "window_boundary", +) +SUPPORTED_RULE = { + "rule_id": "brute_force", + "grouping_key": "source_ip", + "window_boundary": "inclusive", + "signal_kind": "ssh_failed_password", + "counts_as_terminal_auth_failure": True, +} +BASELINE_ALGORITHM = { + "id": "v0.6.activity_segments_best_count_window", + "implementation": "src/detector.cpp", + "segment_boundary": "split when an adjacent event gap is greater than the rule window", + "selection_tie_break": "keep the first maximum encountered in chronological scan order", +} + + +@dataclass(frozen=True) +class BaselineContext: + segments: list[list[dict[str, Any]]] + selections: tuple[CandidateWindow, ...] + episode_count: int + + +def rule_projection(rule: dict[str, Any]) -> dict[str, Any]: + return {key: rule[key] for key in RULE_FIELDS} + + +def canonical_oracle_timestamp(timestamp: datetime) -> str: + return timestamp.astimezone(timezone.utc).isoformat().replace("+00:00", "Z") + + +def canonical_finding_timestamp(timestamp: datetime) -> str: + return timestamp.astimezone(timezone.utc).strftime("%Y-%m-%d %H:%M:%S") + + +def _hash_append(value: int, text: str) -> int: + for byte in text.encode("utf-8"): + value ^= byte + value = (value * 1099511628211) & ((1 << 64) - 1) + value ^= 0xFF + return (value * 1099511628211) & ((1 << 64) - 1) + + +def finding_id(rule: dict[str, Any], candidate: CandidateWindow) -> str: + value = 14695981039346656037 + fields = [ + str(rule["rule_id"]), + str(rule["grouping_key"]), + str(rule["subject"]), + canonical_finding_timestamp(candidate.first_seen), + canonical_finding_timestamp(candidate.last_seen), + str(rule["threshold"]), + str(candidate.event_count), + str(candidate.event_count), + *candidate.event_ids, + ] + for field in fields: + value = _hash_append(value, field) + return f"finding:{rule['rule_id']}:{value:016x}" + + +def _validate_fixture(fixture: dict[str, Any]) -> None: + if fixture.get("format") != "loglens.episode_research_fixture.v1": + raise ValueError("unsupported research fixture format") + rule = fixture["rule"] + subject = rule.get("subject") + threshold = rule.get("threshold") + window_seconds = rule.get("window_seconds") + terminal_failure = rule.get("counts_as_terminal_auth_failure") + if ( + type(subject) is not str + or type(threshold) is not int + or type(window_seconds) is not int + or type(terminal_failure) is not bool + ): + raise ValueError("candidate v1 rule requires canonical scalar types") + if any(rule.get(field) != value for field, value in SUPPORTED_RULE.items()): + raise ValueError( + "candidate v1 supports only brute_force source_ip terminal-failure rules" + ) + if not subject: + raise ValueError("candidate v1 requires a non-empty source_ip subject") + if threshold < 1 or window_seconds < 1: + raise ValueError("candidate v1 threshold and window_seconds must be positive") + if any( + event.get("event_type") != SUPPORTED_RULE["signal_kind"] + or event.get("source_ip") != subject + for event in fixture["events"] + ): + raise ValueError( + "candidate v1 fixture events must match the brute_force rule subject" + ) + + +def _baseline_selections( + segments: Sequence[Sequence[dict[str, Any]]], + threshold: int, + window_seconds: int, +) -> tuple[CandidateWindow, ...]: + selections: list[CandidateWindow] = [] + for events in segments: + start = 0 + best_start = 0 + best_end = 0 + best_count = 0 + for end, event in enumerate(events): + end_timestamp = parse_timestamp(str(event["timestamp"])) + while ( + start < end + and ( + end_timestamp + - parse_timestamp(str(events[start]["timestamp"])) + ).total_seconds() + > window_seconds + ): + start += 1 + count = end - start + 1 + if count > best_count: + best_start, best_end, best_count = start, end, count + if best_count < threshold: + continue + selected = events[best_start : best_end + 1] + selections.append( + CandidateWindow( + event_ids=tuple(str(event["event_id"]) for event in selected), + first_seen=parse_timestamp(str(selected[0]["timestamp"])), + last_seen=parse_timestamp(str(selected[-1]["timestamp"])), + threshold_crossing_event_id=str( + selected[threshold - 1]["event_id"] + ), + ) + ) + return tuple(selections) + + +def _expected_findings( + rule: dict[str, Any], selections: Sequence[CandidateWindow] +) -> list[dict[str, Any]]: + return [ + { + "finding_id": finding_id(rule, candidate), + "episode_index": index, + "rule_id": "brute_force", + "subject_kind": "source_ip", + "subject": str(rule["subject"]), + "grouping_key": "source_ip", + "threshold": int(rule["threshold"]), + "observed_count": candidate.event_count, + "event_count": candidate.event_count, + "window_start": canonical_finding_timestamp(candidate.first_seen), + "window_end": canonical_finding_timestamp(candidate.last_seen), + "evidence_event_ids": list(candidate.event_ids), + "verdict_boundary": "triage_signal_not_compromise_or_attribution", + } + for index, candidate in enumerate(selections, start=1) + ] + + +def _contiguous_events( + event_ids: Sequence[str], segments: Sequence[Sequence[dict[str, Any]]] +) -> list[dict[str, Any]] | None: + target = list(event_ids) + for events in segments: + segment_ids = [str(event["event_id"]) for event in events] + for start in range(len(segment_ids) - len(target) + 1): + if segment_ids[start : start + len(target)] == target: + return list(events[start : start + len(target)]) + return None + + +def _validate_candidate_windows( + baseline: dict[str, Any], + segments: Sequence[Sequence[dict[str, Any]]], + selections: Sequence[CandidateWindow], + threshold: int, + window_seconds: int, +) -> None: + candidate_ids: set[str] = set() + event_sequences: set[tuple[str, ...]] = set() + declared_selected: set[tuple[str, ...]] = set() + expected_selected = {candidate.event_ids for candidate in selections} + for record in baseline["candidate_windows"]: + candidate_id = str(record.get("candidate_id", "")) + event_ids = tuple(str(item) for item in record.get("event_ids", [])) + if not candidate_id or candidate_id in candidate_ids: + raise ValueError("baseline candidate IDs must be non-empty and unique") + if not event_ids or event_ids in event_sequences: + raise ValueError("baseline candidate event sequences must be unique") + candidate_ids.add(candidate_id) + event_sequences.add(event_ids) + events = _contiguous_events(event_ids, segments) + if events is None: + raise ValueError("baseline candidate must be contiguous inside one segment") + first_seen = parse_timestamp(str(events[0]["timestamp"])) + last_seen = parse_timestamp(str(events[-1]["timestamp"])) + try: + recorded_first = parse_timestamp(str(record["first_seen"])) + recorded_last = parse_timestamp(str(record["last_seen"])) + except (KeyError, TypeError, ValueError) as error: + raise ValueError("baseline candidate timestamps are invalid") from error + if ( + len(event_ids) < threshold + or (last_seen - first_seen).total_seconds() > window_seconds + or recorded_first != first_seen + or recorded_last != last_seen + or record.get("event_count") != len(event_ids) + or record.get("threshold_crossing_event_id") + != event_ids[threshold - 1] + or record.get("threshold_met") is not True + ): + raise ValueError("baseline candidate evidence does not match fixture events") + decision = record.get("decision") + if decision == "selected" and event_ids in expected_selected: + declared_selected.add(event_ids) + elif decision != "not_selected" or event_ids in expected_selected: + raise ValueError("baseline candidate decision disagrees with v0.6 selection") + if not str(record.get("decision_reason", "")): + raise ValueError("baseline candidate decision requires a reason") + if declared_selected != expected_selected: + raise ValueError("baseline selected windows do not match v0.6 selection") + + +def validate_fixture_baseline( + fixture: dict[str, Any], baseline: dict[str, Any] +) -> BaselineContext: + _validate_fixture(fixture) + rule = fixture["rule"] + window_seconds = int(rule["window_seconds"]) + threshold = int(rule["threshold"]) + segments = activity_segments(fixture["events"], window_seconds) + if baseline.get("format") != "loglens.episode_baseline_expected.v1": + raise ValueError("unsupported baseline format") + if baseline.get("fixture_id") != fixture.get("fixture_id"): + raise ValueError("baseline fixture_id does not match the research fixture") + if baseline.get("rule") != rule_projection(rule): + raise ValueError("baseline rule does not match the research fixture") + if baseline.get("algorithm") != BASELINE_ALGORITHM: + raise ValueError("baseline algorithm contract is stale or unsupported") + + ordered = ordered_events(fixture["events"]) + derived = baseline["derived_input"] + records = derived["activity_segments"] + if ( + derived.get("event_count") != len(ordered) + or derived.get("activity_segment_count") != len(segments) + or len(records) != len(segments) + ): + raise ValueError("baseline activity segment count does not match derived input") + for index, (events, record) in enumerate(zip(segments, records), start=1): + event_ids = [str(event["event_id"]) for event in events] + gaps = [ + ( + parse_timestamp(str(events[position]["timestamp"])) + - parse_timestamp(str(events[position - 1]["timestamp"])) + ).total_seconds() + for position in range(1, len(events)) + ] + try: + recorded_first = parse_timestamp(str(record["first_seen"])) + recorded_last = parse_timestamp(str(record["last_seen"])) + except (KeyError, TypeError, ValueError) as error: + raise ValueError("baseline segment timestamps are invalid") from error + if ( + record.get("segment_id") != f"segment:{index}" + or record.get("event_ids") != event_ids + or recorded_first != parse_timestamp(str(events[0]["timestamp"])) + or recorded_last != parse_timestamp(str(events[-1]["timestamp"])) + or record.get("max_adjacent_gap_seconds") != int(max(gaps, default=0)) + ): + raise ValueError("baseline derived segment does not match fixture events") + + selections = _baseline_selections(segments, threshold, window_seconds) + _validate_candidate_windows( + baseline, segments, selections, threshold, window_seconds + ) + expected_output = baseline["expected_output"] + expected_findings = _expected_findings(rule, selections) + if ( + expected_output.get("episode_count") != len(expected_findings) + or expected_output.get("findings") != expected_findings + ): + raise ValueError("baseline findings do not match the v0.6 selection") + selected_ids = { + event_id for selection in selections for event_id in selection.event_ids + } + excluded_ids = [ + str(event["event_id"]) + for event in ordered + if str(event["event_id"]) not in selected_ids + ] + decisions = expected_output.get("excluded_event_decisions", []) + if [item.get("event_id") for item in decisions] != excluded_ids or any( + item.get("decision") != "excluded" or not str(item.get("reason_code", "")) + for item in decisions + ): + raise ValueError("baseline excluded decisions do not partition fixture events") + return BaselineContext( + segments=segments, + selections=selections, + episode_count=len(expected_findings), + ) diff --git a/tests/test_episode_candidate_baseline_contract.py b/tests/test_episode_candidate_baseline_contract.py new file mode 100644 index 0000000..4aec78f --- /dev/null +++ b/tests/test_episode_candidate_baseline_contract.py @@ -0,0 +1,136 @@ +import copy +import json +import sys +import unittest +from datetime import timedelta, timezone +from pathlib import Path + + +REPO_ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(REPO_ROOT)) + +from scripts.episode_baseline_contract import ( # noqa: E402 + canonical_oracle_timestamp, + finding_id, + validate_fixture_baseline, +) +from scripts.episode_candidate_core import parse_timestamp # noqa: E402 + + +FIXTURE_ROOT = ( + REPO_ROOT + / "tests" + / "fixtures" + / "episode_semantics_v0.7" + / "continuous_background_two_peaks" +) + + +class EpisodeBaselineContractTests(unittest.TestCase): + def setUp(self) -> None: + self.fixture = json.loads( + (FIXTURE_ROOT / "fixture.json").read_text(encoding="utf-8") + ) + self.baseline = json.loads( + (FIXTURE_ROOT / "baseline.expected.json").read_text(encoding="utf-8") + ) + + def test_committed_baseline_replays_v06_selection_and_identity(self) -> None: + context = validate_fixture_baseline(self.fixture, self.baseline) + + self.assertEqual(context.episode_count, 1) + self.assertEqual(len(context.selections), 1) + self.assertEqual( + finding_id(self.fixture["rule"], context.selections[0]), + "finding:brute_force:584fd14b544a7959", + ) + + def test_equivalent_timezone_offsets_preserve_baseline_identity(self) -> None: + offset_fixture = copy.deepcopy(self.fixture) + offset = timezone(timedelta(hours=1)) + for event in offset_fixture["events"]: + event["timestamp"] = parse_timestamp(event["timestamp"]).astimezone( + offset + ).isoformat() + + original = validate_fixture_baseline(self.fixture, self.baseline) + shifted = validate_fixture_baseline(offset_fixture, self.baseline) + + self.assertEqual( + [canonical_oracle_timestamp(item.first_seen) for item in shifted.selections], + [canonical_oracle_timestamp(item.first_seen) for item in original.selections], + ) + self.assertEqual( + [finding_id(offset_fixture["rule"], item) for item in shifted.selections], + [finding_id(self.fixture["rule"], item) for item in original.selections], + ) + + def test_unsupported_rule_contracts_fail_closed(self) -> None: + mutations = { + "rule": ("rule_id", "multi_user_probing"), + "grouping": ("grouping_key", "username"), + "boundary": ("window_boundary", "exclusive"), + "signal": ("signal_kind", "sudo_command"), + } + + for label, (field, value) in mutations.items(): + with self.subTest(label=label): + fixture = copy.deepcopy(self.fixture) + baseline = copy.deepcopy(self.baseline) + fixture["rule"][field] = value + if field in baseline["rule"]: + baseline["rule"][field] = value + with self.assertRaisesRegex(ValueError, "candidate v1 supports only"): + validate_fixture_baseline(fixture, baseline) + + def test_fixture_events_must_match_the_rule_subject(self) -> None: + fixture = copy.deepcopy(self.fixture) + fixture["events"][0]["source_ip"] = "203.0.113.99" + + with self.assertRaisesRegex(ValueError, "fixture events"): + validate_fixture_baseline(fixture, self.baseline) + + def test_rule_scalar_types_fail_closed(self) -> None: + mutations = { + "string threshold": ("threshold", "5"), + "string window": ("window_seconds", "600"), + "numeric subject": ("subject", 77), + "numeric terminal flag": ("counts_as_terminal_auth_failure", 1), + } + + for label, (field, value) in mutations.items(): + with self.subTest(label=label): + fixture = copy.deepcopy(self.fixture) + baseline = copy.deepcopy(self.baseline) + fixture["rule"][field] = value + if field in baseline["rule"]: + baseline["rule"][field] = value + with self.assertRaisesRegex(ValueError, "canonical scalar types"): + validate_fixture_baseline(fixture, baseline) + + def test_semantically_stale_baseline_fails_closed(self) -> None: + mutations = { + "algorithm": lambda baseline: baseline["algorithm"].__setitem__( + "id", "stale.algorithm" + ), + "finding identity": lambda baseline: baseline["expected_output"][ + "findings" + ][0].__setitem__("finding_id", "finding:brute_force:stale"), + "selected window": lambda baseline: baseline["candidate_windows"][ + 0 + ].__setitem__("last_seen", "2026-03-10T09:01:30Z"), + "segment evidence": lambda baseline: baseline["derived_input"][ + "activity_segments" + ][0].__setitem__("max_adjacent_gap_seconds", 539), + } + + for label, mutate in mutations.items(): + with self.subTest(label=label): + baseline = copy.deepcopy(self.baseline) + mutate(baseline) + with self.assertRaisesRegex(ValueError, "baseline"): + validate_fixture_baseline(self.fixture, baseline) + + +if __name__ == "__main__": + unittest.main()