coulomb-social/scripts/check-owner-communication.py

#!/usr/bin/env python3
"""Browser-check exact owner source snapshots without production configs or sends."""

import argparse
import hashlib
import json
import os
import shutil
import subprocess
import sys
import tempfile
import xml.etree.ElementTree as ET
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
NODE = "scripts/owner_communication/test_browser.py::test_real_owner_registration_and_invitation"


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--user-engine", type=Path, required=True)
    parser.add_argument("--email-connect", type=Path, required=True)
    parser.add_argument("--receipt", type=Path, required=True)
    args = parser.parse_args()
    manifest_path = args.receipt.with_name(args.receipt.name + ".sources.txt")
    if any(p.exists() or p.is_symlink() for p in (args.receipt, manifest_path)):
        raise ValueError("Fresh receipt required")
    hashes = {}
    with tempfile.TemporaryDirectory(prefix="csoc-owner-browser-") as temporary:
        work = Path(temporary)
        for owner, package in [(args.user_engine, "user_engine"),
                               (args.email_connect, "email_connect")]:
            source = owner.resolve(strict=True) / "src" / package
            if (not source.is_dir() or source.is_symlink()
                    or not (source / "__init__.py").is_file()):
                raise ValueError("Real owner source directory required")
            paths = sorted(source.rglob("*.py"))
            if package == "user_engine":
                paths += sorted((source / "custody_canon").glob("*.json"))
            for path in paths:
                if not path.is_file() or any(p.is_symlink() for p in [path, *path.parents]):
                    raise ValueError("Owner source symlink refused")
                name = "src/" + path.relative_to(source.parent).as_posix()
                data = path.read_bytes()
                hashes[name] = hashlib.sha256(data).hexdigest()
                target = work / name
                target.parent.mkdir(parents=True, exist_ok=True)
                target.write_bytes(data)
        for name in ["scripts/owner_communication/test_browser.py",
                     "scripts/owner_communication/browser.mjs"]:
            data = (ROOT / name).read_bytes()
            hashes[name] = hashlib.sha256(data).hexdigest()
            target = work / name
            target.parent.mkdir(parents=True, exist_ok=True)
            target.write_bytes(data)
        config = work / "pytest.ini"
        config.write_text("[pytest]\nmarkers =\n    browser: real browser\n"
                          "    usecase(id): candidate contract\n")
        report, checks = work / "result.xml", work / "checks.json"
        result = subprocess.run(
            [sys.executable, "-m", "pytest", "-q", "-c", str(config), NODE,
             "-p", "no:cacheprovider", "--junitxml", str(report)],
            cwd=work, capture_output=True, timeout=200,
            env={"PATH": os.defpath, "PYTHONPATH": str(work / "src"),
                 "PYTHONDONTWRITEBYTECODE": "1", "PYTEST_DISABLE_PLUGIN_AUTOLOAD": "1",
                 "OWNER_NODE": shutil.which("node") or "node",
                 "OWNER_BROWSER_SCRIPT": str(work / "scripts/owner_communication/browser.mjs"),
                 "OWNER_PLAYWRIGHT_MODULE": str(ROOT / "scripts/node_modules/playwright/index.mjs"),
                 "OWNER_CHECKS_RESULT": str(checks)})
        cases = ET.parse(report).findall(".//testcase") if report.exists() else []
        passed = (result.returncode == 0 and len(cases) == 1 and checks.exists()
                  and cases[0].attrib.get("name") == NODE.split("::")[-1]
                  and not any(c.findall("failure") or c.findall("error") or c.findall("skipped")
                              for c in cases))
        manifest = "".join(f"{name}: {digest}\n"
                           for name, digest in sorted(hashes.items())).encode()
        proof = {"schema": "coulomb.owner-communication-browser.v1", "passed": passed,
                 "browser_node": NODE, "source_manifest": manifest_path.name,
                 "source_manifest_sha256": hashlib.sha256(manifest).hexdigest(),
                 "source_file_count": len(hashes),
                 "runner_sha256": hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),
                 "python_lock_sha256": hashlib.sha256((ROOT / "uv.lock").read_bytes()).hexdigest(),
                 "package_lock_sha256": hashlib.sha256(
                     (ROOT / "scripts/package-lock.json").read_bytes()).hexdigest(),
                 "checks": json.loads(checks.read_bytes()) if passed else {},
                 "production_access": False, "smtp_provider_sends": False,
                 "synthetic_session_and_directory": True,
                 "productive_acceptance": False, "native_identity_reset_verified": False}
        if not passed:
            fd, diagnostic = tempfile.mkstemp(prefix="csoc-owner-browser-private-diagnostics-")
            with os.fdopen(fd, "wb") as stream:
                stream.write(result.stdout + result.stderr)
            proof["private_diagnostics"] = diagnostic
    fd = os.open(manifest_path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
    with os.fdopen(fd, "wb") as stream:
        stream.write(manifest)
    fd = os.open(args.receipt, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
    with os.fdopen(fd, "w") as stream:
        json.dump(proof, stream, indent=2)
        stream.write("\n")
    public = {"passed": passed, "browser_node": NODE}
    if not passed:
        public["private_diagnostics"] = proof["private_diagnostics"]
    print(json.dumps(public))
    return 0 if passed else 1


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except (OSError, ValueError, KeyError, subprocess.SubprocessError, ET.ParseError):
        raise SystemExit("Owner browser check refused; inspect private inputs/diagnostics locally.")