tenant.openagents/omega

1#!/usr/bin/env python3
2"""Scan installed-proof surfaces without revealing the secret or matching paths."""
3
4from __future__ import annotations
5
6import argparse
7import hashlib
8import json
9import os
10import stat
11import sys
12import tempfile
13import re
14from dataclasses import dataclass
15from datetime import datetime, timezone
16from pathlib import Path
17
18CHUNK_SIZE = 1024 * 1024
19MINIMUM_NEEDLE_BYTES = 16
20MAXIMUM_NEEDLE_BYTES = 4096
21
22
23@dataclass
24class SurfaceResult:
25    name: str
26    path_digest: str
27    status: str
28    files_scanned: int = 0
29    symlinks_skipped: int = 0
30    bytes_scanned: int = 0
31    errors: int = 0
32    match_detected: bool = False
33    evidence_digest: str = ""
34
35    def as_dict(self) -> dict[str, object]:
36        return {
37            "name": self.name,
38            "path_digest": self.path_digest,
39            "status": self.status,
40            "files_scanned": self.files_scanned,
41            "symlinks_skipped": self.symlinks_skipped,
42            "bytes_scanned": self.bytes_scanned,
43            "errors": self.errors,
44            "match_detected": self.match_detected,
45            "evidence_digest": self.evidence_digest,
46        }
47
48
49def digest_text(value: str) -> str:
50    return hashlib.sha256(value.encode("utf-8", errors="surrogateescape")).hexdigest()
51
52
53def read_needle(file_descriptor: int) -> bytes:
54    chunks: list[bytes] = []
55    remaining = MAXIMUM_NEEDLE_BYTES + 2
56    while remaining > 0:
57        chunk = os.read(file_descriptor, remaining)
58        if not chunk:
59            break
60        chunks.append(chunk)
61        remaining -= len(chunk)
62    needle = b"".join(chunks).rstrip(b"\r\n")
63    if not MINIMUM_NEEDLE_BYTES <= len(needle) <= MAXIMUM_NEEDLE_BYTES:
64        raise ValueError(
65            f"canary must contain {MINIMUM_NEEDLE_BYTES}..{MAXIMUM_NEEDLE_BYTES} bytes"
66        )
67    return needle
68
69
70def scan_file(path: Path, needle: bytes) -> tuple[bool, int, str, int, int]:
71    content_digest = hashlib.sha256()
72    matched = False
73    bytes_scanned = 0
74    overlap = b""
75    file_descriptor = os.open(path, os.O_RDONLY | os.O_NOFOLLOW)
76    metadata = os.fstat(file_descriptor)
77    if not stat.S_ISREG(metadata.st_mode):
78        os.close(file_descriptor)
79        raise OSError("surface entry is not a regular file")
80    with os.fdopen(file_descriptor, "rb") as handle:
81        while True:
82            chunk = handle.read(CHUNK_SIZE)
83            if not chunk:
84                break
85            bytes_scanned += len(chunk)
86            content_digest.update(chunk)
87            if needle in overlap + chunk:
88                matched = True
89            overlap = (overlap + chunk)[-(len(needle) - 1) :]
90    return (
91        matched,
92        bytes_scanned,
93        content_digest.hexdigest(),
94        stat.S_IMODE(metadata.st_mode),
95        metadata.st_size,
96    )
97
98
99def surface_files(root: Path) -> tuple[list[Path], int]:
100    root_metadata = root.lstat()
101    if stat.S_ISLNK(root_metadata.st_mode):
102        return [], 1
103    if stat.S_ISREG(root_metadata.st_mode):
104        return [root], 0
105    if not stat.S_ISDIR(root_metadata.st_mode):
106        return [], 0
107
108    files: list[Path] = []
109    skipped_symlinks = 0
110    for directory, directory_names, file_names in os.walk(root, followlinks=False):
111        directory_path = Path(directory)
112        retained_directories: list[str] = []
113        for directory_name in directory_names:
114            candidate = directory_path / directory_name
115            if candidate.is_symlink():
116                skipped_symlinks += 1
117            else:
118                retained_directories.append(directory_name)
119        directory_names[:] = retained_directories
120        for file_name in file_names:
121            candidate = directory_path / file_name
122            if candidate.is_symlink():
123                skipped_symlinks += 1
124            else:
125                files.append(candidate)
126    return sorted(files, key=lambda path: os.fsencode(path)), skipped_symlinks
127
128
129def scan_surface(name: str, root: Path, needle: bytes) -> SurfaceResult:
130    result = SurfaceResult(
131        name=name,
132        path_digest=digest_text(os.fspath(root.resolve(strict=False))),
133        status="pass",
134    )
135    if not root.exists() and not root.is_symlink():
136        result.status = "absent"
137        result.evidence_digest = hashlib.sha256(b"absent").hexdigest()
138        return result
139
140    evidence_digest = hashlib.sha256()
141    try:
142        files, result.symlinks_skipped = surface_files(root)
143        for path in files:
144            try:
145                matched, bytes_scanned, file_digest, file_mode, file_size = scan_file(path, needle)
146                result.files_scanned += 1
147                result.bytes_scanned += bytes_scanned
148                result.match_detected = result.match_detected or matched
149                relative = path.name if root.is_file() else os.fspath(path.relative_to(root))
150                evidence_digest.update(digest_text(relative).encode())
151                evidence_digest.update(str(file_mode).encode())
152                evidence_digest.update(str(file_size).encode())
153                evidence_digest.update(file_digest.encode())
154            except (OSError, ValueError):
155                result.errors += 1
156    except OSError:
157        result.errors += 1
158
159    if result.errors:
160        result.status = "error"
161    elif result.match_detected:
162        result.status = "match"
163    result.evidence_digest = evidence_digest.hexdigest()
164    return result
165
166
167def parse_surface(value: str) -> tuple[str, Path]:
168    name, separator, raw_path = value.partition("=")
169    if not separator or not name or not raw_path:
170        raise argparse.ArgumentTypeError("surface must be NAME=PATH")
171    if not all(character.isalnum() or character in "_-" for character in name):
172        raise argparse.ArgumentTypeError("surface name must be alphanumeric, '_' or '-'")
173    return name, Path(raw_path).expanduser()
174
175
176def validate_candidate_digest(value: str) -> str:
177    if re.fullmatch(r"[0-9a-f]{64}", value) is None:
178        raise ValueError("candidate digest must be 64 lowercase hexadecimal characters")
179    return value
180
181
182def build_report(
183    surfaces: list[tuple[str, Path]], needle: bytes, candidate_digest: str
184) -> dict[str, object]:
185    names = [name for name, _ in surfaces]
186    if len(names) != len(set(names)):
187        raise ValueError("surface names must be unique")
188    results = [scan_surface(name, path, needle) for name, path in surfaces]
189    overall_status = "pass"
190    if any(result.status == "error" for result in results):
191        overall_status = "error"
192    elif any(result.match_detected for result in results):
193        overall_status = "match"
194    return {
195        "schema": "openagents.omega.installed-secret-tripwires.v1",
196        "candidate_digest": validate_candidate_digest(candidate_digest),
197        "generated_at": datetime.now(timezone.utc).isoformat(),
198        "status": overall_status,
199        "needle": {
200            "source": "protected_file_descriptor",
201            "length": len(needle),
202            "value_recorded": False,
203        },
204        "surfaces": [result.as_dict() for result in results],
205    }
206
207
208def self_test() -> int:
209    needle = b"nsec1-disposable-proof-canary-0001"
210    with tempfile.TemporaryDirectory() as directory:
211        root = Path(directory)
212        (root / "clean").write_bytes(b"public evidence only")
213        (root / "match").write_bytes(b"prefix" + needle + b"suffix")
214        os.symlink(root / "match", root / "skipped-link")
215        report = build_report(
216            [("clean", root / "clean"), ("leak", root / "match"), ("missing", root / "none")],
217            needle,
218            "a" * 64,
219        )
220        results = {result["name"]: result for result in report["surfaces"]}
221        assert report["status"] == "match"
222        assert results["clean"]["status"] == "pass"
223        assert results["leak"]["status"] == "match"
224        assert results["missing"]["status"] == "absent"
225        linked = scan_surface("linked", root / "skipped-link", needle)
226        assert linked.status == "pass"
227        assert linked.symlinks_skipped == 1
228        assert not linked.match_detected
229    print("secret-tripwire scanner self-test passed")
230    return 0
231
232
233def main() -> int:
234    parser = argparse.ArgumentParser(description=__doc__)
235    parser.add_argument("--needle-fd", type=int, default=0)
236    parser.add_argument("--candidate-digest")
237    parser.add_argument("--surface", action="append", type=parse_surface, default=[])
238    parser.add_argument("--output", type=Path)
239    parser.add_argument("--self-test", action="store_true")
240    arguments = parser.parse_args()
241    if arguments.self_test:
242        return self_test()
243    if arguments.candidate_digest is None:
244        parser.error("--candidate-digest is required")
245    if not arguments.surface:
246        parser.error("at least one --surface NAME=PATH is required")
247    try:
248        needle = read_needle(arguments.needle_fd)
249        report = build_report(arguments.surface, needle, arguments.candidate_digest)
250    except (OSError, ValueError) as error:
251        print(f"error: {error}", file=sys.stderr)
252        return 2
253    serialized = json.dumps(report, indent=2, sort_keys=True) + "\n"
254    if arguments.output:
255        with arguments.output.open("x", encoding="utf-8") as output:
256            output.write(serialized)
257        print(f"wrote {arguments.output}")
258    else:
259        sys.stdout.write(serialized)
260    if report["status"] == "match":
261        return 1
262    if report["status"] == "error":
263        return 2
264    return 0
265
266
267if __name__ == "__main__":
268    raise SystemExit(main())
269