coulomb-social/scripts/check-communication.py

#!/usr/bin/env python3
"""Run exact owner retry/privacy regressions on a private, network-denied snapshot.

Only source and declared synthetic test modules are copied. The owner's configuration,
environment, caches and runtime databases are neither read nor mutated.
"""

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

ROOT = Path(__file__).resolve().parents[1]
CONTRACT = ROOT / "docs/usecases/communication-regressions.json"
GUARD = '''
import socket
import pytest

def forbidden(*args, **kwargs):
    raise RuntimeError("Communication regressions forbid all network access")

def pytest_sessionstart(session):
    socket.socket.connect = forbidden
    socket.socket.connect_ex = forbidden
    socket.getaddrinfo = forbidden
    socket.create_connection = forbidden

def pytest_collection_modifyitems(items):
    for item in items:
        if any(item.get_closest_marker(m) for m in ("skip", "skipif", "xfail")):
            raise pytest.UsageError("Communication contracts cannot skip or xfail")
'''


def check(owner, receipt):
    contract = json.loads(CONTRACT.read_text(encoding="utf-8"))
    if contract["schema_version"] != 1:
        raise ValueError("Unsupported communication contract")
    owner = owner.resolve(strict=True)
    source = owner / "src/email_connect"
    test_files = contract.get("test_files", [contract.get("test_file")])
    if not test_files or any(
        not isinstance(name, str) or Path(name).parent != Path("tests")
        or not Path(name).name.startswith("test_") or Path(name).suffix != ".py"
        for name in test_files
    ):
        raise ValueError("Synthetic tests must be explicit test modules")
    tests = [owner / name for name in test_files]
    if not source.is_dir() or source.is_symlink() or any(
        not test.is_file() or test.is_symlink() for test in tests
    ):
        raise ValueError("Owner source and synthetic tests must be regular local files")
    files = sorted(source.rglob("*.py")) + tests
    if any(p.is_symlink() for p in files):
        raise ValueError("Owner Python sources must not be symlinks")
    # Snapshot exact bytes first. Hash receipts bind the code actually executed,
    # including an uncommitted source change, rather than assuming a HEAD pin.
    hashes = {}
    with tempfile.TemporaryDirectory(prefix="csoc-communication-") as private:
        workspace = Path(private)
        for path in files:
            relative = path.relative_to(owner)
            data = path.read_bytes()
            hashes[str(relative)] = hashlib.sha256(data).hexdigest()
            target = workspace / relative
            target.parent.mkdir(parents=True, exist_ok=True)
            target.write_bytes(data)
        (workspace / "conftest.py").write_text(GUARD, encoding="utf-8")
        (workspace / "pytest.ini").write_text("[pytest]\n", encoding="utf-8")
        report = workspace / "result.xml"
        result = subprocess.run(
            [sys.executable, "-m", "pytest", "-q", "-c", str(workspace / "pytest.ini"),
             *[str(workspace / name) for name in test_files], "-p", "no:cacheprovider",
             "--junitxml", str(report), "--basetemp", str(workspace / "test-data")],
            cwd=workspace, capture_output=True, timeout=120,
            env={"PATH": os.defpath, "PYTHONPATH": str(workspace / "src"),
                 "PYTHONDONTWRITEBYTECODE": "1", "PYTEST_DISABLE_PLUGIN_AUTOLOAD": "1"},
        )
        cases = ET.parse(report).findall(".//testcase") if report.is_file() else []
        covered = {case.attrib["name"].split("[", 1)[0] for case in cases}
        qualified = {
            case.attrib.get("classname", "").split(".")[-1] + "::"
            + case.attrib["name"].split("[", 1)[0] for case in cases
        }
        missing = sorted(set(contract["required_tests"]) - (covered | qualified))
        disabled = sum(len(case.findall("skipped")) for case in cases)
        failed = sum(len(case.findall("failure")) + len(case.findall("error")) for case in cases)
        passed = result.returncode == 0 and bool(cases) and not (missing or disabled or failed)
        evidence = {
            "schema_version": 1, "owner": contract["owner"], "passed": passed,
            "network": "all socket connects and DNS resolution denied",
            "tests": len(cases), "skipped": disabled, "failed": failed,
            "missing_required_tests": missing, "source_sha256": hashes,
            "contract_sha256": hashlib.sha256(CONTRACT.read_bytes()).hexdigest(),
            "acceptance_complete": False,
            "remaining_acceptance": contract["remaining_acceptance"],
        }
        # Do not publish raw pytest output: reports contain no message bodies,
        # handles, bearer values or provider diagnostics, including on failures.
        receipt.parent.mkdir(parents=True, exist_ok=True)
        receipt.write_text(json.dumps(evidence, indent=2) + "\n", encoding="utf-8")
        return evidence


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--owner", type=Path, required=True)
    parser.add_argument("--receipt", type=Path,
                        default=ROOT / "var/test-results/communication.json")
    args = parser.parse_args()
    try:
        evidence = check(args.owner, args.receipt)
    except (OSError, ValueError, KeyError, subprocess.TimeoutExpired, ET.ParseError):
        print("Communication check could not produce a valid isolated receipt", file=sys.stderr)
        return 1
    print(json.dumps({key: evidence[key] for key in
                      ("passed", "tests", "skipped", "failed", "missing_required_tests")}))
    return 0 if evidence["passed"] else 1


if __name__ == "__main__":
    raise SystemExit(main())