feat(lwe): add structured recovery challenge

Author: builtin-pb

2.0/README.md

@@ -126,3 +126,14 @@ tail-frame (≥60) LPIPS-vs-GT over the unpatched baseline, gated by a wall-cloc
 guardrail (so drift can't be bought with more compute). A history-stabilization
 reference reliably beats baseline (validated t≈2.5/22 clips); beating it
 substantially is the open challenge.
+
+## Structured-LWE Public Witness Recovery
+
+This cryptanalysis task publishes 200 structured-LWE instances spanning ten
+balanced matrix/secret structure families.  Its problem ID is
+`lwe_structured_recovery`.  Agents recover any public-valid secret for as many
+instances as possible and submit an immediately updated cumulative JSON ledger;
+each solved instance contributes one point.  The evaluator holds no planted
+secret or private checking key: it deterministically reconstructs the public
+matrix and checks the submitted vector's public secret and centered-error
+predicates.

2.0/problems/lwe_structured_recovery/config.yaml

@@ -0,0 +1,26 @@
+tag: security
+runtime:
+  language: python
+  timeout_seconds: 10800
+  environment: "Public structured-LWE instances; Python 3.12 helper library; CPU only"
+  apt_packages:
+    - build-essential
+    - ca-certificates
+    - fplll-tools
+    - git
+    - libgmp-dev
+    - libmpfr-dev
+    - pkg-config
+    - python3
+    - python3-dev
+  docker:
+    image: ubuntu:24.04
+environment:
+  cpus: 8
+  memory_mb: 32768
+  storage_mb: 32768
+  build_timeout_seconds: 1800
+submission:
+  kind: file
+  path: /app/solution.json
+  max_queue_size: 3

2.0/problems/lwe_structured_recovery/evaluate.sh

@@ -0,0 +1,51 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+SCRIPT_DIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
+if [[ -n "${FCS_PYTHON:-}" ]]; then
+  PYTHON_CANDIDATES=("$FCS_PYTHON")
+else
+  PYTHON_CANDIDATES=(python3.12 python3.11 python3)
+fi
+
+PYTHON_BIN=""
+for CANDIDATE in "${PYTHON_CANDIDATES[@]}"; do
+  if command -v "$CANDIDATE" >/dev/null 2>&1 \
+    && "$CANDIDATE" -c 'import sys; raise SystemExit(sys.version_info < (3, 11))' \
+      >/dev/null 2>&1; then
+    PYTHON_BIN="$CANDIDATE"
+    break
+  fi
+done
+
+if [[ -z "$PYTHON_BIN" ]]; then
+  if command -v apt-get >/dev/null 2>&1; then
+    export DEBIAN_FRONTEND=noninteractive
+    apt-get update -qq
+    apt-get install -y -qq --no-install-recommends python3 >/dev/null
+  fi
+  for CANDIDATE in "${PYTHON_CANDIDATES[@]}"; do
+    if command -v "$CANDIDATE" >/dev/null 2>&1 \
+      && "$CANDIDATE" -c 'import sys; raise SystemExit(sys.version_info < (3, 11))' \
+        >/dev/null 2>&1; then
+      PYTHON_BIN="$CANDIDATE"
+      break
+    fi
+  done
+  if [[ -z "$PYTHON_BIN" ]]; then
+    echo "Error: Python 3.11 or newer is required" >&2
+    exit 1
+  fi
+fi
+
+if [[ $# -gt 0 ]]; then
+  SOLUTION="$1"
+else
+  SOLUTION="/work/execution_env/solution_env/solution.json"
+  CI_REFERENCE="/work/execution_env/solution_env/solution.py"
+  if [[ ! -f "$SOLUTION" && -f "$CI_REFERENCE" ]]; then
+    "$PYTHON_BIN" "$CI_REFERENCE" > "$SOLUTION"
+  fi
+fi
+
+exec "$PYTHON_BIN" "$SCRIPT_DIR/evaluator.py" "$SOLUTION"

2.0/problems/lwe_structured_recovery/evaluator.py

@@ -0,0 +1,158 @@
+"""Frontier-CS entry point for the public structured-LWE evaluator."""
+
+from __future__ import annotations
+
+import hashlib
+import math
+import os
+import stat
+import sys
+from collections.abc import Mapping
+from functools import lru_cache
+from pathlib import Path
+from typing import TypeAlias
+
+
+JsonScalar: TypeAlias = None | bool | int | float | str
+PlainData: TypeAlias = JsonScalar | list["PlainData"] | dict[str, "PlainData"]
+
+try:
+    _SOURCE_PUBLIC_DIR = (
+        Path(__file__).resolve().parent / "harbor" / "app" / "public"
+    )
+    _INITIAL_CATALOG_OVERRIDE = os.environ.get("FCS_STRUCTURED_LWE_CATALOG")
+    _IMPORT_PUBLIC_DIR = Path(
+        _SOURCE_PUBLIC_DIR
+        if _INITIAL_CATALOG_OVERRIDE is not None
+        else os.environ.get("FRONTIER_PUBLIC_DIR", _SOURCE_PUBLIC_DIR)
+    ).resolve()
+    _IMPORT_PUBLIC_DIR_TEXT = str(_IMPORT_PUBLIC_DIR)
+    while _IMPORT_PUBLIC_DIR_TEXT in sys.path:
+        sys.path.remove(_IMPORT_PUBLIC_DIR_TEXT)
+    sys.path.insert(0, _IMPORT_PUBLIC_DIR_TEXT)
+
+    from lwe_challenge.evaluator_core import evaluate_path  # noqa: E402
+    from lwe_challenge.schema import MAX_CATALOG_BYTES, Catalog  # noqa: E402
+except Exception:
+    if __name__ == "__main__":
+        print("infrastructure_error", file=sys.stderr)
+        raise SystemExit(1) from None
+    raise
+
+
+_SIDECAR_BYTES = len(f"{'0' * 64}  catalog.jsonl\n".encode("ascii"))
+
+
+def _catalog_path() -> Path:
+    override = os.environ.get("FCS_STRUCTURED_LWE_CATALOG")
+    if override is not None:
+        return Path(override).resolve()
+    public_dir = Path(
+        os.environ.get("FRONTIER_PUBLIC_DIR", _SOURCE_PUBLIC_DIR)
+    ).resolve()
+    return (public_dir / "catalog.jsonl").resolve()
+
+
+def _catalog() -> Catalog:
+    path = _catalog_path()
+    production = os.environ.get("FCS_STRUCTURED_LWE_CATALOG") is None
+    catalog_bytes = _read_regular_bytes(path, max_bytes=MAX_CATALOG_BYTES)
+    digest = hashlib.sha256(catalog_bytes).hexdigest()
+    if production:
+        expected_sidecar = f"{digest}  catalog.jsonl\n".encode("ascii")
+        try:
+            sidecar = _read_regular_bytes(
+                path.with_name("catalog.sha256"), max_bytes=_SIDECAR_BYTES
+            )
+        except ValueError:
+            raise ValueError("catalog.sha256 is not a valid sidecar") from None
+        if sidecar != expected_sidecar:
+            raise ValueError("catalog.sha256 does not match catalog.jsonl")
+    return _load_catalog(str(path), digest, production)
+
+
+def _read_regular_bytes(path: Path, *, max_bytes: int) -> bytes:
+    flags = (
+        os.O_RDONLY
+        | os.O_NONBLOCK
+        | getattr(os, "O_CLOEXEC", 0)
+        | getattr(os, "O_NOFOLLOW", 0)
+    )
+    fd = os.open(path, flags)
+    try:
+        if not stat.S_ISREG(os.fstat(fd).st_mode):
+            raise ValueError("catalog assets must be regular files")
+        data = bytearray()
+        limit = max_bytes + 1
+        while len(data) < limit:
+            chunk = os.read(fd, min(1024 * 1024, limit - len(data)))
+            if not chunk:
+                break
+            data.extend(chunk)
+    finally:
+        os.close(fd)
+    if len(data) > max_bytes:
+        raise ValueError("catalog asset exceeds its byte limit")
+    return bytes(data)
+
+
+@lru_cache(maxsize=8)
+def _load_catalog(path: str, digest: str, production: bool) -> Catalog:
+    catalog = Catalog.load(path)
+    if catalog.catalog_id != digest:
+        raise RuntimeError("catalog changed while it was being verified")
+    if production and len(catalog.instances) != 200:
+        raise ValueError("production catalog must contain exactly 200 instances")
+    return catalog
+
+
+def _plain_data(value: object) -> PlainData:
+    if isinstance(value, float) and not math.isfinite(value):
+        raise TypeError("public metrics require finite float values")
+    if value is None or isinstance(value, (bool, int, float, str)):
+        return value
+    if isinstance(value, Mapping):
+        copied: dict[str, PlainData] = {}
+        for key, item in value.items():
+            if not isinstance(key, str):
+                raise TypeError("public metrics mappings require string keys")
+            copied[key] = _plain_data(item)
+        return copied
+    if isinstance(value, tuple):
+        return [_plain_data(item) for item in value]
+    raise TypeError("public metrics contain a non-JSON value")
+
+
+def prepare() -> dict[str, object]:
+    catalog = _catalog()
+    return {
+        "instance_count": len(catalog.instances),
+        "catalog_id": catalog.catalog_id,
+    }
+
+
+def evaluate(solution_path: str) -> tuple[float, float, str, dict[str, object]]:
+    catalog = _catalog()
+    result = evaluate_path(solution_path, catalog=catalog)
+    metrics = _plain_data(result.metrics)
+    if not isinstance(metrics, dict):
+        raise TypeError("evaluator metrics must be a mapping")
+    return result.score, result.score_unbounded, result.message, metrics
+
+
+def main(argv: list[str]) -> int:
+    if len(argv) != 2:
+        print("usage: evaluator.py SOLUTION_JSON", file=sys.stderr)
+        return 2
+    try:
+        score, score_unbounded, message, _metrics = evaluate(argv[1])
+    except Exception:
+        print("infrastructure_error", file=sys.stderr)
+        return 1
+    print(message, file=sys.stderr)
+    print(f"{score:.12f} {score_unbounded:.12f}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main(sys.argv))

2.0/problems/lwe_structured_recovery/harbor/app/add_solution.py

@@ -0,0 +1,94 @@
+#!/usr/bin/env python3
+"""Merge one candidate witness into the cumulative solution ledger."""
+
+from __future__ import annotations
+
+import argparse
+import sys
+from collections.abc import Sequence
+from pathlib import Path
+from typing import NoReturn
+
+
+def _load_ledger_module():
+    public_path = str(Path(__file__).resolve().parent / "public")
+    while public_path in sys.path:
+        sys.path.remove(public_path)
+    sys.path.insert(0, public_path)
+    import lwe_challenge.ledger as ledger_module
+
+    return ledger_module
+
+
+if __name__ != "__main__":
+    ledger = _load_ledger_module()
+
+
+def merge_solution(
+    ledger_path: str | Path,
+    instance_id: str,
+    secret: Sequence[int],
+) -> int:
+    """Merge one new witness through the canonical locked transaction."""
+
+    updated = ledger.merge_witness_transaction(
+        ledger_path,
+        instance_id=instance_id,
+        secret=secret,
+        replace=False,
+    )
+    return len(updated.solutions)
+
+
+class _SanitizedArgumentParser(argparse.ArgumentParser):
+    def error(self, _message: str) -> NoReturn:
+        self.exit(2, "error: invalid command line\n")
+
+
+def _parser() -> argparse.ArgumentParser:
+    parser = _SanitizedArgumentParser()
+    parser.add_argument("instance_id", metavar="INSTANCE_ID")
+    parser.add_argument("secret", metavar="SECRET", nargs="?")
+    parser.add_argument("--ledger", default="/app/solution.json")
+    parser.add_argument("--replace", action="store_true")
+    return parser
+
+
+def main() -> int:
+    parser = _parser()
+    args, unknown = parser.parse_known_args()
+    if args.secret is None and len(unknown) == 1:
+        args.secret = unknown[0]
+        unknown = []
+    if args.secret is None or unknown:
+        parser.error("INSTANCE_ID and SECRET are required")
+    try:
+        secret = tuple(int(component, 10) for component in args.secret.split(","))
+    except ValueError:
+        parser.error("SECRET must be a comma-separated integer vector")
+
+    try:
+        if args.replace:
+            updated = ledger.merge_witness_transaction(
+                args.ledger,
+                instance_id=args.instance_id,
+                secret=secret,
+                replace=True,
+            )
+            solved_count = len(updated.solutions)
+        else:
+            solved_count = merge_solution(args.ledger, args.instance_id, secret)
+    except (OSError, TypeError, ValueError):
+        parser.exit(2, "error: unable to update ledger\n")
+
+    print(solved_count)
+    return 0
+
+
+if __name__ == "__main__":
+    try:
+        ledger = _load_ledger_module()
+    except Exception:
+        sys.stderr.write("error: unable to update ledger\n")
+        raise SystemExit(2)
+    raise SystemExit(main())