Evaluate frozen base/adapter programs under audited nobody/seccomp runner with matched reference checks and exact quantum-state fidelity

execution/quantum_sandbox.py

@@ -0,0 +1,311 @@
+"""Fail-closed Linux worker for a restricted subset of quantum Python.
+
+AST screening is combined with nobody UID, cleared environment, process/resource
+limits and seccomp denial of filesystem opens, networking and process execution.
+Libraries are imported/warmed in the parent before any untrusted code runs.
+"""
+
+import ast
+import contextlib
+import ctypes
+import ctypes.util
+import errno
+import importlib
+import io
+import multiprocessing as mp
+import os
+import resource
+
+for variable in ["OPENBLAS_NUM_THREADS", "OMP_NUM_THREADS", "MKL_NUM_THREADS"]:
+    os.environ[variable] = "1"
+
+import numpy as np
+
+MODULES = {
+    "numpy",
+    "numpy.linalg",
+    "math",
+    "cmath",
+    "qiskit",
+    "qiskit.quantum_info",
+    "pennylane",
+    "cirq",
+}
+ATTRIBUTES = set(
+    "array asarray zeros ones eye pi sqrt exp sin cos conj conjugate transpose reshape flatten ravel squeeze astype complex64 complex128 float32 float64 norm kron real imag T shape size data QuantumCircuit Statevector from_instruction h x y z ry rx rz cx cz swap s sdg t tdg sx barrier append device qnode state RY RX RZ Rot PhaseShift Identity PauliX PauliY PauliZ Hadamard CNOT CZ SWAP LineQubit range Circuit Simulator simulate final_state_vector H X Y Z CXPowGate ZPowGate XPowGate HPowGate controlled on inverse quantum_info".split()
+)
+BUILTINS = {
+    "range": lambda *a: bounded_range(*a),
+    "enumerate": enumerate,
+    "zip": zip,
+    "len": len,
+    "list": list,
+    "tuple": tuple,
+    "float": float,
+    "int": int,
+    "complex": complex,
+    "print": print,
+    "abs": abs,
+    "sum": sum,
+    "min": min,
+    "max": max,
+    "reversed": reversed,
+}
+FORBIDDEN = (
+    ast.ClassDef,
+    ast.AsyncFunctionDef,
+    ast.Await,
+    ast.While,
+    ast.Try,
+    ast.With,
+    ast.AsyncWith,
+    ast.Raise,
+    ast.Delete,
+    ast.Global,
+    ast.Nonlocal,
+    ast.Lambda,
+    ast.Yield,
+    ast.YieldFrom,
+    ast.Match,
+)
+LIB = None
+
+
+def bounded_range(*args):
+    value = range(*args)
+    if len(value) > 1024:
+        raise ValueError("range limit")
+    return value
+
+
+def screen(program):
+    if len(program) > 25000:
+        raise ValueError("program length limit")
+    tree = ast.parse(program)
+    if sum(1 for _ in ast.walk(tree)) > 6000:
+        raise ValueError("AST node limit")
+    for node in ast.walk(tree):
+        if isinstance(node, FORBIDDEN):
+            raise ValueError("unsupported syntax: " + type(node).__name__)
+        if isinstance(node, ast.Name) and (
+            node.id.startswith("_")
+            or node.id
+            in {
+                "eval",
+                "exec",
+                "compile",
+                "open",
+                "getattr",
+                "setattr",
+                "globals",
+                "locals",
+                "vars",
+                "type",
+                "object",
+                "input",
+                "help",
+                "breakpoint",
+            }
+        ):
+            raise ValueError("forbidden name")
+        if isinstance(node, ast.Attribute) and node.attr not in ATTRIBUTES:
+            raise ValueError("unsupported attribute: " + node.attr)
+        if isinstance(node, ast.Import):
+            if any(a.name not in MODULES or (a.asname or "").startswith("_") for a in node.names):
+                raise ValueError("unsupported import")
+        if isinstance(node, ast.ImportFrom):
+            if (
+                node.level
+                or node.module not in MODULES
+                or any(
+                    a.name not in ATTRIBUTES or (a.asname or "").startswith("_") for a in node.names
+                )
+            ):
+                raise ValueError("unsupported from import")
+        if isinstance(node, ast.FunctionDef) and node.name.startswith("_"):
+            raise ValueError("forbidden function name")
+    return tree
+
+
+def initialize():
+    global LIB
+    assert os.name == "posix" and os.getuid() == 0, "Linux root parent required for UID isolation"
+    # Resolve/load seccomp before restricting filesystem access.
+    library = ctypes.util.find_library("seccomp")
+    assert library, "libseccomp is required; never fall back to AST-only execution"
+    LIB = ctypes.CDLL(library, use_errno=True)
+    LIB.seccomp_init.argtypes = [ctypes.c_uint32]
+    LIB.seccomp_init.restype = ctypes.c_void_p
+    LIB.seccomp_syscall_resolve_name.argtypes = [ctypes.c_char_p]
+    LIB.seccomp_syscall_resolve_name.restype = ctypes.c_int
+    LIB.seccomp_rule_add.argtypes = [ctypes.c_void_p, ctypes.c_uint32, ctypes.c_int, ctypes.c_uint]
+    LIB.seccomp_load.argtypes = [ctypes.c_void_p]
+    LIB.seccomp_release.argtypes = [ctypes.c_void_p]
+    from qiskit import QuantumCircuit
+    from qiskit.quantum_info import Statevector
+    import pennylane as qml
+    import cirq
+
+    qc = QuantumCircuit(2)
+    qc.h(0)
+    qc.ry(0.3, 1)
+    qc.x(0)
+    qc.z(1)
+    qc.cx(0, 1)
+    Statevector.from_instruction(qc)
+    device = qml.device("default.qubit", wires=[1, 0])
+
+    @qml.qnode(device)
+    def warm():
+        qml.Hadamard(0)
+        qml.RY(0.3, wires=1)
+        qml.PauliX(0)
+        qml.PauliZ(1)
+        qml.CNOT(wires=[0, 1])
+        return qml.state()
+
+    warm()
+    qubits = cirq.LineQubit.range(2)
+    cirq.Simulator(dtype=np.complex128).simulate(
+        cirq.Circuit(
+            cirq.H(qubits[0]),
+            cirq.ry(0.3)(qubits[1]),
+            cirq.X(qubits[0]),
+            cirq.Z(qubits[1]),
+            cirq.CNOT(*qubits),
+        ),
+        qubit_order=list(reversed(qubits)),
+    )
+    # Warm Python import machinery and serialization before fork/lockdown.
+    for module in MODULES:
+        importlib.import_module(module)
+    importlib.import_module("socket")
+    return dict(
+        qiskit=importlib.import_module("qiskit").__version__,
+        pennylane=qml.__version__,
+        cirq=cirq.__version__,
+    )
+
+
+def lockdown():
+    os.environ.clear()
+    os.setgroups([])
+    os.setgid(65534)
+    os.setuid(65534)
+    resource.setrlimit(resource.RLIMIT_CPU, (15, 15))
+    # Native numeric runtimes reserve large virtual address spaces. Physical
+    # memory remains bounded by the HF CPU job container's memory allocation.
+    resource.setrlimit(resource.RLIMIT_AS, (16 * 1024**3, 16 * 1024**3))
+    resource.setrlimit(resource.RLIMIT_FSIZE, (0, 0))
+    resource.setrlimit(resource.RLIMIT_CORE, (0, 0))
+    resource.setrlimit(resource.RLIMIT_NPROC, (32, 32))
+    libc = ctypes.CDLL(None, use_errno=True)
+    assert libc.prctl(38, 1, 0, 0, 0) == 0, "PR_SET_NO_NEW_PRIVS failed"
+    policy = LIB.seccomp_init(0x7FFF0000)  # SCMP_ACT_ALLOW
+    assert policy
+    deny = 0x00050000 | errno.EPERM  # SCMP_ACT_ERRNO(EPERM)
+    for name in [
+        "open",
+        "openat",
+        "openat2",
+        "creat",
+        "socket",
+        "socketpair",
+        "connect",
+        "accept",
+        "accept4",
+        "bind",
+        "listen",
+        "sendto",
+        "sendmsg",
+        "execve",
+        "execveat",
+        "ptrace",
+        "process_vm_readv",
+        "process_vm_writev",
+        "mount",
+        "umount2",
+        "bpf",
+        "userfaultfd",
+    ]:
+        syscall = LIB.seccomp_syscall_resolve_name(name.encode())
+        if syscall >= 0:
+            assert LIB.seccomp_rule_add(policy, deny, syscall, 0) == 0, name
+    assert LIB.seccomp_load(policy) == 0, "seccomp load failed"
+    LIB.seccomp_release(policy)
+
+
+def safe_import(name, globals=None, locals=None, fromlist=(), level=0):
+    if name not in MODULES or level:
+        raise ImportError("module not allowed")
+    module = importlib.import_module(name)
+    return module if fromlist else importlib.import_module(name.split(".")[0])
+
+
+def worker(program, qubits, connection, mode):
+    try:
+        lockdown()
+        if mode == "isolation_check":
+            import socket
+
+            file_blocked = network_blocked = False
+            try:
+                open("/etc/hostname").read()
+            except PermissionError:
+                file_blocked = True
+            try:
+                socket.socket()
+            except PermissionError:
+                network_blocked = True
+            connection.send(
+                dict(
+                    file_opens_blocked=file_blocked,
+                    network_blocked=network_blocked,
+                    uid=os.getuid(),
+                    environment_empty=not os.environ,
+                )
+            )
+            return
+        tree = screen(program)
+        safe = dict(BUILTINS, __import__=safe_import)
+        namespace = {"__builtins__": safe}
+        with contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(io.StringIO()):
+            exec(compile(tree, "<restricted-quantum>", "exec"), namespace, namespace)
+        value = np.asarray(namespace["state"], dtype=np.complex128).reshape(-1)
+        assert value.shape == (2**qubits,) and np.isfinite(value).all()
+        connection.send(
+            dict(
+                status="executed",
+                real=value.real.tolist(),
+                imag=value.imag.tolist(),
+                norm=float(np.linalg.norm(value)),
+            )
+        )
+    except BaseException as error:
+        connection.send(
+            dict(status="rejected_or_error", error=type(error).__name__, detail=str(error)[:300])
+        )
+    finally:
+        connection.close()
+
+
+def run(program, qubits, mode="program"):
+    assert 1 <= qubits <= 8
+    ctx = mp.get_context("fork")
+    receive, send = ctx.Pipe(duplex=False)
+    process = ctx.Process(target=worker, args=(program, qubits, send, mode))
+    process.start()
+    send.close()
+    if receive.poll(20):
+        try:
+            result = receive.recv()
+        except EOFError:
+            result = dict(status="worker_exit", exitcode=process.exitcode)
+    else:
+        result = dict(status="timeout")
+    process.join(1)
+    if process.is_alive():
+        process.kill()
+        process.join(2)
+    receive.close()
+    return result

execution/train_quantum_execute_pilot.py

@@ -0,0 +1,192 @@
+"""Independent frozen-executor evaluation of base/adapted quantum programs."""
+
+import ast
+import hashlib
+import json
+import re
+from collections import defaultdict
+from pathlib import Path
+import numpy as np
+import pyarrow.parquet as pq
+from huggingface_hub import HfApi, hf_hub_download
+import quantum_sandbox as sandbox
+
+REPO = "itsalissonsilva/QuantumCoder-9B"
+
+
+def main():
+    api = HfApi()
+    revision = api.model_info(REPO).sha
+
+    def get(path):
+        return Path(hf_hub_download(REPO, path, revision=revision))
+
+    preflight = json.loads(get("executor_preflight/preflight.json").read_text())
+    assert preflight["all_passed"] and len(preflight["reference_checks"]) == 90
+    source_hash = hashlib.sha256(Path(sandbox.__file__).read_bytes()).hexdigest()
+    assert (
+        source_hash
+        == hashlib.sha256(get("executor_preflight/quantum_sandbox.py").read_bytes()).hexdigest()
+    ), "Do not change the executor after opening model programs"
+    generations = json.loads(get("hf_pilot/generation_report.json").read_text())
+    data = {}
+    for split in ["validation", "test"]:
+        rows = pq.read_table(
+            hf_hub_download(
+                generations["data_repo"],
+                f"data/{split}.parquet",
+                repo_type="dataset",
+                revision=generations["data_revision"],
+            )
+        ).to_pylist()
+        data.update({r["id"]: r for r in rows})
+    ids = [r["id"] for r in generations["conditions"]["base"]]
+    assert len(ids) == 192 and len(set(ids)) == 192
+    assert ids == [r["id"] for r in generations["conditions"]["adapted"]]
+    versions = sandbox.initialize()
+    isolation = sandbox.run("", 1, mode="isolation_check")
+    assert isolation == preflight["isolation"]
+
+    def execute(program, row):
+        # Unwrap only a single full-output code fence. No prose, repairs or
+        # target-informed program edits are allowed.
+        wrapped = re.fullmatch(r"\s*```(?:python)?\s*\n(.*?)\n```\s*", program, flags=re.S)
+        clean = wrapped.group(1) if wrapped else program
+        syntax = True
+        imports = set()
+        try:
+            tree = ast.parse(clean)
+            for node in ast.walk(tree):
+                if isinstance(node, ast.Import):
+                    imports.update(a.name.split(".")[0] for a in node.names)
+                elif isinstance(node, ast.ImportFrom):
+                    imports.add((node.module or "").split(".")[0])
+        except SyntaxError:
+            syntax = False
+        result = sandbox.run(clean, row["qubits"])
+        fidelity = None
+        state_pass = False
+        if result.get("status") == "executed":
+            actual = np.array(result["real"]) + 1j * np.array(result["imag"])
+            target = np.array(row["state_real"]) + 1j * np.array(row["state_imag"])
+            fidelity = float(abs(np.vdot(target, actual)) ** 2)
+            state_pass = abs(result["norm"] - 1) < 1e-6 and fidelity > 1 - 1e-6
+        requested = row["framework"] in imports
+        execution = {k: v for k, v in result.items() if k not in ["real", "imag"]}
+        return dict(
+            syntax_parses=syntax,
+            requested_framework_import=requested,
+            single_fence_unwrapped=wrapped is not None,
+            execution=execution,
+            state_pass=state_pass,
+            fidelity=fidelity,
+            passed=state_pass and requested,
+            program_sha256=hashlib.sha256(program.encode()).hexdigest(),
+        )
+
+    # Every selected reference must pass the same evaluator before scoring models.
+    reference_checks = []
+    for identifier in ids:
+        row = data[identifier]
+        outcome = execute(row["program"], row)
+        assert outcome["passed"], (identifier, outcome)
+        reference_checks.append(dict(id=identifier, **outcome))
+    records = []
+    for condition, outputs in generations["conditions"].items():
+        for output in outputs:
+            row = data[output["id"]]
+            assert row["framework"] == output["framework"] and row["qubits"] == output["qubits"]
+            assert np.allclose(row["state_real"], output["target_real"], atol=1e-12, rtol=0)
+            assert np.allclose(row["state_imag"], output["target_imag"], atol=1e-12, rtol=0)
+            records.append(
+                dict(
+                    condition=condition,
+                    id=row["id"],
+                    group_id=row["group_id"],
+                    split=output["split"],
+                    framework=row["framework"],
+                    generated_tokens=output["generated_tokens"],
+                    possible_token_cap=output["generated_tokens"] >= 512,
+                    **execute(output["program"], row),
+                )
+            )
+        print(
+            json.dumps(
+                dict(
+                    stage="scored",
+                    condition=condition,
+                    passed=sum(r["passed"] for r in records if r["condition"] == condition),
+                )
+            ),
+            flush=True,
+        )
+    summary = {}
+    for condition in ["base", "adapted"]:
+        summary[condition] = {}
+        for split in ["validation", "test"]:
+            chosen = [r for r in records if r["condition"] == condition and r["split"] == split]
+            summary[condition][split] = dict(
+                n=len(chosen),
+                passed=sum(r["passed"] for r in chosen),
+                pass_at_1=float(np.mean([r["passed"] for r in chosen])),
+                state_correct=sum(r["state_pass"] for r in chosen),
+                executed=sum(r["execution"].get("status") == "executed" for r in chosen),
+                syntax_valid=sum(r["syntax_parses"] for r in chosen),
+                requested_import=sum(r["requested_framework_import"] for r in chosen),
+                possible_token_cap=sum(r["possible_token_cap"] for r in chosen),
+            )
+    base = {r["id"]: r for r in records if r["condition"] == "base" and r["split"] == "test"}
+    adapted = {r["id"]: r for r in records if r["condition"] == "adapted" and r["split"] == "test"}
+    keys = sorted(base)
+    delta = np.array([int(adapted[k]["passed"]) - int(base[k]["passed"]) for k in keys])
+    groups = defaultdict(list)
+    for i, k in enumerate(keys):
+        groups[base[k]["group_id"]].append(i)
+    indices = list(groups.values())
+    rng = np.random.default_rng(20261031)
+    boot = [
+        float(delta[np.concatenate([indices[i] for i in draw])].mean())
+        for draw in rng.integers(0, len(indices), size=(2000, len(indices)))
+    ]
+    report = dict(
+        source_revision=revision,
+        executor_sha256=source_hash,
+        versions=versions,
+        isolation=isolation,
+        data_revision=generations["data_revision"],
+        reference_checks=reference_checks,
+        records=records,
+        summary=summary,
+        paired_test=dict(
+            improved=int(np.sum(delta == 1)),
+            regressed=int(np.sum(delta == -1)),
+            mean_pass_delta=float(delta.mean()),
+            group_bootstrap_95=np.quantile(boot, [0.025, 0.975]).tolist(),
+            groups=len(indices),
+        ),
+        limitations="One synthetic held-out-framework experiment. Pass requires requested-framework import, supported-Python execution, unit norm and global-phase-invariant exact-state fidelity >1-1e-6. Import presence is a proxy for framework compliance, not a formal data-flow proof. Unsupported valid Python is rejected and can yield conservative false negatives. Only a full-output code fence may be removed. No repairs, retries or generated-code access to files/network. Original batch-width at_token_cap flag is unreliable and not used; generated_tokens>=512 is reported as a possible cap, not an exact stop reason.",
+    )
+    report["release_gate_pass"] = (
+        summary["adapted"]["test"]["passed"] > summary["base"]["test"]["passed"]
+    )
+    folder = Path("quantum_execution")
+    folder.mkdir(exist_ok=True)
+    (folder / "execution.json").write_text(json.dumps(report, indent=2))
+    (folder / "train_quantum_execute_pilot.py").write_bytes(Path(__file__).read_bytes())
+    (folder / "quantum_sandbox.py").write_bytes(Path(sandbox.__file__).read_bytes())
+    api.upload_folder(
+        repo_id=REPO,
+        folder_path=str(folder),
+        path_in_repo="execution",
+        commit_message="Evaluate frozen base/adapter programs under audited nobody/seccomp runner with matched reference checks and exact quantum-state fidelity",
+    )
+    print(
+        json.dumps(
+            dict(summary=summary, gate=report["release_gate_pass"], paired=report["paired_test"])
+        ),
+        flush=True,
+    )
+
+
+if __name__ == "__main__":
+    main()