4 Commits

23 changed files with 3126 additions and 132 deletions

View File

@@ -1,18 +1,22 @@
# rpctl # rpctl
`rpctl` 是四款 RP 本地控制 CLI,用于在实验服务器本地独立启动、停止和观察单个 RP/profile 的 all5 实验。一个实验可包含一次 snapshot 和多次 delta 子 run。 `rpctl` 是五款 RP 本地控制 CLI,用于在实验服务器本地独立启动、停止和观察单个 RP/profile 的 all5 实验。一个实验可包含一次 snapshot 和多次复用 state 的子 run。
## 部署 ## 部署
```bash ```bash
mkdir -p /root/rpki_4rp_ctl mkdir -p /data/rpki_4rp_ctl
rsync -a rpki_2/rpki/scripts/rpctl/ root@47.77.237.41:/root/rpki_4rp_ctl/ rsync -a rpki_2/rpki/scripts/rpctl/ root@47.251.127.231:/data/rpki_4rp_ctl/
rsync -a specs/develop/20260707_3/feature100_four_rp_interval_archive/runtime_bundle/ root@47.77.237.41:/root/rpki_4rp_ctl_runtime_bundle/ rsync -a artifacts/rpki-prover/rpki-prover-6857d4bf-linux-amd64.docker.tar.gz root@47.251.127.231:/data/
ssh root@47.77.237.41 'cd /root/rpki_4rp_ctl && ./rpctl init --bundle /root/rpki_4rp_ctl_runtime_bundle' ssh root@47.251.127.231 'cd /data/rpki_4rp_ctl && ./rpctl init \
--bundle /data/rpki_4rp_ctl_runtime_bundle \
--rpki-prover-image /data/rpki-prover-6857d4bf-linux-amd64.docker.tar.gz'
``` ```
`--bundle` 目录需要包含 #100 runtime bundle 的 `bin/`、`lib/`、`fixtures/`、`scripts/` 和 `versions.json`。不要使用早期 precheck 目录作为标准 bundle,因为它可能缺少 FORT 运行所需动态库。 `--bundle` 目录需要包含 #100 runtime bundle 的 `bin/`、`lib/`、`fixtures/`、`scripts/` 和 `versions.json`。不要使用早期 precheck 目录作为标准 bundle,因为它可能缺少 FORT 运行所需动态库。
`--rpki-prover-image` 是可选项;不配置时原有四款 RP 仍可运行,但启动 rpki-prover 会明确报错。配置后 `init` 会复制并加载离线镜像,校验宿主机和镜像均为 `amd64`,确认 tag 为 `rpki-prover:6857d4bf`、版本为 `rpki-prover-0.10.1`,并将 image ID 与 archive SHA-256 写入 `config.json`。运行时不会在线拉取镜像。
## 常用命令 ## 常用命令
```bash ```bash
@@ -20,6 +24,8 @@ ssh root@47.77.237.41 'cd /root/rpki_4rp_ctl && ./rpctl init --bundle /root/rpki
./rpctl status ./rpctl status
./rpctl start --rp ours-pp-object-cache --interval 10m --runs 6 --rirs all5 ./rpctl start --rp ours-pp-object-cache --interval 10m --runs 6 --rirs all5
./rpctl start --rp routinator --interval 0 --runs 1 --rirs apnic ./rpctl start --rp routinator --interval 0 --runs 1 --rirs apnic
./rpctl start --rp rpki-prover --interval 0 --runs 2 --rirs all5
./rpctl start --rp rpki-prover-full --interval 0 --runs 1 --rirs all5
./rpctl continue --exp-id <successful-exp-id> --runs 3 --interval 10m ./rpctl continue --exp-id <successful-exp-id> --runs 3 --interval 10m
./rpctl tail ./rpctl tail
./rpctl show ./rpctl show
@@ -40,6 +46,10 @@ ssh root@47.77.237.41 'cd /root/rpki_4rp_ctl && ./rpctl init --bundle /root/rpki
| `routinator` | `routinator-latest-release` | | `routinator` | `routinator-latest-release` |
| `rpki-client` | `rpki-client-latest-release` | | `rpki-client` | `rpki-client-latest-release` |
| `fort` | `fort-latest-release` | | `fort` | `fort-latest-release` |
| `rpki-prover`, `rpki-prover-incremental` | `rpki-prover-incremental` |
| `rpki-prover-full`, `rpki-prover-no-incremental` | `rpki-prover-full` |
`rpki-prover-incremental` 使用上游默认增量算法;`rpki-prover-full` 显式传入 `--no-incremental-validation`。两者的 state 位于各自实验/profile 目录,互不复用。首轮显示为 `snapshot`;后续和 `continue` 追加轮显示为 `warm`,表示保留了上轮 state,实际 RRDP snapshot/delta 决策仍由 rpki-prover 自身完成。
## 数据目录 ## 数据目录
@@ -53,7 +63,9 @@ ssh root@47.77.237.41 'cd /root/rpki_4rp_ctl && ./rpctl init --bundle /root/rpki
runs/<profile>/<exp-id>/profiles/<profile>/runs/run_0001/ runs/<profile>/<exp-id>/profiles/<profile>/runs/run_0001/
``` ```
每个子 run 目录包含 `stdout.log`、`stderr.log`、`process-time.txt`、`exit-code.txt`、`run-meta.json` 以及各 RP 产物。 每个子 run 目录包含 `stdout.log`、`stderr.log`、`exit-code.txt`、`run-meta.json` 以及各 RP 产物。传统四 RP 使用 `process-time.txt`;rpki-prover 额外保存 `container-inspect.json`、`rss-samples.tsv` 和 `vrps.csv`。
rpki-prover 的 `run-meta.json` 和 `view` 同时记录 VRP 原始行数与按 `(ASN, prefix, maxLength)` 去重数。其当前 CLI 不输出 VAP,所以统一显示 `n/a`,不是 `0`。max RSS 是容器全部相关 PID 的 RSS 峰值汇总,不是短暂 Docker client 进程的 RSS。
## 历史查看与删除 ## 历史查看与删除

View File

@@ -4,8 +4,10 @@ from __future__ import annotations
import argparse import argparse
import csv import csv
import gzip import gzip
import hashlib
import json import json
import os import os
import platform
import re import re
import shutil import shutil
import signal import signal
@@ -37,6 +39,8 @@ PROFILES = {
"routinator-latest-release": "routinator", "routinator-latest-release": "routinator",
"rpki-client-latest-release": "rpki-client", "rpki-client-latest-release": "rpki-client",
"fort-latest-release": "fort", "fort-latest-release": "fort",
"rpki-prover-incremental": "rpki-prover",
"rpki-prover-full": "rpki-prover",
} }
ALIASES = { ALIASES = {
@@ -52,8 +56,15 @@ ALIASES = {
"rpki-client-latest-release": "rpki-client-latest-release", "rpki-client-latest-release": "rpki-client-latest-release",
"fort": "fort-latest-release", "fort": "fort-latest-release",
"fort-latest-release": "fort-latest-release", "fort-latest-release": "fort-latest-release",
"rpki-prover": "rpki-prover-incremental",
"rpki-prover-incremental": "rpki-prover-incremental",
"rpki-prover-full": "rpki-prover-full",
"rpki-prover-no-incremental": "rpki-prover-full",
} }
RPKI_PROVER_IMAGE_TAG = "rpki-prover:6857d4bf"
RPKI_PROVER_VERSION = "rpki-prover-0.10.1"
DEFAULT_CONFIG = { DEFAULT_CONFIG = {
"root": "/root/rpki_4rp_ctl", "root": "/root/rpki_4rp_ctl",
"bundle": "", "bundle": "",
@@ -207,6 +218,14 @@ def command_string(argv: list[str]) -> str:
return " ".join(shlex.quote(x) for x in argv) return " ".join(shlex.quote(x) for x in argv)
def sha256_file(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as handle:
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
digest.update(chunk)
return digest.hexdigest()
def check_bundle(root: Path) -> None: def check_bundle(root: Path) -> None:
required = ["bin/rpki", "bin/routinator", "bin/rpki-client", "bin/fort", "fixtures/tal", "scripts/normalize_and_collect.py", "scripts/write_group_report.py", "versions.json"] required = ["bin/rpki", "bin/routinator", "bin/rpki-client", "bin/fort", "fixtures/tal", "scripts/normalize_and_collect.py", "scripts/write_group_report.py", "versions.json"]
missing = [item for item in required if not (root / item).exists()] missing = [item for item in required if not (root / item).exists()]
@@ -232,6 +251,102 @@ def install_from_bundle(bundle: Path) -> None:
item.chmod(item.stat().st_mode | 0o111) item.chmod(item.stat().st_mode | 0o111)
def validate_rpki_prover_host() -> None:
arch = platform.machine().lower()
if arch not in {"x86_64", "amd64"}:
raise SystemExit(f"rpki-prover image requires amd64 host, found: {arch}")
if shutil.which("docker") is None:
raise SystemExit("docker is required for rpki-prover profiles")
result = subprocess.run(
["docker", "info"],
stdout=subprocess.DEVNULL,
stderr=subprocess.PIPE,
text=True,
check=False,
)
if result.returncode != 0:
raise SystemExit(f"docker is not available: {result.stderr.strip()}")
def inspect_rpki_prover_image() -> dict[str, str]:
inspect = subprocess.run(
["docker", "image", "inspect", RPKI_PROVER_IMAGE_TAG],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
check=False,
)
if inspect.returncode != 0:
raise SystemExit(f"loaded image is missing tag {RPKI_PROVER_IMAGE_TAG}: {inspect.stderr.strip()}")
data = json.loads(inspect.stdout)[0]
architecture = str(data.get("Architecture") or "")
if architecture != "amd64":
raise SystemExit(f"rpki-prover image architecture mismatch: expected amd64, found {architecture or '<unknown>'}")
version = subprocess.run(
["docker", "run", "--rm", "--entrypoint", "/opt/app/rpki-prover", RPKI_PROVER_IMAGE_TAG, "--version"],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
check=False,
)
version_text = version.stdout.strip()
if version.returncode != 0 or version_text != RPKI_PROVER_VERSION:
raise SystemExit(
f"rpki-prover version check failed: expected {RPKI_PROVER_VERSION}, "
f"exit={version.returncode} output={version_text or version.stderr.strip()}"
)
return {
"tag": RPKI_PROVER_IMAGE_TAG,
"imageId": str(data.get("Id") or ""),
"architecture": architecture,
"version": version_text,
}
def install_rpki_prover_image(archive: Path) -> dict[str, str]:
if not archive.is_file():
raise SystemExit(f"rpki-prover image archive does not exist: {archive}")
validate_rpki_prover_host()
image_dir = ctl_root() / "images"
image_dir.mkdir(parents=True, exist_ok=True)
destination = image_dir / archive.name
if archive != destination:
shutil.copy2(archive, destination)
load = subprocess.run(
["docker", "load", "--input", str(destination)],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
check=False,
)
if load.returncode != 0:
raise SystemExit(f"failed to load rpki-prover image archive: {load.stderr.strip()}")
metadata = inspect_rpki_prover_image()
metadata.update({
"archive": str(destination),
"archiveSha256": sha256_file(destination),
"loadedAtUtc": utc_now(),
})
return metadata
def require_rpki_prover_image() -> dict[str, Any]:
metadata = load_config().get("rpkiProverImage")
if not isinstance(metadata, dict) or metadata.get("tag") != RPKI_PROVER_IMAGE_TAG:
raise SystemExit(
"rpki-prover image is not initialized; run ./rpctl init --bundle <bundle> "
"--rpki-prover-image <docker.tar.gz>"
)
validate_rpki_prover_host()
inspected = inspect_rpki_prover_image()
if metadata.get("imageId") and metadata.get("imageId") != inspected.get("imageId"):
raise SystemExit(
f"rpki-prover image id changed: configured={metadata.get('imageId')} "
f"loaded={inspected.get('imageId')}"
)
return metadata
def cmd_init(args: argparse.Namespace) -> None: def cmd_init(args: argparse.Namespace) -> None:
ensure_dirs() ensure_dirs()
bundle = Path(args.bundle).resolve() bundle = Path(args.bundle).resolve()
@@ -239,9 +354,16 @@ def cmd_init(args: argparse.Namespace) -> None:
cfg = load_config() cfg = load_config()
cfg["bundle"] = str(bundle) cfg["bundle"] = str(bundle)
cfg["initializedAtUtc"] = utc_now() cfg["initializedAtUtc"] = utc_now()
if args.rpki_prover_image:
cfg["rpkiProverImage"] = install_rpki_prover_image(Path(args.rpki_prover_image).resolve())
save_config(cfg) save_config(cfg)
print(f"initialized root={ctl_root()}") print(f"initialized root={ctl_root()}")
print(f"bundle={bundle}") print(f"bundle={bundle}")
if cfg.get("rpkiProverImage"):
image = cfg["rpkiProverImage"]
print(f"rpki_prover_image={image.get('tag')}")
print(f"rpki_prover_image_id={image.get('imageId')}")
print(f"rpki_prover_archive_sha256={image.get('archiveSha256')}")
def timed_run(run_dir: Path, argv: list[str], env: dict[str, str] | None = None) -> int: def timed_run(run_dir: Path, argv: list[str], env: dict[str, str] | None = None) -> int:
@@ -396,6 +518,240 @@ def build_fort_args(run_dir: Path, state: Path, rirs: list[str]) -> tuple[list[s
return argv, env return argv, env
def prepare_rpki_prover_state(state: Path, rirs: list[str]) -> None:
tal_dir = state / "tals"
tal_dir.mkdir(parents=True, exist_ok=True)
for old_tal in tal_dir.glob("*.tal"):
old_tal.unlink()
for rir in rirs:
shutil.copy2(ctl_root() / "fixtures" / "tal" / RIR_TAL[rir], tal_dir / RIR_TAL[rir])
def rpki_prover_arguments(profile: str) -> list[str]:
argv = [
"--once",
"--vrp-output", "/outputs/vrps.csv",
"--no-rir-tals",
"--rpki-root-directory", "/rpki-data",
"--cpu-count", "4",
"--fetcher-count", "8",
"--log-level", "info",
]
if profile == "rpki-prover-full":
argv.append("--no-incremental-validation")
return argv
def build_rpki_prover_command(profile: str, run_dir: Path, state: Path, container_name: str) -> list[str]:
return [
"docker", "run", "--name", container_name,
"-v", f"{state.resolve()}:/rpki-data",
"-v", f"{run_dir.resolve()}:/outputs",
RPKI_PROVER_IMAGE_TAG,
*rpki_prover_arguments(profile),
]
def container_rss_kb(container_name: str) -> tuple[int, int]:
top = subprocess.run(
["docker", "top", container_name, "-eo", "pid"],
stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL,
text=True,
check=False,
)
total = 0
maximum = 0
for line in top.stdout.splitlines():
pid = line.strip()
if not pid.isdigit():
continue
try:
status = Path(f"/proc/{pid}/status").read_text(encoding="utf-8", errors="replace")
except OSError:
continue
match = re.search(r"^VmRSS:\s+(\d+)\s+kB$", status, re.MULTILINE)
if not match:
continue
rss = int(match.group(1))
total += rss
maximum = max(maximum, rss)
return total, maximum
def timed_rpki_prover_run(
profile: str,
run_dir: Path,
state: Path,
container_name: str,
) -> tuple[int, int, int, int, dict[str, Any]]:
run_dir.mkdir(parents=True, exist_ok=True)
start_utc = utc_now()
start_ns = time.monotonic_ns()
(run_dir / "start-utc.txt").write_text(start_utc + "\n", encoding="utf-8")
create_argv = [
"docker", "create", "--name", container_name,
"-v", f"{state.resolve()}:/rpki-data",
"-v", f"{run_dir.resolve()}:/outputs",
RPKI_PROVER_IMAGE_TAG,
*rpki_prover_arguments(profile),
]
create = subprocess.run(create_argv, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, check=False)
if create.returncode != 0:
(run_dir / "stderr.log").write_text(create.stderr, encoding="utf-8")
(run_dir / "stdout.log").write_text(create.stdout, encoding="utf-8")
(run_dir / "end-utc.txt").write_text(utc_now() + "\n", encoding="utf-8")
(run_dir / "exit-code.txt").write_text(str(create.returncode) + "\n", encoding="utf-8")
return create.returncode, 0, 0, (time.monotonic_ns() - start_ns) // 1_000_000, {}
previous_sigterm = signal.getsignal(signal.SIGTERM)
def terminate_container(signum: int, _frame: Any) -> None:
subprocess.run(
["docker", "rm", "--force", container_name],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
check=False,
)
raise SystemExit(128 + signum)
signal.signal(signal.SIGTERM, terminate_container)
max_total_rss_kb = 0
max_process_rss_kb = 0
inspect_data: dict[str, Any] = {}
samples_path = run_dir / "rss-samples.tsv"
with (run_dir / "stdout.log").open("wb") as stdout, (run_dir / "stderr.log").open("wb") as stderr, samples_path.open("w", encoding="utf-8") as samples:
samples.write("epoch_ms\ttotal_rss_kb\tmax_process_rss_kb\n")
start = subprocess.Popen(["docker", "start", "--attach", container_name], stdout=stdout, stderr=stderr)
while start.poll() is None:
total_rss_kb, process_rss_kb = container_rss_kb(container_name)
max_total_rss_kb = max(max_total_rss_kb, total_rss_kb)
max_process_rss_kb = max(max_process_rss_kb, process_rss_kb)
samples.write(f"{int(time.time() * 1000)}\t{total_rss_kb}\t{process_rss_kb}\n")
samples.flush()
time.sleep(0.25)
start.wait()
try:
inspect = subprocess.run(
["docker", "inspect", container_name],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
check=False,
)
if inspect.returncode == 0:
parsed = json.loads(inspect.stdout)
if parsed:
inspect_data = parsed[0]
write_json(run_dir / "container-inspect.json", inspect_data)
state_data = inspect_data.get("State", {}) if inspect_data else {}
exit_code = int(state_data.get("ExitCode", start.returncode if start.returncode is not None else 1))
finally:
subprocess.run(["docker", "rm", "--force", container_name], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False)
signal.signal(signal.SIGTERM, previous_sigterm)
wall_ms = (time.monotonic_ns() - start_ns) // 1_000_000
(run_dir / "end-utc.txt").write_text(utc_now() + "\n", encoding="utf-8")
(run_dir / "exit-code.txt").write_text(str(exit_code) + "\n", encoding="utf-8")
return exit_code, max_total_rss_kb, max_process_rss_kb, wall_ms, inspect_data
def collect_rpki_prover_vrps(path: Path) -> tuple[int, set[str]]:
unique: set[str] = set()
raw = 0
if not path.is_file():
return raw, unique
with path.open("r", encoding="utf-8", errors="replace", newline="") as handle:
for row in csv.reader(handle):
if not row or row[0].strip().lower() == "asn":
continue
if len(row) < 3:
continue
raw += 1
unique.add(",".join(value.strip() for value in row[:3]))
return raw, unique
def write_rpki_prover_run_metadata(
run_root: Path,
profile: str,
run_seq: int,
mode: str,
wall_ms: int,
max_rss_kb: int,
max_process_rss_kb: int,
exit_code: int,
) -> dict[str, Any]:
run_dir = run_root / "profiles" / profile / "runs" / f"run_{run_seq:04d}"
raw_vrps, unique_vrps = collect_rpki_prover_vrps(run_dir / "vrps.csv")
with gzip.open(run_dir / "normalized-vrps.txt.gz", "wt", encoding="utf-8") as handle:
for row in sorted(unique_vrps):
handle.write(row + "\n")
image = load_config().get("rpkiProverImage", {})
meta = {
"profile": profile,
"rp": "rpki-prover",
"runSeq": run_seq,
"syncMode": mode,
"startUtc": read_text_strip(run_dir / "start-utc.txt"),
"endUtc": read_text_strip(run_dir / "end-utc.txt"),
"exitCode": exit_code,
"timing": {
"wallMs": wall_ms,
"maxRssKb": max_rss_kb,
"maxProcessRssKb": max_process_rss_kb,
},
"counts": {
"vrps": raw_vrps,
"vrpsRaw": raw_vrps,
"vrpsUnique": len(unique_vrps),
"vaps": "n/a",
},
"vapsAvailable": False,
"validationAlgorithm": "full-every-iteration" if profile == "rpki-prover-full" else "incremental",
"image": image,
"remoteRunDir": str(run_dir),
}
write_json(run_dir / "run-meta.json", meta)
return meta
def write_rpki_prover_group_report(run_root: Path, profile: str) -> None:
rows: list[dict[str, Any]] = []
run_dirs = sorted((run_root / "profiles" / profile / "runs").glob("run_*"))
for run_dir in run_dirs:
meta = read_json(run_dir / "run-meta.json", {})
if not meta:
continue
rows.append({
"profile": profile,
"rp": "rpki-prover",
"run_seq": meta.get("runSeq", ""),
"sync_mode": meta.get("syncMode", ""),
"start_utc": meta.get("startUtc", ""),
"end_utc": meta.get("endUtc", ""),
"exit_code": meta.get("exitCode", ""),
"wall_ms": meta.get("timing", {}).get("wallMs", ""),
"max_rss_kb": meta.get("timing", {}).get("maxRssKb", ""),
"max_process_rss_kb": meta.get("timing", {}).get("maxProcessRssKb", ""),
"vrps": meta.get("counts", {}).get("vrpsRaw", ""),
"vrps_unique": meta.get("counts", {}).get("vrpsUnique", ""),
"vaps": "n/a",
"vaps_available": False,
"validation_algorithm": meta.get("validationAlgorithm", ""),
"remote_run_dir": str(run_dir),
})
reports = run_root / "reports"
reports.mkdir(parents=True, exist_ok=True)
if rows:
with (reports / "per_run_metrics.csv").open("w", encoding="utf-8", newline="") as handle:
writer = csv.DictWriter(handle, fieldnames=list(rows[0]))
writer.writeheader()
writer.writerows(rows)
write_json(reports / "per_run_metrics.json", {"generatedAtUtc": utc_now(), "rows": rows})
def normalize_one(run_root: Path, profile: str, run_seq: int) -> None: def normalize_one(run_root: Path, profile: str, run_seq: int) -> None:
script = ctl_root() / "scripts" / "normalize_and_collect.py" script = ctl_root() / "scripts" / "normalize_and_collect.py"
if not script.exists(): if not script.exists():
@@ -460,6 +816,18 @@ def parse_entry_rirs(entry: dict[str, Any]) -> list[str]:
return parse_rirs(str(value)) return parse_rirs(str(value))
def write_experiment_versions(run_root: Path) -> None:
versions = read_json(ctl_root() / "versions.json", {})
image = load_config().get("rpkiProverImage")
if image:
versions["rpki-prover"] = image
profiles = versions.setdefault("profiles", [])
for profile in ["rpki-prover-incremental", "rpki-prover-full"]:
if profile not in profiles:
profiles.append(profile)
write_json(run_root / "versions.json", versions)
def run_profile(profile: str, exp_id: str, runs: int, interval_secs: int, rirs: list[str], append: bool = False) -> int: def run_profile(profile: str, exp_id: str, runs: int, interval_secs: int, rirs: list[str], append: bool = False) -> int:
ensure_dirs() ensure_dirs()
rp = PROFILES[profile] rp = PROFILES[profile]
@@ -467,7 +835,7 @@ def run_profile(profile: str, exp_id: str, runs: int, interval_secs: int, rirs:
profile_root = run_root / "profiles" / profile profile_root = run_root / "profiles" / profile
(run_root / "logs").mkdir(parents=True, exist_ok=True) (run_root / "logs").mkdir(parents=True, exist_ok=True)
(run_root / "reports").mkdir(parents=True, exist_ok=True) (run_root / "reports").mkdir(parents=True, exist_ok=True)
(run_root / "versions.json").write_text((ctl_root() / "versions.json").read_text(encoding="utf-8"), encoding="utf-8") write_experiment_versions(run_root)
now = utc_now() now = utc_now()
if append: if append:
run_info = normalize_experiment_metadata(read_json(run_root / "run-info.json", {})) run_info = normalize_experiment_metadata(read_json(run_root / "run-info.json", {}))
@@ -504,6 +872,9 @@ def run_profile(profile: str, exp_id: str, runs: int, interval_secs: int, rirs:
"experimentRoot": str(run_root), "experimentRoot": str(run_root),
"stateRoot": str(profile_root / "state"), "stateRoot": str(profile_root / "state"),
} }
if rp == "rpki-prover":
run_info["validationAlgorithm"] = cache_profile_name(profile)
run_info["rpkiProverImage"] = load_config().get("rpkiProverImage", {})
write_json(run_root / "run-info.json", run_info) write_json(run_root / "run-info.json", run_info)
status = { status = {
"state": "running", "state": "running",
@@ -535,7 +906,7 @@ def run_profile(profile: str, exp_id: str, runs: int, interval_secs: int, rirs:
time.sleep(sleep_seconds) time.sleep(sleep_seconds)
last_start = time.time() last_start = time.time()
snapshot = run_seq == 1 snapshot = run_seq == 1
mode = "snapshot" if snapshot else "delta" mode = "snapshot" if snapshot else ("warm" if rp == "rpki-prover" else "delta")
run_dir = profile_root / "runs" / f"run_{run_seq:04d}" run_dir = profile_root / "runs" / f"run_{run_seq:04d}"
if run_dir.exists() and any(run_dir.iterdir()): if run_dir.exists() and any(run_dir.iterdir()):
raise RuntimeError(f"refusing to overwrite existing run directory: {run_dir}") raise RuntimeError(f"refusing to overwrite existing run directory: {run_dir}")
@@ -550,13 +921,34 @@ def run_profile(profile: str, exp_id: str, runs: int, interval_secs: int, rirs:
argv, env = build_rpki_client_args(run_dir, state, rirs) argv, env = build_rpki_client_args(run_dir, state, rirs)
elif rp == "fort": elif rp == "fort":
argv, env = build_fort_args(run_dir, state, rirs) argv, env = build_fort_args(run_dir, state, rirs)
elif rp == "rpki-prover":
prepare_rpki_prover_state(state, rirs)
container_suffix = hashlib.sha1(f"{run_root}:{run_seq}".encode("utf-8")).hexdigest()[:12]
container_name = f"rpctl-prover-{container_suffix}"
argv = build_rpki_prover_command(profile, run_dir, state, container_name)
else: else:
raise RuntimeError(rp) raise RuntimeError(rp)
run_dir.mkdir(parents=True, exist_ok=True) run_dir.mkdir(parents=True, exist_ok=True)
(run_dir / "command.txt").write_text(command_string(argv) + "\n", encoding="utf-8") (run_dir / "command.txt").write_text(command_string(argv) + "\n", encoding="utf-8")
code = timed_run(run_dir, argv, env=env) if rp == "rpki-prover":
normalize_one(run_root, profile, run_seq) code, max_rss_kb, max_process_rss_kb, wall_ms, _ = timed_rpki_prover_run(
meta = parse_latest_meta(run_root, profile, run_seq) profile, run_dir, state, container_name
)
meta = write_rpki_prover_run_metadata(
run_root,
profile,
run_seq,
mode,
wall_ms,
max_rss_kb,
max_process_rss_kb,
code,
)
write_rpki_prover_group_report(run_root, profile)
else:
code = timed_run(run_dir, argv, env=env)
normalize_one(run_root, profile, run_seq)
meta = parse_latest_meta(run_root, profile, run_seq)
update_current( update_current(
state="running", state="running",
lastRunSeq=run_seq, lastRunSeq=run_seq,
@@ -571,7 +963,10 @@ def run_profile(profile: str, exp_id: str, runs: int, interval_secs: int, rirs:
if code != 0: if code != 0:
exit_code = code exit_code = code
break break
write_group_report(run_root, interval_secs, completed_seq) if rp == "rpki-prover":
write_rpki_prover_group_report(run_root, profile)
else:
write_group_report(run_root, interval_secs, completed_seq)
final_state = "success" if exit_code == 0 else "failed" final_state = "success" if exit_code == 0 else "failed"
update_current(state=final_state, finishedAtUtc=utc_now(), exitCode=exit_code) update_current(state=final_state, finishedAtUtc=utc_now(), exitCode=exit_code)
run_info = normalize_experiment_metadata(read_json(run_root / "run-info.json", {})) run_info = normalize_experiment_metadata(read_json(run_root / "run-info.json", {}))
@@ -645,6 +1040,10 @@ def cmd_paths(args: argparse.Namespace) -> None:
print(f"runs={runs_dir()}") print(f"runs={runs_dir()}")
print(f"logs={logs_dir()}") print(f"logs={logs_dir()}")
print(f"bundle={cfg.get('bundle','')}") print(f"bundle={cfg.get('bundle','')}")
image = cfg.get("rpkiProverImage", {})
print(f"rpki_prover_image={image.get('tag', '')}")
print(f"rpki_prover_image_id={image.get('imageId', '')}")
print(f"rpki_prover_image_archive={image.get('archive', '')}")
cur = current_live_status() cur = current_live_status()
if cur.get("experimentRoot"): if cur.get("experimentRoot"):
print(f"current_experiment_root={cur['experimentRoot']}") print(f"current_experiment_root={cur['experimentRoot']}")
@@ -668,6 +1067,8 @@ def cmd_start(args: argparse.Namespace) -> None:
raise SystemExit("--runs must be >= 1") raise SystemExit("--runs must be >= 1")
if not (ctl_root() / "versions.json").exists(): if not (ctl_root() / "versions.json").exists():
raise SystemExit("rpctl is not initialized; run ./rpctl init --bundle <bundle-root> first") raise SystemExit("rpctl is not initialized; run ./rpctl init --bundle <bundle-root> first")
if PROFILES[profile] == "rpki-prover":
require_rpki_prover_image()
exp_id = args.exp_id or f"{utc_id()}_{profile}_{args.runs}runs_{interval_secs}s" exp_id = args.exp_id or f"{utc_id()}_{profile}_{args.runs}runs_{interval_secs}s"
runner_args = [ runner_args = [
sys.executable, sys.executable,
@@ -717,6 +1118,8 @@ def cmd_continue(args: argparse.Namespace) -> None:
if entry["state"] != "success": if entry["state"] != "success":
raise SystemExit(f"can only continue a successful experiment: exp_id={args.exp_id} state={entry['state']}") raise SystemExit(f"can only continue a successful experiment: exp_id={args.exp_id} state={entry['state']}")
profile = entry["profile"] profile = entry["profile"]
if PROFILES[profile] == "rpki-prover":
require_rpki_prover_image()
rirs = parse_entry_rirs(entry) rirs = parse_entry_rirs(entry)
interval_secs = parse_interval(args.interval) interval_secs = parse_interval(args.interval)
runner_args = [ runner_args = [
@@ -867,6 +1270,9 @@ def parse_command_summary(command: str) -> str:
keys.append(f"--rsync-scope={tokens[idx + 1]}") keys.append(f"--rsync-scope={tokens[idx + 1]}")
elif token.startswith("--rsync-scope="): elif token.startswith("--rsync-scope="):
keys.append(token) keys.append(token)
if RPKI_PROVER_IMAGE_TAG in token_set:
algorithm = "full-every-iteration" if "--no-incremental-validation" in token_set else "incremental"
keys.append(f"algorithm={algorithm}")
return " ".join(keys) return " ".join(keys)
@@ -945,6 +1351,10 @@ def cache_profile_name(profile: str) -> str:
return "pp+roa+crt" return "pp+roa+crt"
if profile == "ours-no-cache": if profile == "ours-no-cache":
return "off" return "off"
if profile == "rpki-prover-incremental":
return "incremental"
if profile == "rpki-prover-full":
return "full-every-iteration"
return "n/a" return "n/a"
@@ -971,6 +1381,8 @@ def profile_label(profile: str) -> str:
"routinator-latest-release": "routinator", "routinator-latest-release": "routinator",
"rpki-client-latest-release": "rpki-client", "rpki-client-latest-release": "rpki-client",
"fort-latest-release": "fort", "fort-latest-release": "fort",
"rpki-prover-incremental": "prover-inc",
"rpki-prover-full": "prover-full",
} }
return labels.get(profile, profile) return labels.get(profile, profile)
@@ -1043,6 +1455,9 @@ def infer_output_counts(profile: str, run_dir: Path) -> tuple[str, str]:
return vrps, vaps return vrps, vaps
if rp == "fort": if rp == "fort":
return count_csv_data_rows(run_dir / "vrps.csv"), "0" if (run_dir / "vrps.csv").exists() else "" return count_csv_data_rows(run_dir / "vrps.csv"), "0" if (run_dir / "vrps.csv").exists() else ""
if rp == "rpki-prover":
raw, _ = collect_rpki_prover_vrps(run_dir / "vrps.csv")
return str(raw) if (run_dir / "vrps.csv").exists() else "", "n/a" if (run_dir / "vrps.csv").exists() else ""
return "", "" return "", ""
@@ -1355,6 +1770,7 @@ def fallback_run_details(entry: dict[str, Any]) -> list[dict[str, Any]]:
"wall_ms": timing.get("wallMs", ""), "wall_ms": timing.get("wallMs", ""),
"max_rss_kb": timing.get("maxRssKb", ""), "max_rss_kb": timing.get("maxRssKb", ""),
"vrps": first_non_empty(meta.get("counts", {}).get("vrps") if meta else "", inferred_vrps), "vrps": first_non_empty(meta.get("counts", {}).get("vrps") if meta else "", inferred_vrps),
"vrps_unique": first_non_empty(meta.get("counts", {}).get("vrpsUnique") if meta else ""),
"vaps": first_non_empty(meta.get("counts", {}).get("vaps") if meta else "", inferred_vaps), "vaps": first_non_empty(meta.get("counts", {}).get("vaps") if meta else "", inferred_vaps),
"remote_run_dir": str(run_dir), "remote_run_dir": str(run_dir),
}) })
@@ -1391,7 +1807,7 @@ def cmd_view(args: argparse.Namespace) -> None:
print("command=" + entry["command"]) print("command=" + entry["command"])
if rows: if rows:
print("\nper-run:") print("\nper-run:")
headers = ["run", "mode", "exit", "start", "end", "wall_ms", "rss_kb", "vrps", "vaps", "path"] headers = ["run", "mode", "exit", "start", "end", "wall_ms", "rss_kb", "vrps", "vrps_unique", "vaps", "path"]
table_rows = [] table_rows = []
for row in rows: for row in rows:
table_rows.append([ table_rows.append([
@@ -1403,6 +1819,7 @@ def cmd_view(args: argparse.Namespace) -> None:
first_non_empty(row.get("wall_ms"), row.get("wallMs")), first_non_empty(row.get("wall_ms"), row.get("wallMs")),
first_non_empty(row.get("max_rss_kb"), row.get("maxRssKb")), first_non_empty(row.get("max_rss_kb"), row.get("maxRssKb")),
first_non_empty(row.get("vrps"), row.get("VRPs")), first_non_empty(row.get("vrps"), row.get("VRPs")),
first_non_empty(row.get("vrps_unique"), row.get("vrpsUnique")),
first_non_empty(row.get("vaps"), row.get("VAPs")), first_non_empty(row.get("vaps"), row.get("VAPs")),
first_non_empty(row.get("remote_run_dir"), row.get("remoteRunDir")), first_non_empty(row.get("remote_run_dir"), row.get("remoteRunDir")),
]) ])
@@ -1412,7 +1829,7 @@ def cmd_view(args: argparse.Namespace) -> None:
def build_parser() -> argparse.ArgumentParser: def build_parser() -> argparse.ArgumentParser:
formatter = argparse.RawDescriptionHelpFormatter formatter = argparse.RawDescriptionHelpFormatter
parser = argparse.ArgumentParser( parser = argparse.ArgumentParser(
description="Control one local four-RP performance run on this server.", description="Control one local five-RP performance run on this server.",
formatter_class=formatter, formatter_class=formatter,
epilog=""" epilog="""
RP/profile naming: RP/profile naming:
@@ -1435,6 +1852,14 @@ RP/profile naming:
fort fort
Alias of fort-latest-release. Uses archived FORT validator release binary. Alias of fort-latest-release. Uses archived FORT validator release binary.
rpki-prover / rpki-prover-incremental
Uses the archived rpki-prover Docker image with its default incremental
validation algorithm and persistent state.
rpki-prover-full / rpki-prover-no-incremental
Uses the same image with --no-incremental-validation. Its state is kept
separate from the incremental profile.
Cache difference for ours RP: Cache difference for ours RP:
--rp ours adds: --rp ours adds:
--enable-publication-point-validation-cache --enable-publication-point-validation-cache
@@ -1444,10 +1869,12 @@ Cache difference for ours RP:
--rp ours-no-cache does not add those flags. --rp ours-no-cache does not add those flags.
Typical usage: Typical usage:
./rpctl init --bundle /root/rpki_4rp_ctl_runtime_bundle ./rpctl init --bundle /data/rpki_4rp_ctl_runtime_bundle \\
--rpki-prover-image /data/rpki-prover-6857d4bf-linux-amd64.docker.tar.gz
./rpctl list ./rpctl list
./rpctl start --rp ours --interval 10m --runs 6 --rirs all5 ./rpctl start --rp ours --interval 10m --runs 6 --rirs all5
./rpctl start --rp ours-no-cache --interval 10m --runs 6 --rirs all5 ./rpctl start --rp ours-no-cache --interval 10m --runs 6 --rirs all5
./rpctl start --rp rpki-prover --interval 0 --runs 2 --rirs all5
./rpctl status ./rpctl status
./rpctl show ./rpctl show
./rpctl show --failed ./rpctl show --failed
@@ -1458,7 +1885,8 @@ Typical usage:
Data model: Data model:
- Experiment data stays under ./runs/<profile>/<exp-id>/. - Experiment data stays under ./runs/<profile>/<exp-id>/.
- Each experiment starts with a snapshot; later child runs are deltas and reuse the profile state. - Each experiment starts with a snapshot; later child runs reuse the profile state.
- rpki-prover labels reused-state runs as warm because its native storage decides snapshot/delta behavior.
- By default only one RP/profile can run at a time. - By default only one RP/profile can run at a time.
""", """,
) )
@@ -1481,14 +1909,24 @@ The bundle must contain:
scripts/write_group_report.py scripts/write_group_report.py
versions.json versions.json
Optional rpki-prover image archive:
Pass --rpki-prover-image to copy and load the saved linux/amd64 Docker image.
Init verifies image architecture, tag and rpki-prover version without pulling online.
Recommended source on 47.77.237.41: Recommended source on 47.77.237.41:
/root/rpki_4rp_ctl_runtime_bundle /root/rpki_4rp_ctl_runtime_bundle
Example: Example:
./rpctl init --bundle /root/rpki_4rp_ctl_runtime_bundle ./rpctl init --bundle /data/rpki_4rp_ctl_runtime_bundle \\
--rpki-prover-image /data/rpki-prover-6857d4bf-linux-amd64.docker.tar.gz
""", """,
) )
p.add_argument("--bundle", required=True, help="Path to the archived #100 runtime bundle root.") p.add_argument("--bundle", required=True, help="Path to the archived #100 runtime bundle root.")
p.add_argument(
"--rpki-prover-image",
default="",
help="Optional saved rpki-prover linux/amd64 Docker image archive. Existing four-RP profiles do not require it.",
)
p.set_defaults(func=cmd_init) p.set_defaults(func=cmd_init)
p = sub.add_parser( p = sub.add_parser(
@@ -1502,6 +1940,10 @@ Use this command when unsure which value to pass to --rp.
Important ours RP profiles: Important ours RP profiles:
ours-pp-object-cache cache-enabled profile ours-pp-object-cache cache-enabled profile
ours-no-cache no-cache baseline profile ours-no-cache no-cache baseline profile
rpki-prover profiles:
rpki-prover-incremental default incremental validation
rpki-prover-full --no-incremental-validation baseline
""", """,
) )
p.set_defaults(func=cmd_list) p.set_defaults(func=cmd_list)
@@ -1525,6 +1967,8 @@ RP/profile examples:
./rpctl start --rp routinator --interval 15m --runs 2 --rirs apnic ./rpctl start --rp routinator --interval 15m --runs 2 --rirs apnic
./rpctl start --rp rpki-client --interval 0 --runs 1 --rirs apnic,arin ./rpctl start --rp rpki-client --interval 0 --runs 1 --rirs apnic,arin
./rpctl start --rp fort --interval 0 --runs 1 --rirs all5 ./rpctl start --rp fort --interval 0 --runs 1 --rirs all5
./rpctl start --rp rpki-prover --interval 0 --runs 2 --rirs all5
./rpctl start --rp rpki-prover-full --interval 0 --runs 1 --rirs all5
Cache mapping: Cache mapping:
--rp ours enables PP cache + ROA cache + child certificate cache. --rp ours enables PP cache + ROA cache + child certificate cache.
@@ -1537,7 +1981,7 @@ Safety:
p.add_argument( p.add_argument(
"--rp", "--rp",
required=True, required=True,
help="RP/profile or alias. Common values: ours, ours-no-cache, routinator, rpki-client, fort.", help="RP/profile or alias. Common values: ours, ours-no-cache, routinator, rpki-client, fort, rpki-prover, rpki-prover-full.",
) )
p.add_argument( p.add_argument(
"--interval", "--interval",
@@ -1578,6 +2022,7 @@ Run semantics:
Examples: Examples:
./rpctl continue --exp-id 20260709T081122Z_ours-pp-object-cache_2runs_600s --runs 3 --interval 10m ./rpctl continue --exp-id 20260709T081122Z_ours-pp-object-cache_2runs_600s --runs 3 --interval 10m
./rpctl continue --rp routinator --exp-id m18_237_routinator_all5_snapshot_delta_20260710T002726Z --runs 1 --interval 0 ./rpctl continue --rp routinator --exp-id m18_237_routinator_all5_snapshot_delta_20260710T002726Z --runs 1 --interval 0
./rpctl continue --rp rpki-prover --exp-id prover_all5 --runs 1 --interval 0
""", """,
) )
p.add_argument("--exp-id", required=True, help="Existing successful experiment id to append to.") p.add_argument("--exp-id", required=True, help="Existing successful experiment id to append to.")
@@ -1692,7 +2137,7 @@ Examples:
./rpctl show --json ./rpctl show --json
""", """,
) )
p.add_argument("--rp", default="", help="Optional RP/profile filter, e.g. ours, ours-no-cache, routinator, rpki-client, fort.") p.add_argument("--rp", default="", help="Optional RP/profile filter, e.g. ours, ours-no-cache, routinator, rpki-client, fort, rpki-prover.")
p.add_argument("--failed", action="store_true", help="Only show failed, partial, incomplete or exited experiments.") p.add_argument("--failed", action="store_true", help="Only show failed, partial, incomplete or exited experiments.")
p.add_argument("--limit", type=int, default=50, help="Maximum rows to show. 0 means no limit. Default: 50.") p.add_argument("--limit", type=int, default=50, help="Maximum rows to show. 0 means no limit. Default: 50.")
p.add_argument("--paths", action="store_true", help="Also show full experiment root paths. Hidden by default to keep the table readable.") p.add_argument("--paths", action="store_true", help="Also show full experiment root paths. Hidden by default to keep the table readable.")

113
scripts/rpctl/test_rpctl.py Normal file
View File

@@ -0,0 +1,113 @@
from __future__ import annotations
import importlib.machinery
import importlib.util
import json
import tempfile
import unittest
from pathlib import Path
def load_rpctl():
path = Path(__file__).with_name("rpctl")
loader = importlib.machinery.SourceFileLoader("rpctl_module", str(path))
spec = importlib.util.spec_from_loader(loader.name, loader)
module = importlib.util.module_from_spec(spec)
loader.exec_module(module)
return module
class RpkiProverProfileTests(unittest.TestCase):
@classmethod
def setUpClass(cls) -> None:
cls.rpctl = load_rpctl()
def test_profile_aliases_and_algorithms(self) -> None:
self.assertEqual(self.rpctl.normalize_profile("rpki-prover"), "rpki-prover-incremental")
self.assertEqual(self.rpctl.normalize_profile("rpki-prover-no-incremental"), "rpki-prover-full")
self.assertNotIn("--no-incremental-validation", self.rpctl.rpki_prover_arguments("rpki-prover-incremental"))
self.assertIn("--no-incremental-validation", self.rpctl.rpki_prover_arguments("rpki-prover-full"))
incremental = self.rpctl.parse_command_summary(
f"docker run {self.rpctl.RPKI_PROVER_IMAGE_TAG} --once"
)
full = self.rpctl.parse_command_summary(
f"docker run {self.rpctl.RPKI_PROVER_IMAGE_TAG} --once --no-incremental-validation"
)
self.assertEqual(incremental, "algorithm=incremental")
self.assertEqual(full, "algorithm=full-every-iteration")
def test_vrp_count_preserves_raw_and_unique(self) -> None:
with tempfile.TemporaryDirectory() as temp:
csv_path = Path(temp) / "vrps.csv"
csv_path.write_text(
"ASN,IP Prefix,Max Length,Trust Anchor\n"
"AS64496,192.0.2.0/24,24,TA-A\n"
"AS64496,192.0.2.0/24,24,TA-B\n"
"AS64497,2001:db8::/32,48,TA-A\n",
encoding="utf-8",
)
raw, unique = self.rpctl.collect_rpki_prover_vrps(csv_path)
self.assertEqual(raw, 3)
self.assertEqual(len(unique), 2)
def test_run_metadata_marks_vaps_not_applicable(self) -> None:
with tempfile.TemporaryDirectory() as temp:
root = Path(temp)
old_ctl_root = self.rpctl.ctl_root
self.rpctl.ctl_root = lambda: root
try:
self.rpctl.save_config({"rpkiProverImage": {"tag": self.rpctl.RPKI_PROVER_IMAGE_TAG}})
run_root = root / "runs" / "rpki-prover-incremental" / "test"
run_dir = run_root / "profiles" / "rpki-prover-incremental" / "runs" / "run_0001"
run_dir.mkdir(parents=True)
(run_dir / "start-utc.txt").write_text("2026-07-17T00:00:00Z\n", encoding="utf-8")
(run_dir / "end-utc.txt").write_text("2026-07-17T00:01:00Z\n", encoding="utf-8")
(run_dir / "vrps.csv").write_text(
"ASN,IP Prefix,Max Length,Trust Anchor\nAS64496,192.0.2.0/24,24,TA-A\n",
encoding="utf-8",
)
meta = self.rpctl.write_rpki_prover_run_metadata(
run_root, "rpki-prover-incremental", 1, "snapshot", 60_000, 1234, 1200, 0
)
self.assertEqual(meta["counts"]["vrpsRaw"], 1)
self.assertEqual(meta["counts"]["vrpsUnique"], 1)
self.assertEqual(meta["counts"]["vaps"], "n/a")
self.assertFalse(meta["vapsAvailable"])
self.rpctl.write_rpki_prover_group_report(run_root, "rpki-prover-incremental")
report = (run_root / "reports" / "per_run_metrics.csv").read_text(encoding="utf-8")
self.assertIn("vrps_unique", report)
self.assertIn("n/a", report)
finally:
self.rpctl.ctl_root = old_ctl_root
def test_missing_image_configuration_fails_before_docker(self) -> None:
with tempfile.TemporaryDirectory() as temp:
root = Path(temp)
old_ctl_root = self.rpctl.ctl_root
self.rpctl.ctl_root = lambda: root
try:
self.rpctl.save_config({})
with self.assertRaisesRegex(SystemExit, "not initialized"):
self.rpctl.require_rpki_prover_image()
finally:
self.rpctl.ctl_root = old_ctl_root
def test_versions_include_prover_profiles(self) -> None:
with tempfile.TemporaryDirectory() as temp:
root = Path(temp)
old_ctl_root = self.rpctl.ctl_root
self.rpctl.ctl_root = lambda: root
try:
(root / "versions.json").write_text(json.dumps({"profiles": ["fort-latest-release"]}), encoding="utf-8")
self.rpctl.save_config({"rpkiProverImage": {"tag": self.rpctl.RPKI_PROVER_IMAGE_TAG}})
run_root = root / "experiment"
self.rpctl.write_experiment_versions(run_root)
versions = json.loads((run_root / "versions.json").read_text(encoding="utf-8"))
self.assertIn("rpki-prover-incremental", versions["profiles"])
self.assertIn("rpki-prover-full", versions["profiles"])
finally:
self.rpctl.ctl_root = old_ctl_root
if __name__ == "__main__":
unittest.main()

View File

@@ -14,7 +14,7 @@ Usage:
scripts/soak/build_portable_soak_package.sh [--out-dir <path>] [--profile <profile>] scripts/soak/build_portable_soak_package.sh [--out-dir <path>] [--profile <profile>]
Requires release binaries to already exist. Build them first, for example: Requires release binaries to already exist. Build them first, for example:
cargo build --release --bin rpki --bin rpki_daemon --bin db_stats --bin rpki_artifact_metrics --bin rpki_query_service --bin rpki_query_indexer cargo build --release --bin rpki --bin rpki_daemon --bin db_stats --bin rpki_artifact_metrics --bin rpki_query_service --bin rpki_query_indexer --bin verification_ccr_compare
USAGE USAGE
} }
@@ -53,7 +53,7 @@ else
TARGET_BIN_DIR="$REPO_ROOT/target/$PROFILE" TARGET_BIN_DIR="$REPO_ROOT/target/$PROFILE"
fi fi
REQUIRED_BINS=(rpki rpki_daemon db_stats rpki_artifact_metrics rpki_query_service rpki_query_indexer) REQUIRED_BINS=(rpki rpki_daemon db_stats rpki_artifact_metrics rpki_query_service rpki_query_indexer verification_ccr_compare)
OPTIONAL_BINS=( OPTIONAL_BINS=(
ccr_dump ccr_dump
ccr_state_compare ccr_state_compare
@@ -82,7 +82,7 @@ STAGE_DIR="$BUILD_ROOT/$PACKAGE_DIR_NAME"
ARCHIVE_PATH="$OUT_DIR/$PACKAGE_NAME.tar.gz" ARCHIVE_PATH="$OUT_DIR/$PACKAGE_NAME.tar.gz"
rm -rf "$STAGE_DIR" "$ARCHIVE_PATH" rm -rf "$STAGE_DIR" "$ARCHIVE_PATH"
mkdir -p "$STAGE_DIR/bin" "$STAGE_DIR/fixtures" "$STAGE_DIR/scripts" "$STAGE_DIR/scripts/soak" \ mkdir -p "$STAGE_DIR/bin" "$STAGE_DIR/fixtures" "$STAGE_DIR/scripts" "$STAGE_DIR/scripts/soak" "$STAGE_DIR/scripts/verification" \
"$STAGE_DIR/runs" "$STAGE_DIR/state" "$STAGE_DIR/logs" "$STAGE_DIR/tmp" "$STAGE_DIR/runs" "$STAGE_DIR/state" "$STAGE_DIR/logs" "$STAGE_DIR/tmp"
install -m 0755 "$SCRIPT_DIR/run_soak.sh" "$STAGE_DIR/run_soak.sh" install -m 0755 "$SCRIPT_DIR/run_soak.sh" "$STAGE_DIR/run_soak.sh"
@@ -92,6 +92,8 @@ install -m 0755 "$SCRIPT_DIR/fixed_phase_loop.sh" "$STAGE_DIR/scripts/soak/fixed
install -m 0755 "$SCRIPT_DIR/hourly_soak_report.py" "$STAGE_DIR/scripts/soak/hourly_soak_report.py" install -m 0755 "$SCRIPT_DIR/hourly_soak_report.py" "$STAGE_DIR/scripts/soak/hourly_soak_report.py"
install -m 0755 "$SCRIPT_DIR/publish_remote231.sh" "$STAGE_DIR/scripts/soak/publish_remote231.sh" install -m 0755 "$SCRIPT_DIR/publish_remote231.sh" "$STAGE_DIR/scripts/soak/publish_remote231.sh"
install -m 0755 "$SCRIPT_DIR/publish_remote231_full.sh" "$STAGE_DIR/scripts/soak/publish_remote231_full.sh" install -m 0755 "$SCRIPT_DIR/publish_remote231_full.sh" "$STAGE_DIR/scripts/soak/publish_remote231_full.sh"
install -m 0755 "$REPO_ROOT/scripts/verification/run_verification_only.sh" "$STAGE_DIR/scripts/verification/run_verification_only.sh"
install -m 0644 "$REPO_ROOT/scripts/verification/README.md" "$STAGE_DIR/scripts/verification/README.md"
install -m 0644 "$SCRIPT_DIR/portable-soak.env.example" "$STAGE_DIR/.env" install -m 0644 "$SCRIPT_DIR/portable-soak.env.example" "$STAGE_DIR/.env"
install -m 0644 "$SCRIPT_DIR/portable-soak.env.example" "$STAGE_DIR/portable-soak.env.example" install -m 0644 "$SCRIPT_DIR/portable-soak.env.example" "$STAGE_DIR/portable-soak.env.example"

View File

@@ -48,6 +48,7 @@ TMP_DIR="${TMP_DIR:-$RUN_ROOT/tmp}"
RSYNC_MIRROR_ROOT="${RSYNC_MIRROR_ROOT:-$STATE_ROOT/rsync-mirror}" RSYNC_MIRROR_ROOT="${RSYNC_MIRROR_ROOT:-$STATE_ROOT/rsync-mirror}"
INVALID_ROOT="$STATE_ROOT/invalid" INVALID_ROOT="$STATE_ROOT/invalid"
RESET_STAGING_ROOT="$STATE_ROOT/reset-staging" RESET_STAGING_ROOT="$STATE_ROOT/reset-staging"
STATE_EXECUTION_LOCK_PATH="$STATE_ROOT/locks/state-execution.lock"
LIVE_TA_REFRESH_DIR="${LIVE_TA_REFRESH_DIR:-$META_DIR/live-ta-refresh}" LIVE_TA_REFRESH_DIR="${LIVE_TA_REFRESH_DIR:-$META_DIR/live-ta-refresh}"
LIVE_TA_REFRESH_CONNECT_TIMEOUT_SECS="${LIVE_TA_REFRESH_CONNECT_TIMEOUT_SECS:-15}" LIVE_TA_REFRESH_CONNECT_TIMEOUT_SECS="${LIVE_TA_REFRESH_CONNECT_TIMEOUT_SECS:-15}"
LIVE_TA_REFRESH_MAX_TIME_SECS="${LIVE_TA_REFRESH_MAX_TIME_SECS:-120}" LIVE_TA_REFRESH_MAX_TIME_SECS="${LIVE_TA_REFRESH_MAX_TIME_SECS:-120}"
@@ -1080,6 +1081,7 @@ build_child_args() {
CHILD_ARGS+=( CHILD_ARGS+=(
--report-json "{run_out}/report.json" --report-json "{run_out}/report.json"
--validation-contract-out "{run_out}/validation-contract.json"
) )
if is_true "$OUTPUT_COMPACT_REPORT"; then if is_true "$OUTPUT_COMPACT_REPORT"; then
CHILD_ARGS+=(--report-json-compact) CHILD_ARGS+=(--report-json-compact)
@@ -1109,6 +1111,59 @@ build_child_args() {
fi fi
} }
write_current_state_binding() {
local run_dir="$1"
local run_id="$2"
local output_path="$META_DIR/current-state-binding.json"
python3 - "$run_dir" "$run_id" "$output_path" <<'PY'
import hashlib
import json
import os
import sys
import tempfile
from datetime import datetime, timezone
from pathlib import Path
run_dir = Path(sys.argv[1])
run_id = sys.argv[2]
output_path = Path(sys.argv[3])
summary = json.loads((run_dir / "run-summary.json").read_text(encoding="utf-8"))
if summary.get("status") != "success":
raise SystemExit(f"cannot bind unsuccessful source run: {run_id}")
def digest(name):
path = run_dir / name
if not path.is_file():
raise SystemExit(f"current state binding input is missing: {path}")
hasher = hashlib.sha256()
with path.open("rb") as handle:
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
hasher.update(chunk)
return hasher.hexdigest()
binding = {
"schemaVersion": 1,
"currentRunId": run_id,
"sourceCirSha256": digest("input.cir"),
"sourceCcrSha256": digest("result.ccr"),
"validationContractSha256": digest("validation-contract.json"),
"updatedAtRfc3339Utc": datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"),
}
output_path.parent.mkdir(parents=True, exist_ok=True)
fd, tmp_name = tempfile.mkstemp(prefix=output_path.name + ".", dir=output_path.parent)
try:
with os.fdopen(fd, "w", encoding="utf-8") as handle:
json.dump(binding, handle, indent=2, sort_keys=True)
handle.write("\n")
handle.flush()
os.fsync(handle.fileno())
os.replace(tmp_name, output_path)
finally:
if os.path.exists(tmp_name):
os.unlink(tmp_name)
PY
}
copy_inner_run_outputs() { copy_inner_run_outputs() {
local daemon_state_root="$1" local daemon_state_root="$1"
local run_dir="$2" local run_dir="$2"
@@ -1304,6 +1359,12 @@ run_one_round() {
final_status="failed" final_status="failed"
fi fi
fi fi
if [[ "$final_status" == "success" ]]; then
if ! write_current_state_binding "$run_dir" "$run_id"; then
warn "failed to bind successful run to current state: $run_id"
final_status="failed"
fi
fi
write_run_meta "$run_dir/run-meta.json" "$final_status" "$run_index" "$run_id" "$sync_mode" \ write_run_meta "$run_dir/run-meta.json" "$final_status" "$run_index" "$run_id" "$sync_mode" \
"$snapshot_reason" "$previous_run_id" "$previous_success_value" "$started_at" "$completed_at" \ "$snapshot_reason" "$previous_run_id" "$previous_success_value" "$started_at" "$completed_at" \
"$INVALID_DB_PATH" "$INVALID_STATE_PATH" "$INVALID_TMP_PATH" "$daemon_exit_code" "$PACKAGE_ROOT" "$ENV_FILE" \ "$INVALID_DB_PATH" "$INVALID_STATE_PATH" "$INVALID_TMP_PATH" "$daemon_exit_code" "$PACKAGE_ROOT" "$ENV_FILE" \
@@ -1327,6 +1388,7 @@ main() {
require_command python3 require_command python3
require_command date require_command date
require_command find require_command find
require_command flock
if [[ "$TAL_INPUT_MODE" == "file-live-ta" ]]; then if [[ "$TAL_INPUT_MODE" == "file-live-ta" ]]; then
require_command curl require_command curl
validate_positive_int "LIVE_TA_REFRESH_CONNECT_TIMEOUT_SECS" "$LIVE_TA_REFRESH_CONNECT_TIMEOUT_SECS" validate_positive_int "LIVE_TA_REFRESH_CONNECT_TIMEOUT_SECS" "$LIVE_TA_REFRESH_CONNECT_TIMEOUT_SECS"
@@ -1360,7 +1422,7 @@ main() {
fi fi
done done
mkdir -p "$RUNS_ROOT" "$LOG_ROOT" "$DB_DIR" "$META_DIR" "$TMP_DIR" "$INVALID_ROOT" "$RESET_STAGING_ROOT" "$LIVE_TA_REFRESH_DIR" mkdir -p "$RUNS_ROOT" "$LOG_ROOT" "$DB_DIR" "$META_DIR" "$TMP_DIR" "$INVALID_ROOT" "$RESET_STAGING_ROOT" "$LIVE_TA_REFRESH_DIR" "$(dirname "$STATE_EXECUTION_LOCK_PATH")"
if is_true "$ALLOW_RSYNC_MIRROR_REUSE"; then if is_true "$ALLOW_RSYNC_MIRROR_REUSE"; then
mkdir -p "$RSYNC_MIRROR_ROOT" mkdir -p "$RSYNC_MIRROR_ROOT"
fi fi
@@ -1386,6 +1448,9 @@ main() {
local any_failed=0 local any_failed=0
while (( run_forever == 1 || next_index <= stop_index )); do while (( run_forever == 1 || next_index <= stop_index )); do
local state_lock_fd
exec {state_lock_fd}>"$STATE_EXECUTION_LOCK_PATH"
flock "$state_lock_fd"
INVALID_DB_PATH="" INVALID_DB_PATH=""
INVALID_STATE_PATH="" INVALID_STATE_PATH=""
INVALID_TMP_PATH="" INVALID_TMP_PATH=""
@@ -1472,6 +1537,8 @@ main() {
echo "completed run $(printf 'run_%04d' "$next_index") status=failed" >&2 echo "completed run $(printf 'run_%04d' "$next_index") status=failed" >&2
any_failed=1 any_failed=1
fi fi
flock -u "$state_lock_fd"
exec {state_lock_fd}>&-
if (( STOP_AFTER_SECS > 0 )); then if (( STOP_AFTER_SECS > 0 )); then
elapsed_secs=$(( $(date +%s) - started_epoch )) elapsed_secs=$(( $(date +%s) - started_epoch ))
if (( elapsed_secs >= STOP_AFTER_SECS )); then if (( elapsed_secs >= STOP_AFTER_SECS )); then

View File

@@ -0,0 +1,50 @@
# verification-only 使用说明
`verification-only` 用于复核启用验证缓存的最新成功 normal run。它冻结当前仓库视图,关闭 PP、ROA、child certificate cache、prefetch、VCIR 持久化和网络访问,在源 CIR 的同一验证时间重新执行完整验证,并比较 CCR 的 MFT、VRP、VAP、TA、router key 五类 state digest。
## 前提
- 源 run 必须是当前 state DB 对应的最新成功 normal run。
- 源 run 必须包含 `input.cir`、`result.ccr`、`run-summary.json` 和 `validation-contract.json`。
- state root 必须包含 `db/work-db`、`db/repo-bytes.db` 和 `meta/current-state-binding.json`。
- 源 normal run 必须启用至少一种验证缓存;仅启用 transport prefetch 不算验证缓存。
- verification-only 与 normal soak 共用 `state/locks/state-execution.lock`,不能绕过 wrapper 并发执行。
## 执行
在 portable package 根目录执行:
```bash
./scripts/verification/run_verification_only.sh \
--source-run-dir /data/ours-rp/runs/run_0123 \
--source-state-root /data/ours-rp/state \
--out-dir /data/ours-rp/verification-runs/run_0123
```
验证时间自动从源 `input.cir` 读取,不接受命令行覆盖。成功后 scratch work-db 默认删除;排障时可加 `--keep-scratch`。
## 输出
- `input.cir`、`result.ccr`、`report.json`
- `vrps.csv`、`vaps.csv`
- `stage-timing.json`、`validation-events.jsonl`
- `validation-contract.json`、`verification-meta.json`
- `compare/ccr-state-compare.json`、`compare/ccr-state-compare.md`
- `stdout.log`、`stderr.log`、`process-time.txt`、`wrapper-timing.json`
通过条件是 compare JSON 中 `stateDigestMatch=true`。`producedAt` 不参与比较。
## 隔离保证
- 不分配 normal run id,不更新 lifecycle、periodic snapshot 计数和 retention。
- 源 work-db 只用于创建 checkpoint,后续写入只发生在独立 scratch work-db。
- 源 repo-bytes.db 以只读模式打开。
- 输出目录必须不存在,防止覆盖旧验证证据。
## 常见失败
- `current state binding`:所选 run 已不是当前 state 对应的最新成功 run,需选择 binding 指向的 run。
- `differs from current state binding`:源 CIR、CCR 或 contract 已变化,不能继续复核。
- `blob is missing`:repository view 引用的原始 bytes 不完整,验证不会启动。
- `did not enable a validation cache`:源 run 不是缓存正确性对比的有效基准。
- `CCR state digest mismatch`:查看 `compare/` 下 JSON/Markdown,按不匹配状态集合继续排查。

View File

@@ -0,0 +1,102 @@
#!/usr/bin/env bash
set -euo pipefail
usage() {
cat <<'USAGE'
Usage:
run_verification_only.sh \
--source-run-dir <run_dir> \
--source-state-root <state_dir> \
--out-dir <verification_dir> \
[--rpki-bin <path>] [--keep-scratch]
validation time is always read from the source CIR. The wrapper serializes access
with normal soak runs and never writes into the normal run sequence.
USAGE
}
SOURCE_RUN_DIR=""
SOURCE_STATE_ROOT=""
OUT_DIR=""
RPKI_BIN=""
KEEP_SCRATCH=0
while [[ $# -gt 0 ]]; do
case "$1" in
--source-run-dir) SOURCE_RUN_DIR="${2:?missing value}"; shift 2 ;;
--source-state-root) SOURCE_STATE_ROOT="${2:?missing value}"; shift 2 ;;
--out-dir) OUT_DIR="${2:?missing value}"; shift 2 ;;
--rpki-bin) RPKI_BIN="${2:?missing value}"; shift 2 ;;
--keep-scratch) KEEP_SCRATCH=1; shift ;;
-h|--help) usage; exit 0 ;;
*) echo "unknown argument: $1" >&2; usage >&2; exit 2 ;;
esac
done
[[ -n "$SOURCE_RUN_DIR" ]] || { echo "--source-run-dir is required" >&2; exit 2; }
[[ -n "$SOURCE_STATE_ROOT" ]] || { echo "--source-state-root is required" >&2; exit 2; }
[[ -n "$OUT_DIR" ]] || { echo "--out-dir is required" >&2; exit 2; }
if [[ -z "$RPKI_BIN" ]]; then
RPKI_BIN="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)/target/release/rpki"
fi
[[ -x "$RPKI_BIN" ]] || { echo "rpki binary is not executable: $RPKI_BIN" >&2; exit 2; }
command -v flock >/dev/null || { echo "flock is required" >&2; exit 2; }
command -v python3 >/dev/null || { echo "python3 is required" >&2; exit 2; }
now_ms() {
python3 -c 'import time; print(time.time_ns() // 1_000_000)'
}
mkdir -p "$SOURCE_STATE_ROOT/locks" "$(dirname "$OUT_DIR")"
[[ ! -e "$OUT_DIR" ]] || { echo "verification output already exists: $OUT_DIR" >&2; exit 2; }
LOCK_PATH="$SOURCE_STATE_ROOT/locks/state-execution.lock"
STDOUT_TMP="${OUT_DIR}.stdout.tmp"
STDERR_TMP="${OUT_DIR}.stderr.tmp"
PROCESS_TIME_TMP="${OUT_DIR}.process-time.tmp"
WAIT_STARTED_MS="$(now_ms)"
exec {LOCK_FD}>"$LOCK_PATH"
flock "$LOCK_FD"
LOCK_ACQUIRED_MS="$(now_ms)"
ARGS=(
--verification-only
--verification-source-run "$SOURCE_RUN_DIR"
--verification-source-state "$SOURCE_STATE_ROOT"
--verification-out "$OUT_DIR"
)
if [[ "$KEEP_SCRATCH" -eq 1 ]]; then
ARGS+=(--keep-verification-scratch)
fi
set +e
if [[ -x /usr/bin/time ]]; then
RPKI_STATE_EXECUTION_LOCK_HELD=1 /usr/bin/time -v -o "$PROCESS_TIME_TMP" -- \
"$RPKI_BIN" "${ARGS[@]}" >"$STDOUT_TMP" 2>"$STDERR_TMP"
else
RPKI_STATE_EXECUTION_LOCK_HELD=1 "$RPKI_BIN" "${ARGS[@]}" >"$STDOUT_TMP" 2>"$STDERR_TMP"
fi
EXIT_CODE=$?
set -e
LOCK_RELEASED_MS="$(now_ms)"
flock -u "$LOCK_FD"
exec {LOCK_FD}>&-
mkdir -p "$OUT_DIR"
mv "$STDOUT_TMP" "$OUT_DIR/stdout.log"
mv "$STDERR_TMP" "$OUT_DIR/stderr.log"
if [[ -f "$PROCESS_TIME_TMP" ]]; then
mv "$PROCESS_TIME_TMP" "$OUT_DIR/process-time.txt"
fi
python3 - "$OUT_DIR/wrapper-timing.json" "$WAIT_STARTED_MS" "$LOCK_ACQUIRED_MS" "$LOCK_RELEASED_MS" "$EXIT_CODE" <<'PY'
import json
import sys
from pathlib import Path
path = Path(sys.argv[1])
wait_started_ms, lock_acquired_ms, lock_released_ms, exit_code = map(int, sys.argv[2:])
path.write_text(json.dumps({
"lockWaitMs": lock_acquired_ms - wait_started_ms,
"lockHeldMs": lock_released_ms - lock_acquired_ms,
"exitCode": exit_code,
}, indent=2, sort_keys=True) + "\n", encoding="utf-8")
PY
exit "$EXIT_CODE"

View File

@@ -0,0 +1,171 @@
use std::path::PathBuf;
fn usage() -> &'static str {
"Usage: verification_ccr_compare --source-ccr <path> --verification-ccr <path> --out-json <path> [--out-md <path>]"
}
#[derive(Debug, PartialEq, Eq)]
struct Args {
source_ccr: PathBuf,
verification_ccr: PathBuf,
out_json: PathBuf,
out_md: Option<PathBuf>,
}
fn parse_args(argv: &[String]) -> Result<Args, String> {
let mut source_ccr = None;
let mut verification_ccr = None;
let mut out_json = None;
let mut out_md = None;
let mut index = 1usize;
while index < argv.len() {
let target = match argv[index].as_str() {
"--source-ccr" => &mut source_ccr,
"--verification-ccr" => &mut verification_ccr,
"--out-json" => &mut out_json,
"--out-md" => &mut out_md,
"-h" | "--help" => return Err(usage().to_string()),
other => return Err(format!("unknown argument: {other}\n{}", usage())),
};
index += 1;
*target =
Some(PathBuf::from(argv.get(index).ok_or_else(|| {
format!("{} requires a value", argv[index - 1])
})?));
index += 1;
}
let source_ccr = source_ccr.ok_or_else(|| format!("--source-ccr is required\n{}", usage()))?;
let verification_ccr =
verification_ccr.ok_or_else(|| format!("--verification-ccr is required\n{}", usage()))?;
let out_json = out_json.ok_or_else(|| format!("--out-json is required\n{}", usage()))?;
Ok(Args {
source_ccr,
verification_ccr,
out_json,
out_md,
})
}
fn run(args: &Args) -> Result<(), String> {
let comparison =
rpki::verification_only::compare_ccr_files(&args.source_ccr, &args.verification_ccr)?;
rpki::verification_only::write_ccr_comparison(
&args.out_json,
args.out_md.as_deref(),
&comparison,
)?;
if !comparison.state_digest_match {
return Err(format!(
"CCR state digest mismatch: {}",
comparison.mismatched_states.join(",")
));
}
println!("CCR state digests match: {}", args.out_json.display());
Ok(())
}
fn main() -> Result<(), String> {
let argv = std::env::args().collect::<Vec<_>>();
run(&parse_args(&argv)?)
}
#[cfg(test)]
mod tests {
use super::*;
use rpki::ccr::{
CcrContentInfo, CcrDigestAlgorithm, RpkiCanonicalCacheRepresentation,
build_roa_payload_state, encode_content_info,
};
use rpki::data_model::roa::{IpPrefix, RoaAfi};
use rpki::validation::objects::Vrp;
use std::fs;
fn argv(values: &[&str]) -> Vec<String> {
std::iter::once("verification_ccr_compare")
.chain(values.iter().copied())
.map(str::to_string)
.collect()
}
fn sample_ccr(asn: u32) -> Vec<u8> {
let vrps = build_roa_payload_state(&[Vrp {
asn,
prefix: IpPrefix {
afi: RoaAfi::Ipv4,
prefix_len: 24,
addr: [192, 0, 2, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
},
max_length: 24,
}])
.expect("build VRP state");
encode_content_info(&CcrContentInfo::new(RpkiCanonicalCacheRepresentation {
version: 0,
hash_alg: CcrDigestAlgorithm::Sha256,
produced_at: time::OffsetDateTime::UNIX_EPOCH,
mfts: None,
vrps: Some(vrps),
vaps: None,
tas: None,
rks: None,
}))
.expect("encode CCR")
}
#[test]
fn parse_args_accepts_complete_input_and_rejects_bad_forms() {
let parsed = parse_args(&argv(&[
"--source-ccr",
"source.ccr",
"--verification-ccr",
"verification.ccr",
"--out-json",
"result.json",
"--out-md",
"result.md",
]))
.expect("parse");
assert_eq!(parsed.source_ccr, PathBuf::from("source.ccr"));
assert_eq!(parsed.out_md, Some(PathBuf::from("result.md")));
assert!(parse_args(&argv(&[])).unwrap_err().contains("source-ccr"));
assert!(
parse_args(&argv(&["--source-ccr"]))
.unwrap_err()
.contains("requires a value")
);
assert!(
parse_args(&argv(&["--unknown"]))
.unwrap_err()
.contains("unknown argument")
);
assert!(parse_args(&argv(&["--help"])).is_err());
}
#[test]
fn run_writes_matching_outputs_and_rejects_mismatch() {
let temp = tempfile::tempdir().expect("tempdir");
let source = temp.path().join("source.ccr");
let verification = temp.path().join("verification.ccr");
let out_json = temp.path().join("out/result.json");
let out_md = temp.path().join("out/result.md");
fs::write(&source, sample_ccr(64496)).expect("source");
fs::write(&verification, sample_ccr(64496)).expect("verification");
let args = Args {
source_ccr: source.clone(),
verification_ccr: verification.clone(),
out_json: out_json.clone(),
out_md: Some(out_md.clone()),
};
run(&args).expect("matching run");
assert!(out_json.is_file());
assert!(out_md.is_file());
fs::write(&verification, sample_ccr(64497)).expect("mismatch");
let error = run(&args).unwrap_err();
assert!(error.contains("vrps"));
assert!(
fs::read_to_string(out_json)
.expect("comparison JSON")
.contains("mismatchedStates")
);
}
}

View File

@@ -77,6 +77,7 @@ pub struct ExternalRawStoreDb {
pub struct ExternalRepoBytesDb { pub struct ExternalRepoBytesDb {
path: PathBuf, path: PathBuf,
db: Arc<DB>, db: Arc<DB>,
read_only: bool,
} }
impl ExternalRawStoreDb { impl ExternalRawStoreDb {
@@ -189,6 +190,7 @@ impl ExternalRepoBytesDb {
Ok(Self { Ok(Self {
path, path,
db: Arc::new(db), db: Arc::new(db),
read_only: false,
}) })
} }
@@ -201,6 +203,7 @@ impl ExternalRepoBytesDb {
Ok(Self { Ok(Self {
path, path,
db: Arc::new(db), db: Arc::new(db),
read_only: true,
}) })
} }
@@ -208,6 +211,12 @@ impl ExternalRepoBytesDb {
if blobs.is_empty() { if blobs.is_empty() {
return Ok(()); return Ok(());
} }
if self.read_only {
return Err(StorageError::RocksDb(format!(
"repo-bytes DB is read-only: {}",
self.path.display()
)));
}
let mut batch = WriteBatch::default(); let mut batch = WriteBatch::default();
for (sha256_hex, bytes) in blobs { for (sha256_hex, bytes) in blobs {
validate_blob_sha256_hex(sha256_hex)?; validate_blob_sha256_hex(sha256_hex)?;
@@ -221,6 +230,42 @@ impl ExternalRepoBytesDb {
Ok(()) Ok(())
} }
pub(crate) fn is_read_only(&self) -> bool {
self.read_only
}
pub(crate) fn require_existing_blob_bytes_batch(
&self,
blobs: &[(String, Vec<u8>)],
) -> StorageResult<()> {
if blobs.is_empty() {
return Ok(());
}
let hashes = blobs
.iter()
.map(|(sha256_hex, _)| sha256_hex.clone())
.collect::<Vec<_>>();
let existing = self.get_blob_bytes_batch(&hashes)?;
for ((sha256_hex, expected_bytes), actual_bytes) in blobs.iter().zip(existing) {
match actual_bytes {
Some(actual_bytes) if actual_bytes == *expected_bytes => {}
Some(_) => {
return Err(StorageError::InvalidData {
entity: "read_only_repo_bytes",
detail: format!("existing bytes differ for SHA-256 {sha256_hex}"),
});
}
None => {
return Err(StorageError::InvalidData {
entity: "read_only_repo_bytes",
detail: format!("blob is missing for SHA-256 {sha256_hex}"),
});
}
}
}
Ok(())
}
pub fn get_blob_bytes(&self, sha256_hex: &str) -> StorageResult<Option<Vec<u8>>> { pub fn get_blob_bytes(&self, sha256_hex: &str) -> StorageResult<Option<Vec<u8>>> {
validate_blob_sha256_hex(sha256_hex)?; validate_blob_sha256_hex(sha256_hex)?;
let key = repo_bytes_key(sha256_hex); let key = repo_bytes_key(sha256_hex);

View File

@@ -38,7 +38,7 @@ use crate::validation::run_tree_from_tal::{
run_tree_from_tal_url_parallel_phase2_audit, run_tree_from_tal_url_parallel_phase2_audit,
run_tree_from_tal_url_parallel_phase2_audit_with_timing, run_tree_from_tal_url_parallel_phase2_audit_with_timing,
}; };
use crate::validation::tree::TreeRunConfig; use crate::validation::tree::{DEFAULT_MAX_CA_DEPTH, TreeRunConfig};
#[cfg(test)] #[cfg(test)]
use output::write_json; use output::write_json;
use output::{ use output::{
@@ -115,6 +115,12 @@ fn vcir_storage_summary_enabled() -> bool {
#[derive(Clone, Debug, PartialEq, Eq)] #[derive(Clone, Debug, PartialEq, Eq)]
pub struct CliArgs { pub struct CliArgs {
pub verification_only: bool,
pub verification_source_run_dir: Option<PathBuf>,
pub verification_source_state_root: Option<PathBuf>,
pub verification_out_dir: Option<PathBuf>,
pub keep_verification_scratch: bool,
pub validation_contract_out_path: Option<PathBuf>,
pub tal_urls: Vec<String>, pub tal_urls: Vec<String>,
pub tal_paths: Vec<PathBuf>, pub tal_paths: Vec<PathBuf>,
pub ta_paths: Vec<PathBuf>, pub ta_paths: Vec<PathBuf>,
@@ -168,7 +174,7 @@ pub struct CliArgs {
pub rsync_mirror_root: Option<PathBuf>, pub rsync_mirror_root: Option<PathBuf>,
pub rsync_scope_policy: RsyncScopePolicy, pub rsync_scope_policy: RsyncScopePolicy,
pub max_depth: Option<usize>, pub max_ca_depth: usize,
pub max_instances: Option<usize>, pub max_instances: Option<usize>,
pub validation_time: Option<time::OffsetDateTime>, pub validation_time: Option<time::OffsetDateTime>,
@@ -184,8 +190,19 @@ fn usage() -> String {
Usage: Usage:
{bin} --db <path> --tal-url <url> [--tal-url <url> ...] [options] {bin} --db <path> --tal-url <url> [--tal-url <url> ...] [options]
{bin} --db <path> --tal-path <path> --ta-path <path> [--tal-path <path> --ta-path <path> ...] [options] {bin} --db <path> --tal-path <path> --ta-path <path> [--tal-path <path> --ta-path <path> ...] [options]
{bin} --verification-only --verification-source-run <path> --verification-source-state <path> --verification-out <path> [options]
Options: Options:
--verification-only Revalidate the latest successful run without validation caches or network access
--verification-source-run <path>
Source normal run directory containing input.cir/result.ccr/validation-contract.json
--verification-source-state <path>
Source state root containing db/work-db and db/repo-bytes.db
--verification-out <path> Independent output directory for the one-shot verification run
--keep-verification-scratch
Keep the private work-db checkpoint after a successful verification
--validation-contract-out <path>
Write the effective normal-run validation contract (not valid in verification-only mode)
--db <path> RocksDB directory path (required) --db <path> RocksDB directory path (required)
--raw-store-db <path> External raw-by-hash store DB path (optional) --raw-store-db <path> External raw-by-hash store DB path (optional)
--repo-bytes-db <path> External repo object bytes DB path (optional) --repo-bytes-db <path> External repo object bytes DB path (optional)
@@ -259,7 +276,8 @@ Options:
--rsync-timeout-secs <n> rsync I/O timeout seconds (default: 60) --rsync-timeout-secs <n> rsync I/O timeout seconds (default: 60)
--rsync-mirror-root <path> Persist rsync mirrors under this directory (default: disabled) --rsync-mirror-root <path> Persist rsync mirrors under this directory (default: disabled)
--rsync-scope <policy> rsync scope policy: host, publication-point, or module-root (default: module-root) --rsync-scope <policy> rsync scope policy: host, publication-point, or module-root (default: module-root)
--max-depth <n> Max CA instance depth (0 = root only) --max-ca-depth <n> Maximum CA depth from a trust anchor (root = 0, default: {DEFAULT_MAX_CA_DEPTH})
--max-depth <n> Deprecated alias for --max-ca-depth
--max-instances <n> Max number of CA instances to process --max-instances <n> Max number of CA instances to process
--validation-time <rfc3339> Validation time in RFC3339 (default: now UTC) --validation-time <rfc3339> Validation time in RFC3339 (default: now UTC)
--analyze Write timing analysis JSON under target/live/analyze/<timestamp>/ --analyze Write timing analysis JSON under target/live/analyze/<timestamp>/
@@ -272,6 +290,12 @@ Options:
} }
pub fn parse_args(argv: &[String]) -> Result<CliArgs, String> { pub fn parse_args(argv: &[String]) -> Result<CliArgs, String> {
let mut verification_only = false;
let mut verification_source_run_dir: Option<PathBuf> = None;
let mut verification_source_state_root: Option<PathBuf> = None;
let mut verification_out_dir: Option<PathBuf> = None;
let mut keep_verification_scratch = false;
let mut validation_contract_out_path: Option<PathBuf> = None;
let mut tal_urls: Vec<String> = Vec::new(); let mut tal_urls: Vec<String> = Vec::new();
let mut tal_paths: Vec<PathBuf> = Vec::new(); let mut tal_paths: Vec<PathBuf> = Vec::new();
let mut ta_paths: Vec<PathBuf> = Vec::new(); let mut ta_paths: Vec<PathBuf> = Vec::new();
@@ -319,7 +343,8 @@ pub fn parse_args(argv: &[String]) -> Result<CliArgs, String> {
let mut rsync_timeout_secs: u64 = 30; let mut rsync_timeout_secs: u64 = 30;
let mut rsync_mirror_root: Option<PathBuf> = None; let mut rsync_mirror_root: Option<PathBuf> = None;
let mut rsync_scope_policy = RsyncScopePolicy::default(); let mut rsync_scope_policy = RsyncScopePolicy::default();
let mut max_depth: Option<usize> = None; let mut max_ca_depth: Option<usize> = None;
let mut max_ca_depth_option: Option<&'static str> = None;
let mut max_instances: Option<usize> = None; let mut max_instances: Option<usize> = None;
let mut validation_time: Option<time::OffsetDateTime> = None; let mut validation_time: Option<time::OffsetDateTime> = None;
let mut analyze: bool = false; let mut analyze: bool = false;
@@ -331,6 +356,39 @@ pub fn parse_args(argv: &[String]) -> Result<CliArgs, String> {
let arg = argv[i].as_str(); let arg = argv[i].as_str();
match arg { match arg {
"--help" | "-h" => return Err(usage()), "--help" | "-h" => return Err(usage()),
"--verification-only" => {
verification_only = true;
}
"--verification-source-run" => {
i += 1;
verification_source_run_dir = Some(PathBuf::from(
argv.get(i)
.ok_or("--verification-source-run requires a value")?,
));
}
"--verification-source-state" => {
i += 1;
verification_source_state_root = Some(PathBuf::from(
argv.get(i)
.ok_or("--verification-source-state requires a value")?,
));
}
"--verification-out" => {
i += 1;
verification_out_dir = Some(PathBuf::from(
argv.get(i).ok_or("--verification-out requires a value")?,
));
}
"--keep-verification-scratch" => {
keep_verification_scratch = true;
}
"--validation-contract-out" => {
i += 1;
validation_contract_out_path = Some(PathBuf::from(
argv.get(i)
.ok_or("--validation-contract-out requires a value")?,
));
}
"--tal-url" => { "--tal-url" => {
i += 1; i += 1;
let v = argv.get(i).ok_or("--tal-url requires a value")?; let v = argv.get(i).ok_or("--tal-url requires a value")?;
@@ -661,13 +719,26 @@ pub fn parse_args(argv: &[String]) -> Result<CliArgs, String> {
let v = argv.get(i).ok_or("--rsync-scope requires a value")?; let v = argv.get(i).ok_or("--rsync-scope requires a value")?;
rsync_scope_policy = RsyncScopePolicy::parse_cli_value(v)?; rsync_scope_policy = RsyncScopePolicy::parse_cli_value(v)?;
} }
"--max-depth" => { "--max-ca-depth" | "--max-depth" => {
i += 1; i += 1;
let v = argv.get(i).ok_or("--max-depth requires a value")?; let option = arg;
max_depth = Some( let v = argv
.get(i)
.ok_or_else(|| format!("{option} requires a value"))?;
if let Some(previous_option) = max_ca_depth_option {
return Err(format!(
"{option} cannot be combined with {previous_option}; use only --max-ca-depth"
));
}
max_ca_depth = Some(
v.parse::<usize>() v.parse::<usize>()
.map_err(|_| format!("invalid --max-depth: {v}"))?, .map_err(|_| format!("invalid {option}: {v}"))?,
); );
max_ca_depth_option = Some(match option {
"--max-ca-depth" => "--max-ca-depth",
"--max-depth" => "--max-depth",
_ => unreachable!("matched CA depth option"),
});
} }
"--max-instances" => { "--max-instances" => {
i += 1; i += 1;
@@ -702,10 +773,91 @@ pub fn parse_args(argv: &[String]) -> Result<CliArgs, String> {
i += 1; i += 1;
} }
if verification_only {
if verification_source_run_dir.is_none() {
return Err(format!(
"--verification-only requires --verification-source-run\n\n{}",
usage()
));
}
if verification_source_state_root.is_none() {
return Err(format!(
"--verification-only requires --verification-source-state\n\n{}",
usage()
));
}
if verification_out_dir.is_none() {
return Err(format!(
"--verification-only requires --verification-out\n\n{}",
usage()
));
}
if db_path.is_some()
|| raw_store_db.is_some()
|| repo_bytes_db.is_some()
|| !tal_urls.is_empty()
|| !tal_paths.is_empty()
|| !ta_paths.is_empty()
|| policy_path.is_some()
|| strict_policy.is_some()
|| resource_validation_mode.is_some()
|| validation_time.is_some()
|| max_ca_depth_option.is_some()
|| max_instances.is_some()
|| enable_roa_validation_cache
|| enable_child_certificate_validation_cache
|| publication_point_cache_observe_only
|| enable_publication_point_validation_cache
|| enable_transport_request_prefetch
|| rsync_local_dir.is_some()
|| rsync_command.is_some()
|| !http_root_cert_paths.is_empty()
|| payload_replay_archive.is_some()
|| payload_replay_locks.is_some()
|| payload_base_archive.is_some()
|| payload_base_locks.is_some()
|| payload_delta_archive.is_some()
|| payload_delta_locks.is_some()
|| validation_contract_out_path.is_some()
{
return Err(format!(
"verification-only derives DB, TAL, validation time, policy and cache settings from the source run; conflicting normal-run options are not allowed\n\n{}",
usage()
));
}
if report_json_path.is_some()
|| ccr_out_path.is_some()
|| vrps_csv_out_path.is_some()
|| vaps_csv_out_path.is_some()
|| cir_enabled
|| cir_out_path.is_some()
{
return Err(format!(
"verification-only output paths are derived from --verification-out\n\n{}",
usage()
));
}
db_path = Some(
verification_out_dir
.as_ref()
.expect("checked verification out")
.join("scratch/work-db"),
);
} else if verification_source_run_dir.is_some()
|| verification_source_state_root.is_some()
|| verification_out_dir.is_some()
|| keep_verification_scratch
{
return Err(format!(
"verification source/output options require --verification-only\n\n{}",
usage()
));
}
let db_path = db_path.ok_or_else(|| format!("--db is required\n\n{}", usage()))?; let db_path = db_path.ok_or_else(|| format!("--db is required\n\n{}", usage()))?;
let tal_mode_count = (!tal_urls.is_empty()) as u8 + (!tal_paths.is_empty()) as u8; let tal_mode_count = (!tal_urls.is_empty()) as u8 + (!tal_paths.is_empty()) as u8;
if tal_mode_count != 1 { if !verification_only && tal_mode_count != 1 {
return Err(format!( return Err(format!(
"must specify either one-or-more --tal-url or one-or-more --tal-path/--ta-path pairs\n\n{}", "must specify either one-or-more --tal-url or one-or-more --tal-path/--ta-path pairs\n\n{}",
usage() usage()
@@ -759,13 +911,13 @@ pub fn parse_args(argv: &[String]) -> Result<CliArgs, String> {
usage() usage()
)); ));
} }
if !tal_urls.is_empty() && !ta_paths.is_empty() { if !verification_only && !tal_urls.is_empty() && !ta_paths.is_empty() {
return Err(format!( return Err(format!(
"--ta-path cannot be used with --tal-url mode\n\n{}", "--ta-path cannot be used with --tal-url mode\n\n{}",
usage() usage()
)); ));
} }
if !tal_paths.is_empty() { if !verification_only && !tal_paths.is_empty() {
if !ta_paths.is_empty() { if !ta_paths.is_empty() {
if ta_paths.len() != tal_paths.len() { if ta_paths.len() != tal_paths.len() {
return Err(format!( return Err(format!(
@@ -940,6 +1092,12 @@ pub fn parse_args(argv: &[String]) -> Result<CliArgs, String> {
} }
Ok(CliArgs { Ok(CliArgs {
verification_only,
verification_source_run_dir,
verification_source_state_root,
verification_out_dir,
keep_verification_scratch,
validation_contract_out_path,
tal_urls, tal_urls,
tal_paths, tal_paths,
ta_paths, ta_paths,
@@ -989,7 +1147,7 @@ pub fn parse_args(argv: &[String]) -> Result<CliArgs, String> {
rsync_timeout_secs, rsync_timeout_secs,
rsync_mirror_root, rsync_mirror_root,
rsync_scope_policy, rsync_scope_policy,
max_depth, max_ca_depth: max_ca_depth.unwrap_or(DEFAULT_MAX_CA_DEPTH),
max_instances, max_instances,
validation_time, validation_time,
analyze, analyze,
@@ -1854,7 +2012,7 @@ where
H: crate::sync::rrdp::Fetcher + Clone + 'static, H: crate::sync::rrdp::Fetcher + Clone + 'static,
R: crate::fetch::rsync::RsyncFetcher + Clone + 'static, R: crate::fetch::rsync::RsyncFetcher + Clone + 'static,
{ {
if args.tal_inputs.len() > 1 { if args.verification_only || args.tal_inputs.len() > 1 {
return if let Some(t) = timing { return if let Some(t) = timing {
run_tree_from_multiple_tals_parallel_phase2_audit_with_timing( run_tree_from_multiple_tals_parallel_phase2_audit_with_timing(
store, store,
@@ -1995,21 +2153,88 @@ where
} }
pub fn run(argv: &[String]) -> Result<(), String> { pub fn run(argv: &[String]) -> Result<(), String> {
let args = parse_args(argv)?; let mut args = parse_args(argv)?;
let mut prepared_verification = if args.verification_only {
let prepared = crate::verification_only::prepare_verification(
args.verification_source_run_dir
.as_deref()
.expect("validated by parse_args"),
args.verification_source_state_root
.as_deref()
.expect("validated by parse_args"),
args.verification_out_dir
.as_deref()
.expect("validated by parse_args"),
)?;
args.db_path = prepared.artifacts.scratch_work_db.clone();
args.repo_bytes_db = Some(prepared.artifacts.source_repo_bytes_db.clone());
args.report_json_path = Some(prepared.artifacts.report_json.clone());
args.report_json_compact = true;
args.skip_report_build = false;
args.skip_vcir_persist = true;
args.enable_roa_validation_cache = false;
args.enable_child_certificate_validation_cache = false;
args.publication_point_cache_observe_only = false;
args.enable_publication_point_validation_cache = false;
args.enable_transport_request_prefetch = false;
args.ccr_out_path = Some(prepared.artifacts.result_ccr.clone());
args.vrps_csv_out_path = Some(prepared.artifacts.vrps_csv.clone());
args.vaps_csv_out_path = Some(prepared.artifacts.vaps_csv.clone());
args.compare_view_trust_anchor = Some("verification-only".to_string());
args.cir_enabled = true;
args.cir_out_path = Some(prepared.artifacts.result_cir.clone());
args.cir_tal_uris = crate::verification_only::cir_tal_uris(&prepared.cir);
args.tal_inputs = crate::verification_only::tal_inputs_from_cir(&prepared.cir);
args.validation_time = Some(prepared.cir.validation_time);
args.max_ca_depth = prepared.contract.max_ca_depth;
args.max_instances = prepared.contract.max_instances;
args.disable_rrdp = true;
args.rsync_local_dir = None;
Some(prepared)
} else {
None
};
let mut policy = read_policy(args.policy_path.as_deref())?; let mut policy = if let Some(prepared) = prepared_verification.as_ref() {
if let Some(strict_policy) = args.strict_policy { prepared.contract.policy.clone()
policy.strict = strict_policy; } else {
} read_policy(args.policy_path.as_deref())?
if let Some(resource_validation_mode) = args.resource_validation_mode { };
policy.resource_validation_mode = resource_validation_mode; if !args.verification_only {
} if let Some(strict_policy) = args.strict_policy {
if args.disable_rrdp { policy.strict = strict_policy;
}
if let Some(resource_validation_mode) = args.resource_validation_mode {
policy.resource_validation_mode = resource_validation_mode;
}
if args.disable_rrdp {
policy.sync_preference = crate::policy::SyncPreference::RsyncOnly;
}
} else {
policy.sync_preference = crate::policy::SyncPreference::RsyncOnly; policy.sync_preference = crate::policy::SyncPreference::RsyncOnly;
policy.ca_failed_fetch_policy = crate::policy::CaFailedFetchPolicy::StopAllOutput;
} }
let validation_time = args let validation_time = args
.validation_time .validation_time
.unwrap_or_else(time::OffsetDateTime::now_utc); .unwrap_or_else(time::OffsetDateTime::now_utc);
let validation_time =
time::OffsetDateTime::from_unix_timestamp(validation_time.unix_timestamp())
.map_err(|error| format!("normalize validation time failed: {error}"))?;
if let Some(path) = args.validation_contract_out_path.as_deref() {
let contract = crate::verification_only::ValidationContract::for_current_binary(
validation_time,
policy.clone(),
args.max_ca_depth,
args.max_instances,
crate::verification_only::ValidationCacheContract {
publication_point: args.enable_publication_point_validation_cache,
roa: args.enable_roa_validation_cache,
child_certificate: args.enable_child_certificate_validation_cache,
transport_prefetch: args.enable_transport_request_prefetch,
},
)?;
crate::verification_only::write_validation_contract(path, &contract)?;
}
let http_root_certificates_pem = args let http_root_certificates_pem = args
.http_root_cert_paths .http_root_cert_paths
.iter() .iter()
@@ -2019,7 +2244,12 @@ pub fn run(argv: &[String]) -> Result<(), String> {
}) })
.collect::<Result<Vec<_>, _>>()?; .collect::<Result<Vec<_>, _>>()?;
let store = if args.raw_store_db.is_some() || args.repo_bytes_db.is_some() { let store = if let Some(prepared) = prepared_verification.as_mut() {
prepared
.store
.take()
.expect("prepared verification store must be available")
} else if args.raw_store_db.is_some() || args.repo_bytes_db.is_some() {
Arc::new( Arc::new(
RocksStore::open_with_external_stores( RocksStore::open_with_external_stores(
&args.db_path, &args.db_path,
@@ -2032,7 +2262,7 @@ pub fn run(argv: &[String]) -> Result<(), String> {
Arc::new(RocksStore::open(&args.db_path).map_err(|e| e.to_string())?) Arc::new(RocksStore::open(&args.db_path).map_err(|e| e.to_string())?)
}; };
let config = TreeRunConfig { let config = TreeRunConfig {
max_depth: args.max_depth, max_depth: Some(args.max_ca_depth),
max_instances: args.max_instances, max_instances: args.max_instances,
compact_audit: args.skip_report_build compact_audit: args.skip_report_build
&& args.report_json_path.is_none() && args.report_json_path.is_none()
@@ -2122,7 +2352,23 @@ pub fn run(argv: &[String]) -> Result<(), String> {
); );
let validation_started = std::time::Instant::now(); let validation_started = std::time::Instant::now();
let collect_current_repo_objects = false; let collect_current_repo_objects = false;
let out = if delta_replay_mode { let out = if args.verification_only {
let http = crate::verification_only::NetworkDisabledHttpFetcher;
let rsync = crate::fetch::current_repository::CurrentRepositoryViewRsyncFetcher::new(
Arc::clone(&store),
);
run_online_validation_with_fetchers(
Arc::clone(&store),
&policy,
&args,
&http,
&rsync,
validation_time,
&config,
collect_current_repo_objects,
timing.as_ref().map(|(_, t)| t),
)?
} else if delta_replay_mode {
let tal_path = args let tal_path = args
.tal_path .tal_path
.as_ref() .as_ref()
@@ -2651,6 +2897,62 @@ pub fn run(argv: &[String]) -> Result<(), String> {
} }
print_summary_from_shared(validation_time, &shared); print_summary_from_shared(validation_time, &shared);
if let Some(prepared) = prepared_verification.as_ref() {
std::fs::create_dir_all(&prepared.artifacts.compare_dir).map_err(|error| {
format!(
"create verification compare directory failed: {}: {error}",
prepared.artifacts.compare_dir.display()
)
})?;
let comparison = crate::verification_only::compare_ccr_files(
&prepared.artifacts.source_ccr,
&prepared.artifacts.result_ccr,
)?;
crate::verification_only::write_ccr_comparison(
&prepared.artifacts.compare_dir.join("ccr-state-digest.json"),
Some(&prepared.artifacts.compare_dir.join("ccr-state-digest.md")),
&comparison,
)?;
let scratch_retained = !comparison.state_digest_match || args.keep_verification_scratch;
let source_run_id = prepared
.artifacts
.source_run_dir
.file_name()
.and_then(|value| value.to_str())
.unwrap_or("unknown")
.to_string();
crate::verification_only::write_verification_meta(
&prepared.artifacts.verification_meta,
&crate::verification_only::VerificationRunMeta {
schema_version: 1,
status: if comparison.state_digest_match {
"success".to_string()
} else {
"mismatch".to_string()
},
source_run_id,
validation_time: crate::verification_only::format_validation_time(validation_time)?,
wall_ms: total_started.elapsed().as_millis() as u64,
repository_objects_checked: prepared.blob_summary.current_objects,
repository_bytes_checked: prepared.blob_summary.bytes_verified,
state_digest_match: comparison.state_digest_match,
scratch_retained,
},
)?;
if !comparison.state_digest_match {
return Err(format!(
"verification-only CCR state digest mismatch: {}",
comparison.mismatched_states.join(",")
));
}
if !args.keep_verification_scratch {
drop(shared);
drop(store);
std::fs::remove_dir_all(prepared.artifacts.output_dir.join("scratch")).map_err(
|error| format!("remove verification scratch directory failed: {error}"),
)?;
}
}
Ok(()) Ok(())
} }

View File

@@ -21,6 +21,8 @@ fn parse_help_returns_usage() {
assert!(err.contains("--memory-trim-after-validation"), "{err}"); assert!(err.contains("--memory-trim-after-validation"), "{err}");
assert!(err.contains("--enable-roa-validation-cache"), "{err}"); assert!(err.contains("--enable-roa-validation-cache"), "{err}");
assert!(err.contains("--resource-validation-mode"), "{err}"); assert!(err.contains("--resource-validation-mode"), "{err}");
assert!(err.contains("--max-ca-depth"), "{err}");
assert!(err.contains("default: 32"), "{err}");
assert!( assert!(
err.contains("--publication-point-cache-observe-only"), err.contains("--publication-point-cache-observe-only"),
"{err}" "{err}"
@@ -47,6 +49,63 @@ fn parse_rejects_unknown_argument() {
assert!(err.contains("unknown argument"), "{err}"); assert!(err.contains("unknown argument"), "{err}");
} }
#[test]
fn parse_accepts_verification_only_without_manual_validation_time() {
let argv = vec![
"rpki".to_string(),
"--verification-only".to_string(),
"--verification-source-run".to_string(),
"/run/run_0002".to_string(),
"--verification-source-state".to_string(),
"/run/state".to_string(),
"--verification-out".to_string(),
"/verification/check_0002".to_string(),
];
let args = parse_args(&argv).expect("parse verification-only args");
assert!(args.verification_only);
assert!(args.validation_time.is_none());
assert_eq!(
args.db_path,
PathBuf::from("/verification/check_0002/scratch/work-db")
);
}
#[test]
fn parse_rejects_manual_validation_time_in_verification_only() {
let argv = vec![
"rpki".to_string(),
"--verification-only".to_string(),
"--verification-source-run".to_string(),
"/run/run_0002".to_string(),
"--verification-source-state".to_string(),
"/run/state".to_string(),
"--verification-out".to_string(),
"/verification/check_0002".to_string(),
"--validation-time".to_string(),
"2026-07-16T00:00:00Z".to_string(),
];
let error = parse_args(&argv).expect_err("manual time must be rejected");
assert!(error.contains("validation time"), "{error}");
}
#[test]
fn parse_accepts_normal_validation_contract_output() {
let argv = vec![
"rpki".to_string(),
"--db".to_string(),
"db".to_string(),
"--tal-url".to_string(),
"https://example.test/root.tal".to_string(),
"--validation-contract-out".to_string(),
"out/validation-contract.json".to_string(),
];
let args = parse_args(&argv).expect("parse normal contract output");
assert_eq!(
args.validation_contract_out_path,
Some(PathBuf::from("out/validation-contract.json"))
);
}
#[test] #[test]
fn parse_rejects_both_tal_url_and_tal_path() { fn parse_rejects_both_tal_url_and_tal_path() {
let argv = vec![ let argv = vec![
@@ -80,6 +139,51 @@ fn parse_rejects_invalid_max_depth() {
assert!(err.contains("invalid --max-depth"), "{err}"); assert!(err.contains("invalid --max-depth"), "{err}");
} }
#[test]
fn parse_rejects_invalid_max_ca_depth() {
let argv = vec![
"rpki".to_string(),
"--db".to_string(),
"db".to_string(),
"--tal-url".to_string(),
"https://example.test/x.tal".to_string(),
"--max-ca-depth".to_string(),
"nope".to_string(),
];
let err = parse_args(&argv).unwrap_err();
assert!(err.contains("invalid --max-ca-depth"), "{err}");
}
#[test]
fn parse_rejects_both_ca_depth_options() {
let argv = vec![
"rpki".to_string(),
"--db".to_string(),
"db".to_string(),
"--tal-url".to_string(),
"https://example.test/x.tal".to_string(),
"--max-ca-depth".to_string(),
"32".to_string(),
"--max-depth".to_string(),
"12".to_string(),
];
let err = parse_args(&argv).unwrap_err();
assert!(err.contains("cannot be combined"), "{err}");
}
#[test]
fn parse_defaults_max_ca_depth_to_32() {
let argv = vec![
"rpki".to_string(),
"--db".to_string(),
"db".to_string(),
"--tal-url".to_string(),
"https://example.test/x.tal".to_string(),
];
let args = parse_args(&argv).expect("parse");
assert_eq!(args.max_ca_depth, 32);
}
#[test] #[test]
fn parse_accepts_ccr_out_path() { fn parse_accepts_ccr_out_path() {
let argv = vec![ let argv = vec![
@@ -1103,7 +1207,7 @@ fn parse_accepts_offline_mode_requires_ta() {
assert_eq!(args.ta_paths, vec![PathBuf::from("ta.cer")]); assert_eq!(args.ta_paths, vec![PathBuf::from("ta.cer")]);
assert_eq!(args.tal_path.as_deref(), Some(Path::new("a.tal"))); assert_eq!(args.tal_path.as_deref(), Some(Path::new("a.tal")));
assert_eq!(args.ta_path.as_deref(), Some(Path::new("ta.cer"))); assert_eq!(args.ta_path.as_deref(), Some(Path::new("ta.cer")));
assert_eq!(args.max_depth, Some(0)); assert_eq!(args.max_ca_depth, 0);
} }
#[test] #[test]

View File

@@ -0,0 +1,178 @@
use std::sync::Arc;
use crate::fetch::rsync::{
RsyncFetchError, RsyncFetchResult, RsyncFetcher, normalize_rsync_base_uri,
};
use crate::storage::{RepositoryViewState, RocksStore};
#[derive(Clone)]
pub struct CurrentRepositoryViewRsyncFetcher {
store: Arc<RocksStore>,
}
impl CurrentRepositoryViewRsyncFetcher {
pub fn new(store: Arc<RocksStore>) -> Self {
Self { store }
}
}
impl RsyncFetcher for CurrentRepositoryViewRsyncFetcher {
fn fetch_objects(&self, rsync_base_uri: &str) -> RsyncFetchResult<Vec<(String, Vec<u8>)>> {
let base = normalize_rsync_base_uri(rsync_base_uri);
let entries = self
.store
.list_repository_view_entries_with_prefix(&base)
.map_err(|error| {
RsyncFetchError::Fetch(format!(
"list frozen repository view failed for {base}: {error}"
))
})?;
let mut objects = Vec::with_capacity(entries.len());
for entry in entries {
if !matches!(
entry.state,
RepositoryViewState::Present | RepositoryViewState::Replaced
) {
continue;
}
let bytes = self
.store
.load_current_object_bytes_by_uri(&entry.rsync_uri)
.map_err(|error| {
RsyncFetchError::Fetch(format!(
"load frozen repository object failed for {}: {error}",
entry.rsync_uri
))
})?
.ok_or_else(|| {
RsyncFetchError::Fetch(format!(
"frozen repository object missing for {}",
entry.rsync_uri
))
})?;
objects.push((entry.rsync_uri, bytes));
}
objects.sort_by(|left, right| left.0.cmp(&right.0));
if objects.is_empty() {
return Err(RsyncFetchError::Fetch(format!(
"frozen repository view contains no current objects under {base}"
)));
}
Ok(objects)
}
fn fetch_object(&self, rsync_uri: &str) -> RsyncFetchResult<Vec<u8>> {
self.store
.load_current_object_bytes_by_uri(rsync_uri)
.map_err(|error| {
RsyncFetchError::Fetch(format!(
"load frozen repository object failed for {rsync_uri}: {error}"
))
})?
.ok_or_else(|| {
RsyncFetchError::Fetch(format!("frozen repository object not found: {rsync_uri}"))
})
}
fn dedup_key(&self, rsync_base_uri: &str) -> String {
normalize_rsync_base_uri(rsync_base_uri)
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::storage::{RepositoryViewEntry, RocksStore};
#[test]
fn current_repository_fetcher_returns_sorted_present_objects() {
let dir = tempfile::tempdir().expect("tempdir");
let store = Arc::new(RocksStore::open(dir.path()).expect("open store"));
let entries = [
("rsync://example.test/repo/b.roa", b"b".as_slice()),
("rsync://example.test/repo/a.mft", b"a".as_slice()),
];
for (uri, bytes) in entries {
let hash = hex::encode(crate::cir::sha256(bytes));
store
.put_blob_bytes_batch(&[(hash.clone(), bytes.to_vec())])
.expect("put blob");
store
.put_repository_view_entry(&RepositoryViewEntry {
rsync_uri: uri.to_string(),
repository_source: Some("fixture".to_string()),
object_type: Some("roa".to_string()),
state: RepositoryViewState::Present,
current_hash: Some(hash),
})
.expect("put view");
}
let fetcher = CurrentRepositoryViewRsyncFetcher::new(store);
let objects = fetcher
.fetch_objects("rsync://example.test/repo/")
.expect("fetch objects");
assert_eq!(
objects
.iter()
.map(|(uri, _)| uri.as_str())
.collect::<Vec<_>>(),
vec![
"rsync://example.test/repo/a.mft",
"rsync://example.test/repo/b.roa"
]
);
assert_eq!(
fetcher
.fetch_object("rsync://example.test/repo/a.mft")
.expect("fetch one"),
b"a"
);
assert!(
fetcher
.fetch_object("rsync://example.test/repo/missing.roa")
.is_err()
);
assert_eq!(
fetcher.dedup_key("rsync://example.test/repo"),
"rsync://example.test/repo/"
);
assert!(
fetcher
.fetch_objects("rsync://empty.example/repo/")
.unwrap_err()
.to_string()
.contains("no current objects")
);
}
#[test]
fn current_repository_fetcher_ignores_withdrawn_and_reports_missing_blob() {
let dir = tempfile::tempdir().expect("tempdir");
let store = Arc::new(RocksStore::open(dir.path()).expect("open store"));
store
.put_repository_view_entry(&RepositoryViewEntry {
rsync_uri: "rsync://example.test/repo/withdrawn.roa".to_string(),
repository_source: Some("fixture".to_string()),
object_type: Some("roa".to_string()),
state: RepositoryViewState::Withdrawn,
current_hash: None,
})
.expect("put withdrawn view");
let missing_hash = "ab".repeat(32);
store
.put_repository_view_entry(&RepositoryViewEntry {
rsync_uri: "rsync://example.test/repo/missing.roa".to_string(),
repository_source: Some("fixture".to_string()),
object_type: Some("roa".to_string()),
state: RepositoryViewState::Present,
current_hash: Some(missing_hash),
})
.expect("put missing view");
let fetcher = CurrentRepositoryViewRsyncFetcher::new(store);
let error = fetcher
.fetch_objects("rsync://example.test/repo/")
.unwrap_err()
.to_string();
assert!(error.contains("blob bytes missing"));
}
}

View File

@@ -1,3 +1,4 @@
pub mod current_repository;
pub mod http; pub mod http;
pub mod rsync; pub mod rsync;
pub mod rsync_system; pub mod rsync_system;

View File

@@ -42,3 +42,5 @@ pub mod sync;
pub mod tools; pub mod tools;
#[cfg(feature = "full")] #[cfg(feature = "full")]
pub mod validation; pub mod validation;
#[cfg(feature = "full")]
pub mod verification_only;

View File

@@ -8,6 +8,7 @@ use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex}; use std::sync::{Arc, Mutex};
use base64::Engine; use base64::Engine;
use rocksdb::checkpoint::Checkpoint;
use rocksdb::{ColumnFamily, DB, Direction, IteratorMode, Options, WriteBatch}; use rocksdb::{ColumnFamily, DB, Direction, IteratorMode, Options, WriteBatch};
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
@@ -53,6 +54,13 @@ pub enum StorageError {
pub type StorageResult<T> = Result<T, StorageError>; pub type StorageResult<T> = Result<T, StorageError>;
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize)]
pub struct RepositoryBlobVerificationSummary {
pub current_objects: u64,
pub bytes_verified: u64,
pub batches: u64,
}
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize)] #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize)]
pub struct ChildCertificateCacheMmapLookup { pub struct ChildCertificateCacheMmapLookup {
pub projections: Vec<Option<ChildCertificateCacheProjection>>, pub projections: Vec<Option<ChildCertificateCacheProjection>>,
@@ -595,7 +603,7 @@ pub struct VcirFailedFetchReuseIdentity {
pub ta_context_digest: [u8; 32], pub ta_context_digest: [u8; 32],
#[serde(rename = "p")] #[serde(rename = "p")]
#[serde(with = "serde_bytes_32")] #[serde(with = "serde_bytes_32")]
pub parent_context_digest: [u8; 32], pub ca_validation_context_digest: [u8; 32],
#[serde(rename = "f")] #[serde(rename = "f")]
#[serde(with = "serde_bytes_32")] #[serde(with = "serde_bytes_32")]
pub policy_fingerprint: [u8; 32], pub policy_fingerprint: [u8; 32],
@@ -1117,7 +1125,7 @@ pub struct RoaCacheObjectMeta {
#[derive(Clone, Debug, PartialEq, Eq)] #[derive(Clone, Debug, PartialEq, Eq)]
pub struct RoaCacheProjectionContext { pub struct RoaCacheProjectionContext {
pub parent_context_digest: [u8; 32], pub ca_validation_context_digest: [u8; 32],
pub policy_fingerprint: [u8; 32], pub policy_fingerprint: [u8; 32],
pub object_meta: Vec<RoaCacheObjectMeta>, pub object_meta: Vec<RoaCacheObjectMeta>,
} }
@@ -1258,7 +1266,7 @@ pub struct RoaCacheProjection {
pub issuer_ca_sha256_hex: Option<String>, pub issuer_ca_sha256_hex: Option<String>,
#[serde(rename = "p", default, skip_serializing_if = "Option::is_none")] #[serde(rename = "p", default, skip_serializing_if = "Option::is_none")]
#[serde(with = "serde_optional_bytes_32")] #[serde(with = "serde_optional_bytes_32")]
pub parent_context_digest: Option<[u8; 32]>, pub ca_validation_context_digest: Option<[u8; 32]>,
#[serde(rename = "f", default, skip_serializing_if = "Option::is_none")] #[serde(rename = "f", default, skip_serializing_if = "Option::is_none")]
#[serde(with = "serde_optional_bytes_32")] #[serde(with = "serde_optional_bytes_32")]
pub policy_fingerprint: Option<[u8; 32]>, pub policy_fingerprint: Option<[u8; 32]>,
@@ -1400,7 +1408,7 @@ pub struct ChildCertificateCacheProjection {
pub issuer_crl_sha256_hex: String, pub issuer_crl_sha256_hex: String,
#[serde(rename = "pc")] #[serde(rename = "pc")]
#[serde(with = "serde_bytes_32")] #[serde(with = "serde_bytes_32")]
pub parent_context_digest: [u8; 32], pub ca_validation_context_digest: [u8; 32],
#[serde(rename = "pf")] #[serde(rename = "pf")]
#[serde(with = "serde_bytes_32")] #[serde(with = "serde_bytes_32")]
pub validation_policy_fingerprint: [u8; 32], pub validation_policy_fingerprint: [u8; 32],
@@ -1707,7 +1715,7 @@ pub struct PublicationPointCacheProjection {
pub ta_context_digest: [u8; 32], pub ta_context_digest: [u8; 32],
#[serde(rename = "pc")] #[serde(rename = "pc")]
#[serde(with = "serde_bytes_32")] #[serde(with = "serde_bytes_32")]
pub parent_context_digest: [u8; 32], pub ca_validation_context_digest: [u8; 32],
#[serde(rename = "pf")] #[serde(rename = "pf")]
#[serde(with = "serde_bytes_32")] #[serde(with = "serde_bytes_32")]
pub validation_policy_fingerprint: [u8; 32], pub validation_policy_fingerprint: [u8; 32],
@@ -1743,7 +1751,7 @@ impl PublicationPointCacheProjection {
ca_cert_sha256: [u8; 32], ca_cert_sha256: [u8; 32],
manifest_sha256: [u8; 32], manifest_sha256: [u8; 32],
ta_context_digest: [u8; 32], ta_context_digest: [u8; 32],
parent_context_digest: [u8; 32], ca_validation_context_digest: [u8; 32],
validation_policy_fingerprint: [u8; 32], validation_policy_fingerprint: [u8; 32],
) -> StorageResult<Self> { ) -> StorageResult<Self> {
let projection = Self { let projection = Self {
@@ -1756,7 +1764,7 @@ impl PublicationPointCacheProjection {
manifest_sha256, manifest_sha256,
tal_id: vcir.tal_id.clone(), tal_id: vcir.tal_id.clone(),
ta_context_digest, ta_context_digest,
parent_context_digest, ca_validation_context_digest,
validation_policy_fingerprint, validation_policy_fingerprint,
instance_effective_not_before: vcir.last_successful_validation_time.clone(), instance_effective_not_before: vcir.last_successful_validation_time.clone(),
instance_effective_until: vcir.instance_gate.instance_effective_until.clone(), instance_effective_until: vcir.instance_gate.instance_effective_until.clone(),
@@ -1811,7 +1819,7 @@ impl PublicationPointCacheProjection {
tal_id: self.tal_id.clone(), tal_id: self.tal_id.clone(),
ca_subject_name: String::from("publication-point-cache"), ca_subject_name: String::from("publication-point-cache"),
ca_ski: hex::encode(self.ca_cert_sha256), ca_ski: hex::encode(self.ca_cert_sha256),
issuer_ski: hex::encode(self.parent_context_digest), issuer_ski: hex::encode(self.ca_validation_context_digest),
last_successful_validation_time: validation_time.clone(), last_successful_validation_time: validation_time.clone(),
current_manifest_rsync_uri: self.manifest_rsync_uri.clone(), current_manifest_rsync_uri: self.manifest_rsync_uri.clone(),
current_crl_rsync_uri: self current_crl_rsync_uri: self
@@ -2042,7 +2050,7 @@ impl RoaCacheProjection {
manifest_rsync_uri: vcir.manifest_rsync_uri.clone(), manifest_rsync_uri: vcir.manifest_rsync_uri.clone(),
instance_effective_until: vcir.instance_gate.instance_effective_until.clone(), instance_effective_until: vcir.instance_gate.instance_effective_until.clone(),
issuer_ca_sha256_hex, issuer_ca_sha256_hex,
parent_context_digest: Some(context.parent_context_digest), ca_validation_context_digest: Some(context.ca_validation_context_digest),
policy_fingerprint: Some(context.policy_fingerprint), policy_fingerprint: Some(context.policy_fingerprint),
crl_sha256_by_uri, crl_sha256_by_uri,
entries, entries,
@@ -2063,10 +2071,10 @@ impl RoaCacheProjection {
if let Some(hash) = &self.issuer_ca_sha256_hex { if let Some(hash) = &self.issuer_ca_sha256_hex {
validate_sha256_hex("roa_cache_projection.issuer_ca_sha256_hex", hash)?; validate_sha256_hex("roa_cache_projection.issuer_ca_sha256_hex", hash)?;
} }
if self.parent_context_digest.is_some() != self.policy_fingerprint.is_some() { if self.ca_validation_context_digest.is_some() != self.policy_fingerprint.is_some() {
return Err(StorageError::InvalidData { return Err(StorageError::InvalidData {
entity: "roa_cache_projection.context", entity: "roa_cache_projection.context",
detail: "parent_context_digest and policy_fingerprint must be both present or both absent" detail: "ca_validation_context_digest and policy_fingerprint must be both present or both absent"
.to_string(), .to_string(),
}); });
} }
@@ -2893,6 +2901,38 @@ fn write_publication_point_cache_projection_to_batch(
} }
impl RocksStore { impl RocksStore {
pub fn create_read_only_checkpoint(source: &Path, destination: &Path) -> StorageResult<()> {
if destination.exists() {
return Err(StorageError::InvalidData {
entity: "work_db_checkpoint",
detail: format!(
"checkpoint destination already exists: {}",
destination.display()
),
});
}
if let Some(parent) = destination.parent() {
std::fs::create_dir_all(parent)
.map_err(|error| StorageError::RocksDb(error.to_string()))?;
}
let mut options = Options::default();
let blob_mode = work_db_blob_mode_from_env();
let memory_profile = work_db_memory_profile_from_env();
configure_work_db_options(&mut options, blob_mode, memory_profile);
let db = DB::open_cf_descriptors_read_only(
&options,
source,
column_family_descriptors_for_blob_mode(blob_mode),
false,
)
.map_err(|error| StorageError::RocksDb(error.to_string()))?;
let checkpoint =
Checkpoint::new(&db).map_err(|error| StorageError::RocksDb(error.to_string()))?;
checkpoint
.create_checkpoint(destination)
.map_err(|error| StorageError::RocksDb(error.to_string()))
}
pub fn open(path: &Path) -> StorageResult<Self> { pub fn open(path: &Path) -> StorageResult<Self> {
let mut base_opts = Options::default(); let mut base_opts = Options::default();
base_opts.create_if_missing(true); base_opts.create_if_missing(true);
@@ -2931,6 +2971,15 @@ impl RocksStore {
Self::open_with_external_stores(path, None, Some(repo_bytes_path)) Self::open_with_external_stores(path, None, Some(repo_bytes_path))
} }
pub fn open_with_external_repo_bytes_read_only(
path: &Path,
repo_bytes_path: &Path,
) -> StorageResult<Self> {
let mut store = Self::open(path)?;
store.external_repo_bytes = Some(ExternalRepoBytesDb::open_read_only(repo_bytes_path)?);
Ok(store)
}
pub fn open_with_external_stores( pub fn open_with_external_stores(
path: &Path, path: &Path,
raw_store_path: Option<&Path>, raw_store_path: Option<&Path>,
@@ -3147,6 +3196,53 @@ impl RocksStore {
.collect() .collect()
} }
pub fn verify_current_repository_blobs(
&self,
batch_size: usize,
) -> StorageResult<RepositoryBlobVerificationSummary> {
if batch_size == 0 {
return Err(StorageError::InvalidData {
entity: "repository_blob_verification",
detail: "batch_size must be greater than zero".to_string(),
});
}
let cf = self.cf(CF_REPOSITORY_VIEW)?;
let prefix = repository_view_prefix("");
let mode = IteratorMode::From(prefix.as_bytes(), Direction::Forward);
let mut summary = RepositoryBlobVerificationSummary::default();
let mut batch = Vec::with_capacity(batch_size);
for item in self.db.iterator_cf(cf, mode) {
let (key, value) = item.map_err(|error| StorageError::RocksDb(error.to_string()))?;
if !key.starts_with(prefix.as_bytes()) {
break;
}
let entry = decode_cbor::<RepositoryViewEntry>(&value, "repository_view")?;
entry.validate_internal()?;
if !matches!(
entry.state,
RepositoryViewState::Present | RepositoryViewState::Replaced
) {
continue;
}
let hash = entry
.current_hash
.clone()
.ok_or(StorageError::InvalidData {
entity: "repository_blob_verification",
detail: format!("current hash missing for {}", entry.rsync_uri),
})?;
batch.push((entry.rsync_uri, hash));
if batch.len() >= batch_size {
verify_repository_blob_batch(self, &batch, &mut summary)?;
batch.clear();
}
}
if !batch.is_empty() {
verify_repository_blob_batch(self, &batch, &mut summary)?;
}
Ok(summary)
}
pub fn put_raw_by_hash_entry(&self, entry: &RawByHashEntry) -> StorageResult<()> { pub fn put_raw_by_hash_entry(&self, entry: &RawByHashEntry) -> StorageResult<()> {
entry.validate_internal()?; entry.validate_internal()?;
if let Some(raw_store) = self.external_raw_store.as_ref() { if let Some(raw_store) = self.external_raw_store.as_ref() {
@@ -3206,6 +3302,9 @@ impl RocksStore {
return Ok(()); return Ok(());
} }
if let Some(repo_bytes) = self.external_repo_bytes.as_ref() { if let Some(repo_bytes) = self.external_repo_bytes.as_ref() {
if repo_bytes.is_read_only() {
return repo_bytes.require_existing_blob_bytes_batch(blobs);
}
return repo_bytes.put_blob_bytes_batch(blobs); return repo_bytes.put_blob_bytes_batch(blobs);
} }
if let Some(raw_store) = self.external_raw_store.as_ref() { if let Some(raw_store) = self.external_raw_store.as_ref() {
@@ -4682,6 +4781,37 @@ impl RocksStore {
} }
} }
fn verify_repository_blob_batch(
store: &RocksStore,
batch: &[(String, String)],
summary: &mut RepositoryBlobVerificationSummary,
) -> StorageResult<()> {
let hashes = batch
.iter()
.map(|(_, hash)| hash.clone())
.collect::<Vec<_>>();
let blobs = store.get_blob_bytes_batch(&hashes)?;
for ((uri, expected_hash), blob) in batch.iter().zip(blobs.into_iter()) {
let bytes = blob.as_ref().ok_or(StorageError::InvalidData {
entity: "repository_blob_verification",
detail: format!("blob missing for URI {uri} (hash={expected_hash})"),
})?;
let actual_hash = hex::encode(compute_sha256_32(bytes));
if !actual_hash.eq_ignore_ascii_case(expected_hash) {
return Err(StorageError::InvalidData {
entity: "repository_blob_verification",
detail: format!(
"blob hash mismatch for URI {uri}: expected={expected_hash}, actual={actual_hash}"
),
});
}
summary.current_objects += 1;
summary.bytes_verified += bytes.len() as u64;
}
summary.batches += 1;
Ok(())
}
#[cfg(test)] #[cfg(test)]
#[path = "storage/tests.rs"] #[path = "storage/tests.rs"]
mod tests; mod tests;

View File

@@ -29,7 +29,7 @@ fn sample_child_certificate_cache_projection(
issuer_ca_sha256_hex: sha256_hex(b"issuer-ca"), issuer_ca_sha256_hex: sha256_hex(b"issuer-ca"),
issuer_crl_uri: "rsync://example.test/repo/issuer.crl".to_string(), issuer_crl_uri: "rsync://example.test/repo/issuer.crl".to_string(),
issuer_crl_sha256_hex: sha256_hex(b"issuer-crl"), issuer_crl_sha256_hex: sha256_hex(b"issuer-crl"),
parent_context_digest: sha256_32(b"parent-context"), ca_validation_context_digest: sha256_32(b"parent-context"),
validation_policy_fingerprint: sha256_32(b"policy"), validation_policy_fingerprint: sha256_32(b"policy"),
effective_not_before: pack_time(0), effective_not_before: pack_time(0),
effective_until: pack_time(24), effective_until: pack_time(24),
@@ -294,7 +294,7 @@ fn roa_cache_projection_context(vcir: &ValidatedCaInstanceResult) -> RoaCachePro
}) })
.expect("sample VCIR has a ROA output"); .expect("sample VCIR has a ROA output");
RoaCacheProjectionContext { RoaCacheProjectionContext {
parent_context_digest: [0x31; 32], ca_validation_context_digest: [0x31; 32],
policy_fingerprint: [0x32; 32], policy_fingerprint: [0x32; 32],
object_meta: vec![RoaCacheObjectMeta { object_meta: vec![RoaCacheObjectMeta {
source_object_uri: roa_output.source_object_uri.clone(), source_object_uri: roa_output.source_object_uri.clone(),
@@ -2276,3 +2276,121 @@ fn pack_file_can_lazy_load_bytes_from_external_repo_bytes_store() {
assert_eq!(file.bytes_cloned().expect("cloned repo bytes"), bytes); assert_eq!(file.bytes_cloned().expect("cloned repo bytes"), bytes);
assert_eq!(file.compute_sha256().expect("compute sha256"), file.sha256); assert_eq!(file.compute_sha256().expect("compute sha256"), file.sha256);
} }
#[test]
fn read_only_checkpoint_isolated_from_source_work_db() {
let td = tempfile::tempdir().expect("tempdir");
let source_path = td.path().join("source-work-db");
let checkpoint_path = td.path().join("checkpoint-work-db");
let uri = "rsync://example.test/repo/a.roa";
let bytes = b"checkpoint-object";
let hash = sha256_hex(bytes);
{
let source = RocksStore::open(&source_path).expect("open source");
source
.put_blob_bytes_batch(&[(hash.clone(), bytes.to_vec())])
.expect("put blob");
source
.put_repository_view_entry(&RepositoryViewEntry {
rsync_uri: uri.to_string(),
current_hash: Some(hash),
repository_source: Some("fixture".to_string()),
object_type: Some("roa".to_string()),
state: RepositoryViewState::Present,
})
.expect("put view");
}
RocksStore::create_read_only_checkpoint(&source_path, &checkpoint_path)
.expect("create checkpoint");
let checkpoint = RocksStore::open(&checkpoint_path).expect("open checkpoint");
checkpoint
.delete_repository_view_entry(uri)
.expect("delete checkpoint entry");
drop(checkpoint);
let source = RocksStore::open(&source_path).expect("reopen source");
assert!(
source
.get_repository_view_entry(uri)
.expect("get source entry")
.is_some()
);
}
#[test]
fn read_only_external_repo_bytes_rejects_writes() {
let td = tempfile::tempdir().expect("tempdir");
let path = td.path().join("repo-bytes.db");
let hash = sha256_hex(b"repo-object");
{
let writable = ExternalRepoBytesDb::open(&path).expect("open writable repo bytes");
writable
.put_blob_bytes_batch(&[(hash.clone(), b"repo-object".to_vec())])
.expect("seed repo bytes");
}
let read_only = ExternalRepoBytesDb::open_read_only(&path).expect("open read-only repo bytes");
assert_eq!(
read_only
.get_blob_bytes(&hash)
.expect("read repo bytes")
.expect("blob"),
b"repo-object"
);
assert!(
read_only
.put_blob_bytes_batch(&[(hash, b"repo-object".to_vec())])
.is_err()
);
}
#[test]
fn read_only_store_accepts_only_idempotent_existing_blob_apply() {
let td = tempfile::tempdir().expect("tempdir");
let work_db_path = td.path().join("work-db");
let repo_bytes_path = td.path().join("repo-bytes.db");
let bytes = b"repo-object".to_vec();
let hash = sha256_hex(&bytes);
{
let repo_bytes = ExternalRepoBytesDb::open(&repo_bytes_path).expect("open repo bytes");
repo_bytes
.put_blob_bytes_batch(&[(hash.clone(), bytes.clone())])
.expect("seed repo bytes");
}
let store =
RocksStore::open_with_external_repo_bytes_read_only(&work_db_path, &repo_bytes_path)
.expect("open read-only store");
store
.put_blob_bytes_batch(&[(hash.clone(), bytes)])
.expect("idempotent apply");
assert!(
store
.put_blob_bytes_batch(&[(hash, b"different".to_vec())])
.is_err()
);
}
#[test]
fn repository_blob_verification_detects_missing_external_blob() {
let td = tempfile::tempdir().expect("tempdir");
let repo_bytes_path = td.path().join("repo-bytes.db");
ExternalRepoBytesDb::open(&repo_bytes_path).expect("create repo bytes");
let store = RocksStore::open_with_external_repo_bytes_read_only(
&td.path().join("work-db"),
&repo_bytes_path,
)
.expect("open work db");
store
.put_repository_view_entry(&RepositoryViewEntry {
rsync_uri: "rsync://example.test/repo/missing.roa".to_string(),
current_hash: Some(sha256_hex(b"missing")),
repository_source: Some("fixture".to_string()),
object_type: Some("roa".to_string()),
state: RepositoryViewState::Present,
})
.expect("put view");
let error = store
.verify_current_repository_blobs(16)
.expect_err("missing blob must fail");
assert!(error.to_string().contains("missing.roa"), "{error}");
}

View File

@@ -180,7 +180,7 @@ pub struct RoaValidationCacheStats {
pub struct RoaValidationCacheInput<'a> { pub struct RoaValidationCacheInput<'a> {
enabled: bool, enabled: bool,
view: Option<&'a RoaValidationCacheView>, view: Option<&'a RoaValidationCacheView>,
parent_context_digest: Option<[u8; 32]>, ca_validation_context_digest: Option<[u8; 32]>,
policy_fingerprint: Option<[u8; 32]>, policy_fingerprint: Option<[u8; 32]>,
} }
@@ -189,7 +189,7 @@ impl<'a> RoaValidationCacheInput<'a> {
Self { Self {
enabled: false, enabled: false,
view: None, view: None,
parent_context_digest: None, ca_validation_context_digest: None,
policy_fingerprint: None, policy_fingerprint: None,
} }
} }
@@ -198,20 +198,20 @@ impl<'a> RoaValidationCacheInput<'a> {
Self { Self {
enabled: true, enabled: true,
view, view,
parent_context_digest: None, ca_validation_context_digest: None,
policy_fingerprint: None, policy_fingerprint: None,
} }
} }
pub fn enabled_with_context( pub fn enabled_with_context(
view: Option<&'a RoaValidationCacheView>, view: Option<&'a RoaValidationCacheView>,
parent_context_digest: [u8; 32], ca_validation_context_digest: [u8; 32],
policy_fingerprint: [u8; 32], policy_fingerprint: [u8; 32],
) -> Self { ) -> Self {
Self { Self {
enabled: true, enabled: true,
view, view,
parent_context_digest: Some(parent_context_digest), ca_validation_context_digest: Some(ca_validation_context_digest),
policy_fingerprint: Some(policy_fingerprint), policy_fingerprint: Some(policy_fingerprint),
} }
} }
@@ -495,7 +495,7 @@ fn evaluate_roa_cache_crl_gate(
pub struct RoaValidationCacheView { pub struct RoaValidationCacheView {
entries_by_uri: HashMap<String, CachedRoaValidationResult>, entries_by_uri: HashMap<String, CachedRoaValidationResult>,
issuer_ca_sha256_hex: Option<String>, issuer_ca_sha256_hex: Option<String>,
parent_context_digest: Option<[u8; 32]>, ca_validation_context_digest: Option<[u8; 32]>,
policy_fingerprint: Option<[u8; 32]>, policy_fingerprint: Option<[u8; 32]>,
crl_sha256_by_uri: HashMap<String, String>, crl_sha256_by_uri: HashMap<String, String>,
blocked: bool, blocked: bool,
@@ -519,7 +519,7 @@ impl RoaValidationCacheView {
let mut entries_by_uri: HashMap<String, CachedRoaValidationResult> = let mut entries_by_uri: HashMap<String, CachedRoaValidationResult> =
HashMap::with_capacity(projection.entries.len()); HashMap::with_capacity(projection.entries.len());
let issuer_ca_sha256_hex = projection.issuer_ca_sha256_hex.clone(); let issuer_ca_sha256_hex = projection.issuer_ca_sha256_hex.clone();
let parent_context_digest = projection.parent_context_digest; let ca_validation_context_digest = projection.ca_validation_context_digest;
let policy_fingerprint = projection.policy_fingerprint; let policy_fingerprint = projection.policy_fingerprint;
let crl_sha256_by_uri = projection let crl_sha256_by_uri = projection
.crl_sha256_by_uri .crl_sha256_by_uri
@@ -536,7 +536,7 @@ impl RoaValidationCacheView {
return Self { return Self {
entries_by_uri, entries_by_uri,
issuer_ca_sha256_hex, issuer_ca_sha256_hex,
parent_context_digest, ca_validation_context_digest,
policy_fingerprint, policy_fingerprint,
crl_sha256_by_uri, crl_sha256_by_uri,
blocked, blocked,
@@ -580,7 +580,7 @@ impl RoaValidationCacheView {
Self { Self {
entries_by_uri, entries_by_uri,
issuer_ca_sha256_hex, issuer_ca_sha256_hex,
parent_context_digest, ca_validation_context_digest,
policy_fingerprint, policy_fingerprint,
crl_sha256_by_uri, crl_sha256_by_uri,
blocked, blocked,
@@ -590,7 +590,7 @@ impl RoaValidationCacheView {
fn matches_current_context( fn matches_current_context(
&self, &self,
issuer_ca_der: &[u8], issuer_ca_der: &[u8],
parent_context_digest: Option<[u8; 32]>, ca_validation_context_digest: Option<[u8; 32]>,
policy_fingerprint: Option<[u8; 32]>, policy_fingerprint: Option<[u8; 32]>,
) -> bool { ) -> bool {
if self.blocked { if self.blocked {
@@ -604,8 +604,8 @@ impl RoaValidationCacheView {
return false; return false;
} }
if let Some(expected_parent_context) = self.parent_context_digest { if let Some(expected_ca_validation_context) = self.ca_validation_context_digest {
if Some(expected_parent_context) != parent_context_digest { if Some(expected_ca_validation_context) != ca_validation_context_digest {
return false; return false;
} }
} else { } else {
@@ -797,7 +797,7 @@ fn active_roa_cache_view<'a>(
let gate_started = Instant::now(); let gate_started = Instant::now();
if view.matches_current_context( if view.matches_current_context(
issuer_ca_der, issuer_ca_der,
roa_cache.parent_context_digest, roa_cache.ca_validation_context_digest,
roa_cache.policy_fingerprint, roa_cache.policy_fingerprint,
) { ) {
stats.context_gate_nanos = stats stats.context_gate_nanos = stats
@@ -3914,7 +3914,7 @@ mod tests {
OffsetDateTime::parse(value, &Rfc3339).expect("parse fixed test time") OffsetDateTime::parse(value, &Rfc3339).expect("parse fixed test time")
} }
const TEST_PARENT_CONTEXT: [u8; 32] = [0x70; 32]; const TEST_CA_VALIDATION_CONTEXT: [u8; 32] = [0x70; 32];
const TEST_POLICY_FINGERPRINT: [u8; 32] = [0x71; 32]; const TEST_POLICY_FINGERPRINT: [u8; 32] = [0x71; 32];
const TEST_CRL_URI: &str = "rsync://example.test/repo/current.crl"; const TEST_CRL_URI: &str = "rsync://example.test/repo/current.crl";
const TEST_ROA_URI: &str = "rsync://example.test/repo/a.roa"; const TEST_ROA_URI: &str = "rsync://example.test/repo/a.roa";
@@ -3933,7 +3933,7 @@ mod tests {
RoaCacheProjection::from_vcir_with_context( RoaCacheProjection::from_vcir_with_context(
vcir, vcir,
Some(&RoaCacheProjectionContext { Some(&RoaCacheProjectionContext {
parent_context_digest: TEST_PARENT_CONTEXT, ca_validation_context_digest: TEST_CA_VALIDATION_CONTEXT,
policy_fingerprint: TEST_POLICY_FINGERPRINT, policy_fingerprint: TEST_POLICY_FINGERPRINT,
object_meta: vec![RoaCacheObjectMeta { object_meta: vec![RoaCacheObjectMeta {
source_object_uri: TEST_ROA_URI.to_string(), source_object_uri: TEST_ROA_URI.to_string(),
@@ -4071,7 +4071,7 @@ mod tests {
assert!(view.matches_current_context( assert!(view.matches_current_context(
issuer_der, issuer_der,
Some(TEST_PARENT_CONTEXT), Some(TEST_CA_VALIDATION_CONTEXT),
Some(TEST_POLICY_FINGERPRINT) Some(TEST_POLICY_FINGERPRINT)
)); ));
let hit = view.lookup(&file, &mut crl_cache, issuer_der, validation_time); let hit = view.lookup(&file, &mut crl_cache, issuer_der, validation_time);
@@ -4202,7 +4202,10 @@ mod tests {
fixed_time("2026-06-08T00:00:00Z"), fixed_time("2026-06-08T00:00:00Z"),
); );
let projection = sample_roa_cache_projection(&vcir, roa_hash); let projection = sample_roa_cache_projection(&vcir, roa_hash);
assert_eq!(projection.parent_context_digest, Some(TEST_PARENT_CONTEXT)); assert_eq!(
projection.ca_validation_context_digest,
Some(TEST_CA_VALIDATION_CONTEXT)
);
assert_eq!(projection.policy_fingerprint, Some(TEST_POLICY_FINGERPRINT)); assert_eq!(projection.policy_fingerprint, Some(TEST_POLICY_FINGERPRINT));
assert_eq!( assert_eq!(
projection.entries[0].outputs_effective_until_unix, projection.entries[0].outputs_effective_until_unix,
@@ -4245,7 +4248,7 @@ mod tests {
let issuer_changed = RoaValidationCacheView::from_projection(&projection, validation_time); let issuer_changed = RoaValidationCacheView::from_projection(&projection, validation_time);
assert!(!issuer_changed.matches_current_context( assert!(!issuer_changed.matches_current_context(
issuer_der, issuer_der,
Some(TEST_PARENT_CONTEXT), Some(TEST_CA_VALIDATION_CONTEXT),
Some(TEST_POLICY_FINGERPRINT) Some(TEST_POLICY_FINGERPRINT)
)); ));
@@ -4258,7 +4261,7 @@ mod tests {
)); ));
assert!(!parent_changed.matches_current_context( assert!(!parent_changed.matches_current_context(
issuer_der, issuer_der,
Some(TEST_PARENT_CONTEXT), Some(TEST_CA_VALIDATION_CONTEXT),
Some([0x98; 32]) Some([0x98; 32])
)); ));
@@ -4283,7 +4286,7 @@ mod tests {
RoaValidationCacheView::from_projection(&expired_projection, validation_time); RoaValidationCacheView::from_projection(&expired_projection, validation_time);
assert!(!expired_view.matches_current_context( assert!(!expired_view.matches_current_context(
issuer_der, issuer_der,
Some(TEST_PARENT_CONTEXT), Some(TEST_CA_VALIDATION_CONTEXT),
Some(TEST_POLICY_FINGERPRINT) Some(TEST_POLICY_FINGERPRINT)
)); ));
let mut crl_cache = sample_crl_cache(b"current-crl".to_vec()); let mut crl_cache = sample_crl_cache(b"current-crl".to_vec());
@@ -4563,7 +4566,7 @@ mod tests {
assert!(view.matches_current_context( assert!(view.matches_current_context(
issuer_der, issuer_der,
Some(TEST_PARENT_CONTEXT), Some(TEST_CA_VALIDATION_CONTEXT),
Some(TEST_POLICY_FINGERPRINT) Some(TEST_POLICY_FINGERPRINT)
)); ));
assert!(matches!( assert!(matches!(
@@ -4595,43 +4598,43 @@ mod tests {
let mut view = RoaValidationCacheView { let mut view = RoaValidationCacheView {
entries_by_uri: HashMap::new(), entries_by_uri: HashMap::new(),
issuer_ca_sha256_hex: None, issuer_ca_sha256_hex: None,
parent_context_digest: Some(TEST_PARENT_CONTEXT), ca_validation_context_digest: Some(TEST_CA_VALIDATION_CONTEXT),
policy_fingerprint: Some(TEST_POLICY_FINGERPRINT), policy_fingerprint: Some(TEST_POLICY_FINGERPRINT),
crl_sha256_by_uri: HashMap::new(), crl_sha256_by_uri: HashMap::new(),
blocked: false, blocked: false,
}; };
assert!(!view.matches_current_context( assert!(!view.matches_current_context(
b"issuer-ca", b"issuer-ca",
Some(TEST_PARENT_CONTEXT), Some(TEST_CA_VALIDATION_CONTEXT),
Some(TEST_POLICY_FINGERPRINT) Some(TEST_POLICY_FINGERPRINT)
)); ));
view.issuer_ca_sha256_hex = Some("00".repeat(32)); view.issuer_ca_sha256_hex = Some("00".repeat(32));
assert!(!view.matches_current_context( assert!(!view.matches_current_context(
b"issuer-ca", b"issuer-ca",
Some(TEST_PARENT_CONTEXT), Some(TEST_CA_VALIDATION_CONTEXT),
Some(TEST_POLICY_FINGERPRINT) Some(TEST_POLICY_FINGERPRINT)
)); ));
view.issuer_ca_sha256_hex = Some(sha256_hex(b"issuer-ca")); view.issuer_ca_sha256_hex = Some(sha256_hex(b"issuer-ca"));
assert!(view.matches_current_context( assert!(view.matches_current_context(
b"issuer-ca", b"issuer-ca",
Some(TEST_PARENT_CONTEXT), Some(TEST_CA_VALIDATION_CONTEXT),
Some(TEST_POLICY_FINGERPRINT) Some(TEST_POLICY_FINGERPRINT)
)); ));
assert!(!view.matches_current_context(b"issuer-ca", None, Some(TEST_POLICY_FINGERPRINT))); assert!(!view.matches_current_context(b"issuer-ca", None, Some(TEST_POLICY_FINGERPRINT)));
view.parent_context_digest = None; view.ca_validation_context_digest = None;
assert!(!view.matches_current_context( assert!(!view.matches_current_context(
b"issuer-ca", b"issuer-ca",
Some(TEST_PARENT_CONTEXT), Some(TEST_CA_VALIDATION_CONTEXT),
Some(TEST_POLICY_FINGERPRINT) Some(TEST_POLICY_FINGERPRINT)
)); ));
view.parent_context_digest = Some(TEST_PARENT_CONTEXT); view.ca_validation_context_digest = Some(TEST_CA_VALIDATION_CONTEXT);
view.policy_fingerprint = None; view.policy_fingerprint = None;
assert!(!view.matches_current_context( assert!(!view.matches_current_context(
b"issuer-ca", b"issuer-ca",
Some(TEST_PARENT_CONTEXT), Some(TEST_CA_VALIDATION_CONTEXT),
Some(TEST_POLICY_FINGERPRINT) Some(TEST_POLICY_FINGERPRINT)
)); ));
} }
@@ -4661,7 +4664,7 @@ mod tests {
let mut view = RoaValidationCacheView { let mut view = RoaValidationCacheView {
entries_by_uri: HashMap::new(), entries_by_uri: HashMap::new(),
issuer_ca_sha256_hex: Some(sha256_hex(issuer_der)), issuer_ca_sha256_hex: Some(sha256_hex(issuer_der)),
parent_context_digest: Some(TEST_PARENT_CONTEXT), ca_validation_context_digest: Some(TEST_CA_VALIDATION_CONTEXT),
policy_fingerprint: Some(TEST_POLICY_FINGERPRINT), policy_fingerprint: Some(TEST_POLICY_FINGERPRINT),
crl_sha256_by_uri: HashMap::from([(TEST_CRL_URI.to_string(), crl_hash_hex.clone())]), crl_sha256_by_uri: HashMap::from([(TEST_CRL_URI.to_string(), crl_hash_hex.clone())]),
blocked: false, blocked: false,
@@ -4849,7 +4852,7 @@ mod tests {
false, false,
RoaValidationCacheInput::enabled_with_context( RoaValidationCacheInput::enabled_with_context(
Some(&view), Some(&view),
TEST_PARENT_CONTEXT, TEST_CA_VALIDATION_CONTEXT,
TEST_POLICY_FINGERPRINT, TEST_POLICY_FINGERPRINT,
), ),
) { ) {

View File

@@ -10,6 +10,8 @@ use crate::validation::publication_point::PublicationPointSnapshot;
use std::borrow::Cow; use std::borrow::Cow;
use std::sync::{Arc, OnceLock}; use std::sync::{Arc, OnceLock};
pub const DEFAULT_MAX_CA_DEPTH: usize = 32;
#[derive(Clone, Debug, PartialEq, Eq)] #[derive(Clone, Debug, PartialEq, Eq)]
pub struct TreeRunConfig { pub struct TreeRunConfig {
/// Max CA instance depth to process (0 = root only). /// Max CA instance depth to process (0 = root only).
@@ -37,7 +39,7 @@ pub struct TreeRunConfig {
impl Default for TreeRunConfig { impl Default for TreeRunConfig {
fn default() -> Self { fn default() -> Self {
Self { Self {
max_depth: None, max_depth: Some(DEFAULT_MAX_CA_DEPTH),
max_instances: None, max_instances: None,
compact_audit: false, compact_audit: false,
persist_vcir: true, persist_vcir: true,
@@ -51,6 +53,15 @@ impl Default for TreeRunConfig {
} }
} }
pub(crate) fn ca_depth_is_allowed(config: &TreeRunConfig, depth: usize) -> bool {
config.max_depth.is_none_or(|max_depth| depth <= max_depth)
}
pub(crate) fn next_allowed_ca_depth(config: &TreeRunConfig, parent_depth: usize) -> Option<usize> {
let child_depth = parent_depth.checked_add(1)?;
ca_depth_is_allowed(config, child_depth).then_some(child_depth)
}
#[derive(Clone, Debug)] #[derive(Clone, Debug)]
pub enum CaCertificateRef { pub enum CaCertificateRef {
InlineDer(Vec<u8>), InlineDer(Vec<u8>),
@@ -347,10 +358,8 @@ pub fn run_tree_serial_audit_multi_root(
} }
} }
if let Some(max_depth) = config.max_depth { if !ca_depth_is_allowed(config, ca.depth) {
if ca.depth > max_depth { continue;
continue;
}
} }
let res = match runner.run_publication_point(ca) { let res = match runner.run_publication_point(ca) {
@@ -393,6 +402,9 @@ pub fn run_tree_serial_audit_multi_root(
} }
publication_points.push(audit); publication_points.push(audit);
let Some(child_depth) = next_allowed_ca_depth(config, ca.depth) else {
continue;
};
let mut children = res.discovered_children; let mut children = res.discovered_children;
children.sort_by(|a, b| { children.sort_by(|a, b| {
a.handle a.handle
@@ -413,7 +425,7 @@ pub fn run_tree_serial_audit_multi_root(
for child in children { for child in children {
queue.push_back(QueuedCaInstance { queue.push_back(QueuedCaInstance {
id: next_id, id: next_id,
handle: child.handle.with_depth(ca.depth + 1), handle: child.handle.with_depth(child_depth),
parent_id: Some(node.id), parent_id: Some(node.id),
discovered_from: Some(child.discovered_from), discovered_from: Some(child.discovered_from),
}); });

View File

@@ -19,8 +19,8 @@ use crate::validation::objects::{
}; };
use crate::validation::tree::{ use crate::validation::tree::{
CaInstanceHandle, DiscoveredChildCaInstance, PublicationPointRunResult, PublicationPointRunner, CaInstanceHandle, DiscoveredChildCaInstance, PublicationPointRunResult, PublicationPointRunner,
TreeRunAuditOutput, TreeRunConfig, TreeRunError, TreeRunOutput, TreeRunAuditOutput, TreeRunConfig, TreeRunError, TreeRunOutput, ca_depth_is_allowed,
run_tree_serial_audit_multi_root, next_allowed_ca_depth, run_tree_serial_audit_multi_root,
}; };
use crate::validation::tree_runner::{ use crate::validation::tree_runner::{
FreshPublicationPointFinalizeOutput, FreshPublicationPointStage, Rpkiv1PublicationPointRunner, FreshPublicationPointFinalizeOutput, FreshPublicationPointStage, Rpkiv1PublicationPointRunner,
@@ -659,6 +659,7 @@ pub fn run_tree_parallel_phase2_audit_multi_root(
&mut finished, &mut finished,
ready, ready,
ready_queue_len_after_pop, ready_queue_len_after_pop,
config,
config.compact_audit, config.compact_audit,
); );
ready_batch_metrics.record(metrics); ready_batch_metrics.record(metrics);
@@ -916,10 +917,8 @@ fn start_queued_ca_instances(
if !visited_manifest_uris.insert(node.handle.manifest_rsync_uri.clone()) { if !visited_manifest_uris.insert(node.handle.manifest_rsync_uri.clone()) {
continue; continue;
} }
if let Some(max_depth) = config.max_depth { if !ca_depth_is_allowed(config, node.handle.depth) {
if node.handle.depth > max_depth { continue;
continue;
}
} }
*instances_started += 1; *instances_started += 1;
match repo_runtime.request_publication_point_repo(&node.handle, 0) { match repo_runtime.request_publication_point_repo(&node.handle, 0) {
@@ -959,6 +958,7 @@ fn stage_ready_publication_point(
finished: &mut Vec<FinishedPublicationPoint>, finished: &mut Vec<FinishedPublicationPoint>,
ready: ReadyCaInstance, ready: ReadyCaInstance,
ready_queue_len_after_pop: usize, ready_queue_len_after_pop: usize,
config: &TreeRunConfig,
compact_audit: bool, compact_audit: bool,
) -> ReadyStageMetrics { ) -> ReadyStageMetrics {
let publication_point_started = Instant::now(); let publication_point_started = Instant::now();
@@ -991,6 +991,7 @@ fn stage_ready_publication_point(
next_id, next_id,
ca_queue, ca_queue,
&ready.node, &ready.node,
config,
result.discovered_children.clone(), result.discovered_children.clone(),
); );
metrics.child_enqueue_ms = elapsed_ms(child_enqueue_started); metrics.child_enqueue_ms = elapsed_ms(child_enqueue_started);
@@ -1057,6 +1058,7 @@ fn stage_ready_publication_point(
next_id, next_id,
ca_queue, ca_queue,
&ready.node, &ready.node,
config,
result.discovered_children.clone(), result.discovered_children.clone(),
); );
metrics.child_enqueue_ms = elapsed_ms(child_enqueue_started); metrics.child_enqueue_ms = elapsed_ms(child_enqueue_started);
@@ -1142,6 +1144,7 @@ fn stage_ready_publication_point(
next_id, next_id,
ca_queue, ca_queue,
&ready.node, &ready.node,
config,
fresh_stage.discovered_children.clone(), fresh_stage.discovered_children.clone(),
); );
metrics.child_enqueue_ms = elapsed_ms(child_enqueue_started); metrics.child_enqueue_ms = elapsed_ms(child_enqueue_started);
@@ -1175,7 +1178,7 @@ fn stage_ready_publication_point(
let roa_cache = if runner.enable_roa_validation_cache && has_roa { let roa_cache = if runner.enable_roa_validation_cache && has_roa {
RoaValidationCacheInput::enabled_with_context( RoaValidationCacheInput::enabled_with_context(
roa_cache_view.as_ref(), roa_cache_view.as_ref(),
crate::validation::tree_runner::parent_context_digest_for_ca(&ready.node.handle), crate::validation::tree_runner::ca_validation_context_digest_for_ca(&ready.node.handle),
crate::validation::tree_runner::publication_point_cache_policy_fingerprint( crate::validation::tree_runner::publication_point_cache_policy_fingerprint(
runner.policy, runner.policy,
), ),
@@ -1412,8 +1415,13 @@ fn enqueue_discovered_children(
next_id: &mut u64, next_id: &mut u64,
ca_queue: &mut VecDeque<QueuedCaInstance>, ca_queue: &mut VecDeque<QueuedCaInstance>,
parent: &QueuedCaInstance, parent: &QueuedCaInstance,
config: &TreeRunConfig,
mut children: Vec<DiscoveredChildCaInstance>, mut children: Vec<DiscoveredChildCaInstance>,
) { ) {
let Some(child_depth) = next_allowed_ca_depth(config, parent.handle.depth) else {
return;
};
children.sort_by(|a, b| { children.sort_by(|a, b| {
a.handle a.handle
.manifest_rsync_uri .manifest_rsync_uri
@@ -1428,7 +1436,7 @@ fn enqueue_discovered_children(
let _ = runtime.prefetch_discovered_children(&children); let _ = runtime.prefetch_discovered_children(&children);
} }
for child in children { for child in children {
let mut handle = child.handle.with_depth(parent.handle.depth + 1); let mut handle = child.handle.with_depth(child_depth);
handle.parent_manifest_rsync_uri = Some(parent.handle.manifest_rsync_uri.clone()); handle.parent_manifest_rsync_uri = Some(parent.handle.manifest_rsync_uri.clone());
ca_queue.push_back(QueuedCaInstance { ca_queue.push_back(QueuedCaInstance {
id: *next_id, id: *next_id,

View File

@@ -304,7 +304,8 @@ impl<'a> Rpkiv1PublicationPointRunner<'a> {
self.finish_publication_point_cache_miss(ca, "ta_context_mismatch", lookup_started); self.finish_publication_point_cache_miss(ca, "ta_context_mismatch", lookup_started);
return None; return None;
} }
if projection.parent_context_digest != current_identity.parent_context_digest { if projection.ca_validation_context_digest != current_identity.ca_validation_context_digest
{
self.finish_publication_point_cache_miss(ca, "parent_context_mismatch", lookup_started); self.finish_publication_point_cache_miss(ca, "parent_context_mismatch", lookup_started);
return None; return None;
} }
@@ -505,7 +506,7 @@ impl<'a> Rpkiv1PublicationPointRunner<'a> {
ca_cert_sha256, ca_cert_sha256,
manifest_sha256, manifest_sha256,
ta_context_digest: ta_context_digest_for_ca(ca), ta_context_digest: ta_context_digest_for_ca(ca),
parent_context_digest: parent_context_digest_for_ca(ca), ca_validation_context_digest: ca_validation_context_digest_for_ca(ca),
policy_fingerprint: publication_point_cache_policy_fingerprint(self.policy), policy_fingerprint: publication_point_cache_policy_fingerprint(self.policy),
}) })
} }
@@ -938,7 +939,7 @@ impl<'a> Rpkiv1PublicationPointRunner<'a> {
Some(ChildCertificateValidationCacheContext { Some(ChildCertificateValidationCacheContext {
store: self.store, store: self.store,
issuer_ca_sha256: sha256_digest_32(issuer_ca_der.as_ref()), issuer_ca_sha256: sha256_digest_32(issuer_ca_der.as_ref()),
parent_context_digest: parent_context_digest_for_ca(ca), ca_validation_context_digest: ca_validation_context_digest_for_ca(ca),
policy_fingerprint: publication_point_cache_policy_fingerprint(self.policy), policy_fingerprint: publication_point_cache_policy_fingerprint(self.policy),
}) })
} else { } else {
@@ -1530,7 +1531,7 @@ impl<'a> PublicationPointRunner for Rpkiv1PublicationPointRunner<'a> {
let roa_cache = if self.enable_roa_validation_cache && has_roa { let roa_cache = if self.enable_roa_validation_cache && has_roa {
RoaValidationCacheInput::enabled_with_context( RoaValidationCacheInput::enabled_with_context(
roa_cache_view.as_ref(), roa_cache_view.as_ref(),
parent_context_digest_for_ca(ca), ca_validation_context_digest_for_ca(ca),
publication_point_cache_policy_fingerprint(self.policy), publication_point_cache_policy_fingerprint(self.policy),
) )
} else { } else {
@@ -1946,7 +1947,7 @@ struct PublicationPointCacheIdentity {
ca_cert_sha256: [u8; 32], ca_cert_sha256: [u8; 32],
manifest_sha256: [u8; 32], manifest_sha256: [u8; 32],
ta_context_digest: [u8; 32], ta_context_digest: [u8; 32],
parent_context_digest: [u8; 32], ca_validation_context_digest: [u8; 32],
policy_fingerprint: [u8; 32], policy_fingerprint: [u8; 32],
} }
@@ -1982,7 +1983,7 @@ fn ta_context_digest_for_ca(ca: &CaInstanceHandle) -> [u8; 32] {
]) ])
} }
pub(crate) fn parent_context_digest_for_ca(ca: &CaInstanceHandle) -> [u8; 32] { pub(crate) fn ca_validation_context_digest_for_ca(ca: &CaInstanceHandle) -> [u8; 32] {
hash_serialized_parts(&[ hash_serialized_parts(&[
( (
"version", "version",
@@ -2058,7 +2059,7 @@ fn pack_time_window_contains(
struct ChildCertificateValidationCacheContext<'a> { struct ChildCertificateValidationCacheContext<'a> {
store: &'a RocksStore, store: &'a RocksStore,
issuer_ca_sha256: [u8; 32], issuer_ca_sha256: [u8; 32],
parent_context_digest: [u8; 32], ca_validation_context_digest: [u8; 32],
policy_fingerprint: [u8; 32], policy_fingerprint: [u8; 32],
} }
@@ -2066,7 +2067,7 @@ fn child_certificate_cache_key_sha256_hex(
child_cert_uri: &str, child_cert_uri: &str,
child_cert_sha256: &[u8; 32], child_cert_sha256: &[u8; 32],
issuer_ca_sha256: &[u8; 32], issuer_ca_sha256: &[u8; 32],
parent_context_digest: &[u8; 32], ca_validation_context_digest: &[u8; 32],
policy_fingerprint: &[u8; 32], policy_fingerprint: &[u8; 32],
) -> String { ) -> String {
let digest = hash_serialized_parts(&[ let digest = hash_serialized_parts(&[
@@ -2074,7 +2075,11 @@ fn child_certificate_cache_key_sha256_hex(
("child_cert_uri", child_cert_uri.as_bytes().to_vec()), ("child_cert_uri", child_cert_uri.as_bytes().to_vec()),
("child_cert_sha256", child_cert_sha256.to_vec()), ("child_cert_sha256", child_cert_sha256.to_vec()),
("issuer_ca_sha256", issuer_ca_sha256.to_vec()), ("issuer_ca_sha256", issuer_ca_sha256.to_vec()),
("parent_context_digest", parent_context_digest.to_vec()), // Keep the persisted cache-key label stable; this change only renames Rust identifiers.
(
"parent_context_digest",
ca_validation_context_digest.to_vec(),
),
("policy_fingerprint", policy_fingerprint.to_vec()), ("policy_fingerprint", policy_fingerprint.to_vec()),
]); ]);
sha256_hex_from_32(&digest) sha256_hex_from_32(&digest)
@@ -2469,7 +2474,7 @@ fn discover_children_from_fresh_snapshot_with_audit_cached_with_issuer_der<
&f.rsync_uri, &f.rsync_uri,
&f.sha256, &f.sha256,
&cache.issuer_ca_sha256, &cache.issuer_ca_sha256,
&cache.parent_context_digest, &cache.ca_validation_context_digest,
&cache.policy_fingerprint, &cache.policy_fingerprint,
)); ));
} }
@@ -3022,7 +3027,7 @@ fn discover_children_from_fresh_snapshot_with_audit_cached_with_issuer_der<
&f.rsync_uri, &f.rsync_uri,
&f.sha256, &f.sha256,
&cache.issuer_ca_sha256, &cache.issuer_ca_sha256,
&cache.parent_context_digest, &cache.ca_validation_context_digest,
&cache.policy_fingerprint, &cache.policy_fingerprint,
); );
let projection = ChildCertificateCacheProjection { let projection = ChildCertificateCacheProjection {
@@ -3040,7 +3045,8 @@ fn discover_children_from_fresh_snapshot_with_audit_cached_with_issuer_der<
), ),
issuer_crl_uri: issuer_crl_uri.clone(), issuer_crl_uri: issuer_crl_uri.clone(),
issuer_crl_sha256_hex: verified_crl.sha256_hex.clone(), issuer_crl_sha256_hex: verified_crl.sha256_hex.clone(),
parent_context_digest: cache.parent_context_digest, ca_validation_context_digest: cache
.ca_validation_context_digest,
validation_policy_fingerprint: cache.policy_fingerprint, validation_policy_fingerprint: cache.policy_fingerprint,
effective_not_before, effective_not_before,
effective_until, effective_until,
@@ -3239,7 +3245,7 @@ fn discover_children_from_fresh_snapshot_with_audit_cached_with_issuer_der<
&f.rsync_uri, &f.rsync_uri,
&f.sha256, &f.sha256,
&cache.issuer_ca_sha256, &cache.issuer_ca_sha256,
&cache.parent_context_digest, &cache.ca_validation_context_digest,
&cache.policy_fingerprint, &cache.policy_fingerprint,
); );
let projection = ChildCertificateCacheProjection { let projection = ChildCertificateCacheProjection {
@@ -3255,7 +3261,7 @@ fn discover_children_from_fresh_snapshot_with_audit_cached_with_issuer_der<
issuer_ca_sha256_hex: sha256_hex_from_32(&cache.issuer_ca_sha256), issuer_ca_sha256_hex: sha256_hex_from_32(&cache.issuer_ca_sha256),
issuer_crl_uri: issuer_crl_uri.clone(), issuer_crl_uri: issuer_crl_uri.clone(),
issuer_crl_sha256_hex: verified_crl.sha256_hex.clone(), issuer_crl_sha256_hex: verified_crl.sha256_hex.clone(),
parent_context_digest: cache.parent_context_digest, ca_validation_context_digest: cache.ca_validation_context_digest,
validation_policy_fingerprint: cache.policy_fingerprint, validation_policy_fingerprint: cache.policy_fingerprint,
effective_not_before, effective_not_before,
effective_until, effective_until,
@@ -4395,7 +4401,7 @@ fn failed_fetch_reuse_identity_for_fresh_result(
let identity = VcirFailedFetchReuseIdentity { let identity = VcirFailedFetchReuseIdentity {
current_ca_sha256, current_ca_sha256,
ta_context_digest: ta_context_digest_for_ca(ca), ta_context_digest: ta_context_digest_for_ca(ca),
parent_context_digest: parent_context_digest_for_ca(ca), ca_validation_context_digest: ca_validation_context_digest_for_ca(ca),
policy_fingerprint: publication_point_cache_policy_fingerprint(policy), policy_fingerprint: publication_point_cache_policy_fingerprint(policy),
effective_not_before: PackTime::from_utc_offset_datetime(validation_time), effective_not_before: PackTime::from_utc_offset_datetime(validation_time),
effective_until, effective_until,
@@ -4418,7 +4424,7 @@ fn failed_fetch_reuse_identity_matches_current(
}; };
cached.current_ca_sha256 == current_ca_sha256 cached.current_ca_sha256 == current_ca_sha256
&& cached.ta_context_digest == ta_context_digest_for_ca(ca) && cached.ta_context_digest == ta_context_digest_for_ca(ca)
&& cached.parent_context_digest == parent_context_digest_for_ca(ca) && cached.ca_validation_context_digest == ca_validation_context_digest_for_ca(ca)
&& cached.policy_fingerprint == publication_point_cache_policy_fingerprint(policy) && cached.policy_fingerprint == publication_point_cache_policy_fingerprint(policy)
&& cached.effective_until == vcir.instance_gate.instance_effective_until && cached.effective_until == vcir.instance_gate.instance_effective_until
&& cached.contains_validation_time(validation_time) && cached.contains_validation_time(validation_time)
@@ -5347,7 +5353,7 @@ fn persist_vcir_for_fresh_result_with_timing(
.replace_vcir_manifest_replay_meta_and_projection_action_with_failed_fetch_reuse_identity( .replace_vcir_manifest_replay_meta_and_projection_action_with_failed_fetch_reuse_identity(
&vcir, &vcir,
Some(&RoaCacheProjectionContext { Some(&RoaCacheProjectionContext {
parent_context_digest: parent_context_digest_for_ca(ca), ca_validation_context_digest: ca_validation_context_digest_for_ca(ca),
policy_fingerprint: publication_point_cache_policy_fingerprint(policy), policy_fingerprint: publication_point_cache_policy_fingerprint(policy),
object_meta: objects.roa_cache_object_meta.clone(), object_meta: objects.roa_cache_object_meta.clone(),
}), }),
@@ -5378,7 +5384,7 @@ fn build_publication_point_cache_projection_from_fresh(
ca_cert_sha256, ca_cert_sha256,
manifest_sha256, manifest_sha256,
ta_context_digest_for_ca(ca), ta_context_digest_for_ca(ca),
parent_context_digest_for_ca(ca), ca_validation_context_digest_for_ca(ca),
publication_point_cache_policy_fingerprint(policy), publication_point_cache_policy_fingerprint(policy),
) )
.map_err(|e| e.to_string()) .map_err(|e| e.to_string())

View File

@@ -1601,7 +1601,7 @@ fn discover_children_child_certificate_cache_reuses_successful_child_ca() {
let cache_context = ChildCertificateValidationCacheContext { let cache_context = ChildCertificateValidationCacheContext {
store: &store, store: &store,
issuer_ca_sha256: issuer.ca_certificate_sha256_32().unwrap(), issuer_ca_sha256: issuer.ca_certificate_sha256_32().unwrap(),
parent_context_digest: parent_context_digest_for_ca(&issuer), ca_validation_context_digest: ca_validation_context_digest_for_ca(&issuer),
policy_fingerprint: publication_point_cache_policy_fingerprint(&policy), policy_fingerprint: publication_point_cache_policy_fingerprint(&policy),
}; };
let validation_time = time::OffsetDateTime::now_utc(); let validation_time = time::OffsetDateTime::now_utc();
@@ -1625,7 +1625,7 @@ fn discover_children_child_certificate_cache_reuses_successful_child_ca() {
&child_file.rsync_uri, &child_file.rsync_uri,
&child_file.sha256, &child_file.sha256,
&cache_context.issuer_ca_sha256, &cache_context.issuer_ca_sha256,
&cache_context.parent_context_digest, &cache_context.ca_validation_context_digest,
&cache_context.policy_fingerprint, &cache_context.policy_fingerprint,
); );
assert!( assert!(
@@ -1724,7 +1724,7 @@ fn discover_children_child_certificate_cache_rechecks_changed_valid_crl() {
let cache_context = ChildCertificateValidationCacheContext { let cache_context = ChildCertificateValidationCacheContext {
store: &store, store: &store,
issuer_ca_sha256: issuer.ca_certificate_sha256_32().unwrap(), issuer_ca_sha256: issuer.ca_certificate_sha256_32().unwrap(),
parent_context_digest: parent_context_digest_for_ca(&issuer), ca_validation_context_digest: ca_validation_context_digest_for_ca(&issuer),
policy_fingerprint: publication_point_cache_policy_fingerprint(&policy), policy_fingerprint: publication_point_cache_policy_fingerprint(&policy),
}; };
let validation_time = time::OffsetDateTime::now_utc(); let validation_time = time::OffsetDateTime::now_utc();
@@ -1745,7 +1745,7 @@ fn discover_children_child_certificate_cache_rechecks_changed_valid_crl() {
&child_file.rsync_uri, &child_file.rsync_uri,
&child_file.sha256, &child_file.sha256,
&cache_context.issuer_ca_sha256, &cache_context.issuer_ca_sha256,
&cache_context.parent_context_digest, &cache_context.ca_validation_context_digest,
&cache_context.policy_fingerprint, &cache_context.policy_fingerprint,
); );
let projection = store let projection = store
@@ -1832,7 +1832,7 @@ fn discover_children_child_certificate_cache_misses_when_current_crl_invalid() {
let cache_context = ChildCertificateValidationCacheContext { let cache_context = ChildCertificateValidationCacheContext {
store: &store, store: &store,
issuer_ca_sha256: issuer.ca_certificate_sha256_32().unwrap(), issuer_ca_sha256: issuer.ca_certificate_sha256_32().unwrap(),
parent_context_digest: parent_context_digest_for_ca(&issuer), ca_validation_context_digest: ca_validation_context_digest_for_ca(&issuer),
policy_fingerprint: publication_point_cache_policy_fingerprint(&policy), policy_fingerprint: publication_point_cache_policy_fingerprint(&policy),
}; };
let validation_time = time::OffsetDateTime::now_utc(); let validation_time = time::OffsetDateTime::now_utc();
@@ -1907,7 +1907,7 @@ fn discover_children_child_certificate_cache_misses_when_unchanged_crl_expired()
let cache_context = ChildCertificateValidationCacheContext { let cache_context = ChildCertificateValidationCacheContext {
store: &store, store: &store,
issuer_ca_sha256: issuer.ca_certificate_sha256_32().unwrap(), issuer_ca_sha256: issuer.ca_certificate_sha256_32().unwrap(),
parent_context_digest: parent_context_digest_for_ca(&issuer), ca_validation_context_digest: ca_validation_context_digest_for_ca(&issuer),
policy_fingerprint: publication_point_cache_policy_fingerprint(&policy), policy_fingerprint: publication_point_cache_policy_fingerprint(&policy),
}; };
let validation_time = time::OffsetDateTime::now_utc(); let validation_time = time::OffsetDateTime::now_utc();
@@ -2436,7 +2436,7 @@ fn seed_publication_point_cache_projection(
ca.ca_certificate_sha256_32().unwrap(), ca.ca_certificate_sha256_32().unwrap(),
sha256_32(b"manifest-bytes"), sha256_32(b"manifest-bytes"),
ta_context_digest_for_ca(ca), ta_context_digest_for_ca(ca),
parent_context_digest_for_ca(ca), ca_validation_context_digest_for_ca(ca),
publication_point_cache_policy_fingerprint(policy), publication_point_cache_policy_fingerprint(policy),
) )
.expect("build publication point projection"); .expect("build publication point projection");

1040
src/verification_only.rs Normal file

File diff suppressed because it is too large Load Diff

View File

@@ -253,6 +253,7 @@ fn tree_enqueues_children_for_fresh_and_current_instance_vcir_results() {
fn tree_respects_max_depth_and_max_instances() { fn tree_respects_max_depth_and_max_instances() {
let root_manifest = "rsync://example.test/repo/root.mft"; let root_manifest = "rsync://example.test/repo/root.mft";
let child_manifest = "rsync://example.test/repo/child.mft"; let child_manifest = "rsync://example.test/repo/child.mft";
let grandchild_manifest = "rsync://example.test/repo/grandchild.mft";
let runner = MockRunner::default() let runner = MockRunner::default()
.with( .with(
@@ -301,6 +302,32 @@ fn tree_respects_max_depth_and_max_instances() {
audit: PublicationPointAudit::default(), audit: PublicationPointAudit::default(),
cir_fresh_objects: Vec::new(), cir_fresh_objects: Vec::new(),
cir_cached_objects: Vec::new(), cir_cached_objects: Vec::new(),
discovered_children: vec![discovered_child(child_manifest, grandchild_manifest)],
},
)
.with(
grandchild_manifest,
PublicationPointRunResult {
source: PublicationPointSource::Fresh,
snapshot: Some(empty_snapshot(
grandchild_manifest,
"rsync://example.test/repo/grandchild/",
)),
warnings: Vec::new(),
objects: ObjectsOutput {
vrps: Vec::new(),
aspas: Vec::new(),
router_keys: Vec::new(),
local_outputs_cache: Vec::new(),
warnings: Vec::new(),
stats: ObjectsStats::default(),
audit: Vec::new(),
roa_cache_stats: Default::default(),
roa_cache_object_meta: Vec::new(),
},
audit: PublicationPointAudit::default(),
cir_fresh_objects: Vec::new(),
cir_cached_objects: Vec::new(),
discovered_children: Vec::new(), discovered_children: Vec::new(),
}, },
); );
@@ -324,6 +351,62 @@ fn tree_respects_max_depth_and_max_instances() {
.expect("run tree depth-limited"); .expect("run tree depth-limited");
assert_eq!(out.instances_processed, 1); assert_eq!(out.instances_processed, 1);
assert_eq!(out.instances_failed, 0); assert_eq!(out.instances_failed, 0);
assert_eq!(runner.called(), vec![root_manifest]);
let out = run_tree_serial(
ca_handle(root_manifest),
&runner,
&TreeRunConfig {
max_depth: Some(1),
max_instances: None,
compact_audit: false,
persist_vcir: true,
build_ccr_accumulator: true,
enable_roa_validation_cache: false,
enable_child_certificate_validation_cache: false,
publication_point_cache_observe_only: false,
enable_publication_point_validation_cache: false,
enable_transport_request_prefetch: false,
},
)
.expect("run tree depth-one-limited");
assert_eq!(out.instances_processed, 2);
assert_eq!(out.instances_failed, 0);
assert_eq!(
runner.called(),
vec![root_manifest, root_manifest, child_manifest]
);
let out = run_tree_serial(
ca_handle(root_manifest),
&runner,
&TreeRunConfig {
max_depth: Some(2),
max_instances: None,
compact_audit: false,
persist_vcir: true,
build_ccr_accumulator: true,
enable_roa_validation_cache: false,
enable_child_certificate_validation_cache: false,
publication_point_cache_observe_only: false,
enable_publication_point_validation_cache: false,
enable_transport_request_prefetch: false,
},
)
.expect("run tree depth-two-limited");
assert_eq!(out.instances_processed, 3);
assert_eq!(out.instances_failed, 0);
assert_eq!(
runner.called(),
vec![
root_manifest,
root_manifest,
child_manifest,
root_manifest,
child_manifest,
grandchild_manifest,
]
);
let out = run_tree_serial( let out = run_tree_serial(
ca_handle(root_manifest), ca_handle(root_manifest),