diff --git a/.gitattributes b/.gitattributes index 5214164..53e915e 100644 --- a/.gitattributes +++ b/.gitattributes @@ -6,3 +6,4 @@ /demos/**/data/raw/** text eol=lf /demos/**/artifacts/run_manifest.json text eol=lf /demos/**/artifacts/*.md text eol=lf +/tests/fixtures/process_evidence/telemetry_events.jsonl text eol=lf diff --git a/docs/case-study-process-evidence-to-telemetry.md b/docs/case-study-process-evidence-to-telemetry.md index 50e6625..3f1124c 100644 --- a/docs/case-study-process-evidence-to-telemetry.md +++ b/docs/case-study-process-evidence-to-telemetry.md @@ -102,3 +102,7 @@ The pinned implementations, synthetic fixtures and producer goldens are public. - [Consumer event-time model](https://github.com/stacknil/telemetry-lab/blob/98ae3ffd1edb47d90eb42b3a8625be830582997e/docs/event-time-model.md) *Prepared from the retained 19 October 2026 integration evidence. No engineering tests or demos were rerun for the documentation review.* + +Later work: the [20 October executable integration suite](process-evidence-integration.md) +adds offline regression tests and a separate producer-to-consumer reproduction runner. +Its tests and fresh-environment results are separate from the historical observations above. diff --git a/docs/process-evidence-integration.md b/docs/process-evidence-integration.md new file mode 100644 index 0000000..15edd67 --- /dev/null +++ b/docs/process-evidence-integration.md @@ -0,0 +1,119 @@ +# Executable process-evidence integration + +This suite checks a saved-evidence producer against the real telemetry consumer. +It adds executable regression evidence after the [19 October case study](case-study-process-evidence-to-telemetry.md); +it does not rerun or replace that historical experiment's private receipts. + +## Versions and fixture + +- Producer: [systems-foundations](https://github.com/stacknil/systems-foundations), + pinned at `1d31d4287036cf2bbd3aeb8059d7d332bedef598`, + `projects/linux-process-observe`. +- Consumer: this suite is based on `de542cb3cadba776c8e759f786152f9222fdfdb7`. + Reproduce from the reviewed integration revision containing this guide. + The historical case used `98ae3ffd1edb47d90eb42b3a8625be830582997e`. +- Offline input: [`tests/fixtures/process_evidence/telemetry_events.jsonl`](../tests/fixtures/process_evidence/telemetry_events.jsonl), + an unmodified, public synthetic producer golden, with [provenance](../tests/fixtures/process_evidence/provenance.json). + It is 4,108 LF bytes, SHA-256 + `1977b98518497c9a07da6393499cb49e6d0be9140ac5bcd708c2a1cd31a3ad51`. + Its scoped `.gitattributes` entry prevents checkout newline conversion. + +## Layer A: offline regression gate + +Install this checkout's development dependencies in a separate Python 3.11+ +virtual environment, then run: + +```sh +python -m pip install -e ".[dev]" +python -m pip check +python -m pytest tests/test_process_evidence_integration.py -v +``` + +The tests execute the actual loader, normalization, windows, features, rules and +CLI. They never fetch the producer, access private receipts or mock the consumer. +The existing ordinary `pytest` CI job includes this layer without extra network +access during test execution. + +## Layer B: actual producer commands and two consumer runs + +Git and the same environment's development dependencies are required. Obtain +the pinned public producer before running the workflow. From the consumer root: + +```sh +git clone https://github.com/stacknil/systems-foundations.git ../process-evidence-producer +git -C ../process-evidence-producer checkout --detach 1d31d4287036cf2bbd3aeb8059d7d332bedef598 +python scripts/reproduce_process_evidence.py --producer ../process-evidence-producer --output-root ../process-evidence-run +``` + +Use a **new** output directory outside both repositories on every invocation. +The runner refuses collisions and performs no cleanup. It reads immutable Git +objects, exports only the pinned project with Git newline conversion disabled, +and executes the exported source. Dirty files in the supplied producer checkout +cannot select the executed source. The producer uses only the standard library; +its exported `src` is selected explicitly for each subprocess. No producer fetch, +package download, live procfs read or network operation occurs inside the runner. + +The executable sequence is the following, with `PROJECT` referring to the +exported pinned project and `OUT` to the fresh output directory: + +```sh +python -m linux_process_observe snapshot --proc-root "$PROJECT/tests/fixtures/baseline/proc" --ss "$PROJECT/tests/fixtures/baseline/ss.txt" --host-id lab-host --observed-at 2026-07-05T00:00:00Z --output-dir "$OUT/baseline" +python -m linux_process_observe snapshot --proc-root "$PROJECT/tests/fixtures/changed/proc" --ss "$PROJECT/tests/fixtures/changed/ss.txt" --host-id lab-host --observed-at 2026-07-05T00:05:00Z --output-dir "$OUT/changed" +python -m linux_process_observe diff --before-processes "$OUT/baseline/process_snapshot.json" --after-processes "$OUT/changed/process_snapshot.json" --before-links "$OUT/baseline/process_socket_links.json" --after-links "$OUT/changed/process_socket_links.json" --output-dir "$OUT/diff" +python -m linux_process_observe adapt --input "$OUT/diff/process_diff.json" --output "$OUT/diff/telemetry_events.jsonl" +python -m telemetry_lab.cli run window --config "$OUT/one/integration.yaml" +python -m telemetry_lab.cli run window --config "$OUT/two/integration.yaml" +``` + +The runner supplies each module's source path and starts commands from the +disposable directory. Each consumer config is a fresh copy of `configs/default.yaml` +with only `input_path` and `output_dir` changed. `integration_receipt.json` records +actual argv, exit codes, versions and comparisons. Keep generated artifacts outside +Git; paths in local receipts may reveal local directory names. + +## Expected results and boundaries + +| Check | Expected contract | +| --- | --- | +| Producer | Four successful commands; five JSON objects and report text match pinned golden content; JSONL matches exact bytes and parsed fixture records | +| Loading and normalization | Seven rows; five required fields; source, target and nested metadata retained | +| Time | All rows at `2026-07-05T00:05:00Z`; one half-open `[00:05:00Z, 00:06:00Z)` window | +| Features | Eleven numeric fields; seven events, four sources, seven targets; error, severity and five configured category counts all zero | +| Rules | Zero alerts with unchanged default rules, thresholds and cooldown | +| Outputs | Seven artifacts per consumer run; both summary/v1 and manifest/v2 pass their schemas with format checks | +| Repetition | Feature/alert CSV bytes and selected summary/manifest fields stable; config fingerprints legitimately differ because output paths differ | +| Plots | Three PNGs per run decode with nonzero dimensions; no cross-platform byte equality claim | + +`metadata.time_semantics = snapshot_diff_observed_at` describes comparison +observation time. Window membership uses `timestamp`; the consumer does not +infer occurrence time or validate this advisory marker. `source` has the producer +form `host_id:pid:start_time_ticks`, bounded to same-boot evidence without a boot +identifier. Preservation proves neither cross-reboot identity nor authentication. +Executable and endpoint targets remain context, not independent attribution. + +Negative and metamorphic controls cover missing required columns/values, +malformed JSONL and timestamps, changed comparison time, equivalent UTC offsets, +unknown categories, status changes across the existing error threshold, missing +or altered advisory metadata, duplicate timestamps, row ordering, half-open +boundaries and changed raw-byte fingerprints. Unknown categories and numeric +identifiers remain accepted where existing normalization supports them. Metadata +changes can leave aggregation unchanged; duplicate records are counted again. + +Per-file manifest fingerprints identify exact input/config bytes. Legacy aggregate +digests separately include labels and normalized text. LF/CRLF changes therefore +change the exact-byte fingerprint while preserving the legacy digest and features. +These values provide content identification, not authenticated provenance. + +## Architecture and limitations + +Layer B stays an explicit standalone workflow. Adding a second repository checkout +to required PR CI would introduce a network dependency without improving the +offline consumer gate; using mutable producer `main` would weaken reproducibility. +The ordinary CI and all release/publication workflows remain unchanged. + +No consumer admission policy, default detection rule, schema, golden demo output +or producer implementation changes. This suite checks structural and bounded +semantic compatibility and reproducibility, not detection efficacy or production +readiness. It adds no collector, new demo, security attribution or private evidence. +Dependency versions are reported rather than locked; other supported environments +must run these checks rather than infer equivalent PNG bytes or performance. diff --git a/scripts/reproduce_process_evidence.py b/scripts/reproduce_process_evidence.py new file mode 100644 index 0000000..802e0ab --- /dev/null +++ b/scripts/reproduce_process_evidence.py @@ -0,0 +1,182 @@ +"""Execute pinned saved-evidence producer commands and the real consumer twice.""" +from __future__ import annotations + +import argparse +from hashlib import sha256 +from importlib.metadata import version +import io +import json +import os +from pathlib import Path, PurePosixPath +import platform +import subprocess +import sys +import tarfile + +from jsonschema import Draft202012Validator, FormatChecker +from PIL import Image +import yaml + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT / "src")) +from telemetry_lab.io import load_alert_table, load_config, load_feature_table # noqa: E402 + +PIN = "1d31d4287036cf2bbd3aeb8059d7d332bedef598" +PROJECT = "projects/linux-process-observe" +DIGEST = "1977b98518497c9a07da6393499cb49e6d0be9140ac5bcd708c2a1cd31a3ad51" +EXPECTED = dict(event_count=7, error_count=0, error_rate=0.0, unique_sources=4, + unique_targets=7, high_severity_count=0, login_fail_count=0, + login_success_count=0, config_change_count=0, malware_alert_count=0, + policy_denied_count=0) + + +def require(condition, message): + if not condition: + raise ValueError(message) + + +def command(args, *, cwd, env=None): + result = subprocess.run([str(arg) for arg in args], cwd=cwd, env=env, + capture_output=True, timeout=120) + if result.returncode: + raise ValueError("Command failed: " + result.stderr.decode("utf-8", errors="replace")) + return result.stdout + + +def export_producer(repository, destination): + # Read immutable Git objects, never execute or overwrite the caller's checkout. + actual = command(["git", "rev-parse", PIN + "^{commit}"], cwd=repository).decode().strip() + require(actual == PIN, "Pinned producer commit is unavailable") + data = command(["git", "-c", "core.autocrlf=false", "-c", "core.eol=lf", "archive", "--format=tar", PIN, PROJECT], cwd=repository) + with tarfile.open(fileobj=io.BytesIO(data)) as archive: + for member in archive: + path = PurePosixPath(member.name) + require(not path.is_absolute() and ".." not in path.parts and "\\" not in member.name and ":" not in member.name, + "Unsafe producer archive path") + # git archive includes the requested project's ancestor directory. + if member.isdir() and path.parts == ("projects",): + continue + require(path.parts[:2] == ("projects", "linux-process-observe"), "Unexpected producer archive scope") + if member.isdir(): + continue + require(member.isfile(), "Producer archive must contain regular files") + output = destination.joinpath(*path.parts) + output.parent.mkdir(parents=True, exist_ok=True) + with archive.extractfile(member) as source: + output.write_bytes(source.read()) + return destination / PROJECT + + +def check_result(output, input_path, config_path): + summary = json.loads((output / "summary.json").read_text(encoding="utf-8")) + manifest = json.loads((output / "run_manifest.json").read_text(encoding="utf-8")) + for value, schema in ((summary, "telemetry_summary.schema.json"), (manifest, "run_manifest.v2.schema.json")): + contract = json.loads((ROOT / "schemas" / schema).read_text(encoding="utf-8")) + Draft202012Validator.check_schema(contract) + Draft202012Validator(contract, format_checker=FormatChecker()).validate(value) + counts = {key: summary[key] for key in ("normalized_event_count", "window_count", "feature_row_count", "alert_count")} + require(counts == dict(normalized_event_count=7, window_count=1, feature_row_count=1, alert_count=0), "Unexpected summary counts") + features = load_feature_table(output / "features.csv") + require(set(features.columns) == {"window_start", "window_end", *EXPECTED}, "Unexpected feature columns") + require(len(features) == 1 and features.iloc[0][list(EXPECTED)].to_dict() == EXPECTED, "Unexpected aggregate features") + row = features.iloc[0] + require(row.window_start.isoformat() == "2026-07-05T00:05:00+00:00" and row.window_end.isoformat() == "2026-07-05T00:06:00+00:00", "Unexpected observation window") + require(load_alert_table(output / "alerts.csv").empty, "Default rules unexpectedly alerted") + require(manifest["artifact_schema_versions"] == {"run_manifest": "run-manifest/v2", "telemetry_summary": "telemetry-summary/v1"}, "Unexpected artifact versions") + for kind, path in (("input", input_path), ("config", config_path)): + require(manifest[kind + "_file_digests"] == {f"external/{kind}/{path.name}": "sha256:" + sha256(path.read_bytes()).hexdigest()}, "Incorrect exact-byte fingerprint") + normalized = path.read_text(encoding="utf-8").encode("utf-8") + legacy = sha256(path.name.encode() + b"\0" + normalized + b"\0").hexdigest() + require(manifest[kind + "_digest"] == "sha256:" + legacy, "Incorrect legacy digest") + names = {Path(name).name for name in summary["generated_artifacts"]} + expected_names = {"features.csv", "alerts.csv", "summary.json", "run_manifest.json", + "event_count_timeline.png", "error_rate_timeline.png", "alerts_timeline.png"} + require(names == {path.name for path in output.iterdir()} == expected_names, "Missing or unexpected output artifact") + for name in names: + if name.endswith(".png"): + with Image.open(output / name) as image: + require(image.format == "PNG" and min(image.size) > 0, "Invalid plot") + image.verify() + return summary, manifest + + +def reproduce(repository, output): + repository, output = repository.resolve(), output.resolve() + require(not output.is_relative_to(ROOT) and not output.is_relative_to(repository), "Outputs must be outside both repositories") + output.mkdir(parents=True, exist_ok=False) # refuse collisions; never remove existing outputs + producer = export_producer(repository, output / "pinned-source") + env = dict(os.environ, PYTHONDONTWRITEBYTECODE="1", PYTHONNOUSERSITE="1", MPLBACKEND="Agg", + MPLCONFIGDIR=str(output / "plot-config")) + receipts = [] + def execute(module, args, source): + invocation = [sys.executable, "-m", module, *map(str, args)] + stdout = command(invocation, cwd=output, env=dict(env, PYTHONPATH=str(source / "src"))) + receipts.append({"argv": invocation, "exit_code": 0, "stdout": stdout.decode("utf-8", errors="replace")}) + for name, instant in (("baseline", "2026-07-05T00:00:00Z"), ("changed", "2026-07-05T00:05:00Z")): + fixture = producer / "tests/fixtures" / name + execute("linux_process_observe", ["snapshot", "--proc-root", fixture / "proc", "--ss", fixture / "ss.txt", + "--host-id", "lab-host", "--observed-at", instant, "--output-dir", output / name], producer) + execute("linux_process_observe", ["diff", "--before-processes", output / "baseline/process_snapshot.json", + "--after-processes", output / "changed/process_snapshot.json", "--before-links", output / "baseline/process_socket_links.json", + "--after-links", output / "changed/process_socket_links.json", "--output-dir", output / "diff"], producer) + input_path = output / "diff/telemetry_events.jsonl" + execute("linux_process_observe", ["adapt", "--input", output / "diff/process_diff.json", "--output", input_path], producer) + producer_hashes = {} + for folder, names in (("baseline", ("process_snapshot.json", "process_socket_links.json")), + ("changed", ("process_snapshot.json", "process_socket_links.json")), + ("diff", ("process_diff.json", "report.md", "telemetry_events.jsonl"))): + for name in names: + data = (output / folder / name).read_bytes() + golden = (producer / "tests/golden" / folder / name).read_bytes() + # Compare structured contracts; only the JSONL contract is byte-exact. + if name.endswith(".json"): + equal = json.loads(data) == json.loads(golden) + elif name.endswith(".md"): + equal = data.decode().replace("\r\n", "\n") == golden.decode().replace("\r\n", "\n") + else: + equal = data == golden + require(equal, "Producer output differs from pinned golden: " + folder + "/" + name) + producer_hashes[folder + "/" + name] = sha256(data).hexdigest() + require(producer_hashes["diff/telemetry_events.jsonl"] == DIGEST, "Producer JSONL digest changed") + generated = [json.loads(line) for line in input_path.read_text(encoding="utf-8").splitlines()] + canonical = [json.loads(line) for line in (ROOT / "tests/fixtures/process_evidence/telemetry_events.jsonl").read_text(encoding="utf-8").splitlines()] + require(generated == canonical, "Generated producer records differ semantically from the consumer fixture") + results = [] + for name in ("one", "two"): + directory = output / name + directory.mkdir() + config = load_config(ROOT / "configs/default.yaml") + config.update(input_path=str(input_path), output_dir=str(directory / "artifacts")) + path = directory / "integration.yaml" + path.write_text(yaml.safe_dump(config, sort_keys=False), encoding="utf-8", newline="\n") + execute("telemetry_lab.cli", ["run", "window", "--config", path], ROOT) + results.append(check_result(directory / "artifacts", input_path, path)) + for name in ("features.csv", "alerts.csv"): + require((output / "one/artifacts" / name).read_bytes() == (output / "two/artifacts" / name).read_bytes(), "Structured output is unstable") + stable_summary = ("normalized_event_count", "window_count", "feature_row_count", "alert_count", "triggered_rule_names", "triggered_rule_counts", "cooldown_seconds") + require({key: results[0][0][key] for key in stable_summary} == {key: results[1][0][key] for key in stable_summary}, "Summary metrics are unstable") + for key in ("tool_version", "demo_id", "input_digest", "input_file_digests", "artifact_schema_versions", "execution_mode"): + require(results[0][1][key] == results[1][1][key], "Manifest input contract is unstable") + report = dict(producer_commit=PIN, consumer_base="de542cb3cadba776c8e759f786152f9222fdfdb7", + python=platform.python_version(), platform=platform.system(), + packages={name: version(name) for name in ("telemetry-lab", "pandas", "matplotlib", "PyYAML", "jsonschema")}, + commands=receipts, producer_output_sha256=producer_hashes, schema_validations=4, + producer_comparison="five JSON objects; report text with normalized newlines; JSONL exact bytes and parsed records", + structured_outputs="stable; output paths/config fingerprints differ by design", png="six decoded PNG smoke checks; no cross-platform byte claim") + (output / "integration_receipt.json").write_text(json.dumps(report, indent=2) + "\n", encoding="utf-8", newline="\n") + print("PASS: four producer commands, seven golden contracts, two consumer runs, four schema validations") + + +def main(argv=None): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--producer", type=Path, required=True, help="local systems-foundations Git repository containing the exact pin") + parser.add_argument("--output-root", type=Path, required=True, help="new disposable directory outside both repositories") + args = parser.parse_args(argv) + try: + reproduce(args.producer, args.output_root) + except (OSError, ValueError, subprocess.SubprocessError) as error: + parser.exit(1, f"FAIL: {error}\n") + + +if __name__ == "__main__": + main() diff --git a/tests/fixtures/process_evidence/provenance.json b/tests/fixtures/process_evidence/provenance.json new file mode 100644 index 0000000..af95174 --- /dev/null +++ b/tests/fixtures/process_evidence/provenance.json @@ -0,0 +1,8 @@ +{ + "repository": "https://github.com/stacknil/systems-foundations", + "commit": "1d31d4287036cf2bbd3aeb8059d7d332bedef598", + "path": "projects/linux-process-observe/tests/golden/diff/telemetry_events.jsonl", + "sha256": "1977b98518497c9a07da6393499cb49e6d0be9140ac5bcd708c2a1cd31a3ad51", + "bytes": 4108, + "description": "Unmodified public synthetic producer golden; canonical Git blob with LF newlines, not a private execution receipt." +} diff --git a/tests/fixtures/process_evidence/telemetry_events.jsonl b/tests/fixtures/process_evidence/telemetry_events.jsonl new file mode 100644 index 0000000..0870bc2 --- /dev/null +++ b/tests/fixtures/process_evidence/telemetry_events.jsonl @@ -0,0 +1,7 @@ +{"event_type":"process_added","metadata":{"adapter_contract":"stacknil.system-evidence.telemetry.v1","change_type":"added","changes":{},"evidence_schema":"stacknil.system-evidence.v1","evidence_source":"procfs+ss","host_id":"lab-host","identity":"lab-host:330:3300","process_id":"lab-host:330:3300","record_index":1,"record_type":"process_change","time_semantics":"snapshot_diff_observed_at"},"source":"lab-host:330:3300","status":"added","target":"/usr/bin/python3.11","timestamp":"2026-07-05T00:05:00Z"} +{"event_type":"process_removed","metadata":{"adapter_contract":"stacknil.system-evidence.telemetry.v1","change_type":"removed","changes":{},"evidence_schema":"stacknil.system-evidence.v1","evidence_source":"procfs+ss","host_id":"lab-host","identity":"lab-host:220:2200","process_id":"lab-host:220:2200","record_index":2,"record_type":"process_change","time_semantics":"snapshot_diff_observed_at"},"source":"lab-host:220:2200","status":"removed","target":"/opt/example/bin/app-worker","timestamp":"2026-07-05T00:05:00Z"} +{"event_type":"process_modified","metadata":{"adapter_contract":"stacknil.system-evidence.telemetry.v1","change_type":"modified","changes":{"argv":{"after":["/usr/bin/dash","-c","sleep 30"],"before":["/opt/example/bin/task-runner","--once"]},"executable":{"after":"/usr/bin/dash","before":"/opt/example/bin/task-runner"},"name":{"after":"sh","before":"task-runner"}},"evidence_schema":"stacknil.system-evidence.v1","evidence_source":"procfs+ss","host_id":"lab-host","identity":"lab-host:230:2300","process_id":"lab-host:230:2300","record_index":3,"record_type":"process_change","time_semantics":"snapshot_diff_observed_at"},"source":"lab-host:230:2300","status":"modified","target":"/usr/bin/dash","timestamp":"2026-07-05T00:05:00Z"} +{"event_type":"socket_link_added","metadata":{"adapter_contract":"stacknil.system-evidence.telemetry.v1","change_type":"added","changes":{},"evidence_schema":"stacknil.system-evidence.v1","evidence_source":"procfs+ss","host_id":"lab-host","identity":"lab-host:120:1200|tcp|ESTAB|192.0.2.10:22|198.51.100.21:51000","process_id":"lab-host:120:1200","record_index":4,"record_type":"process_socket_link_change","time_semantics":"snapshot_diff_observed_at"},"source":"lab-host:120:1200","status":"added","target":"tcp ESTAB 192.0.2.10:22 -> 198.51.100.21:51000","timestamp":"2026-07-05T00:05:00Z"} +{"event_type":"socket_link_added","metadata":{"adapter_contract":"stacknil.system-evidence.telemetry.v1","change_type":"added","changes":{},"evidence_schema":"stacknil.system-evidence.v1","evidence_source":"procfs+ss","host_id":"lab-host","identity":"lab-host:330:3300|tcp|LISTEN|0.0.0.0:8080|0.0.0.0:*","process_id":"lab-host:330:3300","record_index":5,"record_type":"process_socket_link_change","time_semantics":"snapshot_diff_observed_at"},"source":"lab-host:330:3300","status":"added","target":"tcp LISTEN 0.0.0.0:8080 -> 0.0.0.0:*","timestamp":"2026-07-05T00:05:00Z"} +{"event_type":"socket_link_removed","metadata":{"adapter_contract":"stacknil.system-evidence.telemetry.v1","change_type":"removed","changes":{},"evidence_schema":"stacknil.system-evidence.v1","evidence_source":"procfs+ss","host_id":"lab-host","identity":"lab-host:120:1200|tcp|ESTAB|192.0.2.10:22|198.51.100.20:50000","process_id":"lab-host:120:1200","record_index":6,"record_type":"process_socket_link_change","time_semantics":"snapshot_diff_observed_at"},"source":"lab-host:120:1200","status":"removed","target":"tcp ESTAB 192.0.2.10:22 -> 198.51.100.20:50000","timestamp":"2026-07-05T00:05:00Z"} +{"event_type":"socket_link_removed","metadata":{"adapter_contract":"stacknil.system-evidence.telemetry.v1","change_type":"removed","changes":{},"evidence_schema":"stacknil.system-evidence.v1","evidence_source":"procfs+ss","host_id":"lab-host","identity":"lab-host:220:2200|tcp|LISTEN|127.0.0.1:9000|0.0.0.0:*","process_id":"lab-host:220:2200","record_index":7,"record_type":"process_socket_link_change","time_semantics":"snapshot_diff_observed_at"},"source":"lab-host:220:2200","status":"removed","target":"tcp LISTEN 127.0.0.1:9000 -> 0.0.0.0:*","timestamp":"2026-07-05T00:05:00Z"} diff --git a/tests/test_process_evidence_integration.py b/tests/test_process_evidence_integration.py new file mode 100644 index 0000000..d48bb3d --- /dev/null +++ b/tests/test_process_evidence_integration.py @@ -0,0 +1,243 @@ +"""Offline contract for the pinned public process-evidence producer, not a detector verdict.""" +from __future__ import annotations + +from collections import Counter +from hashlib import sha256 +import json +from pathlib import Path + +from jsonschema import Draft202012Validator, FormatChecker +import pandas as pd +import pytest +import yaml + +from telemetry_lab.cli import main +from telemetry_lab.features import compute_window_features +from telemetry_lab.io import load_alert_table, load_config, load_events, load_feature_table +from telemetry_lab.preprocess import normalize_events +from telemetry_lab.rules import apply_rules +from telemetry_lab.windowing import build_windows + +ROOT = Path(__file__).resolve().parents[1] +FIXTURE = ROOT / "tests/fixtures/process_evidence/telemetry_events.jsonl" +EXPECTED_DIGEST = "1977b98518497c9a07da6393499cb49e6d0be9140ac5bcd708c2a1cd31a3ad51" +FEATURES = dict(event_count=7, error_count=0, error_rate=0.0, unique_sources=4, + unique_targets=7, high_severity_count=0, login_fail_count=0, + login_success_count=0, config_change_count=0, malware_alert_count=0, + policy_denied_count=0) + + +def records(): + return [json.loads(line) for line in FIXTURE.read_text(encoding="utf-8").splitlines()] + + +def write_records(tmp_path, rows): + path = tmp_path / "telemetry_events.jsonl" + path.write_text("".join(json.dumps(row, sort_keys=True, separators=(",", ":")) + "\n" for row in rows), + encoding="utf-8", newline="\n") + return path + + +def pipeline(path): + cfg = load_config(ROOT / "configs/default.yaml") + loaded = load_events(path) + normalized = normalize_events(loaded, error_statuses=cfg["features"]["error_statuses"], + high_severity_levels=cfg["features"]["severity_levels"]) + windows = build_windows(normalized, **cfg["time"]) + features = compute_window_features(normalized, windows, cfg["features"]["count_event_types"]) + return loaded, normalized, features, apply_rules(features, cfg["rules"]) + + +def run_cli(tmp_path, data): + tmp_path.mkdir(parents=True, exist_ok=True) + input_path = tmp_path / "telemetry_events.jsonl" + input_path.write_bytes(data) + cfg = load_config(ROOT / "configs/default.yaml") + cfg.update(input_path=str(input_path), output_dir=str(tmp_path / "output")) + config_path = tmp_path / "integration.yaml" + config_path.write_text(yaml.safe_dump(cfg, sort_keys=False), encoding="utf-8", newline="\n") + main(["run", "window", "--config", str(config_path)]) + summary = json.loads((tmp_path / "output/summary.json").read_text(encoding="utf-8")) + manifest = json.loads((tmp_path / "output/run_manifest.json").read_text(encoding="utf-8")) + return summary, manifest, input_path, config_path + + +def legacy_digest(path): + normalized = path.read_bytes().decode("utf-8").replace("\r\n", "\n").replace("\r", "\n").encode() + return "sha256:" + sha256(path.name.encode() + b"\0" + normalized + b"\0").hexdigest() + + +def test_pinned_fixture_and_normalized_observation_contract(): + provenance = json.loads(FIXTURE.with_name("provenance.json").read_text(encoding="utf-8")) + assert provenance["commit"] == "1d31d4287036cf2bbd3aeb8059d7d332bedef598" + assert provenance["sha256"] == sha256(FIXTURE.read_bytes()).hexdigest() == EXPECTED_DIGEST + assert provenance["bytes"] == len(FIXTURE.read_bytes()) == 4108 + expected = records() + assert all({"timestamp", "event_type", "source", "target", "status"} <= row.keys() for row in expected) + loaded, normalized, features, alerts = pipeline(FIXTURE) + assert len(loaded) == len(normalized) == 7 + assert Counter(zip(normalized.event_type, normalized.status)) == { + ("process_added", "added"): 1, ("process_removed", "removed"): 1, + ("process_modified", "modified"): 1, ("socket_link_added", "added"): 2, + ("socket_link_removed", "removed"): 2, + } + # Equal timestamps do not establish a tie order. Compare evidence by record identity. + actual = {row.metadata["record_index"]: row for row in normalized.itertuples()} + for event in expected: + row = actual[event["metadata"]["record_index"]] + assert (row.source, row.target, row.metadata) == (event["source"], event["target"], event["metadata"]) + assert row.source == row.metadata["process_id"] # same-boot context, not authentication + assert row.metadata["time_semantics"] == "snapshot_diff_observed_at" + assert row.metadata["adapter_contract"] == "stacknil.system-evidence.telemetry.v1" + assert normalized.timestamp.eq(pd.Timestamp("2026-07-05T00:05:00Z")).all() + assert set(features.columns) == {"window_start", "window_end", *FEATURES} + assert len(features) == 1 and features.iloc[0][list(FEATURES)].to_dict() == FEATURES + assert features.iloc[0].window_start == pd.Timestamp("2026-07-05T00:05:00Z") + assert features.iloc[0].window_end == pd.Timestamp("2026-07-05T00:06:00Z") + assert alerts.empty + + +def test_real_cli_outputs_schemas_and_exact_fingerprint_attribution(tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) # paths and schema lookup must not depend on repository cwd + summary, manifest, input_path, config_path = run_cli(tmp_path, FIXTURE.read_bytes()) + assert {key: summary[key] for key in ("normalized_event_count", "window_count", "feature_row_count", "alert_count")} == { + "normalized_event_count": 7, "window_count": 1, "feature_row_count": 1, "alert_count": 0} + assert summary["triggered_rule_names"] == [] and summary["triggered_rule_counts"] == {} + names = {Path(path).name for path in summary["generated_artifacts"]} + assert names == {"features.csv", "alerts.csv", "summary.json", "run_manifest.json", + "event_count_timeline.png", "error_rate_timeline.png", "alerts_timeline.png"} + for artifact, schema in ((summary, "telemetry_summary.schema.json"), (manifest, "run_manifest.v2.schema.json")): + validator = Draft202012Validator(json.loads((ROOT / "schemas" / schema).read_text(encoding="utf-8")), format_checker=FormatChecker()) + validator.validate(artifact) + assert manifest["artifact_schema_versions"] == {"run_manifest": "run-manifest/v2", "telemetry_summary": "telemetry-summary/v1"} + assert manifest["execution_mode"] == "synthetic-local" + assert manifest["input_file_digests"] == {"external/input/telemetry_events.jsonl": "sha256:" + EXPECTED_DIGEST} + assert manifest["config_file_digests"] == {"external/config/integration.yaml": "sha256:" + sha256(config_path.read_bytes()).hexdigest()} + assert manifest["input_digest"] == legacy_digest(input_path) + assert manifest["config_digest"] == legacy_digest(config_path) + features = load_feature_table(tmp_path / "output/features.csv") + assert set(features.columns) == {"window_start", "window_end", *FEATURES} + assert features.iloc[0][list(FEATURES)].to_dict() == FEATURES + assert load_alert_table(tmp_path / "output/alerts.csv").empty + + +@pytest.mark.parametrize("field", ["timestamp", "event_type", "source", "target", "status"]) +def test_missing_required_column_is_rejected(tmp_path, field): + rows = records() + for row in rows: + del row[field] + with pytest.raises(ValueError, match="Missing required event fields"): + load_events(write_records(tmp_path, rows)) + + +@pytest.mark.parametrize("value", [None, "", " "]) +def test_missing_required_row_value_is_rejected(tmp_path, value): + rows = records() + rows[0]["source"] = value + with pytest.raises(ValueError, match="Missing required event values"): + load_events(write_records(tmp_path, rows)) + + +def test_malformed_timestamp_fails_normalization_not_structural_loading(tmp_path): + rows = records() + rows[0]["timestamp"] = "not-a-timestamp" + loaded = load_events(write_records(tmp_path, rows)) + assert len(loaded) == 7 + with pytest.raises(ValueError, match="invalid timestamps"): + normalize_events(loaded) + + +@pytest.mark.parametrize("line", ['{"timestamp":', "[]"]) +def test_bad_jsonl_record_is_rejected_with_line_context(tmp_path, line): + path = tmp_path / "bad.jsonl" + path.write_text(FIXTURE.read_text(encoding="utf-8").splitlines()[0] + "\n" + line + "\n", encoding="utf-8") + with pytest.raises(ValueError, match="line 2"): + load_events(path) + + +@pytest.mark.parametrize("timestamp,seconds", [("2026-07-05T00:05:10Z", 10), ("2026-07-05T08:05:00+08:00", 0)]) +def test_comparison_timestamp_controls_windows_as_an_instant(tmp_path, timestamp, seconds): + rows = records() + for row in rows: + row["timestamp"] = timestamp + _, normalized, actual, alerts = pipeline(write_records(tmp_path, rows)) + expected = pipeline(FIXTURE)[2].copy() + for column in ("window_start", "window_end"): + expected[column] += pd.Timedelta(seconds=seconds) + pd.testing.assert_frame_equal(actual, expected) + assert all(meta["time_semantics"] == "snapshot_diff_observed_at" for meta in normalized.metadata) + assert alerts.empty + + +def test_unknown_category_and_numeric_identifier_are_transparently_processed(tmp_path): + rows = records() + rows[0].update(event_type="future_process_category", source=42) + _, normalized, features, alerts = pipeline(write_records(tmp_path, rows)) + assert "future_process_category" in set(normalized.event_type) and "42" in set(normalized.source) + assert features.iloc[0].event_count == 7 and alerts.empty + assert features.iloc[0][[key for key in FEATURES if key.endswith("_count") and key != "event_count"]].eq(0).all() + + +@pytest.mark.parametrize("count,alerts_expected", [(2, 0), (3, 1)]) +def test_status_change_crosses_unchanged_default_error_threshold(tmp_path, count, alerts_expected): + rows = records() + for row in rows[:count]: + row["status"] = " FAIL " + _, _, features, alerts = pipeline(write_records(tmp_path, rows)) + assert features.iloc[0].error_count == count + assert features.iloc[0].error_rate == pytest.approx(count / 7) + assert len(alerts) == alerts_expected + assert set(alerts.rule_name) == ({"high_error_rate"} if alerts_expected else set()) + + +@pytest.mark.parametrize("marker", [None, "claimed_occurrence_time"]) +def test_time_marker_is_advisory_metadata_not_semantic_admission(tmp_path, marker): + rows = records() + for row in rows: + if marker is None: + del row["metadata"]["time_semantics"] + else: + row["metadata"]["time_semantics"] = marker + _, normalized, features, alerts = pipeline(write_records(tmp_path, rows)) + pd.testing.assert_frame_equal(features, pipeline(FIXTURE)[2]) + assert [meta.get("time_semantics") for meta in normalized.metadata] == [marker] * 7 + assert alerts.empty + + +def test_optional_metadata_absence_and_duplicate_timestamp_order(tmp_path): + rows = list(reversed(records())) + for row in rows: + del row["metadata"] + _, normalized, features, _ = pipeline(write_records(tmp_path, rows)) + assert normalized.metadata.isna().all() + pd.testing.assert_frame_equal(features, pipeline(FIXTURE)[2]) + rows.append(rows[0].copy()) + _, normalized, features, _ = pipeline(write_records(tmp_path, rows)) + assert len(normalized) == features.iloc[0].event_count == 8 + assert features.iloc[0].unique_sources == 4 and features.iloc[0].unique_targets == 7 + + +def test_windows_are_half_open_around_the_observation_boundary(tmp_path): + rows = records() + for timestamp in ("2026-07-05T00:04:59Z", "2026-07-05T00:06:00Z", "2026-07-05T00:06:01Z"): + rows.append(dict(rows[0], timestamp=timestamp)) + _, _, features, _ = pipeline(write_records(tmp_path, rows)) + counts = features.set_index("window_start").event_count + assert len(features) == 8 + assert counts[pd.Timestamp("2026-07-05T00:04:50Z")] == 8 + assert counts[pd.Timestamp("2026-07-05T00:05:00Z")] == 7 + assert counts[pd.Timestamp("2026-07-05T00:06:00Z")] == 2 + + +def test_line_endings_change_exact_input_fingerprint_not_legacy_or_semantics(tmp_path): + data = FIXTURE.read_bytes() + one = run_cli(tmp_path / "one", data) + two = run_cli(tmp_path / "two", data.replace(b"\n", b"\r\n")) + assert one[1]["input_digest"] == two[1]["input_digest"] + for summary, manifest, input_path, config_path in (one, two): + assert summary["normalized_event_count"] == 7 and summary["alert_count"] == 0 + assert manifest["input_file_digests"] == {"external/input/telemetry_events.jsonl": "sha256:" + sha256(input_path.read_bytes()).hexdigest()} + assert manifest["config_file_digests"] == {"external/config/integration.yaml": "sha256:" + sha256(config_path.read_bytes()).hexdigest()} + assert manifest["config_digest"] == legacy_digest(config_path) + assert one[1]["input_file_digests"] != two[1]["input_file_digests"] + pd.testing.assert_frame_equal(load_feature_table(tmp_path / "one/output/features.csv"), load_feature_table(tmp_path / "two/output/features.csv"))