From 1cc08b7785a65b458c0211bb748e0427459d119d Mon Sep 17 00:00:00 2001 From: Sylvester Kaczmarek <16242628+sylvesterkaczmarek@users.noreply.github.com> Date: Sun, 9 Aug 2026 14:03:47 +0100 Subject: [PATCH 1/2] Rebuild as CUDA-Q execution guardrail toolkit --- .github/workflows/ci.yml | 51 +++ .gitignore | 212 +---------- CITATION.cff | 20 +- CONTRIBUTING.md | 31 ++ LICENSE | 2 +- Makefile | 23 ++ README.md | 346 ++++++++++++++---- SECURITY.md | 5 + ...ocial-card-nvidia-cudaq-quantum-guard.png} | Bin docs/audit-format.md | 18 + docs/policy-reference.md | 27 ++ docs/reproducibility.md | 16 + docs/security-model.md | 26 ++ examples/library_integration.py | 34 ++ hybrid_secure_demo.py | 181 --------- policies/local-safe.toml | 13 + policies/remote-explicit.toml | 23 ++ pyproject.toml | 44 +++ requirements.txt | 6 +- src/cudaq_guard/__init__.py | 20 + src/cudaq_guard/__main__.py | 3 + src/cudaq_guard/audit.py | 71 ++++ src/cudaq_guard/cli.py | 247 +++++++++++++ src/cudaq_guard/crypto.py | 29 ++ src/cudaq_guard/doctor.py | 77 ++++ src/cudaq_guard/errors.py | 14 + src/cudaq_guard/guard.py | 137 +++++++ src/cudaq_guard/models.py | 50 +++ src/cudaq_guard/policy.py | 156 ++++++++ src/cudaq_guard/runtime.py | 100 +++++ src/cudaq_guard/workloads.py | 60 +++ tests/test_audit.py | 27 ++ tests/test_cli.py | 28 ++ tests/test_crypto.py | 8 + tests/test_doctor.py | 33 ++ tests/test_guard.py | 130 +++++++ tests/test_policy.py | 91 +++++ 37 files changed, 1887 insertions(+), 472 deletions(-) create mode 100644 .github/workflows/ci.yml create mode 100644 CONTRIBUTING.md create mode 100644 Makefile create mode 100644 SECURITY.md rename assets/social/{github-social-card-nvidia-quantum-hybrid.png => github-social-card-nvidia-cudaq-quantum-guard.png} (100%) create mode 100644 docs/audit-format.md create mode 100644 docs/policy-reference.md create mode 100644 docs/reproducibility.md create mode 100644 docs/security-model.md create mode 100644 examples/library_integration.py delete mode 100644 hybrid_secure_demo.py create mode 100644 policies/local-safe.toml create mode 100644 policies/remote-explicit.toml create mode 100644 pyproject.toml create mode 100644 src/cudaq_guard/__init__.py create mode 100644 src/cudaq_guard/__main__.py create mode 100644 src/cudaq_guard/audit.py create mode 100644 src/cudaq_guard/cli.py create mode 100644 src/cudaq_guard/crypto.py create mode 100644 src/cudaq_guard/doctor.py create mode 100644 src/cudaq_guard/errors.py create mode 100644 src/cudaq_guard/guard.py create mode 100644 src/cudaq_guard/models.py create mode 100644 src/cudaq_guard/policy.py create mode 100644 src/cudaq_guard/runtime.py create mode 100644 src/cudaq_guard/workloads.py create mode 100644 tests/test_audit.py create mode 100644 tests/test_cli.py create mode 100644 tests/test_crypto.py create mode 100644 tests/test_doctor.py create mode 100644 tests/test_guard.py create mode 100644 tests/test_policy.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..8e829a1 --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,51 @@ +name: CI + +on: + push: + pull_request: + +jobs: + unit: + runs-on: ubuntu-latest + strategy: + matrix: + python-version: ["3.11", "3.12", "3.13"] + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python-version }} + cache: pip + - name: Install + run: | + python -m pip install --upgrade pip + python -m pip install -e ".[dev]" + - name: Compile + run: python -m compileall -q src tests examples + - name: Test + run: python -m pytest + - name: Build package + if: matrix.python-version == '3.12' + run: python -m build + + cudaq-smoke: + runs-on: ubuntu-latest + timeout-minutes: 25 + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: "3.12" + cache: pip + - name: Install CUDA-Q + run: | + python -m pip install --upgrade pip + python -m pip install -e ".[cudaq,dev]" + - name: Diagnose + run: cudaq-guard doctor --json + - name: CPU simulator smoke test + run: | + cudaq-guard run ghz --policy policies/local-safe.toml --target qpp-cpu --qubits 3 --shots 256 --seed 7 --audit /tmp/cudaq-guard-audit.jsonl + cudaq-guard audit verify /tmp/cudaq-guard-audit.jsonl + - name: VQE smoke test + run: cudaq-guard run vqe --policy policies/local-safe.toml --target qpp-cpu --steps 7 --seed 7 --audit /tmp/cudaq-guard-audit.jsonl diff --git a/.gitignore b/.gitignore index b7faf40..d0831ec 100644 --- a/.gitignore +++ b/.gitignore @@ -1,207 +1,15 @@ -# Byte-compiled / optimized / DLL files __pycache__/ -*.py[codz] -*$py.class - -# C extensions +*.py[cod] *.so - -# Distribution / packaging -.Python -build/ -develop-eggs/ -dist/ -downloads/ -eggs/ -.eggs/ -lib/ -lib64/ -parts/ -sdist/ -var/ -wheels/ -share/python-wheels/ *.egg-info/ -.installed.cfg -*.egg -MANIFEST - -# PyInstaller -# Usually these files are written by a python script from a template -# before PyInstaller builds the exe, so as to inject date/other infos into it. -*.manifest -*.spec - -# Installer logs -pip-log.txt -pip-delete-this-directory.txt - -# Unit test / coverage reports -htmlcov/ -.tox/ -.nox/ -.coverage -.coverage.* -.cache -nosetests.xml -coverage.xml -*.cover -*.py.cover -.hypothesis/ .pytest_cache/ -cover/ - -# Translations -*.mo -*.pot - -# Django stuff: -*.log -local_settings.py -db.sqlite3 -db.sqlite3-journal - -# Flask stuff: -instance/ -.webassets-cache - -# Scrapy stuff: -.scrapy - -# Sphinx documentation -docs/_build/ - -# PyBuilder -.pybuilder/ -target/ - -# Jupyter Notebook -.ipynb_checkpoints - -# IPython -profile_default/ -ipython_config.py - -# pyenv -# For a library or package, you might want to ignore these files since the code is -# intended to run in multiple environments; otherwise, check them in: -# .python-version - -# pipenv -# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control. -# However, in case of collaboration, if having platform-specific dependencies or dependencies -# having no cross-platform support, pipenv may install dependencies that don't work, or not -# install all needed dependencies. -#Pipfile.lock - -# UV -# Similar to Pipfile.lock, it is generally recommended to include uv.lock in version control. -# This is especially recommended for binary packages to ensure reproducibility, and is more -# commonly ignored for libraries. -#uv.lock - -# poetry -# Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control. -# This is especially recommended for binary packages to ensure reproducibility, and is more -# commonly ignored for libraries. -# https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control -#poetry.lock -#poetry.toml - -# pdm -# Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control. -# pdm recommends including project-wide configuration in pdm.toml, but excluding .pdm-python. -# https://pdm-project.org/en/latest/usage/project/#working-with-version-control -#pdm.lock -#pdm.toml -.pdm-python -.pdm-build/ - -# pixi -# Similar to Pipfile.lock, it is generally recommended to include pixi.lock in version control. -#pixi.lock -# Pixi creates a virtual environment in the .pixi directory, just like venv module creates one -# in the .venv directory. It is recommended not to include this directory in version control. -.pixi - -# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm -__pypackages__/ - -# Celery stuff -celerybeat-schedule -celerybeat.pid - -# SageMath parsed files -*.sage.py - -# Environments -.env -.envrc -.venv -env/ +.venv/ venv/ -ENV/ -env.bak/ -venv.bak/ - -# Spyder project settings -.spyderproject -.spyproject - -# Rope project settings -.ropeproject - -# mkdocs documentation -/site - -# mypy -.mypy_cache/ -.dmypy.json -dmypy.json - -# Pyre type checker -.pyre/ - -# pytype static type analyzer -.pytype/ - -# Cython debug symbols -cython_debug/ - -# PyCharm -# JetBrains specific template is maintained in a separate JetBrains.gitignore that can -# be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore -# and can be added to the global gitignore or merged into this file. For a more nuclear -# option (not recommended) you can uncomment the following to ignore the entire idea folder. -#.idea/ - -# Abstra -# Abstra is an AI-powered process automation framework. -# Ignore directories containing user credentials, local state, and settings. -# Learn more at https://abstra.io/docs -.abstra/ - -# Visual Studio Code -# Visual Studio Code specific template is maintained in a separate VisualStudioCode.gitignore -# that can be found at https://github.com/github/gitignore/blob/main/Global/VisualStudioCode.gitignore -# and can be added to the global gitignore or merged into this file. However, if you prefer, -# you could uncomment the following to ignore the entire vscode folder -# .vscode/ - -# Ruff stuff: -.ruff_cache/ - -# PyPI configuration file -.pypirc - -# Cursor -# Cursor is an AI-powered code editor. `.cursorignore` specifies files/directories to -# exclude from AI features like autocomplete and code analysis. Recommended for sensitive data -# refer to https://docs.cursor.com/context/ignore-files -.cursorignore -.cursorindexingignore - -# Marimo -marimo/_static/ -marimo/_lsp/ -__marimo__/ +dist/ +build/ +.DS_Store +.idea/ +.vscode/ +*.log +runs/ +*.jsonl diff --git a/CITATION.cff b/CITATION.cff index 8095e86..9ab80f4 100644 --- a/CITATION.cff +++ b/CITATION.cff @@ -1,21 +1,25 @@ cff-version: 1.2.0 -title: NVIDIA Quantum Hybrid -message: "If you use this repository, please cite it." +title: NVIDIA CUDA-Q Quantum Guard +message: "If you use or adapt this software, please cite it." +type: software +version: 0.4.0 authors: - family-names: Kaczmarek given-names: Sylvester orcid: https://orcid.org/0000-0002-0393-228X -doi: 10.5281/zenodo.17502919 -date-released: 2025-11-01 -url: https://github.com/sylvesterkaczmarek/nvidia-quantum-hybrid license: MIT -type: software +doi: 10.5281/zenodo.17502919 +url: https://github.com/sylvesterkaczmarek/nvidia-cudaq-quantum-guard +repository-code: https://github.com/sylvesterkaczmarek/nvidia-cudaq-quantum-guard preferred-citation: - type: article - title: NVIDIA Quantum Hybrid + type: software + title: NVIDIA CUDA-Q Quantum Guard authors: - family-names: Kaczmarek given-names: Sylvester orcid: https://orcid.org/0000-0002-0393-228X year: 2025 + publisher: + name: Zenodo doi: 10.5281/zenodo.17502919 + url: https://doi.org/10.5281/zenodo.17502919 diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md new file mode 100644 index 0000000..50bdf23 --- /dev/null +++ b/CONTRIBUTING.md @@ -0,0 +1,31 @@ +# Contributing + +Contributions are welcome when they improve practical CUDA-Q execution control, diagnostics, provenance, or reproducibility. + +## Development setup + +```bash +python -m venv .venv +source .venv/bin/activate +python -m pip install -e ".[dev]" +python -m pytest +``` + +For real CUDA-Q smoke tests: + +```bash +python -m pip install -e ".[cudaq,dev]" +cudaq-guard doctor +make smoke +``` + +## Design rules + +- fail closed rather than silently changing target or policy +- keep credentials out of logs and fixtures +- do not claim a security property that the code does not enforce +- keep core policy/audit code usable without a CUDA-Q installation +- add regression tests for policy semantics and execution adapters +- keep provider-specific assumptions behind explicit configuration + +If a change alters the audit schema or policy semantics, document the compatibility impact in the pull request. diff --git a/LICENSE b/LICENSE index 9c25cf9..85e4e68 100644 --- a/LICENSE +++ b/LICENSE @@ -1,6 +1,6 @@ MIT License -Copyright (c) 2025 Sylvester Kaczmarek +Copyright (c) 2026 Sylvester Kaczmarek Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal diff --git a/Makefile b/Makefile new file mode 100644 index 0000000..5bebbeb --- /dev/null +++ b/Makefile @@ -0,0 +1,23 @@ +.PHONY: install install-cudaq test smoke doctor build clean + +install: + python -m pip install -e ".[dev]" + +install-cudaq: + python -m pip install -e ".[cudaq,dev]" + +test: + python -m pytest + +smoke: + cudaq-guard run ghz --policy policies/local-safe.toml --target qpp-cpu --qubits 3 --shots 256 --seed 7 --audit /tmp/cudaq-guard-audit.jsonl + cudaq-guard audit verify /tmp/cudaq-guard-audit.jsonl + +doctor: + cudaq-guard doctor + +build: + python -m build + +clean: + rm -rf build dist *.egg-info .pytest_cache diff --git a/README.md b/README.md index 33690c3..3616779 100644 --- a/README.md +++ b/README.md @@ -1,125 +1,317 @@ -# NVIDIA Quantum Hybrid -[![Python 3.12+](https://img.shields.io/badge/Python-3.12%2B-blue.svg)](https://www.python.org/downloads/) -[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](LICENSE) +# NVIDIA CUDA-Q Quantum Guard + +![NVIDIA CUDA-Q Quantum Guard](assets/social/github-social-card-nvidia-cudaq-quantum-guard.png) + +[![CI](https://github.com/sylvesterkaczmarek/nvidia-cudaq-quantum-guard/actions/workflows/ci.yml/badge.svg)](https://github.com/sylvesterkaczmarek/nvidia-cudaq-quantum-guard/actions/workflows/ci.yml) +![Python](https://img.shields.io/badge/Python-3.11%2B-3776AB?logo=python&logoColor=white) +![CUDA-Q](https://img.shields.io/badge/CUDA--Q-0.15%2B-76B900?logo=nvidia&logoColor=white) +![License](https://img.shields.io/badge/License-MIT-yellow.svg) [![DOI](https://zenodo.org/badge/DOI/10.5281/zenodo.17502919.svg)](https://doi.org/10.5281/zenodo.17502919) -![NVIDIA Quantum Hybrid](assets/social/github-social-card-nvidia-quantum-hybrid.png) +Policy-controlled execution, diagnostics, backend comparison, and tamper-evident audit trails for NVIDIA CUDA-Q. The `cudaq-guard` CLI is intended for researchers and engineers who want explicit controls around CPU, NVIDIA GPU, multi-QPU, and remote-QPU execution rather than scattering target and safety checks throughout application code. + +This is independent open-source software. It is not an NVIDIA product and is not affiliated with or endorsed by NVIDIA. + +## At a glance + +```mermaid +flowchart LR + A[Application or CLI] --> B[Execution request] + B --> C[Fail-closed TOML policy] + C -->|deny| D[Audited denial] + C -->|allow| E[CUDA-Q target] + E --> F[CPU simulator] + E --> G[NVIDIA GPU] + E --> H[Remote QPU] + F --> I[Result summary] + G --> I + H --> I + I --> J[Tamper-evident JSONL audit] +``` -This repository shows a small classical–quantum–classical workflow with an explicit safety gate. It is written to match the current NVIDIA work on quantum to GPU hybrid computing and to show how AI safety and AI security logic can control access to quantum routines. +CUDA-Q already provides a unified programming model across CPUs, GPUs, and QPUs. This repository focuses on a different layer: **how an application decides what quantum work is allowed to run, where it is allowed to run, and what evidence is retained afterwards.** -Official NVIDIA announcement -https://nvidianews.nvidia.com/news/nvidia-nvqlink-quantum-gpu-computing +## Why this is useful -## Project overview +A backend switch that is convenient in a research notebook can become risky in shared infrastructure. A typo, stale configuration, or unreviewed target option can change cost, data handling, resource use, or the system that receives a job. -- Classical stage prepares and normalises data. -- A policy checks whether the data stays inside an approved envelope. -- If the policy allows it, the quantum step is executed. -- The pipeline emits structured JSON with policy, backend and latency fields. This is suitable for audit, MLOps and security monitoring. +NVIDIA CUDA-Q Quantum Guard adds a small policy-as-code layer for CUDA-Q: -This is a simple reference that connects current hybrid quantum–GPU news with secure and trustworthy AI concepts. +- allowlist `sample`, `observe`, or other execution operations +- allowlist specific CUDA-Q targets +- distinguish local simulator targets from remote or hardware targets +- bound qubits, shots, QPU IDs, asynchronous execution, and target options +- preflight built-in kernels with CUDA-Q resource estimation before execution +- require deterministic simulator seeds where appropriate +- audit both allowed and denied requests +- chain audit records with SHA-256 so later modification is detectable +- diagnose installed CUDA-Q targets, CUDA-Q GPU visibility, and NVIDIA driver visibility +- compare the same guarded workload across CPU and GPU targets -## Why this is useful +## What changed from the original prototype -- Shows how to put a safety/policy check in front of a quantum call, which is what you want in secure AI or space/defence contexts. -- Produces JSON that can be logged, audited or sent to an MLOps/SOC pipeline, so it is easy to demo to non-quantum teams. -- Can be swapped from a simulator to CUDA-Q or a partner QPU without changing the classical logic, so it is future-ready. +The original repository used a one-qubit Qiskit Aer demonstration and described CUDA-Q as a future backend. Version `0.4.0` reverses that architecture: -## Features +- CUDA-Q is now the native quantum runtime +- Qiskit is no longer required +- unknown policies and target options fail closed +- remote execution requires explicit authorization +- simulator randomness can be seeded +- results are called what they are: samples or expectation values, not "quantum confidence" +- the old post-hoc probability "noise" calculation is removed +- policy and execution logic are packaged and tested instead of living in one demo script +- audit records are tamper-evident rather than plain `print()` output -- classical → policy → quantum → classical loop -- Qiskit Aer simulator as default backend -- registry of policies (small, strict) -- JSON output with pipeline version, policy name, quantum backend, latency -- ready to swap to CUDA-Q or to a real neutral-atom or photonic backend +## Install -## Requirements +The policy, audit, and static validation tools have no mandatory third-party dependencies: -```text -numpy -qiskit -qiskit-aer +```bash +git clone https://github.com/sylvesterkaczmarek/nvidia-cudaq-quantum-guard.git +cd nvidia-cudaq-quantum-guard +python -m venv .venv +source .venv/bin/activate +python -m pip install -e ".[dev]" +``` + +Install CUDA-Q support as well: + +```bash +python -m pip install -e ".[cudaq,dev]" +``` + +CUDA-Q itself is distributed by NVIDIA as the `cudaq` Python package. GPU acceleration requires a supported NVIDIA GPU/runtime; CUDA-Q can also run CPU simulation without a GPU. + +## First commands + +Diagnose the machine and list CUDA-Q targets: + +```bash +cudaq-guard doctor +cudaq-guard targets ``` -Install +Run a guarded GHZ workload on the CPU simulator: ```bash -pip install -r requirements.txt +cudaq-guard run ghz \ + --policy policies/local-safe.toml \ + --target qpp-cpu \ + --qubits 4 \ + --shots 1000 \ + --seed 7 ``` -You can also install manually +Verify the resulting audit chain: ```bash -pip install numpy qiskit qiskit-aer +cudaq-guard audit verify runs/audit.jsonl ``` -## Run +On a Linux host with a supported NVIDIA GPU, use the same workload and policy with the GPU target: ```bash -python hybrid_secure_demo.py +cudaq-guard run ghz --policy policies/local-safe.toml --target nvidia --qubits 20 --shots 1000 --seed 7 ``` -The script runs two examples. One passes the policy and calls the quantum circuit. The other fails the policy and does not call the quantum circuit. +Compare targets without changing the workload: + +```bash +cudaq-guard compare --policy policies/local-safe.toml --targets qpp-cpu,nvidia --qubits 20 --shots 1000 --seed 7 +``` + +## Policy example + +`policies/local-safe.toml` permits bounded local simulation but blocks remote hardware: + +```toml +version = 1 +name = "local-safe" +allowed_operations = ["sample", "observe"] +allowed_targets = ["qpp-cpu", "nvidia"] +allow_remote = false +allow_async = false +max_qubits = 28 +max_shots = 100000 +allowed_qpu_ids = [0] +require_seed = true + +[target_options.nvidia] +option = ["mqpu", "fp64"] +``` + +Unknown policy fields, unsupported policy versions, unlisted targets, and unlisted target-option values are rejected rather than silently defaulted. + +See [docs/policy-reference.md](docs/policy-reference.md). + +## Library integration + +The guard can wrap application-owned CUDA-Q kernels rather than only the built-in CLI examples: + +```python +from cudaq_guard import ExecutionRequest, Guard, GuardPolicy +from cudaq_guard.runtime import CudaQRuntime + +runtime = CudaQRuntime() +policy = GuardPolicy.from_toml("policies/local-safe.toml") +guard = Guard(policy, audit_path="runs/audit.jsonl", runtime=runtime) + +request = ExecutionRequest( + workload="my-kernel", + operation="sample", + target="qpp-cpu", + qubits=5, + shots=1000, + seed=7, +) + +result = guard.execute( + request, + lambda rt: rt.sample(my_cudaq_kernel, shots=1000), +) +``` + +A complete example is in [`examples/library_integration.py`](examples/library_integration.py). The example also supplies a `resource_probe`, so CUDA-Q's compiled resource estimate is checked against the declared and policy qubit limits before the kernel executes. + +## Built-in workloads + +### GHZ sampling + +Useful for validating installation, target selection, finite-shot execution, audit behavior, and CPU/GPU comparison. -## File layout +### H2 VQE grid + +A small two-qubit variational workload exercises CUDA-Q `observe` and the classical-quantum loop without adding an optimizer dependency: + +```bash +cudaq-guard run vqe --policy policies/local-safe.toml --target qpp-cpu --steps 25 --seed 7 +``` + +It is an execution-path example, not a chemistry benchmark. + +## Remote and asynchronous execution + +CUDA-Q supports asynchronous submission to multi-QPU simulators and hardware providers. The guard exposes the asynchronous flag to policy so a deployment can explicitly decide whether that mode is allowed. + +For example, `policies/remote-explicit.toml` permits only named remote targets and asynchronous execution. Provider accounts, credentials, costs, and device-specific options remain the user's responsibility. + +Credential-like target options are redacted from local audit records. Do not place provider secrets directly in policy files. + +## Tamper-evident audit + +Every record contains the hash of the preceding record and its own canonical SHA-256 hash: + +```text +genesis -> record 1 -> record 2 -> record 3 -> ... +``` + +This detects local record modification or deletion within the observed chain. It is deliberately described as **tamper-evident**, not tamper-proof: a party able to rewrite the whole file can recompute hashes. Sign or externally anchor audit heads when stronger provenance is required. + +See [docs/audit-format.md](docs/audit-format.md). + +## Security model + +The project provides execution controls around calls that go through the guard. It is **not** a Python sandbox and cannot stop arbitrary application code from importing CUDA-Q and bypassing the guard. + +For higher-assurance deployments, enforce the guard at a process/service boundary and restrict direct provider/runtime access. See [docs/security-model.md](docs/security-model.md). + +## CUDA-Q compatibility + +The implementation targets the current CUDA-Q Python API used for: + +- `cudaq.get_targets()` / `cudaq.get_target()` +- `cudaq.set_target()` +- `cudaq.sample()` and `cudaq.sample_async()` +- `cudaq.observe()` +- `cudaq.set_random_seed()` +- `qpp-cpu` CPU simulation +- `nvidia` GPU simulation, including policy-controlled target options such as `mqpu` + +Primary documentation: + +- [CUDA-Q quick start](https://nvidia.github.io/cuda-quantum/latest/using/quick_start.html) +- [CUDA-Q execution](https://nvidia.github.io/cuda-quantum/latest/using/examples/executing_kernels.html) +- [CUDA-Q simulators](https://nvidia.github.io/cuda-quantum/latest/using/simulators.html) +- [CUDA-Q hardware providers](https://nvidia.github.io/cuda-quantum/latest/using/backends/hardware.html) + +## Repository layout ```text -nvidia-quantum-hybrid/ +nvidia-cudaq-quantum-guard/ +├── .github/workflows/ci.yml +├── docs/ +│ ├── audit-format.md +│ ├── policy-reference.md +│ ├── reproducibility.md +│ └── security-model.md +├── examples/ +│ └── library_integration.py +├── policies/ +│ ├── local-safe.toml +│ └── remote-explicit.toml +├── src/cudaq_guard/ +│ ├── audit.py +│ ├── cli.py +│ ├── doctor.py +│ ├── guard.py +│ ├── policy.py +│ ├── runtime.py +│ └── workloads.py +├── tests/ +├── CITATION.cff +├── LICENSE +├── Makefile +├── pyproject.toml ├── requirements.txt -├── hybrid_secure_demo.py +├── SECURITY.md └── README.md ``` -- `hybrid_secure_demo.py` is the main demo. -- `requirements.txt` keeps the environment minimal. - -## Example output - -```json -{ - "pipeline_version": "0.3-nvidia-quantum-hybrid", - "policy_used": "small", - "policy_tag": "ok", - "features_meta": { - "mean": 1.012, - "std": 0.04, - "anomaly_score": 0.26 - }, - "quantum_called": true, - "quantum_backend": "qiskit_sim", - "quantum_confidence": 0.462, - "noise_level": 0.03, - "reason": "ok", - "explain": "policy=small, mean=1.012, theta=0.222", - "latency_s": 0.152 -} -``` +## Validation + +The unit test suite covers policy denial paths, remote-target gating, target-option allowlists, deterministic policy hashing, audit-chain verification and tamper detection, credential redaction, guarded execution, denial-before-execution, environment diagnostics, and CLI policy checks. + +GitHub Actions additionally installs NVIDIA CUDA-Q on Linux and runs real `qpp-cpu` GHZ and VQE smoke workloads. No real QPU execution occurs in CI. + +## Reproducibility + +See [docs/reproducibility.md](docs/reproducibility.md). Simulator examples use explicit seeds and record the CUDA-Q/Python versions in audit evidence. + +## What this repository does not claim + +- It is not an NVIDIA product or security boundary inside CUDA-Q. +- It does not prove that arbitrary Python code truthfully declared qubit/resource metadata. +- It does not formally verify quantum kernels. +- It does not manage cloud/QPU credentials or provider billing. +- It does not make real QPU execution deterministic. +- A hash chain alone does not make local logs immutable or cryptographically authentic. +- Passing the included tests is not evidence of flight, safety-critical, or high-assurance certification. ## Extending -- edit `run_quantum(...)` to call a different backend -- add more policies to the `POLICIES` dictionary -- expose the function as an HTTP service -- add stronger logging instead of `print` -- add provenance or signature checks before the quantum call +Useful next integrations include signed policies, external audit anchoring, persistent CUDA-Q asynchronous job references, organization-specific provider approval plugins, and scheduler/HPC adapters. Contributions should preserve the fail-closed behavior and avoid silently falling back to a different target. -## Cite this demo +## Requirements -If you use or adapt this repository, please cite +- Python 3.11+ +- CUDA-Q 0.15+ for quantum execution +- Linux x86_64/ARM64 or macOS ARM64 according to CUDA-Q platform support; GPU simulation is Linux-only +- no NVIDIA GPU is required for `qpp-cpu` -> Kaczmarek, S. (2025). *NVIDIA Quantum Hybrid*. Zenodo. https://doi.org/10.5281/zenodo.17502919 +## Cite this repository -[![DOI](https://zenodo.org/badge/DOI/10.5281/zenodo.17502919.svg)](https://doi.org/10.5281/zenodo.17502919) +If you use or adapt this repository, please cite + +> Kaczmarek, S. (2025). *NVIDIA CUDA-Q Quantum Guard*. Zenodo. https://doi.org/10.5281/zenodo.17502919 -**BibTeX** ```bibtex -@software{Kaczmarek_2025_NVIDIA_Quantum_Hybrid, +@software{Kaczmarek_2025_NVIDIA_CUDAQ_Quantum_Guard, author = {Sylvester Kaczmarek}, - title = {{NVIDIA Quantum Hybrid}}, + title = {{NVIDIA CUDA-Q Quantum Guard}}, year = {2025}, publisher = {Zenodo}, - url = {https://github.com/sylvesterkaczmarek/nvidia-quantum-hybrid}, - doi = {10.5281/zenodo.17502919} + doi = {10.5281/zenodo.17502919}, + url = {https://github.com/sylvesterkaczmarek/nvidia-cudaq-quantum-guard} } ``` @@ -127,4 +319,4 @@ If you use or adapt this repository, please cite MIT. See [LICENSE](LICENSE). -© **Sylvester Kaczmarek** · https://www.sylvesterkaczmarek.com +© **Sylvester Kaczmarek** · [https://www.sylvesterkaczmarek.com](https://www.sylvesterkaczmarek.com) diff --git a/SECURITY.md b/SECURITY.md new file mode 100644 index 0000000..5cc3a57 --- /dev/null +++ b/SECURITY.md @@ -0,0 +1,5 @@ +# Security + +Please report security-sensitive issues privately rather than opening a public issue when disclosure could create avoidable risk. + +This project is a guardrail and provenance toolkit, not a secure sandbox. Code that can import CUDA-Q directly can bypass the guard unless the surrounding deployment restricts that path. See [docs/security-model.md](docs/security-model.md) for the intended boundary and limitations. diff --git a/assets/social/github-social-card-nvidia-quantum-hybrid.png b/assets/social/github-social-card-nvidia-cudaq-quantum-guard.png similarity index 100% rename from assets/social/github-social-card-nvidia-quantum-hybrid.png rename to assets/social/github-social-card-nvidia-cudaq-quantum-guard.png diff --git a/docs/audit-format.md b/docs/audit-format.md new file mode 100644 index 0000000..0e45f93 --- /dev/null +++ b/docs/audit-format.md @@ -0,0 +1,18 @@ +# Audit format + +Every guarded decision can be written as one JSON object per line. Records contain request metadata, policy identity, execution status, timings, CUDA-Q version, result summaries, and errors. + +Each record also contains: + +- `previous_hash`: SHA-256 hash of the preceding record, or 64 zeroes for the first record +- `record_hash`: SHA-256 of the canonical current record before `record_hash` is added + +This creates a tamper-evident chain. It does not provide authenticity by itself: a party able to rewrite the entire file can recompute the chain. For stronger provenance, sign the audit head or store it in an external immutable system. + +Verify a chain with: + +```bash +cudaq-guard audit verify runs/audit.jsonl +``` + +Target-option keys containing words such as `token`, `password`, `secret`, or `credential` are redacted from audit records. Provider credentials should still be supplied through the provider's documented credential mechanism rather than command-line options. diff --git a/docs/policy-reference.md b/docs/policy-reference.md new file mode 100644 index 0000000..4f41003 --- /dev/null +++ b/docs/policy-reference.md @@ -0,0 +1,27 @@ +# Policy reference + +CUDA-Q Guard uses a small TOML policy so execution limits are reviewable and version-controlled. + +```toml +version = 1 +name = "local-safe" +allowed_operations = ["sample", "observe"] +allowed_targets = ["qpp-cpu", "nvidia"] +allow_remote = false +allow_async = false +max_qubits = 28 +max_shots = 100000 +allowed_qpu_ids = [0] +require_seed = true + +[target_options.nvidia] +option = ["mqpu", "fp64"] +``` + +A request is denied when any bound is exceeded. Unknown policy keys, unlisted target options, unknown target-option values, and unsupported policy versions fail closed. + +`allow_remote = false` is deliberately conservative. The runtime uses CUDA-Q's target `is_remote()` metadata when available and falls back conservatively for older targets. A remote policy must explicitly allow both remote execution and the provider target name. + +A target-option value of `"*"` allows any value for that explicitly named option key. This is useful for provider-specific machine identifiers, but should be used narrowly. + +The policy evaluates declared workload metadata such as qubit count and operation. Built-in workloads also call CUDA-Q `estimate_resources()` before execution and fail if the compiled kernel exceeds the declared or policy qubit bound. Library users can opt into the same preflight check with `resource_probe`. This is still not a compiler sandbox or formal verification of arbitrary Python control flow. diff --git a/docs/reproducibility.md b/docs/reproducibility.md new file mode 100644 index 0000000..1a1570d --- /dev/null +++ b/docs/reproducibility.md @@ -0,0 +1,16 @@ +# Reproducibility + +Local simulator examples use an explicit CUDA-Q random seed. Policies and requests are hashed into audit records, and the audit record stores the CUDA-Q and Python versions when available. + +Recommended clean run: + +```bash +python -m venv .venv +source .venv/bin/activate +python -m pip install -e ".[cudaq,dev]" +cudaq-guard doctor +cudaq-guard run ghz --policy policies/local-safe.toml --target qpp-cpu --qubits 4 --shots 1000 --seed 7 +cudaq-guard audit verify runs/audit.jsonl +``` + +Finite-shot sampling is statistical even with deterministic simulator seeds across software versions. Bit-for-bit equality across CUDA-Q releases, different GPU architectures, and real QPUs is not claimed. diff --git a/docs/security-model.md b/docs/security-model.md new file mode 100644 index 0000000..fefdc9e --- /dev/null +++ b/docs/security-model.md @@ -0,0 +1,26 @@ +# Security model + +CUDA-Q Guard is an execution-control and provenance layer around CUDA-Q calls. Its useful security boundary is the guarded API or CLI invocation. + +It provides: + +- fail-closed policy evaluation before execution +- target, operation, qubit, shot, asynchronous-execution and QPU-ID bounds +- explicit remote-target authorization +- target-option allowlists +- deterministic-seed requirements where requested +- optional CUDA-Q resource preflight that detects under-declared qubit use before execution +- tamper-evident local audit chains +- redaction of obvious credential-like target options + +It does not provide: + +- a sandbox for arbitrary Python code +- formal verification that arbitrary Python code cannot bypass or misrepresent resource use +- authentication or authorization of operating-system users +- cryptographic signing of audit records +- QPU provider credential management +- formal verification of CUDA-Q kernels +- protection against code that bypasses the guard and calls CUDA-Q directly + +For higher-assurance environments, place the guard behind a process/service boundary, restrict direct CUDA-Q access, use signed policy/configuration, and export audit heads to an independent log or attestation system. diff --git a/examples/library_integration.py b/examples/library_integration.py new file mode 100644 index 0000000..7eaa31c --- /dev/null +++ b/examples/library_integration.py @@ -0,0 +1,34 @@ +"""Minimal library integration example for an application that already uses CUDA-Q.""" + +from cudaq_guard import ExecutionRequest, Guard, GuardPolicy +from cudaq_guard.runtime import CudaQRuntime + +runtime = CudaQRuntime() +cudaq = runtime.cudaq + + +@cudaq.kernel +def bell(): + q = cudaq.qvector(2) + h(q[0]) + x.ctrl(q[0], q[1]) + mz(q) + + +policy = GuardPolicy.from_toml("policies/local-safe.toml") +guard = Guard(policy, audit_path="runs/library-audit.jsonl", runtime=runtime) +request = ExecutionRequest( + workload="bell", + operation="sample", + target="qpp-cpu", + qubits=2, + shots=1000, + seed=7, +) + +counts = guard.execute( + request, + lambda rt: rt.sample(bell, shots=1000), + resource_probe=lambda rt: rt.estimate_resources(bell), +) +print(counts) diff --git a/hybrid_secure_demo.py b/hybrid_secure_demo.py deleted file mode 100644 index aa76b02..0000000 --- a/hybrid_secure_demo.py +++ /dev/null @@ -1,181 +0,0 @@ -""" -NVIDIA Quantum Hybrid — secure AI x quantum demo -Author: Sylvester Kaczmarek -DOI: 10.5281/zenodo.17502919 -License: MIT -""" - -import time -import json -import numpy as np - -from qiskit import QuantumCircuit -from qiskit_aer import AerSimulator - - -PIPELINE_VERSION = "0.3-nvidia-quantum-hybrid" - - -def classical_feature_stage(data): - """ - Classical preprocessing. - Normalise, collect basic stats, compute a tiny anomaly score. - """ - arr = np.array(data, dtype=float) - mean = float(np.mean(arr)) - std = float(np.std(arr) + 1e-8) - norm = (arr - mean) / std - anomaly_score = float(np.mean(np.abs(norm))) - meta = { - "mean": mean, - "std": std, - "anomaly_score": anomaly_score, - } - return norm, meta - - -def policy_envelope_small(norm_features): - """ - Allow values inside [-2.5, 2.5]. - """ - if np.any(np.abs(norm_features) > 2.5): - return False, "small_envelope_violation" - return True, "ok" - - -def policy_envelope_strict(norm_features): - """ - More strict envelope. - """ - if np.any(np.abs(norm_features) > 1.8): - return False, "strict_envelope_violation" - return True, "ok" - - -POLICIES = { - "small": policy_envelope_small, - "strict": policy_envelope_strict, -} - - -def run_policy(norm_features, policy_name="small"): - policy_fn = POLICIES.get(policy_name, policy_envelope_small) - return policy_fn(norm_features) - - -def run_quantum_qiskit(theta): - """ - Simple 1 qubit circuit on Qiskit Aer. - """ - qc = QuantumCircuit(1, 1) - qc.ry(theta, 0) - qc.measure(0, 0) - - sim = AerSimulator() - job = sim.run(qc, shots=1024) - result = job.result() - counts = result.get_counts() - prob_1 = counts.get("1", 0) / 1024.0 - return prob_1 - - -def simulate_noise(prob_1, noise_level=0.03): - """ - Tiny noise model to imitate imperfect hardware. - """ - noisy = prob_1 * (1 - noise_level) + (1 - prob_1) * noise_level - return round(noisy, 4) - - -def run_quantum(theta, backend="qiskit_sim"): - """ - Switchable quantum backend. - For now we only support Qiskit simulator. - """ - if backend == "qiskit_sim": - return run_quantum_qiskit(theta), "qiskit_sim" - - # fallback - return run_quantum_qiskit(theta), "qiskit_sim" - - -def log_event(payload): - """ - Very simple logger. - Replace with proper logging or webhook later. - """ - print("[LOG]", json.dumps(payload)) - - -def run_secure_hybrid(sample, policy_name="small", backend="qiskit_sim", write_file=False): - """ - Full classical -> policy -> quantum -> classical loop. - """ - t0 = time.time() - - # 1. classical features - feats, feat_meta = classical_feature_stage(sample) - - # 2. runtime policy - safe, tag = run_policy(feats, policy_name=policy_name) - - result = { - "pipeline_version": PIPELINE_VERSION, - "policy_used": policy_name, - "policy_tag": tag, - "features_meta": feat_meta, - } - - if not safe: - # quantum step blocked - result.update( - { - "quantum_called": False, - "reason": "blocked_by_policy", - "explain": f"policy={policy_name}, status={tag}", - } - ) - result["latency_s"] = round(time.time() - t0, 4) - log_event(result) - if write_file: - with open("hybrid_output_blocked.json", "w") as f: - json.dump(result, f, indent=2) - return result - - # 3. quantum step allowed - theta = float(np.clip(np.mean(feats) * np.pi, -np.pi, np.pi)) - prob_1, used_backend = run_quantum(theta, backend=backend) - prob_1_noisy = simulate_noise(prob_1) - - # 4. final - result.update( - { - "quantum_called": True, - "quantum_backend": used_backend, - "quantum_confidence": prob_1_noisy, - "noise_level": 0.03, - "reason": "ok", - "explain": f"policy={policy_name}, mean={feat_meta['mean']:.3f}, theta={theta:.3f}", - } - ) - result["latency_s"] = round(time.time() - t0, 4) - - log_event(result) - if write_file: - with open("hybrid_output.json", "w") as f: - json.dump(result, f, indent=2) - return result - - -if __name__ == "__main__": - # sample expected to pass - sample_ok = [1.0, 1.02, 1.05, 0.98, 1.01] - out_ok = run_secure_hybrid(sample_ok, policy_name="small", write_file=True) - print("=== OK sample result ===") - print(json.dumps(out_ok, indent=2)) - - # sample expected to fail - sample_bad = [1.0, 4.5, 1.05, 0.98, 1.01] - out_bad = run_secure_hybrid(sample_bad, policy_name="strict", write_file=True) - print("=== BAD sample result ===") - print(json.dumps(out_bad, indent=2)) diff --git a/policies/local-safe.toml b/policies/local-safe.toml new file mode 100644 index 0000000..c8b5222 --- /dev/null +++ b/policies/local-safe.toml @@ -0,0 +1,13 @@ +version = 1 +name = "local-safe" +allowed_operations = ["sample", "observe"] +allowed_targets = ["qpp-cpu", "nvidia"] +allow_remote = false +allow_async = false +max_qubits = 28 +max_shots = 100000 +allowed_qpu_ids = [0] +require_seed = true + +[target_options.nvidia] +option = ["mqpu", "fp64"] diff --git a/policies/remote-explicit.toml b/policies/remote-explicit.toml new file mode 100644 index 0000000..0063efb --- /dev/null +++ b/policies/remote-explicit.toml @@ -0,0 +1,23 @@ +version = 1 +name = "remote-explicit" +allowed_operations = ["sample", "observe"] +allowed_targets = ["braket", "ionq", "quantinuum", "iqm"] +allow_remote = true +allow_async = true +max_qubits = 32 +max_shots = 10000 +allowed_qpu_ids = [0] +require_seed = false + +[target_options.braket] +machine = ["*"] + +[target_options.quantinuum] +machine = ["*"] +project = ["*"] + +[target_options.ionq] +machine = ["*"] + +[target_options.iqm] +url = ["*"] diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..5894c0d --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,44 @@ +[build-system] +requires = ["setuptools>=69"] +build-backend = "setuptools.build_meta" + +[project] +name = "nvidia-cudaq-quantum-guard" +version = "0.4.0" +description = "Policy-controlled CUDA-Q execution, diagnostics, backend comparison, and tamper-evident audit trails." +readme = "README.md" +requires-python = ">=3.11" +license = {text = "MIT"} +authors = [{name = "Sylvester Kaczmarek"}] +keywords = ["cuda-q", "quantum-computing", "guardrails", "audit", "nvidia"] +classifiers = [ + "Development Status :: 3 - Alpha", + "License :: OSI Approved :: MIT License", + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.11", + "Programming Language :: Python :: 3.12", + "Programming Language :: Python :: 3.13", + "Topic :: Scientific/Engineering", + "Topic :: Security", +] +dependencies = [] + +[project.optional-dependencies] +cudaq = ["cudaq>=0.15,<0.16"] +dev = ["pytest>=8,<10", "build>=1.2,<2"] +all = ["cudaq>=0.15,<0.16", "pytest>=8,<10", "build>=1.2,<2"] + +[project.scripts] +cudaq-guard = "cudaq_guard.cli:main" + +[project.urls] +Repository = "https://github.com/sylvesterkaczmarek/nvidia-cudaq-quantum-guard" +Documentation = "https://github.com/sylvesterkaczmarek/nvidia-cudaq-quantum-guard#readme" +Issues = "https://github.com/sylvesterkaczmarek/nvidia-cudaq-quantum-guard/issues" + +[tool.setuptools.packages.find] +where = ["src"] + +[tool.pytest.ini_options] +addopts = "-q" +testpaths = ["tests"] diff --git a/requirements.txt b/requirements.txt index 213a1b4..97181c1 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,3 +1,3 @@ -numpy -qiskit -qiskit-aer +cudaq>=0.15,<0.16 +pytest>=8,<10 +build>=1.2,<2 diff --git a/src/cudaq_guard/__init__.py b/src/cudaq_guard/__init__.py new file mode 100644 index 0000000..dbadb55 --- /dev/null +++ b/src/cudaq_guard/__init__.py @@ -0,0 +1,20 @@ +"""Policy-controlled execution and audit tooling for NVIDIA CUDA-Q.""" + +from .audit import AuditTrail, verify_audit +from .guard import Guard +from .models import ExecutionRequest, PolicyDecision, TargetInfo +from .policy import GuardPolicy +from .runtime import CudaQRuntime + +__all__ = [ + "AuditTrail", + "CudaQRuntime", + "ExecutionRequest", + "Guard", + "GuardPolicy", + "PolicyDecision", + "TargetInfo", + "verify_audit", +] + +__version__ = "0.4.0" diff --git a/src/cudaq_guard/__main__.py b/src/cudaq_guard/__main__.py new file mode 100644 index 0000000..eb53e2f --- /dev/null +++ b/src/cudaq_guard/__main__.py @@ -0,0 +1,3 @@ +from .cli import main + +raise SystemExit(main()) diff --git a/src/cudaq_guard/audit.py b/src/cudaq_guard/audit.py new file mode 100644 index 0000000..6fccc26 --- /dev/null +++ b/src/cudaq_guard/audit.py @@ -0,0 +1,71 @@ +from __future__ import annotations + +import json +import os +from pathlib import Path +from typing import Any, Iterable + +from .crypto import canonical_json, sha256_json +from .errors import AuditIntegrityError + +GENESIS_HASH = "0" * 64 + + +def _record_hash(record_without_hash: dict[str, Any]) -> str: + return sha256_json(record_without_hash) + + +def read_jsonl(path: str | Path) -> Iterable[dict[str, Any]]: + with open(path, "r", encoding="utf-8") as handle: + for line_no, line in enumerate(handle, 1): + stripped = line.strip() + if not stripped: + continue + try: + value = json.loads(stripped) + except json.JSONDecodeError as exc: + raise AuditIntegrityError(f"invalid JSON at audit line {line_no}") from exc + if not isinstance(value, dict): + raise AuditIntegrityError(f"audit line {line_no} is not an object") + yield value + + +def verify_audit(path: str | Path) -> dict[str, Any]: + previous = GENESIS_HASH + count = 0 + for count, record in enumerate(read_jsonl(path), 1): + expected_previous = record.get("previous_hash") + if expected_previous != previous: + raise AuditIntegrityError(f"audit chain broken at record {count}: previous hash mismatch") + actual_hash = record.get("record_hash") + payload = dict(record) + payload.pop("record_hash", None) + expected_hash = _record_hash(payload) + if actual_hash != expected_hash: + raise AuditIntegrityError(f"audit chain broken at record {count}: record hash mismatch") + previous = actual_hash + return {"valid": True, "records": count, "head_hash": previous} + + +class AuditTrail: + def __init__(self, path: str | Path): + self.path = Path(path) + self.path.parent.mkdir(parents=True, exist_ok=True) + + def _head(self) -> str: + if not self.path.exists() or self.path.stat().st_size == 0: + return GENESIS_HASH + return verify_audit(self.path)["head_hash"] + + def append(self, payload: dict[str, Any]) -> dict[str, Any]: + record = dict(payload) + record["previous_hash"] = self._head() + record["record_hash"] = _record_hash(record) + serialized = canonical_json(record) + "\n" + fd = os.open(self.path, os.O_WRONLY | os.O_CREAT | os.O_APPEND, 0o600) + try: + os.write(fd, serialized.encode("utf-8")) + os.fsync(fd) + finally: + os.close(fd) + return record diff --git a/src/cudaq_guard/cli.py b/src/cudaq_guard/cli.py new file mode 100644 index 0000000..68ef8de --- /dev/null +++ b/src/cudaq_guard/cli.py @@ -0,0 +1,247 @@ +from __future__ import annotations + +import argparse +import json +import sys +import time +from pathlib import Path +from typing import Any + +from .audit import verify_audit +from .doctor import collect_doctor, render_doctor, to_json +from .errors import AuditIntegrityError, CudaQGuardError +from .guard import Guard, summarize_result +from .models import ExecutionRequest +from .policy import GuardPolicy +from .runtime import CudaQRuntime +from .workloads import ghz_kernel, h2_vqe_problem, vqe_grid +from . import __version__ + + +def _target_options(values: list[str]) -> dict[str, str]: + options: dict[str, str] = {} + for value in values: + if "=" not in value: + raise ValueError(f"target option must be KEY=VALUE: {value}") + key, option = value.split("=", 1) + key = key.strip() + if not key: + raise ValueError("target option key cannot be empty") + options[key] = option.strip() + return options + + +def _policy(path: str) -> GuardPolicy: + return GuardPolicy.from_toml(path) + + +def _request(args: argparse.Namespace, operation: str, workload: str, qubits: int, shots: int | None) -> ExecutionRequest: + return ExecutionRequest( + workload=workload, + operation=operation, + target=args.target, + qubits=qubits, + shots=shots, + async_mode=getattr(args, "async_mode", False), + qpu_id=args.qpu_id, + seed=args.seed, + target_options=_target_options(args.target_option), + ) + + +def cmd_doctor(args: argparse.Namespace) -> int: + report = collect_doctor() + print(to_json(report) if args.json else render_doctor(report)) + return 0 if report["cudaq_installed"] else 2 + + +def cmd_targets(args: argparse.Namespace) -> int: + targets = [target.to_dict() for target in CudaQRuntime().available_targets()] + if args.json: + print(json.dumps(targets, indent=2, sort_keys=True)) + else: + for target in targets: + print(f"{target['name']:20} qpus={target['num_qpus']:<3} simulator={target['simulator'] or '-'}") + return 0 + + +def cmd_policy_check(args: argparse.Namespace) -> int: + policy = _policy(args.policy) + request = _request(args, args.operation, "declared-workload", args.qubits, args.shots) + runtime = CudaQRuntime() + static = policy.evaluate(request) + if static.allowed and args.inspect_target: + target = runtime.describe_target(request.target) + decision = policy.evaluate(request, target) + else: + decision = static + print(json.dumps(decision.to_dict(), indent=2, sort_keys=True)) + return 0 if decision.allowed else 3 + + +def cmd_run_ghz(args: argparse.Namespace) -> int: + guard = Guard(_policy(args.policy), audit_path=args.audit) + request = _request(args, "sample", "ghz", args.qubits, args.shots) + def execute(runtime): + kernel = ghz_kernel(runtime.cudaq) + if args.async_mode: + return runtime.sample_async( + kernel, args.qubits, shots=args.shots, qpu_id=args.qpu_id + ).get() + return runtime.sample(kernel, args.qubits, shots=args.shots, qpu_id=args.qpu_id) + + result = guard.execute( + request, + execute, + resource_probe=lambda runtime: runtime.estimate_resources( + ghz_kernel(runtime.cudaq), args.qubits + ), + ) + print(json.dumps(summarize_result(result), indent=2, sort_keys=True)) + return 0 + + +def cmd_run_vqe(args: argparse.Namespace) -> int: + guard = Guard(_policy(args.policy), audit_path=args.audit) + request = _request(args, "observe", "h2-vqe-grid", 2, None) + request = ExecutionRequest(**{**request.to_dict(), "metadata": {"steps": args.steps}}) + result = guard.execute( + request, + lambda runtime: vqe_grid(runtime, steps=args.steps, qpu_id=args.qpu_id), + resource_probe=lambda runtime: runtime.estimate_resources( + h2_vqe_problem(runtime.cudaq)[0], 0.0 + ), + ) + print(json.dumps(result, indent=2, sort_keys=True)) + return 0 + + +def cmd_compare(args: argparse.Namespace) -> int: + targets = [item.strip() for item in args.targets.split(",") if item.strip()] + if not targets: + raise ValueError("at least one target is required") + policy = _policy(args.policy) + rows: list[dict[str, Any]] = [] + for target in targets: + namespace = argparse.Namespace(**vars(args)) + namespace.target = target + request = _request(namespace, "sample", "ghz", args.qubits, args.shots) + guard = Guard(policy, audit_path=args.audit) + started = time.perf_counter() + try: + result = guard.execute( + request, + lambda runtime: ( + runtime.sample_async(ghz_kernel(runtime.cudaq), args.qubits, shots=args.shots, qpu_id=args.qpu_id).get() + if args.async_mode + else runtime.sample(ghz_kernel(runtime.cudaq), args.qubits, shots=args.shots, qpu_id=args.qpu_id) + ), + resource_probe=lambda runtime: runtime.estimate_resources( + ghz_kernel(runtime.cudaq), args.qubits + ), + ) + rows.append( + { + "target": target, + "status": "ok", + "duration_ms": round((time.perf_counter() - started) * 1000, 3), + "result": summarize_result(result), + } + ) + except Exception as exc: + rows.append( + { + "target": target, + "status": "error", + "duration_ms": round((time.perf_counter() - started) * 1000, 3), + "error": f"{type(exc).__name__}: {exc}", + } + ) + print(json.dumps(rows, indent=2, sort_keys=True)) + return 0 if any(row["status"] == "ok" for row in rows) else 4 + + +def cmd_audit_verify(args: argparse.Namespace) -> int: + report = verify_audit(args.path) + print(json.dumps(report, indent=2, sort_keys=True)) + return 0 + + +def _add_execution_args(parser: argparse.ArgumentParser, *, include_async: bool = True) -> None: + parser.add_argument("--policy", default="policies/local-safe.toml") + parser.add_argument("--target", default="qpp-cpu") + parser.add_argument("--target-option", action="append", default=[], metavar="KEY=VALUE") + parser.add_argument("--qpu-id", type=int, default=0) + parser.add_argument("--seed", type=int, default=7) + if include_async: + parser.add_argument("--async", dest="async_mode", action="store_true") + else: + parser.set_defaults(async_mode=False) + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(prog="cudaq-guard", description="Policy-controlled CUDA-Q execution and audit tooling") + parser.add_argument("--version", action="version", version=f"%(prog)s {__version__}") + sub = parser.add_subparsers(dest="command", required=True) + + doctor = sub.add_parser("doctor", help="diagnose CUDA-Q, GPU and target availability") + doctor.add_argument("--json", action="store_true") + doctor.set_defaults(func=cmd_doctor) + + targets = sub.add_parser("targets", help="list CUDA-Q targets visible on this system") + targets.add_argument("--json", action="store_true") + targets.set_defaults(func=cmd_targets) + + policy = sub.add_parser("policy", help="policy utilities") + policy_sub = policy.add_subparsers(dest="policy_command", required=True) + check = policy_sub.add_parser("check", help="evaluate a declared execution request") + _add_execution_args(check) + check.add_argument("--operation", choices=["sample", "observe", "run"], default="sample") + check.add_argument("--qubits", type=int, default=4) + check.add_argument("--shots", type=int, default=1000) + check.add_argument("--inspect-target", action="store_true", help="also inspect CUDA-Q target metadata") + check.set_defaults(func=cmd_policy_check) + + run = sub.add_parser("run", help="run built-in guarded workloads") + run_sub = run.add_subparsers(dest="workload", required=True) + ghz = run_sub.add_parser("ghz", help="sample a GHZ state") + _add_execution_args(ghz) + ghz.add_argument("--qubits", type=int, default=4) + ghz.add_argument("--shots", type=int, default=1000) + ghz.add_argument("--audit", default="runs/audit.jsonl") + ghz.set_defaults(func=cmd_run_ghz) + + vqe = run_sub.add_parser("vqe", help="run a small H2 VQE grid search") + _add_execution_args(vqe, include_async=False) + vqe.add_argument("--steps", type=int, default=25) + vqe.add_argument("--audit", default="runs/audit.jsonl") + vqe.set_defaults(func=cmd_run_vqe) + + compare = sub.add_parser("compare", help="compare the guarded GHZ workload across CUDA-Q targets") + _add_execution_args(compare) + compare.add_argument("--targets", default="qpp-cpu,nvidia") + compare.add_argument("--qubits", type=int, default=20) + compare.add_argument("--shots", type=int, default=1000) + compare.add_argument("--audit", default="runs/compare-audit.jsonl") + compare.set_defaults(func=cmd_compare) + + audit = sub.add_parser("audit", help="tamper-evident audit utilities") + audit_sub = audit.add_subparsers(dest="audit_command", required=True) + verify = audit_sub.add_parser("verify", help="verify an audit hash chain") + verify.add_argument("path") + verify.set_defaults(func=cmd_audit_verify) + return parser + + +def main(argv: list[str] | None = None) -> int: + parser = build_parser() + args = parser.parse_args(argv) + try: + return int(args.func(args)) + except (CudaQGuardError, AuditIntegrityError, ValueError) as exc: + print(f"error: {exc}", file=sys.stderr) + return 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/cudaq_guard/crypto.py b/src/cudaq_guard/crypto.py new file mode 100644 index 0000000..069085c --- /dev/null +++ b/src/cudaq_guard/crypto.py @@ -0,0 +1,29 @@ +from __future__ import annotations + +import hashlib +import json +from typing import Any + + +SENSITIVE_FRAGMENTS = ("password", "passwd", "secret", "token", "credential", "api_key", "apikey") + + +def canonical_json(value: Any) -> str: + return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True) + + +def sha256_json(value: Any) -> str: + return hashlib.sha256(canonical_json(value).encode("utf-8")).hexdigest() + + +def redact_mapping(value: dict[str, Any]) -> dict[str, Any]: + redacted: dict[str, Any] = {} + for key, item in value.items(): + lowered = key.lower() + if any(fragment in lowered for fragment in SENSITIVE_FRAGMENTS): + redacted[key] = "" + elif isinstance(item, dict): + redacted[key] = redact_mapping(item) + else: + redacted[key] = item + return redacted diff --git a/src/cudaq_guard/doctor.py b/src/cudaq_guard/doctor.py new file mode 100644 index 0000000..0b148de --- /dev/null +++ b/src/cudaq_guard/doctor.py @@ -0,0 +1,77 @@ +from __future__ import annotations + +import json +import platform +import shutil +import subprocess +from dataclasses import asdict +from typing import Any + +from .errors import CudaQUnavailableError +from .runtime import CudaQRuntime + + +def collect_doctor(runtime: CudaQRuntime | None = None) -> dict[str, Any]: + runtime = runtime or CudaQRuntime() + report: dict[str, Any] = { + "schema_version": 1, + "python": platform.python_version(), + "system": platform.system(), + "machine": platform.machine(), + "cudaq_installed": False, + "cudaq_version": None, + "cudaq_gpu_count": None, + "targets": [], + "nvidia_smi": None, + } + if shutil.which("nvidia-smi"): + try: + completed = subprocess.run( + ["nvidia-smi", "--query-gpu=name,driver_version", "--format=csv,noheader"], + check=False, + capture_output=True, + text=True, + timeout=5, + ) + if completed.returncode == 0: + report["nvidia_smi"] = [line.strip() for line in completed.stdout.splitlines() if line.strip()] + except (OSError, subprocess.SubprocessError): + report["nvidia_smi"] = None + try: + targets = runtime.available_targets() + except CudaQUnavailableError: + return report + report["cudaq_installed"] = True + report["cudaq_version"] = runtime.version + report["cudaq_gpu_count"] = runtime.available_gpu_count() + report["targets"] = [asdict(target) for target in targets] + return report + + +def render_doctor(report: dict[str, Any]) -> str: + lines = [ + f"Python: {report['python']}", + f"Platform: {report['system']} {report['machine']}", + f"CUDA-Q installed: {'yes' if report['cudaq_installed'] else 'no'}", + ] + if report["cudaq_version"]: + lines.append(f"CUDA-Q version: {report['cudaq_version']}") + if report["cudaq_gpu_count"] is not None: + lines.append(f"CUDA-Q GPUs visible: {report['cudaq_gpu_count']}") + if report["nvidia_smi"]: + lines.append("NVIDIA GPU: " + "; ".join(report["nvidia_smi"])) + targets = report.get("targets", []) + if targets: + lines.append("Targets:") + for target in targets: + kind = "remote/hardware" if target["is_remote"] else "simulator" + lines.append( + f" - {target['name']} ({kind}, qpus={target['num_qpus']}, simulator={target['simulator'] or '-'})" + ) + else: + lines.append("Targets: unavailable") + return "\n".join(lines) + + +def to_json(report: dict[str, Any]) -> str: + return json.dumps(report, indent=2, sort_keys=True) diff --git a/src/cudaq_guard/errors.py b/src/cudaq_guard/errors.py new file mode 100644 index 0000000..298f605 --- /dev/null +++ b/src/cudaq_guard/errors.py @@ -0,0 +1,14 @@ +class CudaQGuardError(RuntimeError): + """Base exception for CUDA-Q Guard.""" + + +class PolicyDeniedError(CudaQGuardError): + """Raised when an execution request is denied by policy.""" + + +class CudaQUnavailableError(CudaQGuardError): + """Raised when CUDA-Q is required but unavailable.""" + + +class AuditIntegrityError(CudaQGuardError): + """Raised when a tamper-evident audit chain does not verify.""" diff --git a/src/cudaq_guard/guard.py b/src/cudaq_guard/guard.py new file mode 100644 index 0000000..233cd3f --- /dev/null +++ b/src/cudaq_guard/guard.py @@ -0,0 +1,137 @@ +from __future__ import annotations + +import platform +import time +import uuid +from datetime import datetime, timezone +from typing import Any, Callable + +from .audit import AuditTrail +from .crypto import redact_mapping, sha256_json +from .errors import PolicyDeniedError +from .models import ExecutionRequest, PolicyDecision +from .policy import GuardPolicy +from .runtime import CudaQRuntime + + +class Guard: + def __init__( + self, + policy: GuardPolicy, + *, + audit_path: str | None = None, + runtime: CudaQRuntime | None = None, + ): + self.policy = policy + self.runtime = runtime or CudaQRuntime() + self.audit = AuditTrail(audit_path) if audit_path else None + + def authorize(self, request: ExecutionRequest) -> PolicyDecision: + # First evaluate without touching CUDA-Q. A disallowed target must not be + # configured just to discover that it is disallowed. + static = self.policy.evaluate(request) + if not static.allowed: + return static + target_info = self.runtime.describe_target(request.target) + return self.policy.evaluate(request, target_info) + + def execute( + self, + request: ExecutionRequest, + fn: Callable[[CudaQRuntime], Any], + *, + resource_probe: Callable[[CudaQRuntime], dict[str, Any]] | None = None, + ) -> Any: + run_id = str(uuid.uuid4()) + started = time.perf_counter() + decision: PolicyDecision + try: + decision = self.authorize(request) + except Exception as exc: + self._audit(run_id, request, None, started, "authorization_error", error=exc) + raise + + if not decision.allowed: + self._audit(run_id, request, decision, started, "denied") + raise PolicyDeniedError( + f"execution denied by policy {decision.policy_name}: {', '.join(decision.violations)}" + ) + + resources: dict[str, Any] | None = None + try: + self.runtime.configure(request.target, request.target_options, request.seed) + if resource_probe is not None: + resources = resource_probe(self.runtime) + resource_decision = self.policy.evaluate_resources(request, resources) + if not resource_decision.allowed: + self._audit( + run_id, request, resource_decision, started, "denied_resource", resources=resources + ) + raise PolicyDeniedError( + f"execution denied by policy {resource_decision.policy_name}: " + f"{', '.join(resource_decision.violations)}" + ) + result = fn(self.runtime) + except PolicyDeniedError: + raise + except Exception as exc: + self._audit(run_id, request, decision, started, "error", error=exc, resources=resources) + raise + + self._audit( + run_id, request, decision, started, "completed", result=result, resources=resources + ) + return result + + def _audit( + self, + run_id: str, + request: ExecutionRequest, + decision: PolicyDecision | None, + started: float, + status: str, + *, + result: Any | None = None, + error: BaseException | None = None, + resources: dict[str, Any] | None = None, + ) -> None: + if self.audit is None: + return + request_dict = request.to_dict() + request_dict["target_options"] = redact_mapping(request.target_options) + record: dict[str, Any] = { + "schema_version": 1, + "run_id": run_id, + "timestamp_utc": datetime.now(timezone.utc).isoformat(), + "status": status, + "duration_ms": round((time.perf_counter() - started) * 1000.0, 3), + "request": request_dict, + "request_hash": sha256_json(request_dict), + "policy": decision.to_dict() if decision else None, + "python": platform.python_version(), + "cudaq_version": self.runtime.version, + } + if resources is not None: + record["resources"] = resources + if result is not None: + record["result"] = summarize_result(result) + if error is not None: + record["error"] = {"type": type(error).__name__, "message": str(error)} + self.audit.append(record) + + +def summarize_result(result: Any) -> dict[str, Any]: + if hasattr(result, "expectation") and callable(result.expectation): + try: + return {"kind": "observe", "expectation": float(result.expectation())} + except Exception: + pass + if hasattr(result, "items"): + try: + counts = {str(k): int(v) for k, v in result.items()} + return {"kind": "sample", "counts": counts, "shots": int(sum(counts.values()))} + except Exception: + pass + if isinstance(result, dict): + return {"kind": "mapping", "value": result} + return {"kind": "text", "value": str(result)} diff --git a/src/cudaq_guard/models.py b/src/cudaq_guard/models.py new file mode 100644 index 0000000..4a02f0d --- /dev/null +++ b/src/cudaq_guard/models.py @@ -0,0 +1,50 @@ +from __future__ import annotations + +from dataclasses import asdict, dataclass, field +from typing import Any + + +@dataclass(frozen=True) +class TargetInfo: + name: str + simulator: str = "" + platform: str = "" + description: str = "" + num_qpus: int = 1 + is_remote: bool = False + + def to_dict(self) -> dict[str, Any]: + return asdict(self) + + +@dataclass(frozen=True) +class ExecutionRequest: + workload: str + operation: str + target: str + qubits: int + shots: int | None = None + async_mode: bool = False + qpu_id: int = 0 + seed: int | None = None + target_options: dict[str, str] = field(default_factory=dict) + metadata: dict[str, Any] = field(default_factory=dict) + + def to_dict(self) -> dict[str, Any]: + return asdict(self) + + +@dataclass(frozen=True) +class PolicyDecision: + allowed: bool + policy_name: str + policy_hash: str + violations: tuple[str, ...] = () + + def to_dict(self) -> dict[str, Any]: + return { + "allowed": self.allowed, + "policy_name": self.policy_name, + "policy_hash": self.policy_hash, + "violations": list(self.violations), + } diff --git a/src/cudaq_guard/policy.py b/src/cudaq_guard/policy.py new file mode 100644 index 0000000..e3a6be6 --- /dev/null +++ b/src/cudaq_guard/policy.py @@ -0,0 +1,156 @@ +from __future__ import annotations + +import tomllib +from dataclasses import asdict, dataclass, field +from pathlib import Path +from typing import Any + +from .crypto import sha256_json +from .models import ExecutionRequest, PolicyDecision, TargetInfo + +_ALLOWED_KEYS = { + "version", + "name", + "allowed_operations", + "allowed_targets", + "allow_remote", + "allow_async", + "max_qubits", + "max_shots", + "allowed_qpu_ids", + "require_seed", + "target_options", +} + + +@dataclass(frozen=True) +class GuardPolicy: + version: int = 1 + name: str = "local-safe" + allowed_operations: tuple[str, ...] = ("sample", "observe") + allowed_targets: tuple[str, ...] = ("qpp-cpu",) + allow_remote: bool = False + allow_async: bool = False + max_qubits: int = 24 + max_shots: int = 100_000 + allowed_qpu_ids: tuple[int, ...] = (0,) + require_seed: bool = True + target_options: dict[str, dict[str, tuple[str, ...]]] = field(default_factory=dict) + + @classmethod + def from_toml(cls, path: str | Path) -> "GuardPolicy": + with open(path, "rb") as handle: + raw = tomllib.load(handle) + unknown = set(raw) - _ALLOWED_KEYS + if unknown: + raise ValueError(f"unknown policy keys: {', '.join(sorted(unknown))}") + + target_options_raw = raw.get("target_options", {}) + if not isinstance(target_options_raw, dict): + raise ValueError("target_options must be a TOML table") + normalized_options: dict[str, dict[str, tuple[str, ...]]] = {} + for target, options in target_options_raw.items(): + if not isinstance(options, dict): + raise ValueError(f"target_options.{target} must be a table") + normalized_options[target] = {} + for key, values in options.items(): + if not isinstance(values, list) or not all(isinstance(v, (str, int, float, bool)) for v in values): + raise ValueError(f"target_options.{target}.{key} must be an array of scalar values") + normalized_options[target][key] = tuple(str(v) for v in values) + + policy = cls( + version=int(raw.get("version", 1)), + name=str(raw.get("name", "local-safe")), + allowed_operations=tuple(str(v) for v in raw.get("allowed_operations", ["sample", "observe"])), + allowed_targets=tuple(str(v) for v in raw.get("allowed_targets", ["qpp-cpu"])), + allow_remote=bool(raw.get("allow_remote", False)), + allow_async=bool(raw.get("allow_async", False)), + max_qubits=int(raw.get("max_qubits", 24)), + max_shots=int(raw.get("max_shots", 100_000)), + allowed_qpu_ids=tuple(int(v) for v in raw.get("allowed_qpu_ids", [0])), + require_seed=bool(raw.get("require_seed", True)), + target_options=normalized_options, + ) + policy._validate_self() + return policy + + def _validate_self(self) -> None: + if self.version != 1: + raise ValueError(f"unsupported policy version: {self.version}") + if not self.name.strip(): + raise ValueError("policy name cannot be empty") + if self.max_qubits < 1: + raise ValueError("max_qubits must be positive") + if self.max_shots < 1: + raise ValueError("max_shots must be positive") + if any(qpu < 0 for qpu in self.allowed_qpu_ids): + raise ValueError("allowed_qpu_ids cannot contain negative values") + + @property + def hash(self) -> str: + data = asdict(self) + data["target_options"] = { + target: {key: list(values) for key, values in sorted(options.items())} + for target, options in sorted(self.target_options.items()) + } + return sha256_json(data) + + def evaluate(self, request: ExecutionRequest, target_info: TargetInfo | None = None) -> PolicyDecision: + violations: list[str] = [] + if request.operation not in self.allowed_operations: + violations.append("operation_not_allowed") + if request.target not in self.allowed_targets: + violations.append("target_not_allowed") + if request.qubits < 1: + violations.append("invalid_qubit_count") + elif request.qubits > self.max_qubits: + violations.append("qubit_limit_exceeded") + if request.shots is not None: + if request.shots < 1: + violations.append("invalid_shot_count") + elif request.shots > self.max_shots: + violations.append("shot_limit_exceeded") + if request.async_mode and not self.allow_async: + violations.append("async_not_allowed") + if request.qpu_id not in self.allowed_qpu_ids: + violations.append("qpu_id_not_allowed") + if self.require_seed and request.seed is None: + violations.append("seed_required") + if target_info is not None and target_info.is_remote and not self.allow_remote: + violations.append("remote_target_not_allowed") + + allowed_options = self.target_options.get(request.target, {}) + for key, value in request.target_options.items(): + if key not in allowed_options: + violations.append(f"target_option_not_allowed:{key}") + continue + allowed_values = allowed_options[key] + if "*" not in allowed_values and str(value) not in allowed_values: + violations.append(f"target_option_value_not_allowed:{key}") + + return PolicyDecision( + allowed=not violations, + policy_name=self.name, + policy_hash=self.hash, + violations=tuple(violations), + ) + + def evaluate_resources(self, request: ExecutionRequest, resources: dict[str, Any]) -> PolicyDecision: + violations: list[str] = [] + try: + actual_qubits = int(resources["num_qubits"]) + except (KeyError, TypeError, ValueError): + actual_qubits = 0 + if actual_qubits < 1: + violations.append("resource_estimate_invalid") + else: + if actual_qubits > self.max_qubits: + violations.append("resource_qubit_limit_exceeded") + if actual_qubits > request.qubits: + violations.append("declared_qubit_limit_exceeded") + return PolicyDecision( + allowed=not violations, + policy_name=self.name, + policy_hash=self.hash, + violations=tuple(violations), + ) diff --git a/src/cudaq_guard/runtime.py b/src/cudaq_guard/runtime.py new file mode 100644 index 0000000..c590964 --- /dev/null +++ b/src/cudaq_guard/runtime.py @@ -0,0 +1,100 @@ +from __future__ import annotations + +import importlib +import importlib.metadata +from typing import Any + +from .errors import CudaQUnavailableError +from .models import TargetInfo + + +class CudaQRuntime: + """Small lazy adapter over NVIDIA CUDA-Q. + + The module is imported only when execution or target discovery is requested, + so policy validation and audit verification remain usable without CUDA-Q. + """ + + def __init__(self, module: Any | None = None): + self._module = module + + @property + def cudaq(self) -> Any: + if self._module is None: + try: + self._module = importlib.import_module("cudaq") + except ImportError as exc: + raise CudaQUnavailableError( + 'CUDA-Q is not installed. Install with: pip install "nvidia-cudaq-quantum-guard[cudaq]"' + ) from exc + return self._module + + @property + def version(self) -> str | None: + try: + return importlib.metadata.version("cudaq") + except importlib.metadata.PackageNotFoundError: + return getattr(self._module, "__version__", None) if self._module is not None else None + + def available_targets(self) -> list[TargetInfo]: + targets = [] + for target in self.cudaq.get_targets(): + name = str(getattr(target, "name", "")) + simulator = str(getattr(target, "simulator", "") or "") + platform = str(getattr(target, "platform", "") or "") + description = str(getattr(target, "description", "") or "") + try: + num_qpus = int(target.num_qpus()) + except Exception: + num_qpus = 1 + remote_attr = getattr(target, "is_remote", None) + try: + is_remote = bool(remote_attr()) if callable(remote_attr) else bool(remote_attr) + except Exception: + is_remote = not bool(simulator) + if remote_attr is None: + is_remote = not bool(simulator) + targets.append(TargetInfo(name, simulator, platform, description, num_qpus, is_remote)) + return targets + + def describe_target(self, name: str) -> TargetInfo: + for target in self.available_targets(): + if target.name == name: + return target + raise ValueError(f"CUDA-Q target is not available: {name}") + + def configure(self, target: str, options: dict[str, str], seed: int | None) -> None: + self.cudaq.set_target(target, **options) + if seed is not None and hasattr(self.cudaq, "set_random_seed"): + self.cudaq.set_random_seed(int(seed)) + + def estimate_resources(self, kernel: Any, *args: Any) -> dict[str, int]: + resources = self.cudaq.estimate_resources(kernel, *args) + return { + "num_qubits": int(getattr(resources, "num_qubits", 0)), + "num_used_qubits": int(getattr(resources, "num_used_qubits", 0)), + "gate_count": int(resources.count()), + "depth": int(getattr(resources, "depth", 0)), + "multi_qubit_gate_count": int(getattr(resources, "multi_qubit_gate_count", 0)), + "multi_qubit_depth": int(getattr(resources, "multi_qubit_depth", 0)), + } + + def available_gpu_count(self) -> int | None: + fn = getattr(self.cudaq, "num_available_gpus", None) + if not callable(fn): + return None + try: + return int(fn()) + except Exception: + return None + + def sample(self, kernel: Any, *args: Any, shots: int, qpu_id: int = 0) -> Any: + if qpu_id != 0: + raise ValueError("synchronous cudaq.sample does not accept qpu_id; use asynchronous execution") + return self.cudaq.sample(kernel, *args, shots_count=shots) + + def sample_async(self, kernel: Any, *args: Any, shots: int, qpu_id: int = 0) -> Any: + return self.cudaq.sample_async(kernel, *args, shots_count=shots, qpu_id=qpu_id) + + def observe(self, kernel: Any, hamiltonian: Any, *args: Any, qpu_id: int = 0) -> Any: + return self.cudaq.observe(kernel, hamiltonian, *args, qpu_id=qpu_id) diff --git a/src/cudaq_guard/workloads.py b/src/cudaq_guard/workloads.py new file mode 100644 index 0000000..8f7d65c --- /dev/null +++ b/src/cudaq_guard/workloads.py @@ -0,0 +1,60 @@ +from __future__ import annotations + +import math +from typing import Any + + +def ghz_kernel(cudaq: Any) -> Any: + @cudaq.kernel + def kernel(qubit_count: int): + q = cudaq.qvector(qubit_count) + h(q[0]) + for index in range(qubit_count - 1): + x.ctrl(q[index], q[index + 1]) + mz(q) + + return kernel + + +def h2_vqe_problem(cudaq: Any) -> tuple[Any, Any]: + from cudaq import spin + + @cudaq.kernel + def ansatz(theta: float): + q = cudaq.qvector(2) + x(q[0]) + ry(theta, q[1]) + x.ctrl(q[1], q[0]) + + hamiltonian = ( + 5.907 + - 2.1433 * spin.x(0) * spin.x(1) + - 2.1433 * spin.y(0) * spin.y(1) + + 0.21829 * spin.z(0) + - 6.125 * spin.z(1) + ) + return ansatz, hamiltonian + + +def vqe_grid(runtime: Any, *, steps: int = 25, qpu_id: int = 0) -> dict[str, Any]: + if steps < 3: + raise ValueError("VQE grid requires at least 3 steps") + kernel, hamiltonian = h2_vqe_problem(runtime.cudaq) + best_theta = 0.0 + best_energy = math.inf + evaluations: list[dict[str, float]] = [] + for index in range(steps): + theta = -math.pi + (2.0 * math.pi * index / (steps - 1)) + result = runtime.observe(kernel, hamiltonian, theta, qpu_id=qpu_id) + energy = float(result.expectation()) + evaluations.append({"theta": theta, "energy": energy}) + if energy < best_energy: + best_energy = energy + best_theta = theta + return { + "kind": "vqe_grid", + "steps": steps, + "best_theta": best_theta, + "best_energy": best_energy, + "evaluations": evaluations, + } diff --git a/tests/test_audit.py b/tests/test_audit.py new file mode 100644 index 0000000..711649f --- /dev/null +++ b/tests/test_audit.py @@ -0,0 +1,27 @@ +import json +from pathlib import Path + +import pytest + +from cudaq_guard.audit import AuditTrail, verify_audit +from cudaq_guard.errors import AuditIntegrityError + + +def test_audit_chain_verifies_and_detects_tampering(tmp_path: Path) -> None: + path = tmp_path / "audit.jsonl" + trail = AuditTrail(path) + first = trail.append({"event": "one", "value": 1}) + second = trail.append({"event": "two", "value": 2}) + assert first["previous_hash"] == "0" * 64 + assert second["previous_hash"] == first["record_hash"] + report = verify_audit(path) + assert report["valid"] is True + assert report["records"] == 2 + + lines = path.read_text(encoding="utf-8").splitlines() + record = json.loads(lines[0]) + record["value"] = 999 + lines[0] = json.dumps(record, sort_keys=True, separators=(",", ":")) + path.write_text("\n".join(lines) + "\n", encoding="utf-8") + with pytest.raises(AuditIntegrityError): + verify_audit(path) diff --git a/tests/test_cli.py b/tests/test_cli.py new file mode 100644 index 0000000..ce6afe2 --- /dev/null +++ b/tests/test_cli.py @@ -0,0 +1,28 @@ +from pathlib import Path + +from cudaq_guard.cli import main + + +def test_policy_check_works_without_cudaq(tmp_path: Path, capsys) -> None: + policy = tmp_path / "policy.toml" + policy.write_text( + '\n'.join([ + 'version = 1', + 'name = "test"', + 'allowed_operations = ["sample"]', + 'allowed_targets = ["qpp-cpu"]', + 'allow_remote = false', + 'allow_async = false', + 'max_qubits = 5', + 'max_shots = 2000', + 'allowed_qpu_ids = [0]', + 'require_seed = true', + ]) + '\n', + encoding="utf-8", + ) + rc = main([ + "policy", "check", "--policy", str(policy), "--operation", "sample", + "--target", "qpp-cpu", "--qubits", "4", "--shots", "1000", "--seed", "7" + ]) + assert rc == 0 + assert '"allowed": true' in capsys.readouterr().out diff --git a/tests/test_crypto.py b/tests/test_crypto.py new file mode 100644 index 0000000..9a8acab --- /dev/null +++ b/tests/test_crypto.py @@ -0,0 +1,8 @@ +from cudaq_guard.crypto import redact_mapping + + +def test_sensitive_target_options_are_redacted() -> None: + value = redact_mapping({"machine": "x", "api_token": "secret", "nested": {"password": "p"}}) + assert value["machine"] == "x" + assert value["api_token"] == "" + assert value["nested"]["password"] == "" diff --git a/tests/test_doctor.py b/tests/test_doctor.py new file mode 100644 index 0000000..063d30e --- /dev/null +++ b/tests/test_doctor.py @@ -0,0 +1,33 @@ +from cudaq_guard.doctor import collect_doctor +from cudaq_guard.runtime import CudaQRuntime + + +class FakeTarget: + name = "qpp-cpu" + simulator = "qpp" + platform = "default" + description = "CPU" + + def num_qpus(self): + return 1 + + def is_remote(self): + return False + + +class FakeCudaQ: + __version__ = "0.fake" + + def get_targets(self): + return [FakeTarget()] + + def num_available_gpus(self): + return 2 + + +def test_doctor_reports_targets_with_fake_runtime() -> None: + report = collect_doctor(CudaQRuntime(FakeCudaQ())) + assert report["cudaq_installed"] is True + assert report["targets"][0]["name"] == "qpp-cpu" + assert report["targets"][0]["is_remote"] is False + assert report["cudaq_gpu_count"] == 2 diff --git a/tests/test_guard.py b/tests/test_guard.py new file mode 100644 index 0000000..00726b1 --- /dev/null +++ b/tests/test_guard.py @@ -0,0 +1,130 @@ +import json +from pathlib import Path + +import pytest + +from cudaq_guard.audit import verify_audit +from cudaq_guard.errors import PolicyDeniedError +from cudaq_guard.guard import Guard +from cudaq_guard.models import ExecutionRequest, TargetInfo +from cudaq_guard.policy import GuardPolicy +from cudaq_guard.runtime import CudaQRuntime + + +class FakeTarget: + name = "qpp-cpu" + simulator = "qpp" + platform = "default" + description = "fake CPU simulator" + + def num_qpus(self): + return 1 + + def is_remote(self): + return False + + +class FakeResources: + def __init__(self, qubits=2): + self.num_qubits = qubits + self.num_used_qubits = qubits + self.depth = 3 + self.multi_qubit_gate_count = max(0, qubits - 1) + self.multi_qubit_depth = max(0, qubits - 1) + + def count(self): + return max(1, self.num_qubits * 2 - 1) + + +class FakeCudaQ: + __version__ = "0.test" + + def __init__(self): + self.target = None + self.seed = None + + def get_targets(self): + return [FakeTarget()] + + def set_target(self, target, **options): + self.target = (target, options) + + def set_random_seed(self, seed): + self.seed = seed + + def estimate_resources(self, kernel, *args): + return FakeResources(int(args[0]) if args else 2) + + def num_available_gpus(self): + return 0 + + +def request(**kwargs): + data = dict( + workload="unit", + operation="sample", + target="qpp-cpu", + qubits=2, + shots=100, + qpu_id=0, + seed=7, + target_options={}, + ) + data.update(kwargs) + return ExecutionRequest(**data) + + +def test_guard_executes_allowed_request_and_audits(tmp_path: Path) -> None: + fake = FakeCudaQ() + runtime = CudaQRuntime(fake) + audit = tmp_path / "audit.jsonl" + guard = Guard(GuardPolicy(max_qubits=4), audit_path=str(audit), runtime=runtime) + result = guard.execute(request(), lambda rt: {"00": 50, "11": 50}) + assert result["00"] == 50 + assert fake.target == ("qpp-cpu", {}) + assert fake.seed == 7 + assert verify_audit(audit)["records"] == 1 + record = json.loads(audit.read_text(encoding="utf-8")) + assert record["status"] == "completed" + assert record["result"]["shots"] == 100 + + +def test_guard_denies_before_execution(tmp_path: Path) -> None: + fake = FakeCudaQ() + runtime = CudaQRuntime(fake) + audit = tmp_path / "audit.jsonl" + guard = Guard(GuardPolicy(max_qubits=2), audit_path=str(audit), runtime=runtime) + called = False + + def work(_): + nonlocal called + called = True + + with pytest.raises(PolicyDeniedError): + guard.execute(request(qubits=3), work) + assert called is False + record = json.loads(audit.read_text(encoding="utf-8")) + assert record["status"] == "denied" + + +def test_guard_resource_probe_blocks_underdeclared_kernel(tmp_path: Path) -> None: + fake = FakeCudaQ() + runtime = CudaQRuntime(fake) + audit = tmp_path / "audit.jsonl" + guard = Guard(GuardPolicy(max_qubits=8), audit_path=str(audit), runtime=runtime) + called = False + + def work(_): + nonlocal called + called = True + + with pytest.raises(PolicyDeniedError, match="declared_qubit_limit_exceeded"): + guard.execute( + request(qubits=2), + work, + resource_probe=lambda rt: rt.estimate_resources(object(), 3), + ) + assert called is False + record = json.loads(audit.read_text(encoding="utf-8")) + assert record["status"] == "denied_resource" + assert record["resources"]["num_qubits"] == 3 diff --git a/tests/test_policy.py b/tests/test_policy.py new file mode 100644 index 0000000..57038a3 --- /dev/null +++ b/tests/test_policy.py @@ -0,0 +1,91 @@ +from pathlib import Path + +import pytest + +from cudaq_guard.models import ExecutionRequest, TargetInfo +from cudaq_guard.policy import GuardPolicy + + +def req(**kwargs): + base = dict( + workload="ghz", + operation="sample", + target="qpp-cpu", + qubits=4, + shots=1000, + qpu_id=0, + seed=7, + target_options={}, + ) + base.update(kwargs) + return ExecutionRequest(**base) + + +def test_local_policy_allows_bounded_cpu_sample() -> None: + policy = GuardPolicy() + decision = policy.evaluate(req(), TargetInfo("qpp-cpu", simulator="qpp", is_remote=False)) + assert decision.allowed + assert not decision.violations + + +@pytest.mark.parametrize( + ("execution_request", "violation"), + [ + (req(operation="run"), "operation_not_allowed"), + (req(target="braket"), "target_not_allowed"), + (req(qubits=25), "qubit_limit_exceeded"), + (req(shots=100001), "shot_limit_exceeded"), + (req(async_mode=True), "async_not_allowed"), + (req(qpu_id=1), "qpu_id_not_allowed"), + (req(seed=None), "seed_required"), + ], +) +def test_policy_denies_out_of_bounds_requests(execution_request: ExecutionRequest, violation: str) -> None: + policy = GuardPolicy(max_qubits=24) + decision = policy.evaluate(execution_request) + assert not decision.allowed + assert violation in decision.violations + + +def test_policy_denies_remote_target_after_runtime_inspection() -> None: + policy = GuardPolicy(allowed_targets=("braket",), allow_remote=False) + request = req(target="braket") + decision = policy.evaluate(request, TargetInfo("braket", simulator="", is_remote=True)) + assert not decision.allowed + assert "remote_target_not_allowed" in decision.violations + + +def test_target_options_must_be_explicitly_allowed() -> None: + policy = GuardPolicy( + allowed_targets=("nvidia",), + target_options={"nvidia": {"option": ("mqpu",)}}, + ) + assert policy.evaluate(req(target="nvidia", target_options={"option": "mqpu"})).allowed + decision = policy.evaluate(req(target="nvidia", target_options={"option": "fp64"})) + assert not decision.allowed + assert "target_option_value_not_allowed:option" in decision.violations + + +def test_target_option_wildcard_allows_explicit_key() -> None: + policy = GuardPolicy( + allowed_targets=("braket",), + allow_remote=True, + require_seed=False, + target_options={"braket": {"machine": ("*",)}}, + ) + decision = policy.evaluate( + req(target="braket", seed=None, target_options={"machine": "arn:example"}), + TargetInfo("braket", simulator="", is_remote=True), + ) + assert decision.allowed + + +def test_policy_hash_is_stable() -> None: + assert GuardPolicy().hash == GuardPolicy().hash + + +def test_unknown_toml_keys_fail_closed(tmp_path: Path) -> None: + path = tmp_path / "bad.toml" + path.write_text('version = 1\nname = "x"\nmagic = true\n', encoding="utf-8") + with pytest.raises(ValueError, match="unknown policy keys"): + GuardPolicy.from_toml(path) From 6be166b42199a6a05b9534b67ec35a57fc869617 Mon Sep 17 00:00:00 2001 From: Sylvester Kaczmarek <16242628+sylvesterkaczmarek@users.noreply.github.com> Date: Sun, 9 Aug 2026 14:07:37 +0100 Subject: [PATCH 2/2] Harden CUDA-Q target classification and result reporting --- .github/workflows/ci.yml | 10 +++++++++- docs/policy-reference.md | 2 +- src/cudaq_guard/guard.py | 22 +++++++++------------- src/cudaq_guard/runtime.py | 10 ++++++---- tests/test_doctor.py | 24 ++++++++++++++++++++++++ tests/test_guard.py | 16 +++++++++++++++- 6 files changed, 64 insertions(+), 20 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 8e829a1..749587c 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -45,7 +45,15 @@ jobs: run: cudaq-guard doctor --json - name: CPU simulator smoke test run: | - cudaq-guard run ghz --policy policies/local-safe.toml --target qpp-cpu --qubits 3 --shots 256 --seed 7 --audit /tmp/cudaq-guard-audit.jsonl + cudaq-guard run ghz --policy policies/local-safe.toml --target qpp-cpu --qubits 3 --shots 256 --seed 7 --audit /tmp/cudaq-guard-audit.jsonl > /tmp/ghz.json + cat /tmp/ghz.json + python - <<'PY' + import json + data = json.load(open('/tmp/ghz.json')) + assert data['kind'] == 'sample' + assert data['shots'] == 256 + assert set(data['counts']).issubset({'000', '111'}) + PY cudaq-guard audit verify /tmp/cudaq-guard-audit.jsonl - name: VQE smoke test run: cudaq-guard run vqe --policy policies/local-safe.toml --target qpp-cpu --steps 7 --seed 7 --audit /tmp/cudaq-guard-audit.jsonl diff --git a/docs/policy-reference.md b/docs/policy-reference.md index 4f41003..4359023 100644 --- a/docs/policy-reference.md +++ b/docs/policy-reference.md @@ -20,7 +20,7 @@ option = ["mqpu", "fp64"] A request is denied when any bound is exceeded. Unknown policy keys, unlisted target options, unknown target-option values, and unsupported policy versions fail closed. -`allow_remote = false` is deliberately conservative. The runtime uses CUDA-Q's target `is_remote()` metadata when available and falls back conservatively for older targets. A remote policy must explicitly allow both remote execution and the provider target name. +`allow_remote = false` is deliberately conservative. Provider target definitions can report `is_remote() = false` before provider-specific configuration is supplied, so the runtime treats a target as remote/hardware when CUDA-Q reports it as remote **or** when it has no local simulator backend. A remote policy must explicitly allow both remote execution and the provider target name. A target-option value of `"*"` allows any value for that explicitly named option key. This is useful for provider-specific machine identifiers, but should be used narrowly. diff --git a/src/cudaq_guard/guard.py b/src/cudaq_guard/guard.py index 233cd3f..c58b9b6 100644 --- a/src/cudaq_guard/guard.py +++ b/src/cudaq_guard/guard.py @@ -27,8 +27,6 @@ def __init__( self.audit = AuditTrail(audit_path) if audit_path else None def authorize(self, request: ExecutionRequest) -> PolicyDecision: - # First evaluate without touching CUDA-Q. A disallowed target must not be - # configured just to discover that it is disallowed. static = self.policy.evaluate(request) if not static.allowed: return static @@ -64,9 +62,7 @@ def execute( resources = resource_probe(self.runtime) resource_decision = self.policy.evaluate_resources(request, resources) if not resource_decision.allowed: - self._audit( - run_id, request, resource_decision, started, "denied_resource", resources=resources - ) + self._audit(run_id, request, resource_decision, started, "denied_resource", resources=resources) raise PolicyDeniedError( f"execution denied by policy {resource_decision.policy_name}: " f"{', '.join(resource_decision.violations)}" @@ -78,9 +74,7 @@ def execute( self._audit(run_id, request, decision, started, "error", error=exc, resources=resources) raise - self._audit( - run_id, request, decision, started, "completed", result=result, resources=resources - ) + self._audit(run_id, request, decision, started, "completed", result=result, resources=resources) return result def _audit( @@ -121,17 +115,19 @@ def _audit( def summarize_result(result: Any) -> dict[str, Any]: - if hasattr(result, "expectation") and callable(result.expectation): - try: - return {"kind": "observe", "expectation": float(result.expectation())} - except Exception: - pass + # CUDA-Q SampleResult exposes both mapping-style counts and expectation(). + # Mapping semantics are therefore checked first so samples are not mislabeled. if hasattr(result, "items"): try: counts = {str(k): int(v) for k, v in result.items()} return {"kind": "sample", "counts": counts, "shots": int(sum(counts.values()))} except Exception: pass + if hasattr(result, "expectation") and callable(result.expectation): + try: + return {"kind": "observe", "expectation": float(result.expectation())} + except Exception: + pass if isinstance(result, dict): return {"kind": "mapping", "value": result} return {"kind": "text", "value": str(result)} diff --git a/src/cudaq_guard/runtime.py b/src/cudaq_guard/runtime.py index c590964..bca38f2 100644 --- a/src/cudaq_guard/runtime.py +++ b/src/cudaq_guard/runtime.py @@ -49,11 +49,13 @@ def available_targets(self) -> list[TargetInfo]: num_qpus = 1 remote_attr = getattr(target, "is_remote", None) try: - is_remote = bool(remote_attr()) if callable(remote_attr) else bool(remote_attr) + reported_remote = bool(remote_attr()) if callable(remote_attr) else bool(remote_attr) except Exception: - is_remote = not bool(simulator) - if remote_attr is None: - is_remote = not bool(simulator) + reported_remote = False + # Provider target definitions in CUDA-Q may report is_remote=False until + # provider-specific configuration is supplied. Treat targets without a + # local simulator backend as remote/hardware conservatively. + is_remote = reported_remote or not bool(simulator) targets.append(TargetInfo(name, simulator, platform, description, num_qpus, is_remote)) return targets diff --git a/tests/test_doctor.py b/tests/test_doctor.py index 063d30e..e268cea 100644 --- a/tests/test_doctor.py +++ b/tests/test_doctor.py @@ -31,3 +31,27 @@ def test_doctor_reports_targets_with_fake_runtime() -> None: assert report["targets"][0]["name"] == "qpp-cpu" assert report["targets"][0]["is_remote"] is False assert report["cudaq_gpu_count"] == 2 + + +class FakeProviderTarget: + name = "braket" + simulator = "" + platform = "default" + description = "provider" + + def num_qpus(self): + return 1 + + def is_remote(self): + return False + + +class FakeCudaQWithProvider(FakeCudaQ): + def get_targets(self): + return [FakeTarget(), FakeProviderTarget()] + + +def test_provider_target_without_local_simulator_is_conservatively_remote() -> None: + report = collect_doctor(CudaQRuntime(FakeCudaQWithProvider())) + provider = next(target for target in report["targets"] if target["name"] == "braket") + assert provider["is_remote"] is True diff --git a/tests/test_guard.py b/tests/test_guard.py index 00726b1..6f7bc3f 100644 --- a/tests/test_guard.py +++ b/tests/test_guard.py @@ -5,7 +5,7 @@ from cudaq_guard.audit import verify_audit from cudaq_guard.errors import PolicyDeniedError -from cudaq_guard.guard import Guard +from cudaq_guard.guard import Guard, summarize_result from cudaq_guard.models import ExecutionRequest, TargetInfo from cudaq_guard.policy import GuardPolicy from cudaq_guard.runtime import CudaQRuntime @@ -128,3 +128,17 @@ def work(_): record = json.loads(audit.read_text(encoding="utf-8")) assert record["status"] == "denied_resource" assert record["resources"]["num_qubits"] == 3 + + +class SampleLike(dict): + def expectation(self): + return 0.25 + + +def test_sample_result_is_not_mislabeled_as_observe() -> None: + summary = summarize_result(SampleLike({"000": 5, "111": 3})) + assert summary == { + "kind": "sample", + "counts": {"000": 5, "111": 3}, + "shots": 8, + }