NASP/rpki
84
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: NASP:feature/142-custom-tal-installer
Branches Tags
NASP:main NASP:dev_1.0.0_yuyr_parallel NASP:dev_1.2_xuxt NASP:feature/142-custom-tal-installer NASP:dev_1.3_xuxt NASP:dev_1.0.0_yuyr NASP:dev_1.1_xuxt NASP:dev_1.0_xuxt
NASP:v1.3.0
..
compare: NASP:a77982deaab240dd1660e89a62a12ac73953f40e
Branches Tags
NASP:dev_1.0.0_yuyr_parallel NASP:dev_1.2_xuxt NASP:feature/142-custom-tal-installer NASP:dev_1.3_xuxt NASP:dev_1.0.0_yuyr NASP:main NASP:dev_1.1_xuxt NASP:dev_1.0_xuxt
NASP:v1.3.0

4 Commits
feature/14 ... a77982deaa

Author SHA1 Message Date
yuyr
a77982deaa 20260717 rpctl接入rpki-prover实验控制 2026-07-17 18:10:06 +08:00
yuyr
4865d3698d 20260716 澄清CA验证上下文摘要命名 2026-07-16 22:54:59 +08:00
yuyr
59623b449b 20260716 增加verification-only缓存正确性复核 2026-07-16 19:08:16 +08:00
yuyr
7aac4855bd 20260715 add configurable max CA depth 2026-07-16 09:29:57 +08:00
35 changed files with 3129 additions and 1461 deletions
Expand all files Collapse all files

15
deploy/docker-installer/.env.example
View File

@@ -37,27 +37,12 @@ RETAIN_RUNS=100
# TAL/TA input mode:
# file-with-ta: use packaged fixture TAL + TA only.
# file-live-ta: use packaged fixture TAL; snapshot waits for live TA refresh, delta refreshes TA in background.
# custom-file-with-ta: use one custom TAL + TA from the read-only custom fixture mount.
# url: pass TAL URL to child process.
TAL_INPUT_MODE=file-live-ta
LIVE_TA_REFRESH_BEFORE_SNAPSHOT=1
LIVE_TA_REFRESH_CONNECT_TIMEOUT_SECS=15
LIVE_TA_REFRESH_MAX_TIME_SECS=120
# Custom TAL patch mode. Keep TAL_INPUT_MODE=file-live-ta and the five-RIR RIRS
# list above for the original package behavior. For a private fixture, set:
# RIRS=custom
# TAL_INPUT_MODE=custom-file-with-ta
# The paths below are container paths; CUSTOM_FIXTURE_HOST_DIR is a host path
# resolved relative to the compose directory when left at its default.
CUSTOM_FIXTURE_HOST_DIR=../custom-fixtures
CUSTOM_TAL_PATH=/opt/ours-rp/custom-fixtures/tal/custom.tal
CUSTOM_TA_PATH=/opt/ours-rp/custom-fixtures/ta/custom-ta.cer
CUSTOM_TAL_URI=https://host.docker.internal:18443/tal/custom.tal
# Space-separated container paths; paths must not contain spaces. Set this in
# custom mode when the RRDP server uses a private/self-signed CA.
HTTP_ROOT_CERT_PATHS=
# Sync and runtime behavior.
RSYNC_SCOPE=module-root
DISABLE_COMPETING_RPS=0

7
deploy/docker-installer/compose/docker-compose.yml
View File

@@ -11,15 +11,8 @@ services:
RUN_ROOT: /var/lib/ours-rp
BIN_DIR: /opt/ours-rp/bin
FIXTURE_DIR: /opt/ours-rp/fixtures
CUSTOM_FIXTURE_DIR: /opt/ours-rp/custom-fixtures
extra_hosts:
- "host.docker.internal:host-gateway"
volumes:
- ../.env:/opt/ours-rp/.env:ro
# Overlay the patched runner while keeping the original runtime image
# and its exact rpki binaries unchanged.
- ../scripts/soak/run_soak.sh:/opt/ours-rp/run_soak.sh:ro
- ${CUSTOM_FIXTURE_HOST_DIR:-../custom-fixtures}:/opt/ours-rp/custom-fixtures:ro
- ${HOST_DATA_DIR:-/var/lib/ours-rp-package-installer}/state:/var/lib/ours-rp/state
- ${HOST_DATA_DIR:-/var/lib/ours-rp-package-installer}/runs:/var/lib/ours-rp/runs
- ${HOST_DATA_DIR:-/var/lib/ours-rp-package-installer}/logs:/var/lib/ours-rp/logs

14
deploy/docker-installer/custom-fixtures/README.md
View File

@@ -1,14 +0,0 @@
# Custom TAL fixture mount
This directory is intentionally shipped as a mount point, not as a repository
of private keys. Run the patch tool `tools/generate_custom_fixture.py` to create
the test-only TAL, TA, RPKI objects, HTTPS CA/server certificates, RRDP files,
and rsync module data below it.
The generated fixture contains `baseline-v1`, `sync-hash-mismatch` and
`baseline-v2` for snapshot / hash / delta tests, plus
`validation-expired`, `validation-max-length`,
`validation-max-length-invalid`, `validation-roa-prefix-outside`,
`validation-out-of-resource`, `validation-ca-over-resource`, and
`validation-nonstandard` for validation cases. Generated keys are test-only
and must not be used in a production RPKI hierarchy.

30
deploy/docker-installer/custom-tal-patch/README.zh-CN.md
View File

@@ -1,30 +0,0 @@
# 自建 TAL/TA 补丁
本目录描述并生成一个叠加到客户 Arm64 Ours RP 组件包上的测试补丁。补丁基线是源码 commit `9f4f4cf069e9fa8f6f97ed0f1065adac38abe82f`，不会替换客户包中的 runtime image 或 `bin/rpki`、`bin/rpki_daemon` 二进制。
补丁做三件事：
1. 在 runner 中增加 `RIRS=custom`、`TAL_INPUT_MODE=custom-file-with-ta`、自定义 TAL/TA 路径和 `HTTP_ROOT_CERT_PATHS`；
2. 通过 Compose 只读挂载 patched runner、`custom-fixtures/` 和 `host.docker.internal`，连接固定 HTTPS RRDP 与 rsync 服务；
3. 提供固定证书链、RPKI 对象生成器和服务启动脚本，不依赖 Barry、Rapport 或公网 RIR。
先停止 Ours RP，再执行：
```bash
./apply_custom_tal_patch.sh \
--component-root /opt/ours-rp-rtr-stack/components/ours-rp \
--enable-custom
```
`--enable-custom` 会修改组件 `.env` 的 custom 模式入口；不加该选项只安装 overlay，不改变当前 `.env`。回滚：
```bash
./rollback_custom_tal_patch.sh \
--component-root /opt/ours-rp-rtr-stack/components/ours-rp
```
补丁应用脚本会校验 `PACKAGE-MANIFEST.env` 的源码 commit 和 `arm64` 架构，并留下 `CUSTOM-TAL-PATCH-MANIFEST.env` 与 `.custom-tal-patch-state/` 备份。升级组件后需要重新检查 patch marker；不要把该 overlay 误当成客户原始安装包。
生成器默认把容器访问地址写成 `host.docker.internal`，固定服务端口为 HTTPS RRDP `18443`、rsync `1873`，rsync module 为 `custom`。可用 `--host localhost` 生成给宿主机直接运行 `bin/rpki` 的本地变体。
用例包括：`baseline-v1`、`sync-hash-mismatch`、`baseline-v2`、`validation-expired`、`validation-max-length`、`validation-max-length-invalid`、`validation-roa-prefix-outside`、`validation-out-of-resource`、`validation-ca-over-resource`、`validation-nonstandard`。其中 `validation-ca-over-resource` 在 `RESOURCE_VALIDATION_MODE=rfc6487` 下验证严格拒绝；`validation-update-03` 是原包默认的兼容处理模式。

187
deploy/docker-installer/custom-tal-patch/apply_custom_tal_patch.sh
View File

@@ -1,187 +0,0 @@
#!/usr/bin/env bash
set -euo pipefail
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
PATCH_ROOT="$(cd "$SCRIPT_DIR/.." && pwd)"
PAYLOAD_ROOT="$SCRIPT_DIR/payload"
STATE_DIR_NAME=".custom-tal-patch-state"
EXPECTED_SOURCE_COMMIT="9f4f4cf069e9fa8f6f97ed0f1065adac38abe82f"
usage() {
cat <<'USAGE'
Usage:
./apply_custom_tal_patch.sh --component-root <installed-ours-rp-root>
./apply_custom_tal_patch.sh --component-root <root> --enable-custom
The default operation installs the overlay but does not change the existing
.env. --enable-custom writes the custom TAL/TA/root-cert example values into
the component .env so the next run uses the fixed local fixture.
Stop the ours RP service before applying or rolling back the overlay.
USAGE
}
die() {
echo "error: $*" >&2
exit 2
}
COMPONENT_ROOT=""
ENABLE_CUSTOM=0
while [[ $# -gt 0 ]]; do
case "$1" in
--component-root)
COMPONENT_ROOT="${2:-}"
shift 2
;;
--enable-custom)
ENABLE_CUSTOM=1
shift
;;
-h|--help)
usage
exit 0
;;
*)
die "unknown option: $1"
;;
esac
done
[[ -n "$COMPONENT_ROOT" ]] || die "--component-root is required"
COMPONENT_ROOT="$(cd "$COMPONENT_ROOT" 2>/dev/null && pwd)" \
|| die "component root does not exist: $COMPONENT_ROOT"
[[ -f "$COMPONENT_ROOT/PACKAGE-MANIFEST.env" ]] \
|| die "missing component PACKAGE-MANIFEST.env: $COMPONENT_ROOT"
[[ -d "$PAYLOAD_ROOT" ]] || die "missing patch payload: $PAYLOAD_ROOT"
# shellcheck disable=SC1091
source "$COMPONENT_ROOT/PACKAGE-MANIFEST.env"
[[ "${source_commit:-}" == "$EXPECTED_SOURCE_COMMIT" ]] \
|| die "base source commit mismatch: ${source_commit:-missing} != $EXPECTED_SOURCE_COMMIT"
[[ "${package_arch:-${PACKAGE_ARCH:-}}" == "arm64" ]] \
|| die "this patch is for the customer Arm64 component, got ${package_arch:-${PACKAGE_ARCH:-missing}}"
STATE_ROOT="$COMPONENT_ROOT/$STATE_DIR_NAME"
BACKUP_ROOT="$STATE_ROOT/original"
marker="$COMPONENT_ROOT/CUSTOM-TAL-PATCH-MANIFEST.env"
set_env_value() {
local env_path="$1"
local key="$2"
local value="$3"
local tmp_path="${env_path}.custom-tal-patch.tmp"
[[ -f "$env_path" ]] || die "missing component .env: $env_path"
awk -v key="$key" -v value="$value" '
BEGIN { found = 0 }
$0 ~ "^" key "=" { print key "=" value; found = 1; next }
{ print }
END { if (!found) print key "=" value }
' "$env_path" > "$tmp_path"
mv "$tmp_path" "$env_path"
}
copy_payload_file() {
local source_path="$1"
local target_path="$2"
mkdir -p "$(dirname "$target_path")"
cp "$source_path" "$target_path"
chmod --reference="$source_path" "$target_path" 2>/dev/null || true
}
append_custom_env_example() {
local target_path="$1"
if ! grep -Eq '^CUSTOM_TAL_URI=' "$target_path"; then
{
printf '\n# #142 custom TAL patch settings; original package values above are preserved.\n'
cat "$SCRIPT_DIR/custom-tal.env.example"
} >> "$target_path"
fi
}
if [[ -e "$STATE_ROOT" ]]; then
die "patch is already applied or an incomplete state exists: $STATE_ROOT"
fi
mkdir -p "$BACKUP_ROOT"
for relative_path in compose/docker-compose.yml .env.example; do
[[ -f "$COMPONENT_ROOT/$relative_path" ]] \
|| die "missing base file: $COMPONENT_ROOT/$relative_path"
mkdir -p "$BACKUP_ROOT/$(dirname "$relative_path")"
cp -a "$COMPONENT_ROOT/$relative_path" "$BACKUP_ROOT/$relative_path"
done
runner_path="$COMPONENT_ROOT/scripts/soak/run_soak.sh"
if [[ -e "$runner_path" ]]; then
mkdir -p "$BACKUP_ROOT/scripts/soak"
cp -a "$runner_path" "$BACKUP_ROOT/scripts/soak/run_soak.sh"
printf 'runner_was_present=1\n' > "$STATE_ROOT/state.env"
else
printf 'runner_was_present=0\n' > "$STATE_ROOT/state.env"
fi
if [[ -d "$COMPONENT_ROOT/custom-fixtures" ]]; then
cp -a "$COMPONENT_ROOT/custom-fixtures" "$BACKUP_ROOT/custom-fixtures"
printf 'custom_fixtures_were_present=1\n' >> "$STATE_ROOT/state.env"
else
printf 'custom_fixtures_were_present=0\n' >> "$STATE_ROOT/state.env"
fi
if [[ -d "$COMPONENT_ROOT/custom-tal-patch" ]]; then
cp -a "$COMPONENT_ROOT/custom-tal-patch" "$BACKUP_ROOT/custom-tal-patch"
printf 'patch_tools_were_present=1\n' >> "$STATE_ROOT/state.env"
else
printf 'patch_tools_were_present=0\n' >> "$STATE_ROOT/state.env"
fi
if [[ -f "$COMPONENT_ROOT/custom-tal.env.example" ]]; then
cp -a "$COMPONENT_ROOT/custom-tal.env.example" "$BACKUP_ROOT/custom-tal.env.example"
printf 'custom_env_example_was_present=1\n' >> "$STATE_ROOT/state.env"
else
printf 'custom_env_example_was_present=0\n' >> "$STATE_ROOT/state.env"
fi
if [[ "$ENABLE_CUSTOM" == "1" ]]; then
[[ -f "$COMPONENT_ROOT/.env" ]] || die "missing component .env: $COMPONENT_ROOT/.env"
cp -a "$COMPONENT_ROOT/.env" "$BACKUP_ROOT/.env"
fi
copy_payload_file "$PAYLOAD_ROOT/compose/docker-compose.yml" "$COMPONENT_ROOT/compose/docker-compose.yml"
append_custom_env_example "$COMPONENT_ROOT/.env.example"
copy_payload_file "$PAYLOAD_ROOT/scripts/soak/run_soak.sh" "$runner_path"
mkdir -p "$COMPONENT_ROOT/custom-fixtures/tal" \
"$COMPONENT_ROOT/custom-fixtures/ta" \
"$COMPONENT_ROOT/custom-fixtures/certs" \
"$COMPONENT_ROOT/custom-fixtures/services"
if [[ -d "$PAYLOAD_ROOT/custom-fixtures" ]]; then
cp -a "$PAYLOAD_ROOT/custom-fixtures/." "$COMPONENT_ROOT/custom-fixtures/"
fi
mkdir -p "$COMPONENT_ROOT/custom-tal-patch/tools"
cp -a "$PAYLOAD_ROOT/tools/." "$COMPONENT_ROOT/custom-tal-patch/tools/"
copy_payload_file "$SCRIPT_DIR/custom-tal.env.example" \
"$COMPONENT_ROOT/custom-tal.env.example"
if [[ "$ENABLE_CUSTOM" == "1" ]]; then
set_env_value "$COMPONENT_ROOT/.env" RIRS custom
set_env_value "$COMPONENT_ROOT/.env" TAL_INPUT_MODE custom-file-with-ta
set_env_value "$COMPONENT_ROOT/.env" CUSTOM_FIXTURE_HOST_DIR ../custom-fixtures
set_env_value "$COMPONENT_ROOT/.env" CUSTOM_TAL_PATH /opt/ours-rp/custom-fixtures/tal/custom.tal
set_env_value "$COMPONENT_ROOT/.env" CUSTOM_TA_PATH /opt/ours-rp/custom-fixtures/ta/custom-ta.cer
set_env_value "$COMPONENT_ROOT/.env" CUSTOM_TAL_URI https://host.docker.internal:18443/tal/custom.tal
set_env_value "$COMPONENT_ROOT/.env" HTTP_ROOT_CERT_PATHS /opt/ours-rp/custom-fixtures/certs/rrdp-ca.pem
printf 'env_was_enabled=1\n' >> "$STATE_ROOT/state.env"
else
printf 'env_was_enabled=0\n' >> "$STATE_ROOT/state.env"
fi
[[ -f "$SCRIPT_DIR/PATCH-MANIFEST.env" ]] \
|| die "missing PATCH-MANIFEST.env"
cp "$SCRIPT_DIR/PATCH-MANIFEST.env" "$marker"
printf 'patch_root=%s\n' "$PATCH_ROOT" >> "$STATE_ROOT/state.env"
printf 'applied_at_utc=%s\n' "$(date -u +%Y-%m-%dT%H:%M:%SZ)" >> "$STATE_ROOT/state.env"
echo "custom TAL patch applied to $COMPONENT_ROOT"
if [[ "$ENABLE_CUSTOM" == "1" ]]; then
echo "custom mode enabled in $COMPONENT_ROOT/.env"
else
echo "custom mode remains disabled; review custom-tal.env.example before enabling it"
fi

15
deploy/docker-installer/custom-tal-patch/custom-tal.env.example
View File

@@ -1,15 +0,0 @@
# Copy these values into the component .env after applying the patch.
# The patch's --enable-custom option writes the same values automatically.
RIRS=custom
TAL_INPUT_MODE=custom-file-with-ta
CUSTOM_FIXTURE_HOST_DIR=../custom-fixtures
CUSTOM_TAL_PATH=/opt/ours-rp/custom-fixtures/tal/custom.tal
CUSTOM_TA_PATH=/opt/ours-rp/custom-fixtures/ta/custom-ta.cer
CUSTOM_TAL_URI=https://host.docker.internal:18443/tal/custom.tal
HTTP_ROOT_CERT_PATHS=/opt/ours-rp/custom-fixtures/certs/rrdp-ca.pem
# Recommended for a finite acceptance run; keep the customer default for a
# long-running test only after the fixture service is supervised.
MAX_RUNS=1
INTERVAL_SECS=0
SOAK_RESTART_POLICY=no

80
deploy/docker-installer/custom-tal-patch/rollback_custom_tal_patch.sh
View File

@@ -1,80 +0,0 @@
#!/usr/bin/env bash
set -euo pipefail
usage() {
cat <<'USAGE'
Usage:
./rollback_custom_tal_patch.sh --component-root <installed-ours-rp-root>
Stop the ours RP service before rolling back the overlay.
USAGE
}
die() {
echo "error: $*" >&2
exit 2
}
COMPONENT_ROOT=""
while [[ $# -gt 0 ]]; do
case "$1" in
--component-root)
COMPONENT_ROOT="${2:-}"
shift 2
;;
-h|--help)
usage
exit 0
;;
*)
die "unknown option: $1"
;;
esac
done
[[ -n "$COMPONENT_ROOT" ]] || die "--component-root is required"
COMPONENT_ROOT="$(cd "$COMPONENT_ROOT" 2>/dev/null && pwd)" \
|| die "component root does not exist: $COMPONENT_ROOT"
STATE_ROOT="$COMPONENT_ROOT/.custom-tal-patch-state"
BACKUP_ROOT="$STATE_ROOT/original"
[[ -d "$STATE_ROOT" ]] || die "patch state not found: $STATE_ROOT"
[[ -f "$BACKUP_ROOT/compose/docker-compose.yml" ]] \
|| die "patch backup is incomplete: $BACKUP_ROOT"
cp -a "$BACKUP_ROOT/compose/docker-compose.yml" "$COMPONENT_ROOT/compose/docker-compose.yml"
cp -a "$BACKUP_ROOT/.env.example" "$COMPONENT_ROOT/.env.example"
if [[ -f "$BACKUP_ROOT/scripts/soak/run_soak.sh" ]]; then
mkdir -p "$COMPONENT_ROOT/scripts/soak"
cp -a "$BACKUP_ROOT/scripts/soak/run_soak.sh" "$COMPONENT_ROOT/scripts/soak/run_soak.sh"
else
rm -f "$COMPONENT_ROOT/scripts/soak/run_soak.sh"
rmdir "$COMPONENT_ROOT/scripts/soak" 2>/dev/null || true
fi
if [[ -d "$BACKUP_ROOT/custom-fixtures" ]]; then
rm -rf "$COMPONENT_ROOT/custom-fixtures"
cp -a "$BACKUP_ROOT/custom-fixtures" "$COMPONENT_ROOT/custom-fixtures"
else
rm -rf "$COMPONENT_ROOT/custom-fixtures"
fi
if [[ -d "$BACKUP_ROOT/custom-tal-patch" ]]; then
rm -rf "$COMPONENT_ROOT/custom-tal-patch"
cp -a "$BACKUP_ROOT/custom-tal-patch" "$COMPONENT_ROOT/custom-tal-patch"
else
rm -rf "$COMPONENT_ROOT/custom-tal-patch"
fi
if [[ -f "$BACKUP_ROOT/custom-tal.env.example" ]]; then
cp -a "$BACKUP_ROOT/custom-tal.env.example" "$COMPONENT_ROOT/custom-tal.env.example"
else
rm -f "$COMPONENT_ROOT/custom-tal.env.example"
fi
if [[ -f "$BACKUP_ROOT/.env" ]]; then
cp -a "$BACKUP_ROOT/.env" "$COMPONENT_ROOT/.env"
fi
rm -f "$COMPONENT_ROOT/CUSTOM-TAL-PATCH-MANIFEST.env"
rm -rf "$STATE_ROOT"
echo "custom TAL patch rolled back from $COMPONENT_ROOT"

636
deploy/docker-installer/custom-tal-patch/tools/generate_custom_fixture.py
View File

@@ -1,636 +0,0 @@
#!/usr/bin/env python3
"""Generate a small self-contained RPKI repository for installer tests.
The generator uses Python cryptography and the OpenSSL CLI for CMS signing.
It does not invoke Barry, Rapport, or a public RIR service.
"""
from __future__ import annotations
import argparse
import base64
import hashlib
import ipaddress
import shutil
import subprocess
import tempfile
from datetime import datetime, timedelta, timezone
from pathlib import Path
from cryptography import x509
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import rsa
from cryptography.hazmat.primitives.serialization import Encoding, NoEncryption, PrivateFormat, PublicFormat
from cryptography.x509 import AccessDescription, AuthorityInformationAccess, DistributionPoint, UniformResourceIdentifier
from cryptography.x509.oid import AuthorityInformationAccessOID, NameOID, ObjectIdentifier
RRDP_HOST = "host.docker.internal"
RRDP_PORT = 18443
RSYNC_PORT = 1873
TAL_URI = f"rsync://{RRDP_HOST}:{RSYNC_PORT}/custom/ta.cer"
TAL_HTTPS_URI = f"https://{RRDP_HOST}:{RRDP_PORT}/ta/custom-ta.cer"
RRDP_URI = f"https://{RRDP_HOST}:{RRDP_PORT}/rrdp/notification.xml"
ROOT_REPO_URI = f"rsync://{RRDP_HOST}:{RSYNC_PORT}/custom/root/"
ROOT_MFT_URI = f"{ROOT_REPO_URI}root.mft"
CHILD_REPO_URI = f"rsync://{RRDP_HOST}:{RSYNC_PORT}/custom/child/"
CHILD_MFT_URI = f"{CHILD_REPO_URI}child.mft"
OID_SIA = ObjectIdentifier("1.3.6.1.5.5.7.1.11")
OID_RESOURCES_IP = ObjectIdentifier("1.3.6.1.5.5.7.1.7")
OID_RESOURCES_AS = ObjectIdentifier("1.3.6.1.5.5.7.1.8")
OID_POLICY_IP_AS = ObjectIdentifier("1.3.6.1.5.5.7.14.2")
OID_ROA_ECONTENT = "1.2.840.113549.1.9.16.1.24"
OID_MANIFEST_ECONTENT = "1.2.840.113549.1.9.16.1.26"
OID_SHA256 = "2.16.840.1.101.3.4.2.1"
SIA_CA_REPOSITORY = "1.3.6.1.5.5.7.48.5"
SIA_RPKI_MANIFEST = "1.3.6.1.5.5.7.48.10"
SIA_SIGNED_OBJECT = "1.3.6.1.5.5.7.48.11"
SIA_RPKI_NOTIFY = "1.3.6.1.5.5.7.48.13"
UTC = timezone.utc
VALID_FROM = datetime(2026, 1, 1, tzinfo=UTC)
VALID_TO = datetime(2027, 12, 31, 23, 59, 59, tzinfo=UTC)
CHILD_PREFIX = ipaddress.ip_network("203.0.113.0/24")
SECOND_PREFIX = ipaddress.ip_network("203.0.113.128/25")
OUTSIDE_PREFIX = ipaddress.ip_network("203.0.114.0/24")
TEST_ASN = 64496
SESSION_ID = "11111111-2222-4333-8444-555555555555"
def tlv(tag: int, value: bytes) -> bytes:
length = len(value)
if length < 128:
encoded_length = bytes([length])
else:
raw = length.to_bytes((length.bit_length() + 7) // 8, "big")
encoded_length = bytes([0x80 | len(raw)]) + raw
return bytes([tag]) + encoded_length + value
def seq(*values: bytes) -> bytes:
return tlv(0x30, b"".join(values))
def integer(value: int) -> bytes:
raw = value.to_bytes(max(1, (value.bit_length() + 7) // 8), "big")
if raw[0] & 0x80:
raw = b"\x00" + raw
return tlv(0x02, raw)
def oid(oid_value: str) -> bytes:
parts = [int(part) for part in oid_value.split(".")]
encoded = bytearray([40 * parts[0] + parts[1]])
for part in parts[2:]:
chunks = [part & 0x7F]
part >>= 7
while part:
chunks.append(0x80 | (part & 0x7F))
part >>= 7
encoded.extend(reversed(chunks))
return tlv(0x06, bytes(encoded))
def octet(value: bytes) -> bytes:
return tlv(0x04, value)
def uri(value: str) -> bytes:
return tlv(0x86, value.encode("ascii"))
def bit_string(value: bytes, unused: int = 0) -> bytes:
return tlv(0x03, bytes([unused]) + value)
def generalized_time(value: datetime) -> bytes:
return tlv(0x18, value.astimezone(UTC).strftime("%Y%m%d%H%M%SZ").encode("ascii"))
def context_zero(value: bytes) -> bytes:
return tlv(0xA0, value)
def ip_choice(networks: list[ipaddress._BaseNetwork]) -> bytes:
entries = []
for network in networks:
length = (network.prefixlen + 7) // 8
raw = network.network_address.packed[:length]
entries.append(bit_string(raw, length * 8 - network.prefixlen))
return seq(*entries)
def ip_resources(networks: list[ipaddress._BaseNetwork]) -> bytes:
families: dict[int, list[ipaddress._BaseNetwork]] = {1: [], 2: []}
for network in networks:
families[1 if network.version == 4 else 2].append(network)
return seq(
*(
seq(octet(afi.to_bytes(2, "big")), ip_choice(items))
for afi, items in families.items()
if items
)
)
def ip_inherit() -> bytes:
return seq(
seq(octet(b"\x00\x01"), tlv(0x05, b"")),
seq(octet(b"\x00\x02"), tlv(0x05, b"")),
)
def as_resources(asns: list[int] | None) -> bytes:
if asns is None:
return seq(context_zero(tlv(0x05, b"")))
return seq(context_zero(seq(*(integer(asn) for asn in asns))))
def sia(entries: list[tuple[str, str]]) -> bytes:
return seq(*(seq(oid(method), uri(location)) for method, location in entries))
def write(path: Path, data: bytes, mode: int | None = None) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_bytes(data)
if mode is not None:
path.chmod(mode)
def pem_cert(cert: x509.Certificate) -> bytes:
return cert.public_bytes(Encoding.PEM)
def pem_key(key: rsa.RSAPrivateKey) -> bytes:
return key.private_bytes(Encoding.PEM, PrivateFormat.TraditionalOpenSSL, NoEncryption())
def write_key(path: Path, key: rsa.RSAPrivateKey) -> None:
write(path, pem_key(key), 0o600)
def sign_cert(
subject: x509.Name,
issuer: x509.Name,
public_key,
signing_key: rsa.RSAPrivateKey,
serial: int,
is_ca: bool,
ski_key: rsa.RSAPrivateKey,
aki_key: rsa.RSAPrivateKey | None = None,
ip_ext: bytes | None = None,
as_ext: bytes | None = None,
ca_repository: str | None = None,
manifest_uri: str | None = None,
notify_uri: str | None = None,
signed_object_uri: str | None = None,
crl_uri: str | None = None,
ca_issuer_uri: str | None = None,
not_before: datetime = VALID_FROM,
not_after: datetime = VALID_TO,
) -> x509.Certificate:
builder = (
x509.CertificateBuilder()
.subject_name(subject)
.issuer_name(issuer)
.public_key(public_key)
.serial_number(serial)
.not_valid_before(not_before)
.not_valid_after(not_after)
)
if is_ca:
builder = builder.add_extension(x509.BasicConstraints(ca=True, path_length=None), critical=True)
if is_ca:
usage = x509.KeyUsage(False, False, False, False, False, True, True, False, False)
else:
usage = x509.KeyUsage(True, False, False, False, False, False, False, False, False)
builder = builder.add_extension(usage, critical=True)
builder = builder.add_extension(x509.SubjectKeyIdentifier.from_public_key(ski_key.public_key()), critical=False)
if aki_key is not None:
builder = builder.add_extension(x509.AuthorityKeyIdentifier.from_issuer_public_key(aki_key.public_key()), critical=False)
builder = builder.add_extension(
x509.CertificatePolicies([x509.PolicyInformation(OID_POLICY_IP_AS, None)]),
critical=True,
)
if crl_uri is not None:
builder = builder.add_extension(
x509.CRLDistributionPoints(
[DistributionPoint([UniformResourceIdentifier(crl_uri)], None, None, None)]
),
critical=False,
)
if ca_issuer_uri is not None:
builder = builder.add_extension(
AuthorityInformationAccess(
[AccessDescription(AuthorityInformationAccessOID.CA_ISSUERS, UniformResourceIdentifier(ca_issuer_uri))]
),
critical=False,
)
sia_entries: list[tuple[str, str]] = []
if ca_repository is not None:
sia_entries.append((SIA_CA_REPOSITORY, ca_repository))
if manifest_uri is not None:
sia_entries.append((SIA_RPKI_MANIFEST, manifest_uri))
if notify_uri is not None:
sia_entries.append((SIA_RPKI_NOTIFY, notify_uri))
if signed_object_uri is not None:
sia_entries.append((SIA_SIGNED_OBJECT, signed_object_uri))
if sia_entries:
builder = builder.add_extension(x509.UnrecognizedExtension(OID_SIA, sia(sia_entries)), critical=False)
if ip_ext is not None:
builder = builder.add_extension(x509.UnrecognizedExtension(OID_RESOURCES_IP, ip_ext), critical=True)
if as_ext is not None:
builder = builder.add_extension(x509.UnrecognizedExtension(OID_RESOURCES_AS, as_ext), critical=True)
return builder.sign(signing_key, hashes.SHA256())
def make_crl(subject: x509.Name, key: rsa.RSAPrivateKey, number: int) -> x509.CertificateRevocationList:
return (
x509.CertificateRevocationListBuilder()
.issuer_name(subject)
.last_update(VALID_FROM + timedelta(days=30))
.next_update(VALID_TO - timedelta(days=30))
.add_extension(x509.AuthorityKeyIdentifier.from_issuer_public_key(key.public_key()), critical=False)
.add_extension(x509.CRLNumber(number), critical=False)
.sign(key, hashes.SHA256())
)
def roa_content(prefix: ipaddress._BaseNetwork, max_length: int) -> bytes:
length = (prefix.prefixlen + 7) // 8
raw = prefix.network_address.packed[:length]
address = seq(bit_string(raw, length * 8 - prefix.prefixlen), integer(max_length))
family = seq(octet((1 if prefix.version == 4 else 2).to_bytes(2, "big")), seq(address))
return seq(integer(TEST_ASN), seq(family))
def manifest_content(number: int, files: list[tuple[str, bytes]]) -> bytes:
entries = [
seq(tlv(0x16, name.encode("ascii")), bit_string(hashlib.sha256(content).digest()))
for name, content in files
]
return seq(
integer(number),
generalized_time(VALID_FROM + timedelta(days=30)),
generalized_time(VALID_TO - timedelta(days=30)),
oid(OID_SHA256),
seq(*entries),
)
def write_cms(content: bytes, content_type: str, signer_cert: x509.Certificate, signer_key: rsa.RSAPrivateKey, output: Path, work: Path) -> None:
content_path = work / f"{output.name}.content.der"
signer_path = work / f"{output.name}.signer.pem"
key_path = work / f"{output.name}.key.pem"
write(content_path, content)
write(signer_path, pem_cert(signer_cert))
write_key(key_path, signer_key)
subprocess.run(
[
"openssl", "cms", "-sign", "-binary", "-in", str(content_path),
"-signer", str(signer_path), "-inkey", str(key_path),
"-outform", "DER", "-nodetach", "-nosmimecap", "-keyid", "-md", "sha256",
"-econtent_type", content_type, "-out", str(output),
],
check=True,
stdout=subprocess.DEVNULL,
stderr=subprocess.PIPE,
)
def make_roa(
name: str,
prefix,
max_length: int,
not_before: datetime,
not_after: datetime,
child_name,
child_key,
common: Path,
work: Path,
certificate_prefix=None,
) -> None:
key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
cert = sign_cert(
x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, f"#142 EE {name}")]),
child_name,
key.public_key(),
child_key,
x509.random_serial_number(),
False,
key,
child_key,
ip_ext=ip_resources([certificate_prefix or prefix]),
signed_object_uri=f"{CHILD_REPO_URI}{name}",
crl_uri=f"{CHILD_REPO_URI}child.crl",
ca_issuer_uri=f"{CHILD_REPO_URI}child.cer",
not_before=not_before,
not_after=not_after,
)
write_cms(roa_content(prefix, max_length), OID_ROA_ECONTENT, cert, key, common / name, work)
def make_manifest(name: str, number: int, files: list[tuple[str, bytes]], child_name, child_key, common: Path, work: Path) -> bytes:
key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
cert = sign_cert(
x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, f"#142 Manifest EE {name}")]),
child_name,
key.public_key(),
child_key,
x509.random_serial_number(),
False,
key,
child_key,
ip_ext=ip_inherit(),
as_ext=as_resources(None),
signed_object_uri=CHILD_MFT_URI if name != "root" else ROOT_MFT_URI,
crl_uri=f"{CHILD_REPO_URI}child.crl" if name != "root" else f"{ROOT_REPO_URI}root.crl",
ca_issuer_uri=f"{CHILD_REPO_URI}child.cer" if name != "root" else f"{ROOT_REPO_URI}ta.cer",
)
output = common / f"{name}.mft"
write_cms(manifest_content(number, files), OID_MANIFEST_ECONTENT, cert, key, output, work)
return output.read_bytes()
def publish(uri: str, path: Path) -> str:
encoded = base64.b64encode(path.read_bytes()).decode("ascii")
return f' <publish uri="{uri}">{encoded}</publish>\n'
def snapshot(case_root: Path, names: list[str], serial: int, extra_root_names: list[str] | None = None) -> bytes:
repository = case_root / "repository"
objects = [
(f"{ROOT_REPO_URI}root.crl", repository / "root/root.crl"),
(f"{ROOT_REPO_URI}child.cer", repository / "root/child.cer"),
*[
(f"{ROOT_REPO_URI}{name}", repository / f"root/{name}")
for name in extra_root_names or []
],
(f"{CHILD_REPO_URI}child.crl", repository / "child/child.crl"),
*[(f"{CHILD_REPO_URI}{name}", repository / f"child/{name}") for name in names],
(ROOT_MFT_URI, repository / "root/root.mft"),
(CHILD_MFT_URI, repository / "child/child.mft"),
]
body = "".join(publish(uri, path) for uri, path in objects)
return (
f'<snapshot xmlns="http://www.ripe.net/rpki/rrdp" session_id="{SESSION_ID}" serial="{serial}">\n'
f"{body}</snapshot>\n"
).encode()
def notification(serial: int, snap: bytes, delta: bytes | None) -> bytes:
lines = [
f'<notification xmlns="http://www.ripe.net/rpki/rrdp" version="1" session_id="{SESSION_ID}" serial="{serial}">',
f' <snapshot uri="https://{RRDP_HOST}:{RRDP_PORT}/rrdp/snapshot.xml" hash="{hashlib.sha256(snap).hexdigest()}"/>',
]
if delta is not None:
lines.append(
f' <delta serial="{serial}" uri="https://{RRDP_HOST}:{RRDP_PORT}/rrdp/delta-{serial}.xml" hash="{hashlib.sha256(delta).hexdigest()}"/>'
)
return ("\n".join(lines) + "\n</notification>\n").encode()
def corrupt_first_hash(xml: bytes) -> bytes:
marker = b'hash="'
start = xml.index(marker) + len(marker)
end = xml.index(b'"', start)
value = bytearray(xml[start:end])
value[0] = ord("0") if value[0] != ord("0") else ord("1")
return xml[:start] + bytes(value) + xml[end:]
def delta(add_name: str, add_path: Path, new_mft: Path) -> bytes:
body = publish(CHILD_MFT_URI, new_mft)
body += publish(f"{CHILD_REPO_URI}{add_name}", add_path)
return f'<delta xmlns="http://www.ripe.net/rpki/rrdp" session_id="{SESSION_ID}" serial="2">\n{body}</delta>\n'.encode()
def build_case(
output: Path,
common: Path,
tal: bytes,
ta: bytes,
name: str,
number: int,
names: list[str],
child_mft: bytes,
change: bytes | None = None,
root_mft: bytes | None = None,
extra_root_objects: list[tuple[str, bytes]] | None = None,
notification_hash_mismatch: bool = False,
) -> None:
case_root = output / "cases" / name
repo = case_root / "repository"
(repo / "root").mkdir(parents=True, exist_ok=True)
(repo / "child").mkdir(parents=True, exist_ok=True)
shutil.copy2(common / "ta.cer", repo / "ta.cer")
for item in ("root.crl", "child.cer"):
shutil.copy2(common / item, repo / "root" / item)
(repo / "root/root.mft").write_bytes(root_mft if root_mft is not None else (common / "root.mft").read_bytes())
for item, content in extra_root_objects or []:
(repo / "root" / item).write_bytes(content)
for item in (
"child.crl", "valid.roa", "valid2.roa", "expired.roa", "max-length.roa",
"max-length-invalid.roa", "roa-prefix-outside.roa", "out-of-resource.roa", "bad-format.roa",
):
if (common / item).exists():
shutil.copy2(common / item, repo / "child" / item)
(repo / "child/child.mft").write_bytes(child_mft)
snap = snapshot(case_root, names, number, [item for item, _ in extra_root_objects or []])
(case_root / "http/rrdp").mkdir(parents=True, exist_ok=True)
(case_root / "http/tal").mkdir(parents=True, exist_ok=True)
(case_root / "http/ta").mkdir(parents=True, exist_ok=True)
(case_root / "http/rrdp/snapshot.xml").write_bytes(snap)
notification_xml = notification(number, snap, change)
if notification_hash_mismatch:
notification_xml = corrupt_first_hash(notification_xml)
(case_root / "http/rrdp/notification.xml").write_bytes(notification_xml)
if change is not None:
(case_root / "http/rrdp/delta-2.xml").write_bytes(change)
(case_root / "http/tal/custom.tal").write_bytes(tal)
(case_root / "http/ta/custom-ta.cer").write_bytes(ta)
(case_root / "CASE-MANIFEST.txt").write_text(
f"case={name}\nserial={number}\nobjects={','.join(names)}\nsnapshot_sha256={hashlib.sha256(snap).hexdigest()}\n",
encoding="utf-8",
)
def make_https_certificates(output: Path) -> None:
ca_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
ca_name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "#142 fixed RRDP test CA")])
ca_cert = (
x509.CertificateBuilder()
.subject_name(ca_name).issuer_name(ca_name).public_key(ca_key.public_key())
.serial_number(x509.random_serial_number()).not_valid_before(VALID_FROM).not_valid_after(VALID_TO)
.add_extension(x509.BasicConstraints(ca=True, path_length=1), critical=True)
.add_extension(x509.KeyUsage(False, False, False, False, False, True, True, False, False), critical=True)
.sign(ca_key, hashes.SHA256())
)
server_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
server_name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, RRDP_HOST)])
try:
server_host = x509.IPAddress(ipaddress.ip_address(RRDP_HOST))
except ValueError:
server_host = x509.DNSName(RRDP_HOST)
san_entries = [server_host]
if RRDP_HOST != "localhost":
san_entries.append(x509.DNSName("localhost"))
server_cert = (
x509.CertificateBuilder()
.subject_name(server_name).issuer_name(ca_cert.subject).public_key(server_key.public_key())
.serial_number(x509.random_serial_number()).not_valid_before(VALID_FROM).not_valid_after(VALID_TO)
.add_extension(x509.BasicConstraints(ca=False, path_length=None), critical=True)
.add_extension(x509.KeyUsage(True, False, False, False, False, False, False, False, False), critical=True)
.add_extension(x509.SubjectAlternativeName(san_entries), critical=False)
.sign(ca_key, hashes.SHA256())
)
write(output / "certs/rrdp-ca.pem", pem_cert(ca_cert))
write_key(output / "certs/rrdp-ca.key", ca_key)
write(output / "certs/rrdp-server.pem", pem_cert(server_cert))
write_key(output / "certs/rrdp-server.key", server_key)
def main() -> None:
global RRDP_HOST, TAL_URI, TAL_HTTPS_URI, RRDP_URI
global ROOT_REPO_URI, ROOT_MFT_URI, CHILD_REPO_URI, CHILD_MFT_URI
parser = argparse.ArgumentParser()
parser.add_argument("--output", type=Path, required=True)
parser.add_argument(
"--host",
default=RRDP_HOST,
help="host embedded in TAL, certificate SIA and RRDP data (default: host.docker.internal)",
)
args = parser.parse_args()
RRDP_HOST = args.host
TAL_URI = f"rsync://{RRDP_HOST}:{RSYNC_PORT}/custom/ta.cer"
TAL_HTTPS_URI = f"https://{RRDP_HOST}:{RRDP_PORT}/ta/custom-ta.cer"
RRDP_URI = f"https://{RRDP_HOST}:{RRDP_PORT}/rrdp/notification.xml"
ROOT_REPO_URI = f"rsync://{RRDP_HOST}:{RSYNC_PORT}/custom/root/"
ROOT_MFT_URI = f"{ROOT_REPO_URI}root.mft"
CHILD_REPO_URI = f"rsync://{RRDP_HOST}:{RSYNC_PORT}/custom/child/"
CHILD_MFT_URI = f"{CHILD_REPO_URI}child.mft"
output = args.output.resolve()
if output.exists():
shutil.rmtree(output)
for directory in ("tal", "ta", "certs", "services", "keys"):
(output / directory).mkdir(parents=True)
root_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
child_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
root_name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "#142 Custom RPKI Root")])
child_name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "#142 Custom RPKI Child")])
root_cert = sign_cert(
root_name, root_name, root_key.public_key(), root_key, 0x14200001, True, root_key,
ip_ext=ip_resources([CHILD_PREFIX]),
as_ext=as_resources([TEST_ASN]), ca_repository=ROOT_REPO_URI, manifest_uri=ROOT_MFT_URI,
notify_uri=RRDP_URI,
)
child_cert = sign_cert(
child_name, root_name, child_key.public_key(), root_key, 0x14200002, True, child_key, root_key,
ip_ext=ip_resources([CHILD_PREFIX]), as_ext=as_resources([TEST_ASN]),
ca_repository=CHILD_REPO_URI, manifest_uri=CHILD_MFT_URI, notify_uri=RRDP_URI,
crl_uri=f"{ROOT_REPO_URI}root.crl", ca_issuer_uri=f"{ROOT_REPO_URI}child.cer",
)
bad_child_name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "#142 Custom RPKI Overclaiming Child")])
bad_child_repo_uri = f"rsync://{RRDP_HOST}:{RSYNC_PORT}/custom/bad-child/"
bad_child_cert = sign_cert(
bad_child_name, root_name, child_key.public_key(), root_key, 0x14200003, True, child_key, root_key,
ip_ext=ip_resources([OUTSIDE_PREFIX]), as_ext=as_resources([TEST_ASN]),
ca_repository=bad_child_repo_uri, manifest_uri=f"{bad_child_repo_uri}bad-child.mft", notify_uri=RRDP_URI,
crl_uri=f"{ROOT_REPO_URI}root.crl", ca_issuer_uri=f"{ROOT_REPO_URI}child.cer",
)
root_crl = make_crl(root_name, root_key, 1)
child_crl = make_crl(child_name, child_key, 1)
tal = (
f"{TAL_URI}\n{TAL_HTTPS_URI}\n\n"
+ base64.b64encode(root_key.public_key().public_bytes(Encoding.DER, PublicFormat.SubjectPublicKeyInfo)).decode()
+ "\n"
).encode()
write(output / "tal/custom.tal", tal)
write(output / "ta/custom-ta.cer", root_cert.public_bytes(Encoding.DER))
with tempfile.TemporaryDirectory(prefix="custom-rpki-cms-") as temp:
work = Path(temp)
common = output / ".common"
common.mkdir()
write(common / "ta.cer", root_cert.public_bytes(Encoding.DER))
write(common / "child.cer", child_cert.public_bytes(Encoding.DER))
write(common / "bad-child.cer", bad_child_cert.public_bytes(Encoding.DER))
write(common / "root.crl", root_crl.public_bytes(Encoding.DER))
write(common / "child.crl", child_crl.public_bytes(Encoding.DER))
for name, prefix, max_length, before, after, certificate_prefix in (
("valid.roa", CHILD_PREFIX, 24, VALID_FROM + timedelta(days=31), VALID_TO - timedelta(days=31), None),
("valid2.roa", SECOND_PREFIX, 25, VALID_FROM + timedelta(days=31), VALID_TO - timedelta(days=31), None),
("expired.roa", CHILD_PREFIX, 24, VALID_FROM - timedelta(days=30), VALID_FROM - timedelta(days=1), None),
("max-length.roa", CHILD_PREFIX, 25, VALID_FROM + timedelta(days=31), VALID_TO - timedelta(days=31), None),
("max-length-invalid.roa", CHILD_PREFIX, 33, VALID_FROM + timedelta(days=31), VALID_TO - timedelta(days=31), None),
("roa-prefix-outside.roa", OUTSIDE_PREFIX, 24, VALID_FROM + timedelta(days=31), VALID_TO - timedelta(days=31), CHILD_PREFIX),
("out-of-resource.roa", OUTSIDE_PREFIX, 24, VALID_FROM + timedelta(days=31), VALID_TO - timedelta(days=31), None),
):
make_roa(
name, prefix, max_length, before, after, child_name, child_key, common, work,
certificate_prefix=certificate_prefix,
)
bad = bytearray((common / "valid.roa").read_bytes())
bad[-1] ^= 1
write(common / "bad-format.roa", bytes(bad))
root_mft = make_manifest(
"root", 1, [
("root.crl", (common / "root.crl").read_bytes()),
("child.cer", (common / "child.cer").read_bytes()),
], root_name, root_key, common, work,
)
write(common / "root.mft", root_mft)
bad_root_mft = make_manifest(
"root", 1, [
("root.crl", (common / "root.crl").read_bytes()),
("child.cer", (common / "child.cer").read_bytes()),
("bad-child.cer", (common / "bad-child.cer").read_bytes()),
], root_name, root_key, common, work,
)
write(common / "root.mft", root_mft)
case_specs = (
("baseline-v1", 1, ["child.crl", "valid.roa"]),
("sync-hash-mismatch", 1, ["child.crl", "valid.roa"]),
("baseline-v2", 2, ["child.crl", "valid.roa", "valid2.roa"]),
("validation-expired", 1, ["child.crl", "expired.roa"]),
("validation-max-length", 1, ["child.crl", "max-length.roa"]),
("validation-max-length-invalid", 1, ["child.crl", "max-length-invalid.roa"]),
("validation-roa-prefix-outside", 1, ["child.crl", "roa-prefix-outside.roa"]),
("validation-out-of-resource", 1, ["child.crl", "out-of-resource.roa"]),
("validation-ca-over-resource", 1, ["child.crl", "valid.roa"]),
("validation-nonstandard", 1, ["child.crl", "bad-format.roa"]),
)
child_mfts = {}
for name, number, names in case_specs:
child_mfts[name] = make_manifest(
name, number, [(item, (common / item).read_bytes()) for item in names],
child_name, child_key, common, work,
)
for name, number, names in case_specs:
change = None
if name == "baseline-v2":
change = delta("valid2.roa", common / "valid2.roa", common / "baseline-v2.mft")
build_case(
output, common, tal, root_cert.public_bytes(Encoding.DER), name, number, names,
child_mfts[name], change,
root_mft=bad_root_mft if name == "validation-ca-over-resource" else None,
extra_root_objects=[("bad-child.cer", (common / "bad-child.cer").read_bytes())]
if name == "validation-ca-over-resource" else None,
notification_hash_mismatch=name == "sync-hash-mismatch",
)
write_key(output / "keys/root.key", root_key)
write_key(output / "keys/child.key", child_key)
make_https_certificates(output)
shutil.rmtree(output / ".common")
(output / "FIXTURE-MANIFEST.txt").write_text(
"fixture_schema_version=1\nfixture_kind=custom-tal-fixed-local-rpki\n"
f"tal_uri={TAL_URI}\nrrdp_uri={RRDP_URI}\nrsync_module=custom\n"
f"rrdp_port={RRDP_PORT}\nrsync_port={RSYNC_PORT}\n"
"cases=baseline-v1,sync-hash-mismatch,baseline-v2,validation-expired,validation-max-length,validation-max-length-invalid,"
"validation-roa-prefix-outside,validation-out-of-resource,validation-ca-over-resource,validation-nonstandard\n",
encoding="utf-8",
)
print(f"generated custom fixture: {output}")
if __name__ == "__main__":
main()

36
deploy/docker-installer/custom-tal-patch/tools/serve_https.py
View File

@@ -1,36 +0,0 @@
#!/usr/bin/env python3
"""Serve a fixture directory over HTTPS with request logging."""
from __future__ import annotations
import argparse
import http.server
import ssl
from pathlib import Path
class RequestHandler(http.server.SimpleHTTPRequestHandler):
def log_message(self, format: str, *args) -> None:
print("%s %s" % (self.log_date_time_string(), format % args), flush=True)
def main() -> None:
parser = argparse.ArgumentParser()
parser.add_argument("--directory", type=Path, required=True)
parser.add_argument("--certfile", type=Path, required=True)
parser.add_argument("--keyfile", type=Path, required=True)
parser.add_argument("--host", default="0.0.0.0")
parser.add_argument("--port", type=int, default=18443)
args = parser.parse_args()
handler = lambda *handler_args, directory=str(args.directory): RequestHandler(
*handler_args, directory=directory
)
server = http.server.ThreadingHTTPServer((args.host, args.port), handler)
context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
context.load_cert_chain(certfile=args.certfile, keyfile=args.keyfile)
server.socket = context.wrap_socket(server.socket, server_side=True)
print(f"https fixture server listening on {args.host}:{args.port} root={args.directory}", flush=True)
server.serve_forever()
if __name__ == "__main__":
main()

87
deploy/docker-installer/custom-tal-patch/tools/start_fixed_services.sh
View File

@@ -1,87 +0,0 @@
#!/usr/bin/env bash
set -euo pipefail
SCRIPT_DIR="$(cd "$(dirname "$BASH_SOURCE")" && pwd)"
FIXTURE_ROOT=""
CASE_NAME="baseline-v1"
RRDP_PORT="$(printenv RRDP_PORT 2>/dev/null || true)"
RSYNC_PORT="$(printenv RSYNC_PORT 2>/dev/null || true)"
PID_ROOT=""
[[ -n "$RRDP_PORT" ]] || RRDP_PORT=18443
[[ -n "$RSYNC_PORT" ]] || RSYNC_PORT=1873
usage() {
cat <<'USAGE'
Usage:
./start_fixed_services.sh --fixture-root <generated-fixtures> [--case NAME]
Cases: baseline-v1, sync-hash-mismatch, baseline-v2, validation-expired,
validation-max-length, validation-max-length-invalid,
validation-roa-prefix-outside, validation-out-of-resource,
validation-ca-over-resource, validation-nonstandard
Environment: RRDP_PORT (default 18443), RSYNC_PORT (default 1873)
USAGE
}
die() {
echo "error: $*" >&2
exit 2
}
while [[ $# -gt 0 ]]; do
case "$1" in
--fixture-root) FIXTURE_ROOT="$2"; shift 2 ;;
--case) CASE_NAME="$2"; shift 2 ;;
--pid-root) PID_ROOT="$2"; shift 2 ;;
-h|--help) usage; exit 0 ;;
*) die "unknown option: $1" ;;
esac
done
[[ -n "$FIXTURE_ROOT" ]] || die "--fixture-root is required"
FIXTURE_ROOT="$(cd "$FIXTURE_ROOT" 2>/dev/null && pwd)" || die "fixture root not found"
CASE_ROOT="$FIXTURE_ROOT/cases/$CASE_NAME"
[[ -d "$CASE_ROOT/repository" && -d "$CASE_ROOT/http" ]] || die "fixture case not found: $CASE_NAME"
if [[ -z "$PID_ROOT" ]]; then
PID_ROOT="$FIXTURE_ROOT/services/$CASE_NAME"
fi
mkdir -p "$PID_ROOT"
[[ ! -f "$PID_ROOT/pids.env" ]] || die "services already appear to be running: $PID_ROOT/pids.env"
rm -f "$PID_ROOT/rsyncd.pid" "$PID_ROOT/rsyncd.lock"
RSYNCD_CONF="$PID_ROOT/rsyncd.conf"
RSYNC_UID="$(id -un)"
RSYNC_GID="$(id -gn)"
cat > "$RSYNCD_CONF" <<EOF
uid = $RSYNC_UID
gid = $RSYNC_GID
use chroot = no
read only = yes
port = $RSYNC_PORT
pid file = $PID_ROOT/rsyncd.pid
lock file = $PID_ROOT/rsyncd.lock
log file = $PID_ROOT/rsyncd.log
[custom]
path = $CASE_ROOT/repository
comment = #142 fixed custom RPKI repository
read only = yes
EOF
rsync --daemon --no-detach --config="$RSYNCD_CONF" >"$PID_ROOT/rsyncd.stdout.log" 2>&1 &
RSYNC_PID=$!
python3 "$SCRIPT_DIR/serve_https.py" \
--directory "$CASE_ROOT/http" \
--certfile "$FIXTURE_ROOT/certs/rrdp-server.pem" \
--keyfile "$FIXTURE_ROOT/certs/rrdp-server.key" \
--port "$RRDP_PORT" \
>"$PID_ROOT/https.log" 2>&1 &
HTTPS_PID=$!
printf 'case=%s\nrrdp_pid=%s\nrsync_pid=%s\nrrdp_port=%s\nrsync_port=%s\n' \
"$CASE_NAME" "$HTTPS_PID" "$RSYNC_PID" "$RRDP_PORT" "$RSYNC_PORT" > "$PID_ROOT/pids.env"
sleep 1
kill -0 "$HTTPS_PID" 2>/dev/null || die "HTTPS fixture server exited; see $PID_ROOT/https.log"
kill -0 "$RSYNC_PID" 2>/dev/null || die "rsync fixture server exited; see $PID_ROOT/rsyncd.stdout.log"
echo "fixed RRDP/rsync services started case=$CASE_NAME rrdp=$RRDP_PORT rsync=$RSYNC_PORT pid_root=$PID_ROOT"

23
deploy/docker-installer/custom-tal-patch/tools/stop_fixed_services.sh
View File

@@ -1,23 +0,0 @@
#!/usr/bin/env bash
set -euo pipefail
PID_ROOT="$1"
[[ -n "$PID_ROOT" ]] || { echo "Usage: $0 <pid-root>" >&2; exit 2; }
[[ -f "$PID_ROOT/pids.env" ]] || { echo "services are not running: $PID_ROOT" >&2; exit 2; }
# shellcheck disable=SC1090
source "$PID_ROOT/pids.env"
service_pids=("${rrdp_pid:-}" "${rsync_pid:-}")
for pid in "${service_pids[@]}"; do
if [[ "$pid" =~ ^[0-9]+$ ]] && kill -0 "$pid" 2>/dev/null; then
kill "$pid" 2>/dev/null || true
fi
done
for pid in "${service_pids[@]}"; do
[[ "$pid" =~ ^[0-9]+$ ]] || continue
for _ in $(seq 1 50); do
kill -0 "$pid" 2>/dev/null || break
sleep 0.1
done
done
rm -f "$PID_ROOT/pids.env"
echo "fixed RRDP/rsync services stopped: $PID_ROOT"

146
scripts/docker/build_custom_tal_patch.sh
View File

@@ -1,146 +0,0 @@
#!/usr/bin/env bash
set -euo pipefail
SCRIPT_DIR="$(cd "$(dirname "$BASH_SOURCE")" && pwd)"
REPO_ROOT="$(cd "$SCRIPT_DIR/../.." && pwd)"
PATCH_SOURCE="$REPO_ROOT/deploy/docker-installer/custom-tal-patch"
BASE_COMPONENT_PACKAGE=""
OUT_DIR="$(printenv OUT_DIR 2>/dev/null || true)"
PREFIX="$(printenv PREFIX 2>/dev/null || true)"
ALLOW_DIRTY=0
[[ -n "$OUT_DIR" ]] || OUT_DIR="$REPO_ROOT/target/custom-tal-patch"
[[ -n "$PREFIX" ]] || PREFIX="ours-rp-custom-tal-patch"
usage() {
cat <<'USAGE'
Usage:
scripts/docker/build_custom_tal_patch.sh \
--base-component-package <ours-rp-installer-*.tar.gz> \
[--out-dir <dir>] [--prefix <name>] [--allow-dirty]
The patch targets the Arm64 component package from the customer stack. It
overlays the runner, Compose, fixture mount and test-service tools; it does not
rebuild or replace the runtime image.
USAGE
}
die() {
echo "error: $*" >&2
exit 2
}
while [[ $# -gt 0 ]]; do
case "$1" in
--base-component-package) BASE_COMPONENT_PACKAGE="$2"; shift 2 ;;
--out-dir) OUT_DIR="$2"; shift 2 ;;
--prefix) PREFIX="$2"; shift 2 ;;
--allow-dirty) ALLOW_DIRTY=1; shift ;;
-h|--help) usage; exit 0 ;;
*) die "unknown option: $1" ;;
esac
done
[[ -f "$BASE_COMPONENT_PACKAGE" ]] || die "missing base component package: $BASE_COMPONENT_PACKAGE"
[[ -d "$PATCH_SOURCE" ]] || die "missing patch source: $PATCH_SOURCE"
SOURCE_COMMIT="$(git -C "$REPO_ROOT" rev-parse HEAD)"
SOURCE_COMMIT_SHORT="$(git -C "$REPO_ROOT" rev-parse --short=8 HEAD)"
SOURCE_DIRTY=false
if [[ -n "$(git -C "$REPO_ROOT" status --short)" ]]; then
SOURCE_DIRTY=true
fi
if [[ "$SOURCE_DIRTY" == true && "$ALLOW_DIRTY" != 1 ]]; then
die "source worktree is dirty; use --allow-dirty for a development patch"
fi
package_root="$(tar -tzf "$BASE_COMPONENT_PACKAGE" | awk -F/ '!seen { print $1; seen = 1 }')"
[[ -n "$package_root" ]] || die "cannot determine component package root"
manifest_text="$(tar -xOf "$BASE_COMPONENT_PACKAGE" "$package_root/PACKAGE-MANIFEST.env")" \
|| die "base component package has no PACKAGE-MANIFEST.env"
read_manifest_value() {
printf '%s\n' "$manifest_text" | awk -F= -v key="$1" '$1 == key { print substr($0, index($0, "=") + 1); exit }'
}
base_source_commit="$(read_manifest_value source_commit)"
base_package_arch="$(read_manifest_value package_arch)"
base_runtime_image="$(read_manifest_value runtime_image)"
[[ "$base_source_commit" == 9f4f4cf069e9fa8f6f97ed0f1065adac38abe82f ]] \
|| die "base source commit mismatch: $base_source_commit"
[[ "$base_package_arch" == arm64 ]] || die "base package must be arm64: $base_package_arch"
timestamp="$(date -u +%Y%m%dT%H%M%SZ)"
patch_name="$PREFIX-$SOURCE_COMMIT_SHORT"
if [[ "$SOURCE_DIRTY" == true ]]; then
patch_name="$patch_name-dirty"
fi
patch_name="$patch_name-$timestamp"
stage="$OUT_DIR/$patch_name"
tar_path="$OUT_DIR/$patch_name.tar.gz"
base_sha256="$(sha256sum "$BASE_COMPONENT_PACKAGE" | awk '{ print $1 }')"
rm -rf "$stage"
mkdir -p "$stage/payload/compose" "$stage/payload/scripts/soak" \
"$stage/payload/custom-fixtures" "$stage/payload/tools"
cp "$PATCH_SOURCE/apply_custom_tal_patch.sh" "$stage/"
cp "$PATCH_SOURCE/rollback_custom_tal_patch.sh" "$stage/"
cp "$PATCH_SOURCE/custom-tal.env.example" "$stage/"
cp "$REPO_ROOT/deploy/docker-installer/compose/docker-compose.yml" "$stage/payload/compose/"
cp "$REPO_ROOT/scripts/soak/run_soak.sh" "$stage/payload/scripts/soak/"
cp "$REPO_ROOT/deploy/docker-installer/custom-fixtures/README.md" "$stage/payload/custom-fixtures/"
cp "$PATCH_SOURCE/tools/generate_custom_fixture.py" "$stage/payload/tools/"
cp "$PATCH_SOURCE/tools/serve_https.py" "$stage/payload/tools/"
cp "$PATCH_SOURCE/tools/start_fixed_services.sh" "$stage/payload/tools/"
cp "$PATCH_SOURCE/tools/stop_fixed_services.sh" "$stage/payload/tools/"
chmod +x "$stage/apply_custom_tal_patch.sh" "$stage/rollback_custom_tal_patch.sh" \
"$stage/payload/scripts/soak/run_soak.sh" "$stage/payload/tools/"*.sh \
"$stage/payload/tools/"*.py
{
echo "patch_schema_version=1"
echo "patch_name=$patch_name"
echo "created_at_utc=$timestamp"
echo "source_commit=$SOURCE_COMMIT"
echo "source_commit_short=$SOURCE_COMMIT_SHORT"
echo "source_dirty=$SOURCE_DIRTY"
echo "base_component_archive=$(basename "$BASE_COMPONENT_PACKAGE")"
echo "base_component_archive_sha256=$base_sha256"
echo "base_component_root=$package_root"
echo "base_source_commit=$base_source_commit"
echo "base_package_arch=$base_package_arch"
echo "base_runtime_image=$base_runtime_image"
echo "overlay_runtime_image_unchanged=true"
echo "custom_mode=RIRS=custom,TAL_INPUT_MODE=custom-file-with-ta"
echo "rrdp_port=18443"
echo "rsync_port=1873"
echo "payload_root=payload"
while IFS= read -r relative_path; do
safe_name="$(printf '%s' "$relative_path" | tr '/.' '__')"
file_sha256="$(sha256sum "$stage/payload/$relative_path" | awk '{ print $1 }')"
echo "payload_"$safe_name"_sha256=$file_sha256"
done < <(cd "$stage/payload" && find . -type f -printf '%P\n' | sort)
} > "$stage/PATCH-MANIFEST.env"
cat > "$stage/PATCH-SUMMARY.txt" <<EOF
patch_name: $patch_name
base_component_archive: $(basename "$BASE_COMPONENT_PACKAGE")
base_component_archive_sha256: $base_sha256
base_source_commit: $base_source_commit
source_commit: $SOURCE_COMMIT
source_dirty: $SOURCE_DIRTY
runtime_image: $base_runtime_image
runtime_image_changed: false
custom_fixture_generator: payload/tools/generate_custom_fixture.py
fixed_services: payload/tools/start_fixed_services.sh
EOF
tar -C "$OUT_DIR" -czf "$tar_path" "$patch_name"
patch_sha256="$(sha256sum "$tar_path" | awk '{ print $1 }')"
{
echo "patch=$tar_path"
echo "patch_dir=$stage"
echo "patch_sha256=$patch_sha256"
echo "manifest=$stage/PATCH-MANIFEST.env"
echo "base_component_archive_sha256=$base_sha256"
echo "source_commit=$SOURCE_COMMIT"
echo "source_dirty=$SOURCE_DIRTY"
} > "$OUT_DIR/$patch_name.summary.env"
echo "patch built: $tar_path"

24
scripts/rpctl/README.md
View File

@@ -1,18 +1,22 @@
# rpctl
`rpctl` 是四款 RP 本地控制 CLI，用于在实验服务器本地独立启动、停止和观察单个 RP/profile 的 all5 实验。一个实验可包含一次 snapshot 和多次 delta 子 run。
`rpctl` 是五款 RP 本地控制 CLI，用于在实验服务器本地独立启动、停止和观察单个 RP/profile 的 all5 实验。一个实验可包含一次 snapshot 和多次复用 state 的子 run。
## 部署
```bash
mkdir -p /root/rpki_4rp_ctl
rsync -a rpki_2/rpki/scripts/rpctl/ root@47.77.237.41:/root/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/
ssh root@47.77.237.41 'cd /root/rpki_4rp_ctl && ./rpctl init --bundle /root/rpki_4rp_ctl_runtime_bundle'
mkdir -p /data/rpki_4rp_ctl
rsync -a rpki_2/rpki/scripts/rpctl/ root@47.251.127.231:/data/rpki_4rp_ctl/
rsync -a artifacts/rpki-prover/rpki-prover-6857d4bf-linux-amd64.docker.tar.gz root@47.251.127.231:/data/
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 运行所需动态库。
`--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
@@ -20,6 +24,8 @@ ssh root@47.77.237.41 'cd /root/rpki_4rp_ctl && ./rpctl init --bundle /root/rpki
./rpctl status
./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 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 tail
./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` |
| `rpki-client` | `rpki-client-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/
```
每个子 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。
## 历史查看与删除

471
scripts/rpctl/rpctl
View File

@@ -4,8 +4,10 @@ from __future__ import annotations
import argparse
import csv
import gzip
import hashlib
import json
import os
import platform
import re
import shutil
import signal
@@ -37,6 +39,8 @@ PROFILES = {
"routinator-latest-release": "routinator",
"rpki-client-latest-release": "rpki-client",
"fort-latest-release": "fort",
"rpki-prover-incremental": "rpki-prover",
"rpki-prover-full": "rpki-prover",
}
ALIASES = {
@@ -52,8 +56,15 @@ ALIASES = {
"rpki-client-latest-release": "rpki-client-latest-release",
"fort": "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 = {
"root": "/root/rpki_4rp_ctl",
"bundle": "",
@@ -207,6 +218,14 @@ def command_string(argv: list[str]) -> str:
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:
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()]
@@ -232,6 +251,102 @@ def install_from_bundle(bundle: Path) -> None:
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:
ensure_dirs()
bundle = Path(args.bundle).resolve()
@@ -239,9 +354,16 @@ def cmd_init(args: argparse.Namespace) -> None:
cfg = load_config()
cfg["bundle"] = str(bundle)
cfg["initializedAtUtc"] = utc_now()
if args.rpki_prover_image:
cfg["rpkiProverImage"] = install_rpki_prover_image(Path(args.rpki_prover_image).resolve())
save_config(cfg)
print(f"initialized root={ctl_root()}")
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:
@@ -396,6 +518,240 @@ def build_fort_args(run_dir: Path, state: Path, rirs: list[str]) -> tuple[list[s
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:
script = ctl_root() / "scripts" / "normalize_and_collect.py"
if not script.exists():
@@ -460,6 +816,18 @@ def parse_entry_rirs(entry: dict[str, Any]) -> list[str]:
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:
ensure_dirs()
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
(run_root / "logs").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()
if append:
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),
"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)
status = {
"state": "running",
@@ -535,7 +906,7 @@ def run_profile(profile: str, exp_id: str, runs: int, interval_secs: int, rirs:
time.sleep(sleep_seconds)
last_start = time.time()
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}"
if run_dir.exists() and any(run_dir.iterdir()):
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)
elif rp == "fort":
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:
raise RuntimeError(rp)
run_dir.mkdir(parents=True, exist_ok=True)
(run_dir / "command.txt").write_text(command_string(argv) + "\n", encoding="utf-8")
code = timed_run(run_dir, argv, env=env)
normalize_one(run_root, profile, run_seq)
meta = parse_latest_meta(run_root, profile, run_seq)
if rp == "rpki-prover":
code, max_rss_kb, max_process_rss_kb, wall_ms, _ = timed_rpki_prover_run(
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(
state="running",
lastRunSeq=run_seq,
@@ -571,7 +963,10 @@ def run_profile(profile: str, exp_id: str, runs: int, interval_secs: int, rirs:
if code != 0:
exit_code = code
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"
update_current(state=final_state, finishedAtUtc=utc_now(), exitCode=exit_code)
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"logs={logs_dir()}")
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()
if cur.get("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")
if not (ctl_root() / "versions.json").exists():
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"
runner_args = [
sys.executable,
@@ -717,6 +1118,8 @@ def cmd_continue(args: argparse.Namespace) -> None:
if entry["state"] != "success":
raise SystemExit(f"can only continue a successful experiment: exp_id={args.exp_id} state={entry['state']}")
profile = entry["profile"]
if PROFILES[profile] == "rpki-prover":
require_rpki_prover_image()
rirs = parse_entry_rirs(entry)
interval_secs = parse_interval(args.interval)
runner_args = [
@@ -867,6 +1270,9 @@ def parse_command_summary(command: str) -> str:
keys.append(f"--rsync-scope={tokens[idx + 1]}")
elif token.startswith("--rsync-scope="):
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)
@@ -945,6 +1351,10 @@ def cache_profile_name(profile: str) -> str:
return "pp+roa+crt"
if profile == "ours-no-cache":
return "off"
if profile == "rpki-prover-incremental":
return "incremental"
if profile == "rpki-prover-full":
return "full-every-iteration"
return "n/a"
@@ -971,6 +1381,8 @@ def profile_label(profile: str) -> str:
"routinator-latest-release": "routinator",
"rpki-client-latest-release": "rpki-client",
"fort-latest-release": "fort",
"rpki-prover-incremental": "prover-inc",
"rpki-prover-full": "prover-full",
}
return labels.get(profile, profile)
@@ -1043,6 +1455,9 @@ def infer_output_counts(profile: str, run_dir: Path) -> tuple[str, str]:
return vrps, vaps
if rp == "fort":
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 "", ""
@@ -1355,6 +1770,7 @@ def fallback_run_details(entry: dict[str, Any]) -> list[dict[str, Any]]:
"wall_ms": timing.get("wallMs", ""),
"max_rss_kb": timing.get("maxRssKb", ""),
"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),
"remote_run_dir": str(run_dir),
})
@@ -1391,7 +1807,7 @@ def cmd_view(args: argparse.Namespace) -> None:
print("command=" + entry["command"])
if rows:
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 = []
for row in rows:
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("max_rss_kb"), row.get("maxRssKb")),
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("remote_run_dir"), row.get("remoteRunDir")),
])
@@ -1412,7 +1829,7 @@ def cmd_view(args: argparse.Namespace) -> None:
def build_parser() -> argparse.ArgumentParser:
formatter = argparse.RawDescriptionHelpFormatter
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,
epilog="""
RP/profile naming:
@@ -1435,6 +1852,14 @@ RP/profile naming:
fort
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:
--rp ours adds:
--enable-publication-point-validation-cache
@@ -1444,10 +1869,12 @@ Cache difference for ours RP:
--rp ours-no-cache does not add those flags.
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 start --rp ours --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 show
./rpctl show --failed
@@ -1458,7 +1885,8 @@ Typical usage:
Data model:
- 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.
""",
)
@@ -1481,14 +1909,24 @@ The bundle must contain:
scripts/write_group_report.py
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:
/root/rpki_4rp_ctl_runtime_bundle
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(
"--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 = sub.add_parser(
@@ -1502,6 +1940,10 @@ Use this command when unsure which value to pass to --rp.
Important ours RP profiles:
ours-pp-object-cache cache-enabled 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)
@@ -1525,6 +1967,8 @@ RP/profile examples:
./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 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:
--rp ours enables PP cache + ROA cache + child certificate cache.
@@ -1537,7 +1981,7 @@ Safety:
p.add_argument(
"--rp",
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(
"--interval",
@@ -1578,6 +2022,7 @@ Run semantics:
Examples:
./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 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.")
@@ -1692,7 +2137,7 @@ Examples:
./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("--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.")

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()

8
scripts/soak/build_portable_soak_package.sh
View File

@@ -14,7 +14,7 @@ Usage:
scripts/soak/build_portable_soak_package.sh [--out-dir <path>] [--profile <profile>]
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
}
@@ -53,7 +53,7 @@ else
TARGET_BIN_DIR="$REPO_ROOT/target/$PROFILE"
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=(
ccr_dump
ccr_state_compare
@@ -82,7 +82,7 @@ STAGE_DIR="$BUILD_ROOT/$PACKAGE_DIR_NAME"
ARCHIVE_PATH="$OUT_DIR/$PACKAGE_NAME.tar.gz"
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"
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/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 "$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/portable-soak.env.example"

125
scripts/soak/run_soak.sh
View File

@@ -15,10 +15,6 @@ INTERVAL_SECS="${INTERVAL_SECS:-0}"
STOP_AFTER_SECS="${STOP_AFTER_SECS:-0}"
RIRS="${RIRS:-afrinic,apnic,arin,lacnic,ripe}"
TAL_INPUT_MODE="${TAL_INPUT_MODE:-file-with-ta}"
CUSTOM_TAL_PATH="${CUSTOM_TAL_PATH:-$PACKAGE_ROOT/custom-fixtures/tal/custom.tal}"
CUSTOM_TA_PATH="${CUSTOM_TA_PATH:-$PACKAGE_ROOT/custom-fixtures/ta/custom-ta.cer}"
CUSTOM_TAL_URI="${CUSTOM_TAL_URI:-https://host.docker.internal:18443/tal/custom.tal}"
HTTP_ROOT_CERT_PATHS="${HTTP_ROOT_CERT_PATHS:-}"
RESOURCE_VALIDATION_MODE="${RESOURCE_VALIDATION_MODE:-validation-update-03}"
RUN_ROOT="${RUN_ROOT:-$PACKAGE_ROOT}"
RETAIN_RUNS="${RETAIN_RUNS:-10}"
@@ -52,6 +48,7 @@ TMP_DIR="${TMP_DIR:-$RUN_ROOT/tmp}"
RSYNC_MIRROR_ROOT="${RSYNC_MIRROR_ROOT:-$STATE_ROOT/rsync-mirror}"
INVALID_ROOT="$STATE_ROOT/invalid"
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_CONNECT_TIMEOUT_SECS="${LIVE_TA_REFRESH_CONNECT_TIMEOUT_SECS:-15}"
LIVE_TA_REFRESH_MAX_TIME_SECS="${LIVE_TA_REFRESH_MAX_TIME_SECS:-120}"
@@ -122,10 +119,10 @@ validate_rsync_scope() {
validate_tal_input_mode() {
case "$TAL_INPUT_MODE" in
file-with-ta|file-live-ta|custom-file-with-ta|url)
file-with-ta|file-live-ta|url)
;;
*)
die "TAL_INPUT_MODE must be file-with-ta, file-live-ta, custom-file-with-ta or url: $TAL_INPUT_MODE"
die "TAL_INPUT_MODE must be file-with-ta, file-live-ta or url: $TAL_INPUT_MODE"
;;
esac
}
@@ -159,46 +156,14 @@ parse_rirs() {
afrinic|apnic|arin|lacnic|ripe)
RIR_LIST+=("$normalized")
;;
custom)
[[ "$TAL_INPUT_MODE" == "custom-file-with-ta" ]] \
|| die "RIRS=custom requires TAL_INPUT_MODE=custom-file-with-ta"
RIR_LIST+=("$normalized")
;;
*)
die "invalid RIRS entry: $raw_token; allowed: afrinic,apnic,arin,lacnic,ripe,custom (custom mode only)"
die "invalid RIRS entry: $raw_token; allowed: afrinic,apnic,arin,lacnic,ripe"
;;
esac
done
[[ "${#RIR_LIST[@]}" -gt 0 ]] || die "RIRS must contain at least one RIR"
}
validate_custom_inputs() {
local root_cert_path
local -a root_cert_paths=()
if [[ "$TAL_INPUT_MODE" == "custom-file-with-ta" ]]; then
[[ "${#RIR_LIST[@]}" -eq 1 && "${RIR_LIST[0]}" == "custom" ]] \
|| die "custom-file-with-ta requires exactly RIRS=custom"
[[ -s "$CUSTOM_TAL_PATH" ]] \
|| die "missing custom TAL: $CUSTOM_TAL_PATH"
[[ -s "$CUSTOM_TA_PATH" ]] \
|| die "missing custom TA DER: $CUSTOM_TA_PATH"
[[ -n "$CUSTOM_TAL_URI" ]] \
|| die "CUSTOM_TAL_URI is required in custom-file-with-ta mode"
elif [[ "$RIRS" == *custom* ]]; then
die "RIRS=custom requires TAL_INPUT_MODE=custom-file-with-ta"
fi
if [[ -n "$HTTP_ROOT_CERT_PATHS" ]]; then
# shellcheck disable=SC2206
root_cert_paths=( $HTTP_ROOT_CERT_PATHS )
for root_cert_path in "${root_cert_paths[@]}"; do
[[ -s "$root_cert_path" ]] \
|| die "missing HTTP root certificate: $root_cert_path"
done
fi
}
tal_file_for_rir() {
case "$1" in
afrinic) printf '%s' "$FIXTURE_DIR/tal/afrinic.tal" ;;
@@ -206,7 +171,6 @@ tal_file_for_rir() {
arin) printf '%s' "$FIXTURE_DIR/tal/arin.tal" ;;
lacnic) printf '%s' "$FIXTURE_DIR/tal/lacnic.tal" ;;
ripe) printf '%s' "$FIXTURE_DIR/tal/ripe-ncc.tal" ;;
custom) printf '%s' "$CUSTOM_TAL_PATH" ;;
*) die "unknown RIR: $1" ;;
esac
}
@@ -218,7 +182,6 @@ ta_file_for_rir() {
arin) printf '%s' "$FIXTURE_DIR/ta/arin-ta.cer" ;;
lacnic) printf '%s' "$FIXTURE_DIR/ta/lacnic-ta.cer" ;;
ripe) printf '%s' "$FIXTURE_DIR/ta/ripe-ncc-ta.cer" ;;
custom) printf '%s' "$CUSTOM_TA_PATH" ;;
*) die "unknown RIR: $1" ;;
esac
}
@@ -230,7 +193,6 @@ tal_url_for_rir() {
arin) printf '%s' "https://www.arin.net/resources/manage/rpki/arin.tal" ;;
lacnic) printf '%s' "https://www.lacnic.net/innovaportal/file/4983/1/lacnic.tal" ;;
ripe) printf '%s' "https://tal.rpki.ripe.net/ripe-ncc.tal" ;;
custom) printf '%s' "$CUSTOM_TAL_URI" ;;
*) die "unknown RIR: $1" ;;
esac
}
@@ -1096,16 +1058,6 @@ build_child_args() {
CHILD_ARGS+=(--rsync-mirror-root "$TMP_DIR/rsync-mirror-{run_id}")
fi
local root_cert_path
local -a root_cert_paths=()
if [[ -n "$HTTP_ROOT_CERT_PATHS" ]]; then
# shellcheck disable=SC2206
root_cert_paths=( $HTTP_ROOT_CERT_PATHS )
for root_cert_path in "${root_cert_paths[@]}"; do
CHILD_ARGS+=(--http-root-cert "$root_cert_path")
done
fi
CHILD_ARGS+=(
--parallel-phase2-ready-batch-size 256
--parallel-phase2-ready-batch-wall-time-budget-ms 100
@@ -1129,6 +1081,7 @@ build_child_args() {
CHILD_ARGS+=(
--report-json "{run_out}/report.json"
--validation-contract-out "{run_out}/validation-contract.json"
)
if is_true "$OUTPUT_COMPACT_REPORT"; then
CHILD_ARGS+=(--report-json-compact)
@@ -1158,6 +1111,59 @@ build_child_args() {
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() {
local daemon_state_root="$1"
local run_dir="$2"
@@ -1353,6 +1359,12 @@ run_one_round() {
final_status="failed"
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" \
"$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" \
@@ -1376,6 +1388,7 @@ main() {
require_command python3
require_command date
require_command find
require_command flock
if [[ "$TAL_INPUT_MODE" == "file-live-ta" ]]; then
require_command curl
validate_positive_int "LIVE_TA_REFRESH_CONNECT_TIMEOUT_SECS" "$LIVE_TA_REFRESH_CONNECT_TIMEOUT_SECS"
@@ -1393,7 +1406,6 @@ main() {
validate_positive_int "DB_STATS_EXACT_EVERY" "$DB_STATS_EXACT_EVERY"
fi
parse_rirs
validate_custom_inputs
[[ -x "$RPKI_BIN" ]] || die "missing executable: $RPKI_BIN"
[[ -x "$RPKI_DAEMON_BIN" ]] || die "missing executable: $RPKI_DAEMON_BIN"
@@ -1410,7 +1422,7 @@ main() {
fi
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
mkdir -p "$RSYNC_MIRROR_ROOT"
fi
@@ -1436,6 +1448,9 @@ main() {
local any_failed=0
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_STATE_PATH=""
INVALID_TMP_PATH=""
@@ -1522,6 +1537,8 @@ main() {
echo "completed run $(printf 'run_%04d' "$next_index") status=failed" >&2
any_failed=1
fi
flock -u "$state_lock_fd"
exec {state_lock_fd}>&-
if (( STOP_AFTER_SECS > 0 )); then
elapsed_secs=$(( $(date +%s) - started_epoch ))
if (( elapsed_secs >= STOP_AFTER_SECS )); then

50
scripts/verification/README.md Normal file
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，按不匹配状态集合继续排查。

102
scripts/verification/run_verification_only.sh Executable file
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"

171
src/bin/verification_ccr_compare.rs Normal file
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")
);
}
}

45
src/blob_store.rs
View File

@@ -77,6 +77,7 @@ pub struct ExternalRawStoreDb {
pub struct ExternalRepoBytesDb {
path: PathBuf,
db: Arc<DB>,
read_only: bool,
}
impl ExternalRawStoreDb {
@@ -189,6 +190,7 @@ impl ExternalRepoBytesDb {
Ok(Self {
path,
db: Arc::new(db),
read_only: false,
})
}
@@ -201,6 +203,7 @@ impl ExternalRepoBytesDb {
Ok(Self {
path,
db: Arc::new(db),
read_only: true,
})
}
@@ -208,6 +211,12 @@ impl ExternalRepoBytesDb {
if blobs.is_empty() {
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();
for (sha256_hex, bytes) in blobs {
validate_blob_sha256_hex(sha256_hex)?;
@@ -221,6 +230,42 @@ impl ExternalRepoBytesDb {
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>>> {
validate_blob_sha256_hex(sha256_hex)?;
let key = repo_bytes_key(sha256_hex);

352
src/cli.rs
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_with_timing,
};
use crate::validation::tree::TreeRunConfig;
use crate::validation::tree::{DEFAULT_MAX_CA_DEPTH, TreeRunConfig};
#[cfg(test)]
use output::write_json;
use output::{
@@ -115,6 +115,12 @@ fn vcir_storage_summary_enabled() -> bool {
#[derive(Clone, Debug, PartialEq, Eq)]
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_paths: Vec<PathBuf>,
pub ta_paths: Vec<PathBuf>,
@@ -168,7 +174,7 @@ pub struct CliArgs {
pub rsync_mirror_root: Option<PathBuf>,
pub rsync_scope_policy: RsyncScopePolicy,
pub max_depth: Option<usize>,
pub max_ca_depth: usize,
pub max_instances: Option<usize>,
pub validation_time: Option<time::OffsetDateTime>,
@@ -184,8 +190,19 @@ fn usage() -> String {
Usage:
{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} --verification-only --verification-source-run <path> --verification-source-state <path> --verification-out <path> [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)
--raw-store-db <path> External raw-by-hash store 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-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)
--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
--validation-time <rfc3339> Validation time in RFC3339 (default: now UTC)
--analyze Write timing analysis JSON under target/live/analyze/<timestamp>/
@@ -272,6 +290,12 @@ Options:
}
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_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_mirror_root: Option<PathBuf> = None;
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 validation_time: Option<time::OffsetDateTime> = None;
let mut analyze: bool = false;
@@ -331,6 +356,39 @@ pub fn parse_args(argv: &[String]) -> Result<CliArgs, String> {
let arg = argv[i].as_str();
match arg {
"--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" => {
i += 1;
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")?;
rsync_scope_policy = RsyncScopePolicy::parse_cli_value(v)?;
}
"--max-depth" => {
"--max-ca-depth" | "--max-depth" => {
i += 1;
let v = argv.get(i).ok_or("--max-depth requires a value")?;
max_depth = Some(
let option = arg;
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>()
.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" => {
i += 1;
@@ -702,10 +773,91 @@ pub fn parse_args(argv: &[String]) -> Result<CliArgs, String> {
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 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!(
"must specify either one-or-more --tal-url or one-or-more --tal-path/--ta-path pairs\n\n{}",
usage()
@@ -759,13 +911,13 @@ pub fn parse_args(argv: &[String]) -> Result<CliArgs, String> {
usage()
));
}
if !tal_urls.is_empty() && !ta_paths.is_empty() {
if !verification_only && !tal_urls.is_empty() && !ta_paths.is_empty() {
return Err(format!(
"--ta-path cannot be used with --tal-url mode\n\n{}",
usage()
));
}
if !tal_paths.is_empty() {
if !verification_only && !tal_paths.is_empty() {
if !ta_paths.is_empty() {
if ta_paths.len() != tal_paths.len() {
return Err(format!(
@@ -940,6 +1092,12 @@ pub fn parse_args(argv: &[String]) -> Result<CliArgs, String> {
}
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_paths,
ta_paths,
@@ -989,7 +1147,7 @@ pub fn parse_args(argv: &[String]) -> Result<CliArgs, String> {
rsync_timeout_secs,
rsync_mirror_root,
rsync_scope_policy,
max_depth,
max_ca_depth: max_ca_depth.unwrap_or(DEFAULT_MAX_CA_DEPTH),
max_instances,
validation_time,
analyze,
@@ -1854,7 +2012,7 @@ where
H: crate::sync::rrdp::Fetcher + 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 {
run_tree_from_multiple_tals_parallel_phase2_audit_with_timing(
store,
@@ -1995,21 +2153,88 @@ where
}
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())?;
if let Some(strict_policy) = args.strict_policy {
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 {
let mut policy = if let Some(prepared) = prepared_verification.as_ref() {
prepared.contract.policy.clone()
} else {
read_policy(args.policy_path.as_deref())?
};
if !args.verification_only {
if let Some(strict_policy) = args.strict_policy {
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.ca_failed_fetch_policy = crate::policy::CaFailedFetchPolicy::StopAllOutput;
}
let validation_time = args
.validation_time
.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
.http_root_cert_paths
.iter()
@@ -2019,7 +2244,12 @@ pub fn run(argv: &[String]) -> Result<(), String> {
})
.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(
RocksStore::open_with_external_stores(
&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())?)
};
let config = TreeRunConfig {
max_depth: args.max_depth,
max_depth: Some(args.max_ca_depth),
max_instances: args.max_instances,
compact_audit: args.skip_report_build
&& 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 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
.tal_path
.as_ref()
@@ -2651,6 +2897,62 @@ pub fn run(argv: &[String]) -> Result<(), String> {
}
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(())
}

106
src/cli/tests.rs
View File

@@ -21,6 +21,8 @@ fn parse_help_returns_usage() {
assert!(err.contains("--memory-trim-after-validation"), "{err}");
assert!(err.contains("--enable-roa-validation-cache"), "{err}");
assert!(err.contains("--resource-validation-mode"), "{err}");
assert!(err.contains("--max-ca-depth"), "{err}");
assert!(err.contains("default: 32"), "{err}");
assert!(
err.contains("--publication-point-cache-observe-only"),
"{err}"
@@ -47,6 +49,63 @@ fn parse_rejects_unknown_argument() {
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]
fn parse_rejects_both_tal_url_and_tal_path() {
let argv = vec![
@@ -80,6 +139,51 @@ fn parse_rejects_invalid_max_depth() {
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]
fn parse_accepts_ccr_out_path() {
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.tal_path.as_deref(), Some(Path::new("a.tal")));
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]

178
src/fetch/current_repository.rs Normal file
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"));
}
}

1
src/fetch/mod.rs
View File

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

2
src/lib.rs
View File

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

152
src/storage.rs
View File

@@ -8,6 +8,7 @@ use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};
use base64::Engine;
use rocksdb::checkpoint::Checkpoint;
use rocksdb::{ColumnFamily, DB, Direction, IteratorMode, Options, WriteBatch};
use serde::{Deserialize, Serialize};
@@ -53,6 +54,13 @@ pub enum 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)]
pub struct ChildCertificateCacheMmapLookup {
pub projections: Vec<Option<ChildCertificateCacheProjection>>,
@@ -595,7 +603,7 @@ pub struct VcirFailedFetchReuseIdentity {
pub ta_context_digest: [u8; 32],
#[serde(rename = "p")]
#[serde(with = "serde_bytes_32")]
pub parent_context_digest: [u8; 32],
pub ca_validation_context_digest: [u8; 32],
#[serde(rename = "f")]
#[serde(with = "serde_bytes_32")]
pub policy_fingerprint: [u8; 32],
@@ -1117,7 +1125,7 @@ pub struct RoaCacheObjectMeta {
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct RoaCacheProjectionContext {
pub parent_context_digest: [u8; 32],
pub ca_validation_context_digest: [u8; 32],
pub policy_fingerprint: [u8; 32],
pub object_meta: Vec<RoaCacheObjectMeta>,
}
@@ -1258,7 +1266,7 @@ pub struct RoaCacheProjection {
pub issuer_ca_sha256_hex: Option<String>,
#[serde(rename = "p", default, skip_serializing_if = "Option::is_none")]
#[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(with = "serde_optional_bytes_32")]
pub policy_fingerprint: Option<[u8; 32]>,
@@ -1400,7 +1408,7 @@ pub struct ChildCertificateCacheProjection {
pub issuer_crl_sha256_hex: String,
#[serde(rename = "pc")]
#[serde(with = "serde_bytes_32")]
pub parent_context_digest: [u8; 32],
pub ca_validation_context_digest: [u8; 32],
#[serde(rename = "pf")]
#[serde(with = "serde_bytes_32")]
pub validation_policy_fingerprint: [u8; 32],
@@ -1707,7 +1715,7 @@ pub struct PublicationPointCacheProjection {
pub ta_context_digest: [u8; 32],
#[serde(rename = "pc")]
#[serde(with = "serde_bytes_32")]
pub parent_context_digest: [u8; 32],
pub ca_validation_context_digest: [u8; 32],
#[serde(rename = "pf")]
#[serde(with = "serde_bytes_32")]
pub validation_policy_fingerprint: [u8; 32],
@@ -1743,7 +1751,7 @@ impl PublicationPointCacheProjection {
ca_cert_sha256: [u8; 32],
manifest_sha256: [u8; 32],
ta_context_digest: [u8; 32],
parent_context_digest: [u8; 32],
ca_validation_context_digest: [u8; 32],
validation_policy_fingerprint: [u8; 32],
) -> StorageResult<Self> {
let projection = Self {
@@ -1756,7 +1764,7 @@ impl PublicationPointCacheProjection {
manifest_sha256,
tal_id: vcir.tal_id.clone(),
ta_context_digest,
parent_context_digest,
ca_validation_context_digest,
validation_policy_fingerprint,
instance_effective_not_before: vcir.last_successful_validation_time.clone(),
instance_effective_until: vcir.instance_gate.instance_effective_until.clone(),
@@ -1811,7 +1819,7 @@ impl PublicationPointCacheProjection {
tal_id: self.tal_id.clone(),
ca_subject_name: String::from("publication-point-cache"),
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(),
current_manifest_rsync_uri: self.manifest_rsync_uri.clone(),
current_crl_rsync_uri: self
@@ -2042,7 +2050,7 @@ impl RoaCacheProjection {
manifest_rsync_uri: vcir.manifest_rsync_uri.clone(),
instance_effective_until: vcir.instance_gate.instance_effective_until.clone(),
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),
crl_sha256_by_uri,
entries,
@@ -2063,10 +2071,10 @@ impl RoaCacheProjection {
if let Some(hash) = &self.issuer_ca_sha256_hex {
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 {
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(),
});
}
@@ -2893,6 +2901,38 @@ fn write_publication_point_cache_projection_to_batch(
}
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> {
let mut base_opts = Options::default();
base_opts.create_if_missing(true);
@@ -2931,6 +2971,15 @@ impl RocksStore {
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(
path: &Path,
raw_store_path: Option<&Path>,
@@ -3147,6 +3196,53 @@ impl RocksStore {
.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<()> {
entry.validate_internal()?;
if let Some(raw_store) = self.external_raw_store.as_ref() {
@@ -3206,6 +3302,9 @@ impl RocksStore {
return Ok(());
}
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);
}
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)]
#[path = "storage/tests.rs"]
mod tests;

122
src/storage/tests.rs
View File

@@ -29,7 +29,7 @@ fn sample_child_certificate_cache_projection(
issuer_ca_sha256_hex: sha256_hex(b"issuer-ca"),
issuer_crl_uri: "rsync://example.test/repo/issuer.crl".to_string(),
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"),
effective_not_before: pack_time(0),
effective_until: pack_time(24),
@@ -294,7 +294,7 @@ fn roa_cache_projection_context(vcir: &ValidatedCaInstanceResult) -> RoaCachePro
})
.expect("sample VCIR has a ROA output");
RoaCacheProjectionContext {
parent_context_digest: [0x31; 32],
ca_validation_context_digest: [0x31; 32],
policy_fingerprint: [0x32; 32],
object_meta: vec![RoaCacheObjectMeta {
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.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}");
}

65
src/validation/objects.rs
View File

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

24
src/validation/tree.rs
View File

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

24
src/validation/tree_parallel.rs
View File

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

42
src/validation/tree_runner.rs
View File

@@ -304,7 +304,8 @@ impl<'a> Rpkiv1PublicationPointRunner<'a> {
self.finish_publication_point_cache_miss(ca, "ta_context_mismatch", lookup_started);
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);
return None;
}
@@ -505,7 +506,7 @@ impl<'a> Rpkiv1PublicationPointRunner<'a> {
ca_cert_sha256,
manifest_sha256,
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),
})
}
@@ -938,7 +939,7 @@ impl<'a> Rpkiv1PublicationPointRunner<'a> {
Some(ChildCertificateValidationCacheContext {
store: self.store,
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),
})
} else {
@@ -1530,7 +1531,7 @@ impl<'a> PublicationPointRunner for Rpkiv1PublicationPointRunner<'a> {
let roa_cache = if self.enable_roa_validation_cache && has_roa {
RoaValidationCacheInput::enabled_with_context(
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),
)
} else {
@@ -1946,7 +1947,7 @@ struct PublicationPointCacheIdentity {
ca_cert_sha256: [u8; 32],
manifest_sha256: [u8; 32],
ta_context_digest: [u8; 32],
parent_context_digest: [u8; 32],
ca_validation_context_digest: [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(&[
(
"version",
@@ -2058,7 +2059,7 @@ fn pack_time_window_contains(
struct ChildCertificateValidationCacheContext<'a> {
store: &'a RocksStore,
issuer_ca_sha256: [u8; 32],
parent_context_digest: [u8; 32],
ca_validation_context_digest: [u8; 32],
policy_fingerprint: [u8; 32],
}
@@ -2066,7 +2067,7 @@ fn child_certificate_cache_key_sha256_hex(
child_cert_uri: &str,
child_cert_sha256: &[u8; 32],
issuer_ca_sha256: &[u8; 32],
parent_context_digest: &[u8; 32],
ca_validation_context_digest: &[u8; 32],
policy_fingerprint: &[u8; 32],
) -> String {
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_sha256", child_cert_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()),
]);
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.sha256,
&cache.issuer_ca_sha256,
&cache.parent_context_digest,
&cache.ca_validation_context_digest,
&cache.policy_fingerprint,
));
}
@@ -3022,7 +3027,7 @@ fn discover_children_from_fresh_snapshot_with_audit_cached_with_issuer_der<
&f.rsync_uri,
&f.sha256,
&cache.issuer_ca_sha256,
&cache.parent_context_digest,
&cache.ca_validation_context_digest,
&cache.policy_fingerprint,
);
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_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,
effective_not_before,
effective_until,
@@ -3239,7 +3245,7 @@ fn discover_children_from_fresh_snapshot_with_audit_cached_with_issuer_der<
&f.rsync_uri,
&f.sha256,
&cache.issuer_ca_sha256,
&cache.parent_context_digest,
&cache.ca_validation_context_digest,
&cache.policy_fingerprint,
);
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_crl_uri: issuer_crl_uri.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,
effective_not_before,
effective_until,
@@ -4395,7 +4401,7 @@ fn failed_fetch_reuse_identity_for_fresh_result(
let identity = VcirFailedFetchReuseIdentity {
current_ca_sha256,
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),
effective_not_before: PackTime::from_utc_offset_datetime(validation_time),
effective_until,
@@ -4418,7 +4424,7 @@ fn failed_fetch_reuse_identity_matches_current(
};
cached.current_ca_sha256 == current_ca_sha256
&& 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.effective_until == vcir.instance_gate.instance_effective_until
&& 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(
&vcir,
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),
object_meta: objects.roa_cache_object_meta.clone(),
}),
@@ -5378,7 +5384,7 @@ fn build_publication_point_cache_projection_from_fresh(
ca_cert_sha256,
manifest_sha256,
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),
)
.map_err(|e| e.to_string())

14
src/validation/tree_runner/tests.rs
View File

@@ -1601,7 +1601,7 @@ fn discover_children_child_certificate_cache_reuses_successful_child_ca() {
let cache_context = ChildCertificateValidationCacheContext {
store: &store,
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),
};
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.sha256,
&cache_context.issuer_ca_sha256,
&cache_context.parent_context_digest,
&cache_context.ca_validation_context_digest,
&cache_context.policy_fingerprint,
);
assert!(
@@ -1724,7 +1724,7 @@ fn discover_children_child_certificate_cache_rechecks_changed_valid_crl() {
let cache_context = ChildCertificateValidationCacheContext {
store: &store,
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),
};
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.sha256,
&cache_context.issuer_ca_sha256,
&cache_context.parent_context_digest,
&cache_context.ca_validation_context_digest,
&cache_context.policy_fingerprint,
);
let projection = store
@@ -1832,7 +1832,7 @@ fn discover_children_child_certificate_cache_misses_when_current_crl_invalid() {
let cache_context = ChildCertificateValidationCacheContext {
store: &store,
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),
};
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 {
store: &store,
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),
};
let validation_time = time::OffsetDateTime::now_utc();
@@ -2436,7 +2436,7 @@ fn seed_publication_point_cache_projection(
ca.ca_certificate_sha256_32().unwrap(),
sha256_32(b"manifest-bytes"),
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),
)
.expect("build publication point projection");

1040
src/verification_only.rs Normal file
View File

File diff suppressed because it is too large Load Diff

83
tests/test_tree_traversal_m14.rs
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() {
let root_manifest = "rsync://example.test/repo/root.mft";
let child_manifest = "rsync://example.test/repo/child.mft";
let grandchild_manifest = "rsync://example.test/repo/grandchild.mft";
let runner = MockRunner::default()
.with(
@@ -301,6 +302,32 @@ fn tree_respects_max_depth_and_max_instances() {
audit: PublicationPointAudit::default(),
cir_fresh_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(),
},
);
@@ -324,6 +351,62 @@ fn tree_respects_max_depth_and_max_instances() {
.expect("run tree depth-limited");
assert_eq!(out.instances_processed, 1);
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(
ca_handle(root_manifest),