From an installed release to verified denial evidence¶

import pathlib
import urllib.request


class DownloadRejected(ValueError):
    pass


def download(
    url: str, destination: pathlib.Path, *, max_bytes: int,
    accept: str = "application/octet-stream",
) -> None:
    if max_bytes <= 0:
        raise ValueError("download ceiling must be positive")
    if destination.exists():
        raise DownloadRejected("download destination already exists")
    scratch = destination.with_name(f".{destination.name}.downloading")
    if scratch.exists():
        raise DownloadRejected("download scratch destination already exists")

    request = urllib.request.Request(
        url,
        headers={"Accept": accept, "User-Agent": "assay-release-verifier"},
    )
    try:
        # The driver validates the exact GitHub release URL before this call.
        with urllib.request.urlopen(request, timeout=60) as response:  # noqa: S310
            declared = response.headers.get("Content-Length")
            if declared is not None and (not declared.isdigit() or int(declared) > max_bytes):
                raise DownloadRejected("download content length exceeds ceiling")
            total = 0
            with scratch.open("xb") as output:
                while chunk := response.read(min(65536, max_bytes - total + 1)):
                    total += len(chunk)
                    if total > max_bytes:
                        raise DownloadRejected("download stream exceeds ceiling")
                    output.write(chunk)
            if total == 0:
                raise DownloadRejected("download yielded no bytes")
        scratch.replace(destination)
    except BaseException:
        scratch.unlink(missing_ok=True)
        raise


import hashlib
import json
import platform
import subprocess

version = subprocess.check_output(["assay", "version"], text=True).strip()
targets = {
    ("Linux", "x86_64"): "x86_64-unknown-linux-gnu",
    ("Linux", "aarch64"): "aarch64-unknown-linux-gnu",
    ("Linux", "arm64"): "aarch64-unknown-linux-gnu",
    ("Darwin", "arm64"): "aarch64-apple-darwin",
    ("Darwin", "x86_64"): "x86_64-apple-darwin",
    ("Windows", "AMD64"): "x86_64-pc-windows-msvc",
}
target = targets[(platform.system(), platform.machine())]
extension = ".zip" if platform.system() == "Windows" else ".tar.gz"
asset = "assay-v" + version + "-" + target + extension
url = "https://github.com/Rul1an/assay/releases/download/v" + version + "/" + asset
download(url, pathlib.Path(asset), max_bytes=134217728)
download(url + ".sha256", pathlib.Path(asset + ".sha256"), max_bytes=16384)
with open(asset, "rb") as stream:
    observed = hashlib.sha256()
    for chunk in iter(lambda: stream.read(65536), b""):
        observed.update(chunk)
    digest = observed.hexdigest()
if pathlib.Path(asset + ".sha256").read_text().strip() != digest + "  " + asset:
    raise ValueError("Archive sidecar mismatch")
pathlib.Path("acquisition.json").write_text(json.dumps({"archive": asset}))

import json
import pathlib
import queue
import subprocess
import sys
import threading

server, example = sys.argv[1:]
example = pathlib.Path(example).resolve()
argv = [server, "proxy-enforce", "--upstream-command", sys.executable,
        "--upstream-arg", "-u", "--upstream-arg", str(example / "mock_github_mcp.py"),
        "--enforce-policy", str(example / "policies/no-allowance.yaml"),
        "--declared-mcp-manifest", str(example / "baseline-approved.json"),
        "--enforcement-decision-out", "decisions.ndjson",
        "--denied-call-observation-out", "denied-observations.ndjson"]
messages = [
    {"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {
        "protocolVersion": "2024-11-05", "capabilities": {},
        "clientInfo": {"name": "installed-release-example", "version": "1"}}},
    {"jsonrpc": "2.0", "id": 9, "method": "tools/call", "params": {
        "name": "github.add_deploy_key", "arguments": {"owner": "acme", "repo": "prod-app"}}},
]
with open("proxy.stderr", "w") as errors:
    child = subprocess.Popen(argv, stdin=subprocess.PIPE, stdout=subprocess.PIPE,
                             stderr=errors, text=True)
    replies = queue.Queue()
    def read_replies():
        for line in child.stdout:
            replies.put(line)
        replies.put(None)
    threading.Thread(target=read_replies, daemon=True).start()
    try:
        for message in messages:
            child.stdin.write(json.dumps(message) + "\n")
        child.stdin.flush()
        while True:
            line = replies.get(timeout=30)
            if line is None:
                raise RuntimeError("proxy ended without the denied response")
            response = json.loads(line)
            if response.get("id") == 9:
                assert response["error"]["code"] == -31999
                assert response["error"]["data"]["reason"] == "no_declared_allowance"
                break
        child.stdin.close()
        assert child.wait(timeout=10) == 0
    finally:
        if child.poll() is None:
            child.kill()
        child.wait()