Merge pull request #415 from pq-code-package/acvp-on-the-fly

ACVP: Download testvectors transparently on the fly (Port mlkem-native PR#1119)

Author: mkannwischer

.github/workflows/base.yml

@@ -63,6 +63,31 @@ jobs:
       - name: tests func
         run: |
           ./scripts/tests func
+  quickcheck-acvp:
+    strategy:
+      fail-fast: false
+      matrix:
+        external:
+         - ${{ github.repository_owner != 'pq-code-package' }}
+        target:
+         - runner: pqcp-arm64
+           name: 'aarch64'
+         - runner: ubuntu-latest
+           name: 'x86_64'
+        acvp-version: [v1.1.0.38, v1.1.0.39, v1.1.0.40]
+        exclude:
+          - {external: true,
+             target: {
+               runner: pqcp-arm64,
+               name: 'aarch64'
+             }}
+    name: Quickcheck ACVP (${{ matrix.target.name }}, ${{ matrix.acvp-version }})
+    runs-on: ${{ matrix.target.runner }}
+    steps:
+      - uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2
+      - name: Run ACVP test
+        run: |
+          ./scripts/tests acvp --version ${{ matrix.acvp-version }}
   quickcheck_bench:
     strategy:
       fail-fast: false

.gitignore

@@ -2,3 +2,5 @@
 
 test/build
 .direnv
+# Downloaded ACVP test data
+test/.acvp-data/

Makefile

@@ -61,7 +61,7 @@ run_func_87: func_87
 run_func: run_func_44 run_func_65 run_func_87
 
 run_acvp: acvp
-	python3 ./test/acvp_client.py
+	python3 ./test/acvp_client.py $(if $(ACVP_VERSION),--version $(ACVP_VERSION))
 
 func_44: $(MLDSA44_DIR)/bin/test_mldsa44
 	$(Q)echo "  FUNC       ML-DSA-44:   $^"

scripts/tests

@@ -426,6 +426,10 @@ class Tests:
             if stack_flags:
                 env_update["STACK_ANALYSIS_FLAGS"] = " ".join(stack_flags)
 
+        # Add ACVP version for ACVP tests
+        if test_type == TEST_TYPES.ACVP and hasattr(self.args, "version"):
+            env_update["ACVP_VERSION"] = self.args.version
+
         env = os.environ.copy()
         env.update(env_update)
 
@@ -910,6 +914,12 @@ def cli():
         "acvp", help="Run ACVP client", parents=[common_parser]
     )
 
+    acvp_parser.add_argument(
+        "--version",
+        default="v1.1.0.40",
+        help="ACVP test vector version (default: v1.1.0.40)",
+    )
+
     # examples arguments
     examples_parser = cmd_subparsers.add_parser(
         "examples", help="Run examples", parents=[common_parser]

test/acvp_client.py

@@ -1,3 +1,4 @@
+#!/usr/bin/env python3
 # Copyright (c) The mldsa-native project authors
 # SPDX-License-Identifier: Apache-2.0 OR ISC OR MIT
 
@@ -8,28 +9,78 @@
 #
 # Invokes `acvp_mldsa{lvl}` under the hood.
 
+import argparse
 import os
 import json
 import sys
 import subprocess
+import urllib.request
+from pathlib import Path
 
 # Check if we need to use a wrapper for execution (e.g. QEMU)
 exec_prefix = os.environ.get("EXEC_WRAPPER", "")
 exec_prefix = [exec_prefix] if exec_prefix != "" else []
 
-acvp_dir = "test/acvp_data"
-acvp_keygen_json = f"{acvp_dir}/acvp_keygen_internalProjection.json"
-acvp_sigGen_json = f"{acvp_dir}/acvp_sigGen_internalProjection.json"
-acvp_sigVer_json = f"{acvp_dir}/acvp_sigVer_internalProjection.json"
 
-with open(acvp_keygen_json, "r") as f:
-    acvp_keygen_data = json.load(f)
+def download_acvp_files(version="v1.1.0.40"):
+    """Download ACVP test files for the specified version if not present."""
+    base_url = f"https://raw.githubusercontent.com/usnistgov/ACVP-Server/{version}/gen-val/json-files"
 
-with open(acvp_sigGen_json, "r") as f:
-    acvp_sigGen_data = json.load(f)
+    # Files we need to download for ML-KEM
+    files_to_download = [
+        "ML-DSA-keyGen-FIPS204/internalProjection.json",
+        "ML-DSA-sigGen-FIPS204/internalProjection.json",
+        "ML-DSA-sigGen-FIPS204/internalProjection.json",
+    ]
 
-with open(acvp_sigVer_json, "r") as f:
-    acvp_sigVer_data = json.load(f)
+    # Create directory structure
+    data_dir = Path(f"test/.acvp-data/{version}/files")
+    data_dir.mkdir(parents=True, exist_ok=True)
+
+    for file_path in files_to_download:
+        local_file = data_dir / file_path
+        local_file.parent.mkdir(parents=True, exist_ok=True)
+
+        if not local_file.exists():
+            url = f"{base_url}/{file_path}"
+            print(f"Downloading {file_path}...", file=sys.stderr)
+            try:
+                urllib.request.urlretrieve(url, local_file)
+                # Verify the file is valid JSON
+                with open(local_file, "r") as f:
+                    json.load(f)
+            except json.JSONDecodeError as e:
+                print(
+                    f"Error: Downloaded file {file_path} is not valid JSON: {e}",
+                    file=sys.stderr,
+                )
+                local_file.unlink(missing_ok=True)
+                return False
+            except Exception as e:
+                print(f"Error downloading {file_path}: {e}", file=sys.stderr)
+                local_file.unlink(missing_ok=True)
+                return False
+
+    return True
+
+
+def loadAcvpData(internalProjection):
+    with open(internalProjection, "r") as f:
+        internalProjectionData = json.load(f)
+    return (internalProjection, internalProjectionData)
+
+
+def loadDefaultAcvpData(version="v1.1.0.40"):
+    data_dir = f"test/.acvp-data/{version}/files"
+    acvp_jsons_for_version = [
+        f"{data_dir}/ML-DSA-keyGen-FIPS204/internalProjection.json",
+        f"{data_dir}/ML-DSA-sigGen-FIPS204/internalProjection.json",
+        f"{data_dir}/ML-DSA-sigGen-FIPS204/internalProjection.json",
+    ]
+    acvp_data = []
+    for internalProjection in acvp_jsons_for_version:
+        acvp_data.append(loadAcvpData(internalProjection))
+    return acvp_data
 
 
 def err(msg, **kwargs):
@@ -173,14 +224,62 @@ def run_sigVer_test(tg, tc):
     info("OK")
 
 
-for tg in acvp_keygen_data["testGroups"]:
-    for tc in tg["tests"]:
-        run_keyGen_test(tg, tc)
+def runTestSingle(internalProjectionName, internalProjection):
+    info(f"Running ACVP tests for {internalProjectionName}")
+
+    assert internalProjection["algorithm"] == "ML-DSA"
+    assert (
+        internalProjection["mode"] == "keyGen"
+        or internalProjection["mode"] == "sigGen"
+        or internalProjection["mode"] == "sigVer"
+    )
+
+    # copy top level fields into the results
+    results = internalProjection.copy()
+
+    results["testGroups"] = []
+    for tg in internalProjection["testGroups"]:
+        tgResult = {
+            "tgId": tg["tgId"],
+            "tests": [],
+        }
+        results["testGroups"].append(tgResult)
+        for tc in tg["tests"]:
+            if internalProjection["mode"] == "keyGen":
+                result = run_keyGen_test(tg, tc)
+            elif internalProjection["mode"] == "sigGen":
+                result = run_sigGen_test(tg, tc)
+            elif internalProjection["mode"] == "sigVer":
+                result = run_sigVer_test(tg, tc)
+            tgResult["tests"].append(result)
+
+
+def runTest(data):
+    for internalProjectionName, internalProjection in data:
+        runTestSingle(internalProjectionName, internalProjection)
+    info("ALL GOOD!")
+
+
+def test(version="v1.1.0.40"):
+    # load data from downloaded files
+    data = loadDefaultAcvpData(version)
+
+    runTest(data)
+
+
+parser = argparse.ArgumentParser()
+
+parser.add_argument(
+    "--version",
+    "-v",
+    default="v1.1.0.40",
+    help="ACVP test vector version (default: v1.1.0.40)",
+)
+args = parser.parse_args()
 
-for tg in acvp_sigGen_data["testGroups"]:
-    for tc in tg["tests"]:
-        run_sigGen_test(tg, tc)
+# Download files if needed
+if not download_acvp_files(args.version):
+    print("Failed to download ACVP test files", file=sys.stderr)
+    sys.exit(1)
 
-for tg in acvp_sigVer_data["testGroups"]:
-    for tc in tg["tests"]:
-        run_sigVer_test(tg, tc)
+test(args.version)