tenant.openagents/omega

1#!/usr/bin/env python3
2"""Collect and validate candidate-bound Omega process/network evidence."""
3
4from __future__ import annotations
5
6import argparse
7import ctypes
8import hashlib
9import ipaddress
10import json
11import os
12import re
13import socket
14import subprocess
15import sys
16import time
17from datetime import datetime, timezone
18from pathlib import Path
19from typing import Any
20
21
22ROOT = Path(__file__).resolve().parent.parent
23ALLOWLIST_PATH = ROOT / "crates/app_identity/fixtures/endpoint_allowlist.json"
24SCHEMA = "openagents.omega.network-capture.v1"
25COLLECTOR_VERSION = 2
26EXPECTED_ROOT_EXECUTABLE = Path("/Applications/Omega.app/Contents/MacOS/omega")
27EXPECTED_SIGNING_AUTHORITY = (
28    "Developer ID Application: OpenAgents, Inc. (HQWSG26L43)"
29)
30EXPECTED_TEAM_ID = "HQWSG26L43"
31LOCAL_HOSTS = {"localhost", "localhost.localdomain"}
32
33
34class CaptureError(RuntimeError):
35    pass
36
37
38def sha256_file(path: Path) -> str:
39    digest = hashlib.sha256()
40    with path.open("rb") as handle:
41        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
42            digest.update(chunk)
43    return digest.hexdigest()
44
45
46def load_json(path: Path, description: str) -> dict[str, Any]:
47    try:
48        value = json.loads(path.read_text(encoding="utf-8"))
49    except (OSError, json.JSONDecodeError) as error:
50        raise CaptureError(f"cannot read {description} {path}: {error}") from error
51    if not isinstance(value, dict):
52        raise CaptureError(f"{description} must be a JSON object")
53    return value
54
55
56def validate_digest(value: str) -> str:
57    if re.fullmatch(r"[0-9a-f]{64}", value) is None:
58        raise CaptureError("candidate digest must be 64 lowercase hexadecimal characters")
59    return value
60
61
62def run(command: list[str]) -> subprocess.CompletedProcess[str]:
63    return subprocess.run(command, capture_output=True, text=True, check=False)
64
65
66def collector_binding() -> dict[str, Any]:
67    collector = Path(__file__).resolve()
68    return {
69        "path": "script/omega-network-capture-journey",
70        "version": COLLECTOR_VERSION,
71        "sha256": sha256_file(collector),
72    }
73
74
75def process_executable_path(process_id: int) -> Path:
76    try:
77        libproc = ctypes.CDLL("/usr/lib/libproc.dylib", use_errno=True)
78    except OSError as error:
79        raise CaptureError("cannot load macOS process inspection authority") from error
80    buffer = ctypes.create_string_buffer(4096)
81    libproc.proc_pidpath.argtypes = [ctypes.c_int, ctypes.c_void_p, ctypes.c_uint32]
82    libproc.proc_pidpath.restype = ctypes.c_int
83    length = libproc.proc_pidpath(process_id, buffer, len(buffer))
84    if length <= 0:
85        raise CaptureError(f"cannot resolve executable path for PID {process_id}")
86    try:
87        raw_path = buffer.raw[:length].split(b"\0", 1)[0]
88        return Path(os.fsdecode(raw_path)).resolve(strict=True)
89    except OSError as error:
90        raise CaptureError(f"cannot canonicalize executable path for PID {process_id}") from error
91
92
93def process_start_identity(process_id: int) -> str:
94    result = run(["ps", "-p", str(process_id), "-o", "lstart="])
95    value = result.stdout.strip()
96    if result.returncode != 0 or not value:
97        raise CaptureError(f"cannot resolve start identity for PID {process_id}")
98    return value
99
100
101def codesign_binding(path: Path) -> dict[str, Any]:
102    verification = run(
103        ["/usr/bin/codesign", "--verify", "--deep", "--strict", "--verbose=4", str(path)]
104    )
105    if verification.returncode != 0:
106        raise CaptureError("installed Omega root executable failed strict codesign verification")
107    metadata = run(["/usr/bin/codesign", "-dv", "--verbose=4", str(path)])
108    if metadata.returncode != 0:
109        raise CaptureError("cannot read installed Omega root executable codesign metadata")
110    detail = "\n".join((metadata.stdout, metadata.stderr))
111    authorities = re.findall(r"(?m)^Authority=(.+)$", detail)
112    team_match = re.search(r"(?m)^TeamIdentifier=(.+)$", detail)
113    team_identifier = team_match.group(1) if team_match else None
114    if not authorities or authorities[0] != EXPECTED_SIGNING_AUTHORITY:
115        raise CaptureError("installed Omega root executable has unexpected signing authority")
116    if team_identifier != EXPECTED_TEAM_ID:
117        raise CaptureError("installed Omega root executable has unexpected signing team")
118    return {
119        "strict_verification": "passed",
120        "verification_arguments": ["--verify", "--deep", "--strict", "--verbose=4"],
121        "authorities": authorities,
122        "team_identifier": team_identifier,
123    }
124
125
126def installed_binary_binding(path: Path = EXPECTED_ROOT_EXECUTABLE) -> dict[str, Any]:
127    if path.is_symlink() or not path.is_file():
128        raise CaptureError(f"installed Omega root executable is missing or unsafe: {path}")
129    canonical = path.resolve(strict=True)
130    if canonical != EXPECTED_ROOT_EXECUTABLE:
131        raise CaptureError("installed Omega root executable canonical path is not exact")
132    metadata = canonical.stat()
133    binary_sha256 = sha256_file(canonical)
134    signing = codesign_binding(canonical)
135    confirmed_metadata = canonical.stat()
136    identity_fields = ("st_dev", "st_ino", "st_size", "st_mtime_ns", "st_ctime_ns")
137    if any(
138        getattr(metadata, field) != getattr(confirmed_metadata, field)
139        for field in identity_fields
140    ):
141        raise CaptureError("installed Omega root executable changed during inspection")
142    return {
143        "canonical_path": str(canonical),
144        "sha256": binary_sha256,
145        "device": metadata.st_dev,
146        "inode": metadata.st_ino,
147        "size": metadata.st_size,
148        "mtime_ns": metadata.st_mtime_ns,
149        "ctime_ns": metadata.st_ctime_ns,
150        "codesign": signing,
151    }
152
153
154def root_process_observation(
155    process_id: int, binary: dict[str, Any]
156) -> dict[str, Any]:
157    process_start = process_start_identity(process_id)
158    executable = process_executable_path(process_id)
159    metadata = executable.stat()
160    stable_file_identity = (
161        str(executable) == binary["canonical_path"]
162        and metadata.st_dev == binary["device"]
163        and metadata.st_ino == binary["inode"]
164        and metadata.st_size == binary["size"]
165        and metadata.st_mtime_ns == binary["mtime_ns"]
166        and metadata.st_ctime_ns == binary["ctime_ns"]
167    )
168    executable_sha256 = (
169        binary["sha256"] if stable_file_identity else sha256_file(executable)
170    )
171    if process_start_identity(process_id) != process_start:
172        raise CaptureError(f"PID {process_id} was reused during identity inspection")
173    return {
174        "pid": process_id,
175        "process_start": process_start,
176        "executable_path": str(executable),
177        "binary_sha256": executable_sha256,
178        "device": metadata.st_dev,
179        "inode": metadata.st_ino,
180        "size": metadata.st_size,
181        "mtime_ns": metadata.st_mtime_ns,
182        "ctime_ns": metadata.st_ctime_ns,
183        "matches_installed_binary": stable_file_identity
184        and executable_sha256 == binary["sha256"],
185    }
186
187
188def process_table() -> dict[int, tuple[int, str]]:
189    result = run(["ps", "-axo", "pid=,ppid=,comm="])
190    if result.returncode != 0:
191        raise CaptureError("cannot inspect the process table")
192    processes: dict[int, tuple[int, str]] = {}
193    for line in result.stdout.splitlines():
194        fields = line.strip().split(maxsplit=2)
195        if len(fields) != 3:
196            continue
197        try:
198            process_id = int(fields[0])
199            parent_id = int(fields[1])
200        except ValueError:
201            continue
202        processes[process_id] = (parent_id, Path(fields[2]).name)
203    return processes
204
205
206def descendants(root_pid: int, table: dict[int, tuple[int, str]]) -> set[int]:
207    selected = {root_pid}
208    changed = True
209    while changed:
210        changed = False
211        for process_id, (parent_id, _command) in table.items():
212            if parent_id in selected and process_id not in selected:
213                selected.add(process_id)
214                changed = True
215    return selected
216
217
218def parse_remote_endpoint(value: str) -> tuple[str, int] | None:
219    if "->" not in value:
220        return None
221    remote = value.split("->", 1)[1].split(" (", 1)[0]
222    host, separator, port_text = remote.rpartition(":")
223    if not separator or not port_text.isdigit():
224        return None
225    host = host.strip("[]").rstrip(".").lower()
226    if not host:
227        return None
228    return host, int(port_text)
229
230
231def collect_connections(process_id: int) -> tuple[list[tuple[str, int, str]], str | None]:
232    result = run(
233        ["/usr/sbin/lsof", "-a", "-p", str(process_id), "-i", "-P", "-F", "pPnT"]
234    )
235    if result.returncode not in (0, 1):
236        return [], f"lsof_exit_{result.returncode}"
237    protocol = "unknown"
238    connections: list[tuple[str, int, str]] = []
239    for line in result.stdout.splitlines():
240        if line.startswith("P"):
241            protocol = line[1:].lower()
242        elif line.startswith("n"):
243            endpoint = parse_remote_endpoint(line[1:])
244            if endpoint is not None:
245                connections.append((endpoint[0], endpoint[1], protocol))
246    return connections, None
247
248
249def host_matches(host: str, expected: str) -> bool:
250    return host == expected or host.endswith(f".{expected}")
251
252
253def resolve_approved_hosts(allowlist: dict[str, Any]) -> dict[str, list[str]]:
254    resolutions: dict[str, list[str]] = {}
255    for entry in allowlist.get("entries", []):
256        host = entry.get("host")
257        if entry.get("disposition") != "approved" or not isinstance(host, str):
258            continue
259        try:
260            addresses = {
261                result[4][0]
262                for result in socket.getaddrinfo(host, None, type=socket.SOCK_STREAM)
263            }
264        except socket.gaierror:
265            addresses = set()
266        resolutions[host] = sorted(addresses)
267    return resolutions
268
269
270def classify_hosts(
271    destinations: list[dict[str, Any]],
272    allowlist: dict[str, Any],
273    approved_host_resolutions: dict[str, list[str]],
274) -> tuple[list[str], list[str], list[str]]:
275    forbidden_hosts = allowlist.get("normal_start_forbidden_hosts", [])
276    approved_hosts = {
277        entry.get("host")
278        for entry in allowlist.get("entries", [])
279        if entry.get("disposition") == "approved"
280    }
281    forbidden: set[str] = set()
282    unreviewed: set[str] = set()
283    unresolved: set[str] = set()
284    approved_addresses = {
285        address
286        for addresses in approved_host_resolutions.values()
287        for address in addresses
288    }
289    for destination in destinations:
290        host = destination["host"]
291        if host in LOCAL_HOSTS:
292            continue
293        try:
294            address = ipaddress.ip_address(host)
295        except ValueError:
296            pass
297        else:
298            if address.is_loopback or host in approved_addresses:
299                continue
300            unresolved.add(host)
301            continue
302        if any(host_matches(host, expected) for expected in forbidden_hosts):
303            forbidden.add(host)
304        elif not any(host_matches(host, expected) for expected in approved_hosts):
305            unreviewed.add(host)
306    return sorted(forbidden), sorted(unreviewed), sorted(unresolved)
307
308
309def collect(arguments: argparse.Namespace) -> int:
310    candidate_digest = validate_digest(arguments.candidate_digest)
311    binary = installed_binary_binding()
312    initial_root = root_process_observation(arguments.pid, binary)
313    if not initial_root["matches_installed_binary"]:
314        raise CaptureError(
315            "root PID does not execute the exact installed Omega root executable"
316        )
317    allowlist = load_json(ALLOWLIST_PATH, "endpoint allowlist")
318    approved_host_resolutions = resolve_approved_hosts(allowlist)
319    started_at = datetime.now(timezone.utc)
320    deadline = time.monotonic() + arguments.duration
321    observed_processes: dict[int, dict[str, Any]] = {}
322    observed_destinations: dict[tuple[int, str, int, str], dict[str, Any]] = {}
323    root_identity_observations: list[dict[str, Any]] = []
324    errors: set[str] = set()
325    sample_index = 0
326
327    while True:
328        table = process_table()
329        if arguments.pid not in table:
330            errors.add("root_process_not_running")
331        else:
332            try:
333                observation = root_process_observation(arguments.pid, binary)
334            except CaptureError:
335                errors.add("root_process_identity_unreadable")
336            else:
337                observation["sample_index"] = sample_index
338                root_identity_observations.append(observation)
339                if observation["process_start"] != initial_root["process_start"]:
340                    errors.add("root_pid_reused")
341                if observation["executable_path"] != initial_root["executable_path"]:
342                    errors.add("root_executable_path_changed")
343                if not observation["matches_installed_binary"]:
344                    errors.add("root_executable_binding_changed")
345        for process_id in descendants(arguments.pid, table):
346            process = table.get(process_id)
347            if process is None:
348                continue
349            parent_id, command = process
350            observed_processes[process_id] = {
351                "pid": process_id,
352                "ppid": parent_id,
353                "command_basename": command,
354            }
355            if process_id == arguments.pid and root_identity_observations:
356                observed_processes[process_id].update(
357                    {
358                        "executable_path": root_identity_observations[-1][
359                            "executable_path"
360                        ],
361                        "process_start": root_identity_observations[-1]["process_start"],
362                    }
363                )
364            connections, error = collect_connections(process_id)
365            if error is not None:
366                errors.add(error)
367            for host, port, protocol in connections:
368                key = (process_id, host, port, protocol)
369                observed_destinations[key] = {
370                    "pid": process_id,
371                    "host": host,
372                    "port": port,
373                    "protocol": protocol,
374                }
375        if time.monotonic() >= deadline:
376            break
377        sample_index += 1
378        time.sleep(min(arguments.interval, max(0.0, deadline - time.monotonic())))
379
380    processes = sorted(observed_processes.values(), key=lambda item: item["pid"])
381    destinations = sorted(
382        observed_destinations.values(),
383        key=lambda item: (item["host"], item["port"], item["pid"]),
384    )
385    forbidden, unreviewed, unresolved = classify_hosts(
386        destinations, allowlist, approved_host_resolutions
387    )
388    blocked_reasons = sorted(errors)
389    if not processes:
390        blocked_reasons.append("no_candidate_process_ancestry_captured")
391    if arguments.journey == "online_first_codex" and not destinations:
392        blocked_reasons.append("no_online_destination_captured")
393    if forbidden:
394        blocked_reasons.append("forbidden_host_observed")
395    if unreviewed:
396        blocked_reasons.append("unreviewed_host_observed")
397    if unresolved:
398        blocked_reasons.append("destination_hostname_unresolved")
399    blocked_reasons = sorted(set(blocked_reasons))
400    report = {
401        "schema": SCHEMA,
402        "candidate_digest": candidate_digest,
403        "allowlist": {
404            "path": "crates/app_identity/fixtures/endpoint_allowlist.json",
405            "sha256": sha256_file(ALLOWLIST_PATH),
406            "enforcement": allowlist.get("enforcement"),
407        },
408        "approved_host_resolutions": approved_host_resolutions,
409        "journey": arguments.journey,
410        "started_at": started_at.isoformat(),
411        "completed_at": datetime.now(timezone.utc).isoformat(),
412        "root_pid": arguments.pid,
413        "root_process": initial_root,
414        "root_identity_observations": root_identity_observations,
415        "installed_binary": binary,
416        "collector": {
417            **collector_binding(),
418            "duration_seconds": arguments.duration,
419            "interval_seconds": arguments.interval,
420            "process_source": "ps",
421            "destination_source": "lsof_with_reverse_dns",
422            "packet_payload_recorded": False,
423        },
424        "status": "passed" if not blocked_reasons else "blocked",
425        "blocked_reasons": blocked_reasons,
426        "processes": processes,
427        "destinations": destinations,
428        "forbidden_hosts_detected": forbidden,
429        "unreviewed_hosts": unreviewed,
430        "unresolved_destinations": unresolved,
431    }
432    arguments.output.parent.mkdir(parents=True, exist_ok=True)
433    with arguments.output.open("x", encoding="utf-8") as handle:
434        json.dump(report, handle, indent=2, sort_keys=True)
435        handle.write("\n")
436    print(f"wrote network capture: {arguments.output} ({report['status']})")
437    return 0 if report["status"] == "passed" else 3
438
439
440def validate_collector(report: dict[str, Any]) -> None:
441    collector = report.get("collector")
442    expected = collector_binding()
443    if not isinstance(collector, dict):
444        raise CaptureError("network capture has no collector binding")
445    for field in ("path", "version", "sha256"):
446        if collector.get(field) != expected[field]:
447            raise CaptureError("network capture collector binding differs")
448
449
450def validate_root_binding(
451    report: dict[str, Any], *, verify_live_binary: bool
452) -> None:
453    binary = report.get("installed_binary")
454    root = report.get("root_process")
455    observations = report.get("root_identity_observations")
456    if not isinstance(binary, dict) or not isinstance(root, dict):
457        raise CaptureError("network capture has no installed root binary binding")
458    if not isinstance(observations, list) or not observations:
459        raise CaptureError("network capture has no root PID identity observations")
460    if binary.get("canonical_path") != str(EXPECTED_ROOT_EXECUTABLE):
461        raise CaptureError("network capture root executable canonical path differs")
462    if re.fullmatch(r"[0-9a-f]{64}", str(binary.get("sha256", ""))) is None:
463        raise CaptureError("network capture installed binary digest is invalid")
464    for field in ("device", "inode", "size", "mtime_ns", "ctime_ns"):
465        if not isinstance(binary.get(field), int) or binary[field] < 0:
466            raise CaptureError("network capture installed binary identity is invalid")
467    signing = binary.get("codesign")
468    if not isinstance(signing, dict):
469        raise CaptureError("network capture has no strict codesign binding")
470    authorities = signing.get("authorities")
471    if (
472        signing.get("strict_verification") != "passed"
473        or signing.get("verification_arguments")
474        != ["--verify", "--deep", "--strict", "--verbose=4"]
475        or not isinstance(authorities, list)
476        or not authorities
477        or authorities[0] != EXPECTED_SIGNING_AUTHORITY
478        or signing.get("team_identifier") != EXPECTED_TEAM_ID
479    ):
480        raise CaptureError("network capture strict codesign binding is invalid")
481    if verify_live_binary and installed_binary_binding() != binary:
482        raise CaptureError("network capture no longer matches the installed root binary")
483
484    root_pid = report.get("root_pid")
485    if not isinstance(root_pid, int) or root_pid <= 0 or root.get("pid") != root_pid:
486        raise CaptureError("network capture root PID binding is invalid")
487    expected_identity = {
488        "executable_path": binary["canonical_path"],
489        "binary_sha256": binary["sha256"],
490        "device": binary["device"],
491        "inode": binary["inode"],
492        "size": binary["size"],
493        "mtime_ns": binary["mtime_ns"],
494        "ctime_ns": binary["ctime_ns"],
495    }
496    process_start = root.get("process_start")
497    if not isinstance(process_start, str) or not process_start:
498        raise CaptureError("network capture root process start identity is invalid")
499    for field, expected in expected_identity.items():
500        if root.get(field) != expected:
501            raise CaptureError("network capture root process path or binary differs")
502
503    for index, observation in enumerate(observations):
504        if not isinstance(observation, dict):
505            raise CaptureError("network capture root PID observation is invalid")
506        if observation.get("sample_index") != index:
507            raise CaptureError("network capture root PID observations are incomplete")
508        if observation.get("pid") != root_pid:
509            raise CaptureError("network capture root PID changed")
510        if observation.get("process_start") != process_start:
511            raise CaptureError("network capture detected PID reuse")
512        for field, expected in expected_identity.items():
513            if observation.get(field) != expected:
514                raise CaptureError("network capture detected root executable path mismatch")
515        if observation.get("matches_installed_binary") is not True:
516            raise CaptureError("network capture root executable did not match installation")
517
518    processes = report.get("processes")
519    if not isinstance(processes, list):
520        raise CaptureError("network capture process ancestry is invalid")
521    root_rows = [
522        item
523        for item in processes
524        if isinstance(item, dict) and item.get("pid") == root_pid
525    ]
526    if len(root_rows) != 1:
527        raise CaptureError("network capture process ancestry has no unique root PID")
528    if (
529        root_rows[0].get("executable_path") != binary["canonical_path"]
530        or root_rows[0].get("process_start") != process_start
531    ):
532        raise CaptureError("network capture process ancestry root path differs")
533
534
535def validate_report(path: Path, candidate_digest: str) -> dict[str, Any]:
536    report = load_json(path, "network capture")
537    if report.get("schema") != SCHEMA:
538        raise CaptureError("network capture schema is not supported")
539    if report.get("candidate_digest") != validate_digest(candidate_digest):
540        raise CaptureError("network capture binds a different candidate")
541    validate_collector(report)
542    validate_root_binding(report, verify_live_binary=True)
543    allowlist = load_json(ALLOWLIST_PATH, "endpoint allowlist")
544    binding = report.get("allowlist", {})
545    if binding.get("sha256") != sha256_file(ALLOWLIST_PATH):
546        raise CaptureError("network capture binds a different endpoint allowlist")
547    if binding.get("enforcement") != allowlist.get("enforcement"):
548        raise CaptureError("network capture allowlist enforcement mode differs")
549    approved_hosts = {
550        entry.get("host")
551        for entry in allowlist.get("entries", [])
552        if entry.get("disposition") == "approved"
553    }
554    approved_host_resolutions = report.get("approved_host_resolutions")
555    if not isinstance(approved_host_resolutions, dict):
556        raise CaptureError("network capture has no approved-host DNS evidence")
557    for host, addresses in approved_host_resolutions.items():
558        if host not in approved_hosts or not isinstance(addresses, list):
559            raise CaptureError("network capture has invalid approved-host DNS evidence")
560        for address in addresses:
561            try:
562                parsed_address = ipaddress.ip_address(address)
563            except ValueError as error:
564                raise CaptureError(
565                    "network capture has invalid approved-host DNS evidence"
566                ) from error
567            if not parsed_address.is_global:
568                raise CaptureError(
569                    "network capture approved-host DNS evidence is not global"
570                )
571    if report.get("journey") != "online_first_codex":
572        raise CaptureError("installed proof requires the online first-Codex journey")
573    if report.get("status") != "passed" or report.get("blocked_reasons") != []:
574        raise CaptureError("network capture is blocked")
575    if not isinstance(report.get("processes"), list) or not report["processes"]:
576        raise CaptureError("network capture has no process ancestry")
577    if not isinstance(report.get("destinations"), list) or not report["destinations"]:
578        raise CaptureError("network capture has no observed destinations")
579    forbidden, unreviewed, unresolved = classify_hosts(
580        report["destinations"], allowlist, approved_host_resolutions
581    )
582    if (
583        report.get("forbidden_hosts_detected") != forbidden
584        or report.get("unreviewed_hosts") != unreviewed
585        or report.get("unresolved_destinations") != unresolved
586    ):
587        raise CaptureError("network capture destination classification is inconsistent")
588    for field in ("forbidden_hosts_detected", "unreviewed_hosts", "unresolved_destinations"):
589        if report.get(field) != []:
590            raise CaptureError(f"network capture {field} is not empty")
591    return report
592
593
594def validate_command(arguments: argparse.Namespace) -> int:
595    validate_report(arguments.input, arguments.candidate_digest)
596    print(f"network capture valid: {arguments.input}")
597    return 0
598
599
600def self_test(_arguments: argparse.Namespace) -> int:
601    allowlist = load_json(ALLOWLIST_PATH, "endpoint allowlist")
602    destinations = [
603        {"host": "registry.npmjs.org"},
604        {"host": "api.zed.dev"},
605        {"host": "203.0.113.1"},
606        {"host": "127.0.0.1"},
607        {"host": "localhost"},
608        {"host": "2001:db8::2"},
609        {"host": "unexpected.example"},
610    ]
611    forbidden, unreviewed, unresolved = classify_hosts(
612        destinations,
613        allowlist,
614        {"cdn.agentclientprotocol.com": ["2001:db8::2"]},
615    )
616    if forbidden != ["api.zed.dev"]:
617        raise CaptureError("self-test did not identify the forbidden Zed host")
618    if unreviewed != ["unexpected.example"]:
619        raise CaptureError("self-test did not identify the unreviewed host")
620    if unresolved != ["203.0.113.1"]:
621        raise CaptureError("self-test did not block an unresolved IP destination")
622    if parse_remote_endpoint("localhost:123->registry.npmjs.org:443 (ESTABLISHED)") != (
623        "registry.npmjs.org",
624        443,
625    ):
626        raise CaptureError("self-test endpoint parsing failed")
627
628    binary = {
629        "canonical_path": str(EXPECTED_ROOT_EXECUTABLE),
630        "sha256": "a" * 64,
631        "device": 1,
632        "inode": 2,
633        "size": 3,
634        "mtime_ns": 4,
635        "ctime_ns": 5,
636        "codesign": {
637            "strict_verification": "passed",
638            "verification_arguments": [
639                "--verify",
640                "--deep",
641                "--strict",
642                "--verbose=4",
643            ],
644            "authorities": [EXPECTED_SIGNING_AUTHORITY],
645            "team_identifier": EXPECTED_TEAM_ID,
646        },
647    }
648    identity = {
649        "pid": 42,
650        "process_start": "Fri Jul 24 12:00:00 2026",
651        "executable_path": str(EXPECTED_ROOT_EXECUTABLE),
652        "binary_sha256": "a" * 64,
653        "device": 1,
654        "inode": 2,
655        "size": 3,
656        "mtime_ns": 4,
657        "ctime_ns": 5,
658        "matches_installed_binary": True,
659    }
660    report = {
661        "root_pid": 42,
662        "installed_binary": binary,
663        "root_process": identity,
664        "root_identity_observations": [{**identity, "sample_index": 0}],
665        "processes": [
666            {
667                "pid": 42,
668                "ppid": 1,
669                "command_basename": "omega",
670                "executable_path": str(EXPECTED_ROOT_EXECUTABLE),
671                "process_start": identity["process_start"],
672            }
673        ],
674        "collector": collector_binding(),
675    }
676    validate_collector(report)
677    validate_root_binding(report, verify_live_binary=False)
678    for field, tampered in (
679        ("process_start", "Fri Jul 24 12:00:01 2026"),
680        ("executable_path", "/tmp/fake-omega"),
681    ):
682        forged = json.loads(json.dumps(report))
683        forged["root_identity_observations"][0][field] = tampered
684        try:
685            validate_root_binding(forged, verify_live_binary=False)
686        except CaptureError:
687            pass
688        else:
689            raise CaptureError(f"self-test accepted tampered root {field}")
690    forged = json.loads(json.dumps(report))
691    forged["installed_binary"]["codesign"]["team_identifier"] = "FAKE"
692    try:
693        validate_root_binding(forged, verify_live_binary=False)
694    except CaptureError:
695        pass
696    else:
697        raise CaptureError("self-test accepted tampered codesign team")
698    forged = json.loads(json.dumps(report))
699    forged["collector"]["sha256"] = "b" * 64
700    try:
701        validate_collector(forged)
702    except CaptureError:
703        pass
704    else:
705        raise CaptureError("self-test accepted tampered collector digest")
706    print("Omega network capture self-test passed")
707    return 0
708
709
710def parse_arguments() -> argparse.Namespace:
711    parser = argparse.ArgumentParser(description=__doc__)
712    subparsers = parser.add_subparsers(dest="command", required=True)
713    capture = subparsers.add_parser("capture")
714    capture.add_argument("--candidate-digest", required=True)
715    capture.add_argument("--pid", type=int, required=True)
716    capture.add_argument(
717        "--journey", choices=("offline_first_start", "online_first_codex"), required=True
718    )
719    capture.add_argument("--duration", type=float, default=30.0)
720    capture.add_argument("--interval", type=float, default=0.5)
721    capture.add_argument("--output", type=Path, required=True)
722    capture.set_defaults(handler=collect)
723    validate = subparsers.add_parser("validate")
724    validate.add_argument("--candidate-digest", required=True)
725    validate.add_argument("--input", type=Path, required=True)
726    validate.set_defaults(handler=validate_command)
727    self_test_parser = subparsers.add_parser("self-test")
728    self_test_parser.set_defaults(handler=self_test)
729    return parser.parse_args()
730
731
732def main() -> int:
733    arguments = parse_arguments()
734    if getattr(arguments, "duration", 1) <= 0 or getattr(arguments, "interval", 1) <= 0:
735        print("error: duration and interval must be positive", file=sys.stderr)
736        return 2
737    try:
738        return arguments.handler(arguments)
739    except CaptureError as error:
740        print(f"error: {error}", file=sys.stderr)
741        return 2
742
743
744if __name__ == "__main__":
745    raise SystemExit(main())
746