970698af83
End-to-end pipeline now produces a labeled envelope from a single command.
Drives the orchestrator through an 8-phase XMRig-shaped schedule and
renders a 3-panel envelope (CPU%, RSS, IO write rate) with phase bands
sourced from labels.jsonl. Real telemetry, simulated load — validates the
collection + labeling shape before a real VM is involved.
Components:
- tools/load_mimic.py phase-driven load generator. Reads phase
commands on stdin; CPU/IO behavior matches
the named phase (clean=idle, armed=light burst,
infecting=disk burst+CPU, infected_running=
CPU saturation+stratum-shaped writes,
dormant=quieter than clean).
- tools/run_envelope_demo.py spawns load_mimic, drives EpisodeRunner with
a default 85s schedule that includes the
classic infected_running → dormant → re-entry
pattern.
- tools/plot_envelope.py reads telemetry + labels from an episode dir,
writes envelope.png with colored phase bands.
orchestrator: EpisodeRunner now takes an optional phase_schedule and an
on_phase callback. Walks the schedule emitting one label per transition.
Backwards-compatible — existing single-phase tests still green.
Doc fix (user pushback): README + architecture + threat-model no longer
imply the Pi5 is the deployment target. Pi5's actual role here is the
WireGuard-side collector for episode tarballs. Deployment target is
generic ("constrained Linux device"). The "gateway observer" concept
remains a deployment pattern, decoupled from the Pi5's collector role.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
118 lines
3.3 KiB
Python
118 lines
3.3 KiB
Python
"""Run a synthetic envelope demo end-to-end.
|
|
|
|
Spawns ``tools/load_mimic.py`` as a subprocess, drives the orchestrator
|
|
through an XMRig-shaped phase schedule, and writes a complete episode
|
|
directory. Real telemetry from the host /proc collector; load is simulated
|
|
so we can validate the pipeline before a real VM is involved.
|
|
|
|
Usage:
|
|
|
|
uv run python tools/run_envelope_demo.py [--data-root data]
|
|
|
|
After it finishes, plot with:
|
|
|
|
uv run python tools/plot_envelope.py <episode_dir>
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import logging
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
# Allow running as a script: add repo root to sys.path before project imports.
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
|
|
|
from orchestrator.episode import EpisodeConfig, EpisodeRunner # noqa: E402
|
|
|
|
|
|
# (phase, duration_seconds). Total ~85s.
|
|
DEFAULT_SCHEDULE = [
|
|
("clean", 10.0),
|
|
("armed", 2.0),
|
|
("infecting", 3.0),
|
|
("infected_running", 25.0),
|
|
("dormant", 15.0),
|
|
("infected_running", 20.0),
|
|
("dormant", 5.0),
|
|
("clean", 5.0),
|
|
]
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(prog="run_envelope_demo")
|
|
parser.add_argument("--data-root", default="data")
|
|
parser.add_argument("--interval-ms", type=int, default=100)
|
|
args = parser.parse_args()
|
|
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format="%(asctime)s %(levelname)s %(name)s %(message)s",
|
|
)
|
|
|
|
here = Path(__file__).parent
|
|
load_mimic = here / "load_mimic.py"
|
|
|
|
proc = subprocess.Popen(
|
|
[sys.executable, str(load_mimic)],
|
|
stdin=subprocess.PIPE,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.STDOUT,
|
|
bufsize=1,
|
|
text=True,
|
|
)
|
|
|
|
# First line: "load_mimic pid=<n> scratch=<path>"
|
|
first = proc.stdout.readline().strip()
|
|
if not first.startswith("load_mimic pid="):
|
|
proc.kill()
|
|
print(f"unexpected load_mimic startup: {first!r}", file=sys.stderr)
|
|
return 2
|
|
pid = int(first.split()[1].split("=", 1)[1])
|
|
print(first)
|
|
|
|
def on_phase(p: str) -> None:
|
|
proc.stdin.write(p + "\n")
|
|
proc.stdin.flush()
|
|
# Drain the echo line so the pipe doesn't fill up.
|
|
proc.stdout.readline()
|
|
print(f" >>> phase = {p}")
|
|
|
|
cfg = EpisodeConfig(
|
|
target_pid=pid,
|
|
duration_s=sum(d for _, d in DEFAULT_SCHEDULE),
|
|
interval_ms=args.interval_ms,
|
|
data_root=Path(args.data_root),
|
|
phase_schedule=DEFAULT_SCHEDULE,
|
|
image_name="load_mimic.py",
|
|
snapshot_name="(synthetic)",
|
|
)
|
|
|
|
try:
|
|
result = EpisodeRunner(cfg, on_phase=on_phase).run()
|
|
finally:
|
|
try:
|
|
proc.stdin.write("quit\n")
|
|
proc.stdin.flush()
|
|
except Exception:
|
|
pass
|
|
try:
|
|
proc.wait(timeout=5)
|
|
except subprocess.TimeoutExpired:
|
|
proc.kill()
|
|
|
|
print()
|
|
print(f"episode_id = {result.episode_id}")
|
|
print(f"path = {result.episode_dir}")
|
|
print(f"rows_proc = {result.rows_proc}")
|
|
print(f"phases = {result.phases_observed}")
|
|
print()
|
|
print("To plot:")
|
|
print(f" uv run python tools/plot_envelope.py {result.episode_dir}")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|