129 lines
4.3 KiB
Python
129 lines
4.3 KiB
Python
#!/usr/bin/env python3
|
|
"""Check Phase 3A source boundaries, generated provenance, and secret canaries."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import pathlib
|
|
import re
|
|
import sys
|
|
|
|
|
|
ROOT = pathlib.Path(__file__).resolve().parents[1]
|
|
FORBIDDEN_WIRE_FIELDS = {
|
|
"provider_url",
|
|
"vm_address",
|
|
"machine_id",
|
|
"machine_address",
|
|
"direct_host",
|
|
"private_key",
|
|
"pairing_key",
|
|
"rtsp_url",
|
|
"gamestream",
|
|
"apollo_url",
|
|
"windows_password",
|
|
}
|
|
SECRET_PATTERNS = (
|
|
re.compile(rb"-----BEGIN [A-Z ]+PRIVATE KEY-----"),
|
|
re.compile(rb"\bAKIA[0-9A-Z]{16}\b"),
|
|
re.compile(rb"\bgh[pousr]_[A-Za-z0-9]{20,}\b"),
|
|
re.compile(rb"\bsk-[A-Za-z0-9]{20,}\b"),
|
|
)
|
|
ALLOWED_SECRET_PROPERTIES = {
|
|
("ProviderSessionWork", "client_private_key_pem"),
|
|
}
|
|
|
|
|
|
def fail(message: str) -> None:
|
|
raise ValueError(message)
|
|
|
|
|
|
def check_generated_provenance() -> None:
|
|
schema_path = ROOT / "schemas/control-v1.schema.json"
|
|
generator_path = ROOT / "tools/generate.py"
|
|
manifest = json.loads((ROOT / "gen/manifest.json").read_text(encoding="utf-8"))
|
|
schema_hash = hashlib.sha256(schema_path.read_bytes()).hexdigest()
|
|
generator_hash = hashlib.sha256(generator_path.read_bytes()).hexdigest()
|
|
if manifest.get("schema_sha256") != schema_hash:
|
|
fail("generated manifest schema hash is stale")
|
|
if manifest.get("generator_sha256") != generator_hash:
|
|
fail("generated manifest generator hash is stale")
|
|
if manifest.get("protocol_version") != (ROOT / "VERSION").read_text(encoding="utf-8").strip():
|
|
fail("generated manifest protocol version is stale")
|
|
|
|
|
|
def check_manifest_schema() -> None:
|
|
schema = json.loads((ROOT / "schemas/control-v1.schema.json").read_text(encoding="utf-8"))
|
|
definitions = schema.get("$defs", {})
|
|
for name, definition in definitions.items():
|
|
for field in definition.get("properties", {}):
|
|
if any(forbidden in field.lower() for forbidden in FORBIDDEN_WIRE_FIELDS):
|
|
if (name, field) not in ALLOWED_SECRET_PROPERTIES:
|
|
fail(f"{name} exposes forbidden wire field {field}")
|
|
|
|
|
|
def check_proto_boundaries(path: pathlib.Path) -> None:
|
|
message = ""
|
|
depth = 0
|
|
for line in path.read_text(encoding="utf-8").splitlines():
|
|
match = re.match(r"\s*message\s+([A-Za-z0-9_]+)\s*\{", line)
|
|
if match and depth == 0:
|
|
message = match.group(1)
|
|
if any(field in line.lower() for field in FORBIDDEN_WIRE_FIELDS):
|
|
allowed = (
|
|
message == "ProviderSessionWork"
|
|
and re.fullmatch(r"\s*string\s+client_private_key_pem\s*=\s*[0-9]+;\s*", line)
|
|
)
|
|
if not allowed:
|
|
fail(f"{path.relative_to(ROOT)} exposes forbidden wire field in {message or 'file scope'}")
|
|
depth += line.count("{") - line.count("}")
|
|
if depth == 0:
|
|
message = ""
|
|
|
|
|
|
def check_text_boundaries() -> None:
|
|
paths = [
|
|
ROOT / "openapi/control-v1.yaml",
|
|
ROOT / "proto/versevdi/control/v1/control.proto",
|
|
ROOT / "frames/datagram-v1.md",
|
|
ROOT / "frames/datagram-v2.md",
|
|
ROOT / "frames/registry.json",
|
|
ROOT / "registries/features.json",
|
|
ROOT / "registries/datagrams.json",
|
|
]
|
|
for path in paths:
|
|
text = path.read_text(encoding="utf-8").lower()
|
|
for field in FORBIDDEN_WIRE_FIELDS:
|
|
if field in text:
|
|
fail(f"{path.relative_to(ROOT)} contains forbidden wire field {field}")
|
|
|
|
check_proto_boundaries(ROOT / "proto/versevdi/tunnel/v1/tunnel.proto")
|
|
|
|
|
|
def check_secret_canaries() -> None:
|
|
for path in ROOT.rglob("*"):
|
|
if not path.is_file() or ".git" in path.parts or path.name in {"LICENSE"}:
|
|
continue
|
|
data = path.read_bytes()
|
|
for pattern in SECRET_PATTERNS:
|
|
if pattern.search(data):
|
|
fail(f"secret canary matched in {path.relative_to(ROOT)}")
|
|
|
|
|
|
def main() -> int:
|
|
check_generated_provenance()
|
|
check_manifest_schema()
|
|
check_text_boundaries()
|
|
check_secret_canaries()
|
|
print("Protocol scope and provenance validation passed")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
raise SystemExit(main())
|
|
except (OSError, ValueError, json.JSONDecodeError) as exc:
|
|
print(f"scope: {exc}", file=sys.stderr)
|
|
raise SystemExit(1)
|