From 591bed90a8f61b9b80154415f14a48dfb56a8f58 Mon Sep 17 00:00:00 2001 From: yuyr Date: Thu, 10 Sep 2026 09:53:45 +0800 Subject: [PATCH] Harden first-release validation and dependency hygiene --- .dockerignore | 1 + .github/workflows/ci.yml | 26 +- CONTRIBUTING.md | 4 + Cargo.lock | 4 +- Cargo.toml | 2 +- README.md | 5 + SECURITY.md | 19 ++ THIRD_PARTY_NOTICES.md | 2 +- docs/conformance-matrix.md | 16 +- docs/dependencies.md | 56 ++++ docs/development.md | 13 +- docs/releasing.md | 37 +++ docs/third-party-licenses.txt | 2 +- src/model/rc/certificate_validation.rs | 15 +- src/model/rc/parsed_validation.rs | 7 +- src/model/signed_object/signed_object_impl.rs | 7 +- src/repository/storage/models_core.rs | 91 ++++-- src/repository/storage/models_summary.rs | 11 +- src/repository/storage/store_lifecycle.rs | 18 +- src/repository/storage/store_manifest.rs | 5 +- src/repository/storage/store_repository.rs | 2 - .../storage/store_transport_rrdp.rs | 145 +++++++-- .../sync/repo/tests_parts/setup_and_sync.rs | 9 +- .../sync/rrdp/models_and_parsing.rs | 14 +- src/repository/sync/rrdp/snapshot_apply.rs | 61 +++- .../sync/rrdp/tests_parts/delta_apply.rs | 10 +- src/repository/sync/rrdp/tests_parts/sync.rs | 35 ++- src/runtime/post_validation.rs | 29 +- src/runtime/report.rs | 28 +- src/runtime/run.rs | 153 +++++----- src/scheduler/repo_runtime/phase1_runtime.rs | 6 +- .../repo_runtime/runtime_trait_impl.rs | 1 - src/scheduler/repo_runtime/types_and_trait.rs | 1 - .../repo_scheduler/transport_state.rs | 286 +++++++++--------- src/scheduler/repo_worker/executors.rs | 83 +++-- src/scheduler/repo_worker/pools.rs | 14 +- src/ta_constraints/implementation.rs | 2 - src/validation/ca_path/certificate_checks.rs | 7 +- src/validation/ca_path/ip_resources.rs | 9 +- src/validation/manifest/helpers.rs | 4 +- src/validation/manifest/models_and_process.rs | 4 +- src/validation/objects/object_validation.rs | 132 ++++++-- src/validation/objects/outputs.rs | 43 ++- src/validation/objects/parallel_stage.rs | 38 +-- src/validation/objects/resource_validation.rs | 23 +- src/validation/objects/serial_processing.rs | 249 ++++++++++++--- src/validation/run_tree_from_tal/discovery.rs | 4 +- src/validation/tree_parallel/ready_stage.rs | 7 +- src/validation/tree_parallel/state.rs | 1 - .../tree_runner/audit_projection.rs | 10 +- src/validation/tree_runner/discovery.rs | 139 ++++++--- src/validation/tree_runner/ephemeral_state.rs | 59 ++-- src/validation/tree_runner/fresh_pipeline.rs | 51 ++-- .../tree_runner/publication_point_runner.rs | 19 +- src/validation/tree_runner/timing.rs | 1 - tests/cli_errors.rs | 64 ++++ tests/synthetic_docker_e2e.sh | 11 +- tools/check_format.py | 9 + tools/dependency_notices.py | 86 ++++++ tools/license_supplements.json | 14 + tools/test_dependency_notices.py | 59 ++++ 61 files changed, 1641 insertions(+), 622 deletions(-) create mode 100644 SECURITY.md create mode 100644 docs/dependencies.md create mode 100644 docs/releasing.md create mode 100644 tests/cli_errors.rs create mode 100644 tools/check_format.py create mode 100644 tools/dependency_notices.py create mode 100644 tools/license_supplements.json create mode 100644 tools/test_dependency_notices.py diff --git a/.dockerignore b/.dockerignore index 1fce1cd..085d6a9 100644 --- a/.dockerignore +++ b/.dockerignore @@ -4,6 +4,7 @@ target out state tests +tools docs !docs/third-party-licenses.txt docker diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 0a86e2a..d85a3f4 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -7,26 +7,46 @@ on: permissions: contents: read +concurrency: + group: ci-${{ github.workflow }}-${{ github.ref }} + cancel-in-progress: true + jobs: rust: runs-on: ubuntu-latest + timeout-minutes: 60 steps: - uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 - uses: dtolnay/rust-toolchain@6d653ac52d5f748757700b3b0ec16abf11083d92 with: toolchain: 1.92.0 components: rustfmt, clippy - - run: sudo apt-get update && sudo apt-get install -y --no-install-recommends build-essential clang libclang-dev pkg-config python3-cryptography openssl - - run: cargo fmt --all --check + - run: sudo apt-get update && sudo apt-get install -y --no-install-recommends build-essential clang libclang-dev pkg-config python3-cryptography openssl ripgrep curl rsync time + - run: python3 tools/check_format.py + - run: python3 -m unittest discover -s tools -p 'test_*.py' - run: cargo check --locked --no-default-features - run: cargo test --locked - run: cargo clippy --locked --all-targets -- -D warnings - run: cargo build --locked --release + - run: python3 tools/dependency_notices.py --check docker: runs-on: ubuntu-latest + timeout-minutes: 60 steps: - uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 - - run: sudo apt-get update && sudo apt-get install -y --no-install-recommends python3-cryptography openssl + - run: sudo apt-get update && sudo apt-get install -y --no-install-recommends python3-cryptography openssl ripgrep - run: docker build -f docker/Dockerfile -t panda-rpki:ci . - run: docker run --rm panda-rpki:ci --help - run: bash tests/synthetic_docker_e2e.sh + env: + PANDA_RPKI_TEST_IMAGE: panda-rpki:ci + audit: + runs-on: ubuntu-latest + timeout-minutes: 20 + steps: + - uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 + - uses: dtolnay/rust-toolchain@6d653ac52d5f748757700b3b0ec16abf11083d92 + with: + toolchain: 1.92.0 + - run: cargo install cargo-audit --locked --version 0.22.0 + - run: cargo audit --json diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 5ffb682..f2d749c 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -12,3 +12,7 @@ Documentation and code comments use English. Do not commit generated repository objects, keys, state, logs or build outputs. Keep third-party attribution intact. Submit only code you are entitled to contribute under the [project license](LICENSE) and identify externally sourced code and its license in the pull request. + +Use the [security policy](SECURITY.md) for suspected vulnerabilities, not public +issues. Changes to Cargo.lock must update and check dependency notices using +the [dependency maintenance workflow](docs/dependencies.md). diff --git a/Cargo.lock b/Cargo.lock index c427e5b..f13e172 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1054,9 +1054,9 @@ dependencies = [ [[package]] name = "quick-xml" -version = "0.37.5" +version = "0.41.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "331e97a1af0bf59823e6eadffe373d7b27f485be8748f71471c662c1f269b7fb" +checksum = "e660451e55124f798a69a5af3f49ccfbefbd41910eefd25caf2393e1f3473ec1" dependencies = [ "memchr", ] diff --git a/Cargo.toml b/Cargo.toml index 3c18d11..64bdfeb 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -25,7 +25,7 @@ toml = "0.8.20" rocksdb = { version = "0.22.0", default-features = false, features = ["lz4"] } serde_cbor = "0.11.2" roxmltree = "0.20.0" -quick-xml = "0.37.2" +quick-xml = "0.41.0" uuid = { version = "1.7.0", features = ["v4"] } reqwest = { version = "0.12.12", default-features = false, features = ["blocking", "rustls-tls", "gzip", "brotli", "deflate"] } tempfile = "3.16.0" diff --git a/README.md b/README.md index 0dedc56..cbed536 100644 --- a/README.md +++ b/README.md @@ -4,6 +4,10 @@ Panda RPKI is an RPKI relying party written in Rust. It synchronizes repositorie over RRDP, validates signed objects, and exports routing data as CSV and Canonical Cache Representation (CCR). Run a single cycle or use the daemon for continuous operation. +Version 0.1.0 is an unreleased candidate, validated on Linux x86-64. Other +platforms are not yet verified. Review the [standards and limitations](docs/conformance-matrix.md) +and [security policy](SECURITY.md) before evaluating it for routing use. + ## Quick start Requirements: Linux, Docker, and a TAL with its matching DER trust-anchor @@ -52,6 +56,7 @@ matching TA certificates. No RTR server is included. See the - [Docker and Compose](docs/docker.md): single-anchor, all-five and daemon deployments. - [Development and testing](docs/development.md). - [Contributing](CONTRIBUTING.md) and [changelog](CHANGELOG.md). +- [Release checklist](docs/releasing.md). ## Build from source diff --git a/SECURITY.md b/SECURITY.md new file mode 100644 index 0000000..976726a --- /dev/null +++ b/SECURITY.md @@ -0,0 +1,19 @@ +# Security policy + +## Reporting vulnerabilities + +Do not report suspected vulnerabilities, exploit inputs, or sensitive logs in +public issues or pull requests. The planned reporting channel is GitHub Private +Vulnerability Reporting on the official public repository. + +That repository and reporting channel are not available yet. Public release is +blocked until maintainers configure and test the private reporting channel and +replace this paragraph with its actual link. No response-time commitment is +currently offered. Do not assume that a public issue is a private channel. + +## Supported versions + +Version 0.1.0 is an unreleased candidate. There is no supported production +release yet. Dependency scans and regression tests are release checks, not a +guarantee that the validator is free of vulnerabilities. Validate suitability +for your deployment before using its results for routing decisions. diff --git a/THIRD_PARTY_NOTICES.md b/THIRD_PARTY_NOTICES.md index fe7f117..6226ad7 100644 --- a/THIRD_PARTY_NOTICES.md +++ b/THIRD_PARTY_NOTICES.md @@ -126,7 +126,7 @@ No production trust-anchor or repository data is included in this source tree. | potential_utf | 0.1.6 | Unicode-3.0 | | powerfmt | 0.2.0 | MIT OR Apache-2.0 | | proc-macro2 | 1.0.107 | MIT OR Apache-2.0 | -| quick-xml | 0.37.5 | MIT | +| quick-xml | 0.41.0 | MIT | | quinn | 0.11.11 | MIT OR Apache-2.0 | | quinn-proto | 0.11.17 | MIT OR Apache-2.0 | | quinn-udp | 0.5.15 | MIT OR Apache-2.0 | diff --git a/docs/conformance-matrix.md b/docs/conformance-matrix.md index d4e049b..dfd6020 100644 --- a/docs/conformance-matrix.md +++ b/docs/conformance-matrix.md @@ -4,16 +4,28 @@ | --- | --- | --- | --- | | TAL syntax and TA public-key binding | RFC 8630 §2 | `src/validation/from_tal.rs`, `src/model/ta.rs` | parser and synthetic integration tests | | TA and issued CA profile/signature/resource checks | RFC 6487 §4 | `src/validation/`, `src/model/` | profile unit tests and synthetic integration tests | -| Manifest, manifest file hashes, CRL freshness, and ROA checks | RFC 6486, RFC 6487, RFC 6482 | `src/validation/`, `src/model/` | RRDP unit tests and self-contained `tests/synthetic_docker_e2e.sh` | +| Manifest profile, file hashes, and freshness checks | RFC 9286 §§4, 6 | `src/model/manifest.rs`, `src/validation/manifest.rs` | manifest profile and validation unit tests | +| CRL profile and freshness checks | RFC 6487 §5 | `src/model/crl.rs`, `src/validation/` | CRL profile and validation tests | +| ROA profile and payload checks | RFC 9582 §§3–4 | `src/model/roa.rs`, `src/validation/` | ROA profile tests and synthetic CSV assertions | +| ASPA profile and provider-list checks | `draft-ietf-sidrops-aspa-profile-21` §§2–4 | `src/model/aspa.rs`, `src/validation/` | ASPA profile tests and synthetic VAP assertions | +| BGPsec router-certificate profile checks | RFC 8209 §3 | `src/model/router_cert.rs`, `src/validation/` | router-certificate profile tests | +| CMS signed-object profile | RFC 6488 §§2–3; RFC 9589 §4 | `src/model/signed_object.rs` | signed-object profile tests | | RRDP notification, snapshot, delta/replace/withdraw and fallback | RFC 8182 §§3.4–3.5 | `src/repository/sync/rrdp.rs`, `src/repository/sync/repo.rs` | snapshot/delta state-restart unit tests and synthetic Docker tests | | RRDP direct-reference and redirect origin checks | RFC 9674 §3.2 | `src/repository/fetch/http.rs`, `src/repository/sync/rrdp.rs` | cross-origin unit tests | -| Repeated TAL/TA pairing and stable aggregate output | RFC 8630 §2; RFC 6813 data model | `src/cli/mod.rs`, `src/cli/validate.rs`, `src/runtime/` | CLI parser checks; locked multi-TAL E2E | +| Repeated TAL/TA pairing and aggregate output | RFC 8630 §2 for each TAL; aggregation is an implementation choice | `src/cli/mod.rs`, `src/cli/validate.rs`, `src/runtime/` | CLI parser checks and multi-TAL integration evidence | | Bounded repository and object workers (default 8 each, shared across TALs) | operational support for the standard validation pipeline | `src/scheduler/`, `src/runtime/` | worker-pool unit tests | | Resumable RRDP protocol state | RFC 8182 §3.4 | `src/repository/storage/`, `src/repository/sync/rrdp.rs` | atomic state index and restart delta test | | CCR manifest, ROA, ASPA, trust-anchor and router-key states | `draft-ietf-sidrops-rpki-ccr-11` §§2–4 | `src/ccr/` | encode/decode tests and integration artifacts | | TA Constraints rule normalization and EE resource checks | `draft-ietf-sidrops-constraining-rpki-trust-anchors-01` §§3–4 | `src/ta_constraints.rs`, `src/validation/` | parser/normalization tests; constrained validation E2E | | VRP CSV output | RFC 6810 data semantics | `src/cli/validate.rs`, `src/runtime/` | self-contained Docker E2E CSV/summary assertions | +References identify the implemented profiles, not a certification of complete +RFC compliance. Draft revisions are intentionally explicit and are not claims +of compatibility with later revisions. The CLI uses HTTPS RRDP, requires local +TAL/TA pairs, and does not provide RTR or automatic TA-certificate refresh. +CCR tests establish encoding and payload consistency, not independent +revalidation of the entire trust chain by another relying party. + The worker pool and interval index are implementation mechanisms, not new wire protocols; their tests must also prove deterministic output and unchanged validation decisions. Only the table entries above are in v0.1.0 scope. diff --git a/docs/dependencies.md b/docs/dependencies.md new file mode 100644 index 0000000..ed526fb --- /dev/null +++ b/docs/dependencies.md @@ -0,0 +1,56 @@ +# Dependency maintenance + +Use the committed Cargo.lock. Do not run broad dependency updates as part of +unrelated changes. Review runtime compatibility and licensing for each update. + +## Security audit + +```bash +cargo install cargo-audit --locked --version 0.22.0 +cargo audit --json +``` + +The audit fetches the RustSec advisory database and needs network access. Record +the tool version and database revision with release evidence. Database download +failure is not a clean scan. Review informational warnings as well as errors; +unresolved affected vulnerabilities block release. Cargo auditing does not scan +Debian packages or fully establish the security of native bundled code. + +### Reviewed maintenance warning + +`serde_cbor 0.11.2` has the informational unmaintained advisory +[RUSTSEC-2021-0127](https://rustsec.org/advisories/RUSTSEC-2021-0127.html). +It remains used for persisted repository metadata. Replacing it needs a separate +storage-compatibility review, not an untested codec substitution during release +cleanup. The warning remains visible in audit output; no advisory is ignored. +Operators should protect the state directory from untrusted modification. + +## License notices + +```bash +python3 tools/dependency_notices.py --check +# After a reviewed Cargo.lock change: +python3 tools/dependency_notices.py +``` + +Requirements: Python 3, Cargo, curl, and HTTPS access to the Cargo registry and +GitHub's raw file service. The tool reads the entire locked dependency graph, +including build-time and other-platform dependencies, and preserves upstream +license and notice text. `--check` does not modify tracked files. + +Some package archives omit their license file. `tools/license_supplements.json` +records exact upstream commit URLs and SHA-256 checksums, derived from the +package's `.cargo_vcs_info.json`. Version changes need a new source review; +missing texts or checksum mismatches fail rather than silently omit attribution. +Do not edit original copyright or license wording to match project formatting. + +## Coverage + +With cargo-llvm-cov and the toolchain's llvm-tools-preview component installed: + +```bash +cargo llvm-cov --locked --all-targets --json --output-path target/coverage.json +``` + +Record actual line coverage and tool versions. The release target is 90%. +Do not exclude production modules or change the denominator to meet the target. diff --git a/docs/development.md b/docs/development.md index f8dea59..d957223 100644 --- a/docs/development.md +++ b/docs/development.md @@ -1,10 +1,12 @@ # Development and testing Use Rust 1.92 or newer on Linux and the [native dependencies](getting-started.md). -Tests additionally need Python 3, Python `cryptography`, and OpenSSL's CLI. +Tests additionally need Python 3, Python `cryptography`, OpenSSL's CLI, and +`ripgrep` (`rg`). Install them with `sudo apt-get install python3-cryptography openssl ripgrep`. ```bash -cargo fmt --all --check +python3 tools/check_format.py +python3 -m unittest discover -s tools -p 'test_*.py' cargo check --locked --no-default-features cargo test --locked cargo clippy --locked --all-targets -- -D warnings @@ -31,3 +33,10 @@ Generated keys and outputs are temporary and unsuitable as production inputs. Internal Rust modules are not a stable library API. The supported interface is the CLI and documented output files. Add behavior tests and update documentation when changing these interfaces. + +The format checker also checks files included through `include!`, which are not +all traversed by `cargo fmt`. To format them, run `rustfmt --edition 2024` on +the reported files, then rerun the checker. Preserve upstream license texts. + +For dependency and release checks see [Releasing](releasing.md) and +[Dependency maintenance](dependencies.md). diff --git a/docs/releasing.md b/docs/releasing.md new file mode 100644 index 0000000..f3482af --- /dev/null +++ b/docs/releasing.md @@ -0,0 +1,37 @@ +# Release checklist + +This checklist does not publish packages or images automatically. Keep +`publish = false` unless a separate crates.io publication decision is made. + +## Before the first public release + +- Confirm rights to publish all project code and preserve upstream attribution. +- Establish the official public repository and enable and test GitHub Private + Vulnerability Reporting. Update [SECURITY.md](../SECURITY.md) with the real link. +- Require CI checks on the protected main branch. Verify an actual successful + run on the hosting platform; local tests do not prove hosted CI is enabled. +- Run the [development checks](development.md), dependency audit, license checks, + and coverage measurement from a clean candidate checkout. The line-coverage + target is 90%; record any shortfall as an unresolved release item, without + excluding production files to improve the number. +- Review every security advisory. Fix affected dependencies with minimal + compatible changes, or block release. Do not silently suppress advisories. +- Confirm no credentials, production inputs, repository state or internal + verification data are tracked. Check documentation commands and links. + +## Tag and artifacts + +After the checks pass, replace `Unreleased` with the release date, review the +version and known limitations, and tag the exact tested commit. Rebuild source, +binary and container artifacts from that commit. Publish SHA-256 checksums, +build/toolchain details, dependency inventory (SBOM), and source revision with +the artifacts. Include project and third-party licenses in binary distributions. + +Only Linux x86-64 has been validated for this candidate. Do not label other +architectures or operating systems supported without testing them. Pin and +record container base-image digests for the release build; mutable tags alone +are not sufficient to reproduce an image. + +If runtime code or dependencies change, repeat end-to-end snapshot/delta and +artifact verification. Documentation-only changes do not establish new runtime +performance measurements; retain the revision associated with earlier results. diff --git a/docs/third-party-licenses.txt b/docs/third-party-licenses.txt index 7754ce3..85b54b6 100644 --- a/docs/third-party-licenses.txt +++ b/docs/third-party-licenses.txt @@ -22157,7 +22157,7 @@ DEALINGS IN THE SOFTWARE. ======================================================================== -quick-xml 0.37.5 — LICENSE-MIT.md +quick-xml 0.41.0 — LICENSE-MIT.md ======================================================================== The MIT License (MIT) diff --git a/src/model/rc/certificate_validation.rs b/src/model/rc/certificate_validation.rs index 2ab3e7f..2be6bb4 100644 --- a/src/model/rc/certificate_validation.rs +++ b/src/model/rc/certificate_validation.rs @@ -102,13 +102,14 @@ impl ResourceCertificate { return Err(ResourceCertificateProfileError::CertificatePoliciesTooManyQualifiers); } if let Some(qualifier_oid) = policies.qualifier_oids.first() - && qualifier_oid != OID_QT_CPS { - return Err( - ResourceCertificateProfileError::CertificatePoliciesInvalidQualifier( - qualifier_oid.clone(), - ), - ); - } + && qualifier_oid != OID_QT_CPS + { + return Err( + ResourceCertificateProfileError::CertificatePoliciesInvalidQualifier( + qualifier_oid.clone(), + ), + ); + } if self .tbs diff --git a/src/model/rc/parsed_validation.rs b/src/model/rc/parsed_validation.rs index 6f04803..feee86b 100644 --- a/src/model/rc/parsed_validation.rs +++ b/src/model/rc/parsed_validation.rs @@ -115,9 +115,10 @@ impl RcExtensionsParsed { if is_self_signed { if let (Some(keyid), Some(ski)) = (keyid.as_ref(), subject_key_identifier.as_ref()) - && keyid != ski { - return Err(ResourceCertificateProfileError::AkiSelfSignedNotEqualSki); - } + && keyid != ski + { + return Err(ResourceCertificateProfileError::AkiSelfSignedNotEqualSki); + } } else if keyid.is_none() { return Err(ResourceCertificateProfileError::AkiMissing); } diff --git a/src/model/signed_object/signed_object_impl.rs b/src/model/signed_object/signed_object_impl.rs index 18ff917..132d066 100644 --- a/src/model/signed_object/signed_object_impl.rs +++ b/src/model/signed_object/signed_object_impl.rs @@ -53,11 +53,8 @@ impl RpkiSignedObject { /// Verify the CMS signature using the embedded EE certificate public key. pub fn verify_signature(&self) -> Result<(), SignedObjectVerifyError> { let ee = &self.signed_data.certificates[0]; - - self.verify_signature_with_rsa_components( - &ee.rsa_public_modulus, - &ee.rsa_public_exponent, - ) + + self.verify_signature_with_rsa_components(&ee.rsa_public_modulus, &ee.rsa_public_exponent) } /// Verify the CMS signature using a DER-encoded SubjectPublicKeyInfo. diff --git a/src/repository/storage/models_core.rs b/src/repository/storage/models_core.rs index c95bcb2..343591e 100644 --- a/src/repository/storage/models_core.rs +++ b/src/repository/storage/models_core.rs @@ -7,9 +7,15 @@ pub enum StorageError { #[error("missing column family: {0}")] MissingColumnFamily(&'static str), #[error("cbor codec error for {entity}: {detail}")] - Codec { entity: &'static str, detail: String }, + Codec { + entity: &'static str, + detail: String, + }, #[error("invalid {entity}: {detail}")] - InvalidData { entity: &'static str, detail: String }, + InvalidData { + entity: &'static str, + detail: String, + }, } pub type StorageResult = Result; @@ -52,10 +58,14 @@ impl RepositoryViewEntry { } match self.state { RepositoryViewState::Present | RepositoryViewState::Replaced => { - let hash = self.current_hash.as_deref().ok_or(StorageError::InvalidData { - entity: "repository_view", - detail: "current_hash is required when state is present or replaced".to_string(), - })?; + let hash = self + .current_hash + .as_deref() + .ok_or(StorageError::InvalidData { + entity: "repository_view", + detail: "current_hash is required when state is present or replaced" + .to_string(), + })?; validate_sha256_hex("repository_view.current_hash", hash)?; } RepositoryViewState::Withdrawn => { @@ -79,23 +89,38 @@ pub struct RawByHashEntry { impl RawByHashEntry { pub fn from_bytes(sha256_hex: impl Into, bytes: Vec) -> Self { - Self { sha256_hex: sha256_hex.into(), bytes, origin_uris: Vec::new(), object_type: None, encoding: None } + Self { + sha256_hex: sha256_hex.into(), + bytes, + origin_uris: Vec::new(), + object_type: None, + encoding: None, + } } pub fn validate_internal(&self) -> StorageResult<()> { validate_sha256_hex("raw_by_hash.sha256_hex", &self.sha256_hex)?; if self.bytes.is_empty() { - return Err(StorageError::InvalidData { entity: "raw_by_hash", detail: "bytes must not be empty".to_string() }); + return Err(StorageError::InvalidData { + entity: "raw_by_hash", + detail: "bytes must not be empty".to_string(), + }); } let computed = hex::encode(compute_sha256_32(&self.bytes)); if computed != self.sha256_hex.to_ascii_lowercase() { - return Err(StorageError::InvalidData { entity: "raw_by_hash", detail: "sha256_hex does not match bytes".to_string() }); + return Err(StorageError::InvalidData { + entity: "raw_by_hash", + detail: "sha256_hex does not match bytes".to_string(), + }); } let mut seen = HashSet::with_capacity(self.origin_uris.len()); for uri in &self.origin_uris { validate_non_empty("raw_by_hash.origin_uris[]", uri)?; if !seen.insert(uri.as_str()) { - return Err(StorageError::InvalidData { entity: "raw_by_hash", detail: format!("duplicate origin URI: {uri}") }); + return Err(StorageError::InvalidData { + entity: "raw_by_hash", + detail: format!("duplicate origin URI: {uri}"), + }); } } Ok(()) @@ -118,11 +143,24 @@ pub struct ValidatedManifestMeta { impl ValidatedManifestMeta { pub fn validate_internal(&self) -> StorageResult<()> { - validate_manifest_number_be("validated_manifest_meta.validated_manifest_number", &self.validated_manifest_number)?; - let this_update = parse_time("validated_manifest_meta.validated_manifest_this_update", &self.validated_manifest_this_update)?; - let next_update = parse_time("validated_manifest_meta.validated_manifest_next_update", &self.validated_manifest_next_update)?; + validate_manifest_number_be( + "validated_manifest_meta.validated_manifest_number", + &self.validated_manifest_number, + )?; + let this_update = parse_time( + "validated_manifest_meta.validated_manifest_this_update", + &self.validated_manifest_this_update, + )?; + let next_update = parse_time( + "validated_manifest_meta.validated_manifest_next_update", + &self.validated_manifest_next_update, + )?; if next_update < this_update { - return Err(StorageError::InvalidData { entity: "validated_manifest_meta", detail: "validated_manifest_next_update must be >= validated_manifest_this_update".to_string() }); + return Err(StorageError::InvalidData { + entity: "validated_manifest_meta", + detail: "validated_manifest_next_update must be >= validated_manifest_this_update" + .to_string(), + }); } Ok(()) } @@ -141,11 +179,26 @@ pub struct ManifestAntiRollbackMeta { impl ManifestAntiRollbackMeta { pub fn validate_internal(&self) -> StorageResult<()> { - validate_non_empty("manifest_anti_rollback.manifest_rsync_uri", &self.manifest_rsync_uri)?; - validate_manifest_number_be("manifest_anti_rollback.manifest_number_be", &self.manifest_number_be)?; - parse_time("manifest_anti_rollback.manifest_this_update", &self.manifest_this_update)?; - validate_sha256_digest_bytes("manifest_anti_rollback.manifest_sha256", &self.manifest_sha256)?; - parse_time("manifest_anti_rollback.updated_at_validation_time", &self.updated_at_validation_time)?; + validate_non_empty( + "manifest_anti_rollback.manifest_rsync_uri", + &self.manifest_rsync_uri, + )?; + validate_manifest_number_be( + "manifest_anti_rollback.manifest_number_be", + &self.manifest_number_be, + )?; + parse_time( + "manifest_anti_rollback.manifest_this_update", + &self.manifest_this_update, + )?; + validate_sha256_digest_bytes( + "manifest_anti_rollback.manifest_sha256", + &self.manifest_sha256, + )?; + parse_time( + "manifest_anti_rollback.updated_at_validation_time", + &self.updated_at_validation_time, + )?; Ok(()) } } diff --git a/src/repository/storage/models_summary.rs b/src/repository/storage/models_summary.rs index 357fff4..d782e78 100644 --- a/src/repository/storage/models_summary.rs +++ b/src/repository/storage/models_summary.rs @@ -68,10 +68,13 @@ impl RrdpSourceMemberRecord { &self.last_confirmed_session_id, )?; if self.present { - let hash = self.current_hash.as_deref().ok_or(StorageError::InvalidData { - entity: "rrdp_source_member", - detail: "current_hash is required when present=true".to_string(), - })?; + let hash = self + .current_hash + .as_deref() + .ok_or(StorageError::InvalidData { + entity: "rrdp_source_member", + detail: "current_hash is required when present=true".to_string(), + })?; validate_sha256_hex("rrdp_source_member.current_hash", hash)?; } else if let Some(hash) = &self.current_hash { validate_sha256_hex("rrdp_source_member.current_hash", hash)?; diff --git a/src/repository/storage/store_lifecycle.rs b/src/repository/storage/store_lifecycle.rs index bb0f6a9..f7347af 100644 --- a/src/repository/storage/store_lifecycle.rs +++ b/src/repository/storage/store_lifecycle.rs @@ -19,13 +19,8 @@ impl RocksStore { let mut options = Options::default(); configure_work_db_options(&mut options); let descriptors = column_family_descriptors(); - let db = DB::open_cf_descriptors_read_only( - &options, - source, - descriptors, - false, - ) - .map_err(|error| StorageError::RocksDb(error.to_string()))?; + let db = DB::open_cf_descriptors_read_only(&options, source, descriptors, false) + .map_err(|error| StorageError::RocksDb(error.to_string()))?; let checkpoint = Checkpoint::new(&db).map_err(|error| StorageError::RocksDb(error.to_string()))?; checkpoint @@ -41,12 +36,8 @@ impl RocksStore { configure_work_db_options(&mut base_opts); let descriptors = column_family_descriptors(); - let db = DB::open_cf_descriptors( - &base_opts, - path, - descriptors, - ) - .map_err(|e| StorageError::RocksDb(e.to_string()))?; + let db = DB::open_cf_descriptors(&base_opts, path, descriptors) + .map_err(|e| StorageError::RocksDb(e.to_string()))?; Ok(Self { db, @@ -124,7 +115,6 @@ impl RocksStore { .cf_handle(name) .ok_or(StorageError::MissingColumnFamily(name)) } - } fn reject_unsupported_column_families(path: &Path) -> StorageResult<()> { diff --git a/src/repository/storage/store_manifest.rs b/src/repository/storage/store_manifest.rs index 23a6019..1a4096b 100644 --- a/src/repository/storage/store_manifest.rs +++ b/src/repository/storage/store_manifest.rs @@ -20,7 +20,10 @@ impl RocksStore { Ok(Some(metadata)) } - pub fn put_manifest_anti_rollback_meta(&self, metadata: &ManifestAntiRollbackMeta) -> StorageResult<()> { + pub fn put_manifest_anti_rollback_meta( + &self, + metadata: &ManifestAntiRollbackMeta, + ) -> StorageResult<()> { metadata.validate_internal()?; let cf = self.cf(CF_MANIFEST_ANTI_ROLLBACK)?; let key = manifest_anti_rollback_key(&metadata.manifest_rsync_uri); diff --git a/src/repository/storage/store_repository.rs b/src/repository/storage/store_repository.rs index 3a48a67..ccdea1d 100644 --- a/src/repository/storage/store_repository.rs +++ b/src/repository/storage/store_repository.rs @@ -1,7 +1,6 @@ // Repository-view and raw-object/blob storage operations. impl RocksStore { - pub fn put_repository_view_entry(&self, entry: &RepositoryViewEntry) -> StorageResult<()> { entry.validate_internal()?; let cf = self.cf(CF_REPOSITORY_VIEW)?; @@ -359,5 +358,4 @@ impl RocksStore { } Ok(out) } - } diff --git a/src/repository/storage/store_transport_rrdp.rs b/src/repository/storage/store_transport_rrdp.rs index 5ecac53..81a3afb 100644 --- a/src/repository/storage/store_transport_rrdp.rs +++ b/src/repository/storage/store_transport_rrdp.rs @@ -6,72 +6,144 @@ impl RocksStore { let cf = self.cf(CF_RRDP_SOURCE)?; let key = rrdp_source_key(&record.notify_uri); let value = encode_cbor(record, "rrdp_source")?; - self.db.put_cf(cf, key.as_bytes(), value).map_err(|e| StorageError::RocksDb(e.to_string()))?; + self.db + .put_cf(cf, key.as_bytes(), value) + .map_err(|e| StorageError::RocksDb(e.to_string()))?; Ok(()) } - pub fn get_rrdp_source_record(&self, notify_uri: &str) -> StorageResult> { + pub fn get_rrdp_source_record( + &self, + notify_uri: &str, + ) -> StorageResult> { let cf = self.cf(CF_RRDP_SOURCE)?; let key = rrdp_source_key(notify_uri); - let Some(bytes) = self.db.get_cf(cf, key.as_bytes()).map_err(|e| StorageError::RocksDb(e.to_string()))? else { return Ok(None) }; + let Some(bytes) = self + .db + .get_cf(cf, key.as_bytes()) + .map_err(|e| StorageError::RocksDb(e.to_string()))? + else { + return Ok(None); + }; let record = decode_cbor::(&bytes, "rrdp_source")?; record.validate_internal()?; Ok(Some(record)) } - pub fn put_rrdp_source_member_record(&self, record: &RrdpSourceMemberRecord) -> StorageResult<()> { + pub fn put_rrdp_source_member_record( + &self, + record: &RrdpSourceMemberRecord, + ) -> StorageResult<()> { record.validate_internal()?; let cf = self.cf(CF_RRDP_SOURCE_MEMBER)?; let key = rrdp_source_member_key(&record.notify_uri, &record.rsync_uri); let value = encode_cbor(record, "rrdp_source_member")?; - self.db.put_cf(cf, key.as_bytes(), value).map_err(|e| StorageError::RocksDb(e.to_string()))?; + self.db + .put_cf(cf, key.as_bytes(), value) + .map_err(|e| StorageError::RocksDb(e.to_string()))?; Ok(()) } - pub fn get_rrdp_source_member_record(&self, notify_uri: &str, rsync_uri: &str) -> StorageResult> { + pub fn get_rrdp_source_member_record( + &self, + notify_uri: &str, + rsync_uri: &str, + ) -> StorageResult> { let cf = self.cf(CF_RRDP_SOURCE_MEMBER)?; let key = rrdp_source_member_key(notify_uri, rsync_uri); - let Some(bytes) = self.db.get_cf(cf, key.as_bytes()).map_err(|e| StorageError::RocksDb(e.to_string()))? else { return Ok(None) }; + let Some(bytes) = self + .db + .get_cf(cf, key.as_bytes()) + .map_err(|e| StorageError::RocksDb(e.to_string()))? + else { + return Ok(None); + }; let record = decode_cbor::(&bytes, "rrdp_source_member")?; record.validate_internal()?; Ok(Some(record)) } - pub fn list_rrdp_source_member_records(&self, notify_uri: &str) -> StorageResult> { + pub fn list_rrdp_source_member_records( + &self, + notify_uri: &str, + ) -> StorageResult> { let cf = self.cf(CF_RRDP_SOURCE_MEMBER)?; let prefix = rrdp_source_member_prefix(notify_uri); let mode = IteratorMode::From(prefix.as_bytes(), Direction::Forward); - self.db.iterator_cf(cf, mode).take_while(|res| match res { Ok((key, _)) => key.starts_with(prefix.as_bytes()), Err(_) => false }).map(|res| { - let (_key, value) = res.map_err(|e| StorageError::RocksDb(e.to_string()))?; - let record = decode_cbor::(&value, "rrdp_source_member")?; - record.validate_internal()?; - Ok(record) - }).collect() + self.db + .iterator_cf(cf, mode) + .take_while(|res| match res { + Ok((key, _)) => key.starts_with(prefix.as_bytes()), + Err(_) => false, + }) + .map(|res| { + let (_key, value) = res.map_err(|e| StorageError::RocksDb(e.to_string()))?; + let record = decode_cbor::(&value, "rrdp_source_member")?; + record.validate_internal()?; + Ok(record) + }) + .collect() } - pub fn list_current_rrdp_source_members(&self, notify_uri: &str) -> StorageResult> { + pub fn list_current_rrdp_source_members( + &self, + notify_uri: &str, + ) -> StorageResult> { let mut records = self.list_rrdp_source_member_records(notify_uri)?; records.retain(|record| record.present); records.sort_by(|a, b| a.rsync_uri.cmp(&b.rsync_uri)); Ok(records) } - pub fn is_current_rrdp_source_member(&self, notify_uri: &str, rsync_uri: &str) -> StorageResult { - Ok(matches!(self.get_rrdp_source_member_record(notify_uri, rsync_uri)?, Some(record) if record.present)) + pub fn is_current_rrdp_source_member( + &self, + notify_uri: &str, + rsync_uri: &str, + ) -> StorageResult { + Ok( + matches!(self.get_rrdp_source_member_record(notify_uri, rsync_uri)?, Some(record) if record.present), + ) } - pub fn load_current_object_bytes_by_uri(&self, rsync_uri: &str) -> StorageResult>> { - Ok(self.load_current_object_with_hash_by_uri(rsync_uri)?.map(|object| object.bytes)) + pub fn load_current_object_bytes_by_uri( + &self, + rsync_uri: &str, + ) -> StorageResult>> { + Ok(self + .load_current_object_with_hash_by_uri(rsync_uri)? + .map(|object| object.bytes)) } - pub fn load_current_object_with_hash_by_uri(&self, rsync_uri: &str) -> StorageResult> { - let Some(view) = self.get_repository_view_entry(rsync_uri)? else { return Ok(None) }; + pub fn load_current_object_with_hash_by_uri( + &self, + rsync_uri: &str, + ) -> StorageResult> { + let Some(view) = self.get_repository_view_entry(rsync_uri)? else { + return Ok(None); + }; match view.state { RepositoryViewState::Withdrawn => Ok(None), RepositoryViewState::Present | RepositoryViewState::Replaced => { - let hash = view.current_hash.as_deref().ok_or(StorageError::InvalidData { entity: "repository_view", detail: format!("current_hash missing for current object URI: {rsync_uri}") })?; - let bytes = self.get_blob_bytes(hash)?.ok_or(StorageError::InvalidData { entity: "repository_view", detail: format!("blob bytes missing for current object URI: {rsync_uri} (hash={hash})") })?; - Ok(Some(CurrentObjectWithHash { current_hash_hex: hash.to_ascii_lowercase(), current_hash: decode_sha256_hex_32("repository_view.current_hash", hash)?, bytes })) + let hash = view + .current_hash + .as_deref() + .ok_or(StorageError::InvalidData { + entity: "repository_view", + detail: format!("current_hash missing for current object URI: {rsync_uri}"), + })?; + let bytes = self + .get_blob_bytes(hash)? + .ok_or(StorageError::InvalidData { + entity: "repository_view", + detail: format!( + "blob bytes missing for current object URI: {rsync_uri} (hash={hash})" + ), + })?; + Ok(Some(CurrentObjectWithHash { + current_hash_hex: hash.to_ascii_lowercase(), + current_hash: decode_sha256_hex_32("repository_view.current_hash", hash)?, + bytes, + })) } } } @@ -81,14 +153,25 @@ impl RocksStore { let cf = self.cf(CF_RRDP_URI_OWNER)?; let key = rrdp_uri_owner_key(&record.rsync_uri); let value = encode_cbor(record, "rrdp_uri_owner")?; - self.db.put_cf(cf, key.as_bytes(), value).map_err(|e| StorageError::RocksDb(e.to_string()))?; + self.db + .put_cf(cf, key.as_bytes(), value) + .map_err(|e| StorageError::RocksDb(e.to_string()))?; Ok(()) } - pub fn get_rrdp_uri_owner_record(&self, rsync_uri: &str) -> StorageResult> { + pub fn get_rrdp_uri_owner_record( + &self, + rsync_uri: &str, + ) -> StorageResult> { let cf = self.cf(CF_RRDP_URI_OWNER)?; let key = rrdp_uri_owner_key(rsync_uri); - let Some(bytes) = self.db.get_cf(cf, key.as_bytes()).map_err(|e| StorageError::RocksDb(e.to_string()))? else { return Ok(None) }; + let Some(bytes) = self + .db + .get_cf(cf, key.as_bytes()) + .map_err(|e| StorageError::RocksDb(e.to_string()))? + else { + return Ok(None); + }; let record = decode_cbor::(&bytes, "rrdp_uri_owner")?; record.validate_internal()?; Ok(Some(record)) @@ -97,12 +180,16 @@ impl RocksStore { pub fn delete_rrdp_uri_owner_record(&self, rsync_uri: &str) -> StorageResult<()> { let cf = self.cf(CF_RRDP_URI_OWNER)?; let key = rrdp_uri_owner_key(rsync_uri); - self.db.delete_cf(cf, key.as_bytes()).map_err(|e| StorageError::RocksDb(e.to_string()))?; + self.db + .delete_cf(cf, key.as_bytes()) + .map_err(|e| StorageError::RocksDb(e.to_string()))?; Ok(()) } pub fn write_batch(&self, batch: WriteBatch) -> StorageResult<()> { - self.db.write(batch).map_err(|e| StorageError::RocksDb(e.to_string()))?; + self.db + .write(batch) + .map_err(|e| StorageError::RocksDb(e.to_string()))?; Ok(()) } } diff --git a/src/repository/sync/repo/tests_parts/setup_and_sync.rs b/src/repository/sync/repo/tests_parts/setup_and_sync.rs index 191006a..ed439d4 100644 --- a/src/repository/sync/repo/tests_parts/setup_and_sync.rs +++ b/src/repository/sync/repo/tests_parts/setup_and_sync.rs @@ -5,7 +5,9 @@ use crate::output::analysis::timing::{TimingHandle, TimingMeta}; use crate::repository::fetch::rsync::LocalDirRsyncFetcher; use crate::repository::storage::RepositoryViewState; use crate::repository::sync::rrdp::Fetcher as HttpFetcher; -use crate::repository::sync::store_projection::{build_repository_view_present_entry, compute_sha256_hex}; +use crate::repository::sync::store_projection::{ + build_repository_view_present_entry, compute_sha256_hex, +}; use sha2::Digest; use std::sync::atomic::{AtomicUsize, Ordering}; @@ -252,7 +254,10 @@ fn rsync_second_sync_marks_missing_repository_view_entries_withdrawn() { .get_repository_view_entry("rsync://example.test/repo/c.crl") .expect("get added repo view") .expect("added entry exists"); - assert_eq!(added.state, crate::repository::storage::RepositoryViewState::Present); + assert_eq!( + added.state, + crate::repository::storage::RepositoryViewState::Present + ); } #[test] diff --git a/src/repository/sync/rrdp/models_and_parsing.rs b/src/repository/sync/rrdp/models_and_parsing.rs index 7d33bc8..81a7fdc 100644 --- a/src/repository/sync/rrdp/models_and_parsing.rs +++ b/src/repository/sync/rrdp/models_and_parsing.rs @@ -340,9 +340,10 @@ pub(crate) fn load_rrdp_local_state( if let Some(record) = store .get_rrdp_source_record(notification_uri) .map_err(|e| e.to_string())? - && let (Some(session_id), Some(serial)) = (record.last_session_id, record.last_serial) { - return Ok(Some(RrdpState { session_id, serial })); - } + && let (Some(session_id), Some(serial)) = (record.last_session_id, record.last_serial) + { + return Ok(Some(RrdpState { session_id, serial })); + } Ok(None) } @@ -617,9 +618,10 @@ pub fn parse_delta_file(xml: &[u8]) -> Result { let hash_sha256 = parse_sha256_hex_delta_withdraw(hash)?; if let Some(s) = collect_element_text(&child) - && !strip_all_ascii_whitespace(&s).is_empty() { - return Err(RrdpError::DeltaWithdrawUnexpectedContent); - } + && !strip_all_ascii_whitespace(&s).is_empty() + { + return Err(RrdpError::DeltaWithdrawUnexpectedContent); + } elements.push(DeltaElement::Withdraw { uri, hash_sha256 }); } diff --git a/src/repository/sync/rrdp/snapshot_apply.rs b/src/repository/sync/rrdp/snapshot_apply.rs index 0d20171..f7efd4d 100644 --- a/src/repository/sync/rrdp/snapshot_apply.rs +++ b/src/repository/sync/rrdp/snapshot_apply.rs @@ -19,6 +19,20 @@ use super::{ RrdpResourceKind, RrdpSyncError, parse_u64_str, strip_all_ascii_whitespace, }; +fn decode_attribute( + attr: &quick_xml::events::attributes::Attribute<'_>, + decoder: quick_xml::encoding::Decoder, +) -> Result { + // Preserve the existing decode + unescape behavior without adopting the + // newer API's additional XML attribute whitespace normalization. + let decoded = decoder + .decode(attr.value.as_ref()) + .map_err(|e| RrdpError::Xml(e.to_string()))?; + quick_xml::escape::unescape(&decoded) + .map(|value| value.into_owned()) + .map_err(|e| RrdpError::Xml(e.to_string())) +} + #[cfg(test)] pub(super) fn apply_snapshot( store: &RocksStore, @@ -91,10 +105,7 @@ pub(super) fn apply_snapshot_from_bufread( Err(e) => return Err(RrdpError::Xml(e.to_string()).into()), }; let key = attr.key.as_ref(); - let value = attr - .decode_and_unescape_value(reader.decoder()) - .map_err(|e| RrdpError::Xml(e.to_string()))? - .into_owned(); + let value = decode_attribute(&attr, reader.decoder())?; match key { b"xmlns" => xmlns = value, b"version" => version = value, @@ -136,11 +147,7 @@ pub(super) fn apply_snapshot_from_bufread( Err(e) => return Err(RrdpError::Xml(e.to_string()).into()), }; if attr.key.as_ref() == b"uri" { - uri = Some( - attr.decode_and_unescape_value(reader.decoder()) - .map_err(|e| RrdpError::Xml(e.to_string()))? - .into_owned(), - ); + uri = Some(decode_attribute(&attr, reader.decoder())?); } } let uri = uri.ok_or(RrdpError::PublishUriMissing)?; @@ -186,6 +193,17 @@ pub(super) fn apply_snapshot_from_bufread( current_publish_text.push_str(&text); } } + Ok(Event::GeneralRef(_)) => { + // Earlier reader versions left references inside the raw text, + // where base64 decoding rejected them. Do not silently drop a + // reference now that the reader emits a separate event. + if in_publish && publish_nested_depth == 0 { + return Err(RrdpError::PublishBase64( + "entity references are not valid base64 text".into(), + ) + .into()); + } + } Ok(Event::CData(e)) => { if in_publish && publish_nested_depth == 0 { let text = reader @@ -452,3 +470,28 @@ pub(super) fn fetch_snapshot_into_tempfile( .map_err(|e| RrdpSyncError::Fetch(format!("tempfile rewind failed: {e}")))?; Ok((tmp, bytes_written)) } + +#[cfg(test)] +mod attribute_tests { + use super::*; + + #[test] + fn attribute_decoding_preserves_whitespace_and_unescapes() { + let reader = Reader::from_str(""); + let attr = quick_xml::events::attributes::Attribute::from(( + b"uri".as_slice(), + b"a\tb&c".as_slice(), + )); + assert_eq!(decode_attribute(&attr, reader.decoder()).unwrap(), "a\tb&c"); + } + + #[test] + fn attribute_decoding_rejects_unknown_entities() { + let reader = Reader::from_str(""); + let attr = quick_xml::events::attributes::Attribute::from(( + b"uri".as_slice(), + b"a&unknown;".as_slice(), + )); + assert!(decode_attribute(&attr, reader.decoder()).is_err()); + } +} diff --git a/src/repository/sync/rrdp/tests_parts/delta_apply.rs b/src/repository/sync/rrdp/tests_parts/delta_apply.rs index 1d1afd8..8d5650e 100644 --- a/src/repository/sync/rrdp/tests_parts/delta_apply.rs +++ b/src/repository/sync/rrdp/tests_parts/delta_apply.rs @@ -99,12 +99,18 @@ fn apply_delta_applies_publish_replace_and_withdraw_with_membership_checks() { .get_repository_view_entry("rsync://example.net/repo/a.mft") .expect("get a view") .expect("a view exists"); - assert_eq!(a_view.state, crate::repository::storage::RepositoryViewState::Withdrawn); + assert_eq!( + a_view.state, + crate::repository::storage::RepositoryViewState::Withdrawn + ); let b_view = store .get_repository_view_entry("rsync://example.net/repo/b.roa") .expect("get b view") .expect("b view exists"); - assert_eq!(b_view.state, crate::repository::storage::RepositoryViewState::Present); + assert_eq!( + b_view.state, + crate::repository::storage::RepositoryViewState::Present + ); assert_eq!( b_view.current_hash.as_deref(), Some(hex::encode(sha2::Sha256::digest(b"b2")).as_str()) diff --git a/src/repository/sync/rrdp/tests_parts/sync.rs b/src/repository/sync/rrdp/tests_parts/sync.rs index fd48d79..ea51a1f 100644 --- a/src/repository/sync/rrdp/tests_parts/sync.rs +++ b/src/repository/sync/rrdp/tests_parts/sync.rs @@ -1,5 +1,30 @@ // RRDP test group: sync. +#[test] +fn snapshot_entity_reference_is_not_silently_dropped_from_base64() { + for content in ["YQ==&", "YQ==", "YQ==&unknown;"] { + let temp = tempfile::tempdir().unwrap(); + let store = RocksStore::open(temp.path()).unwrap(); + let sid = Uuid::parse_str("550e8400-e29b-41d4-a716-446655440000").unwrap(); + let xml = format!( + "{content}" + ); + let error = super::snapshot_apply::apply_snapshot( + &store, + "https://example.net/notification.xml", + None, + xml.as_bytes(), + sid, + 1, + ) + .expect_err("an entity must not disappear and turn invalid base64 into valid data"); + assert!(matches!( + error, + RrdpSyncError::Rrdp(RrdpError::PublishBase64(_)) + )); + } +} + #[test] fn sync_from_notification_snapshot_rejects_cross_source_owner_conflict() { let tmp = tempfile::tempdir().expect("tempdir"); @@ -177,7 +202,10 @@ fn sync_from_notification_snapshot_applies_snapshot_and_stores_state() { .get_repository_view_entry("rsync://example.net/repo/a.mft") .expect("get repository view") .expect("repository view exists"); - assert_eq!(view.state, crate::repository::storage::RepositoryViewState::Present); + assert_eq!( + view.state, + crate::repository::storage::RepositoryViewState::Present + ); assert_eq!(view.repository_source.as_deref(), Some(notif_uri)); let current_bytes = store @@ -202,7 +230,10 @@ fn sync_from_notification_snapshot_applies_snapshot_and_stores_state() { .expect("get owner") .expect("owner exists"); assert_eq!(owner.notify_uri, notif_uri); - assert_eq!(owner.owner_state, crate::repository::storage::RrdpUriOwnerState::Active); + assert_eq!( + owner.owner_state, + crate::repository::storage::RrdpUriOwnerState::Active + ); } #[test] diff --git a/src/runtime/post_validation.rs b/src/runtime/post_validation.rs index d778003..28c543f 100644 --- a/src/runtime/post_validation.rs +++ b/src/runtime/post_validation.rs @@ -26,17 +26,25 @@ fn unique_rrdp_repos_from_publication_points( fn print_summary_from_shared(validation_time: time::OffsetDateTime, shared: &PostValidationShared) { let warning_count = shared.tree_warnings.len() - + shared.publication_points.iter().map(|pp| pp.warnings.len()).sum::(); - crate::logging::emit(crate::logging::Level::Info, "validation_summary", || serde_json::json!({ - "validation_time": validation_time.unix_timestamp(), - "publication_points": shared.instances_processed, "failed": shared.instances_failed, - "repositories": unique_rrdp_repos_from_publication_points(&shared.publication_points), - "vrps": shared.vrps.len(), "aspas": shared.aspas.len(), "warnings": warning_count, - })); + + shared + .publication_points + .iter() + .map(|pp| pp.warnings.len()) + .sum::(); + crate::logging::emit(crate::logging::Level::Info, "validation_summary", || { + serde_json::json!({ + "validation_time": validation_time.unix_timestamp(), + "publication_points": shared.instances_processed, "failed": shared.instances_failed, + "repositories": unique_rrdp_repos_from_publication_points(&shared.publication_points), + "vrps": shared.vrps.len(), "aspas": shared.aspas.len(), "warnings": warning_count, + }) + }); if warning_count > 0 { - crate::logging::emit(crate::logging::Level::Warn, "validation_warnings", || serde_json::json!({ - "count": warning_count, "failed_publication_points": shared.instances_failed, - })); + crate::logging::emit(crate::logging::Level::Warn, "validation_warnings", || { + serde_json::json!({ + "count": warning_count, "failed_publication_points": shared.instances_failed, + }) + }); } } @@ -97,7 +105,6 @@ impl PostValidationShared { ccr_accumulator, } } - } #[derive(Default)] diff --git a/src/runtime/report.rs b/src/runtime/report.rs index 3d6ed23..a08b55b 100644 --- a/src/runtime/report.rs +++ b/src/runtime/report.rs @@ -18,15 +18,31 @@ where // One scheduler entry preserves explicit TAL identities for one or many roots. if let Some(t) = timing { run_tree_from_multiple_tals_parallel_phase2_audit_with_timing( - store, policy, args.tal_inputs.clone(), http, rsync, validation_time, - config, args.parallel_phase1_config.clone(), args.parallel_phase2_config.clone(), - collect_current_repo_objects, t, + store, + policy, + args.tal_inputs.clone(), + http, + rsync, + validation_time, + config, + args.parallel_phase1_config.clone(), + args.parallel_phase2_config.clone(), + collect_current_repo_objects, + t, ) } else { run_tree_from_multiple_tals_parallel_phase2_audit( - store, policy, args.tal_inputs.clone(), http, rsync, validation_time, - config, args.parallel_phase1_config.clone(), args.parallel_phase2_config.clone(), + store, + policy, + args.tal_inputs.clone(), + http, + rsync, + validation_time, + config, + args.parallel_phase1_config.clone(), + args.parallel_phase2_config.clone(), collect_current_repo_objects, ) - }.map_err(|error| error.to_string()) + } + .map_err(|error| error.to_string()) } diff --git a/src/runtime/run.rs b/src/runtime/run.rs index 2a1d599..c68363c 100644 --- a/src/runtime/run.rs +++ b/src/runtime/run.rs @@ -13,7 +13,11 @@ pub(crate) fn run_config(args: RunConfig) -> Result<(), String> { } policy.ta_constraints = args.ta_constraints.clone(); for warning in policy.ta_constraints.configuration_warnings() { - crate::logging::emit(crate::logging::Level::Warn, "constraint_warning", || serde_json::json!({"reason": warning})); + crate::logging::emit( + crate::logging::Level::Warn, + "constraint_warning", + || serde_json::json!({"reason": warning}), + ); } let validation_time = args .validation_time @@ -45,8 +49,7 @@ pub(crate) fn run_config(args: RunConfig) -> Result<(), String> { let config = TreeRunConfig { max_depth: Some(args.max_ca_depth), max_instances: args.max_instances, - compact_audit: args.skip_report_build - && args.report_json_path.is_none(), + compact_audit: args.skip_report_build && args.report_json_path.is_none(), build_ccr_accumulator: args.ccr_out_path.is_some(), }; @@ -66,7 +69,7 @@ pub(crate) fn run_config(args: RunConfig) -> Result<(), String> { let fmt = time::format_description::parse_borrowed::<2>( "[year][month][day]T[hour][minute][second]Z", ) - .map_err(|e| format!("format description parse failed: {e}"))?; + .map_err(|e| format!("format description parse failed: {e}"))?; time::OffsetDateTime::now_utc() .format(&fmt) .map_err(|e| format!("format timestamp failed: {e}"))? @@ -157,7 +160,11 @@ pub(crate) fn run_config(args: RunConfig) -> Result<(), String> { }; let validation_ms = validation_started.elapsed().as_millis() as u64; - crate::logging::emit(crate::logging::Level::Info, "validation_phase_completed", || serde_json::json!({"elapsed_ms": validation_ms})); + crate::logging::emit( + crate::logging::Level::Info, + "validation_phase_completed", + || serde_json::json!({"elapsed_ms": validation_ms}), + ); let shared = PostValidationShared::from_run_output(out); record_memory_checkpoint( &mut memory_checkpoints, @@ -222,56 +229,50 @@ pub(crate) fn run_config(args: RunConfig) -> Result<(), String> { .compare_view_trust_anchor .as_deref() .unwrap_or("unknown"); - let (report_result, ccr_result, compare_view_result) = - std::thread::scope(|scope| { - // Reborrow `shared` as a plain reference so the scoped output tasks - // capture the reference instead of moving fields out of the owned value. - let shared = &shared; - let report_handle = if args.skip_report_build { - None - } else { - Some(scope.spawn(|| { - run_report_task( - &policy, - validation_time, - shared, - args.report_json_path.as_deref(), - report_json_format, - ) - })) - }; - let ccr_handle = scope.spawn(|| { - run_ccr_task( + let (report_result, ccr_result, compare_view_result) = std::thread::scope(|scope| { + // Reborrow `shared` as a plain reference so the scoped output tasks + // capture the reference instead of moving fields out of the owned value. + let shared = &shared; + let report_handle = if args.skip_report_build { + None + } else { + Some(scope.spawn(|| { + run_report_task( + &policy, + validation_time, shared, - args.ccr_out_path.as_deref(), - ccr_produced_at, + args.report_json_path.as_deref(), + report_json_format, ) - }); - let compare_view_handle = scope.spawn(|| { - run_compare_view_task( - shared, - args.vrps_csv_out_path.as_deref(), - args.vaps_csv_out_path.as_deref(), - compare_view_trust_anchor, - ) - }); - let report_result = match report_handle { - Some(handle) => handle - .join() - .map_err(|_| "report task panicked".to_string()) - .and_then(|result| result), - None => Ok(ReportTaskOutput::skipped()), - }; - let ccr_result = ccr_handle - .join() - .map_err(|_| "ccr task panicked".to_string()) - .and_then(|result| result); - let compare_view_result = compare_view_handle - .join() - .map_err(|_| "compare view task panicked".to_string()) - .and_then(|result| result); - (report_result, ccr_result, compare_view_result) + })) + }; + let ccr_handle = + scope.spawn(|| run_ccr_task(shared, args.ccr_out_path.as_deref(), ccr_produced_at)); + let compare_view_handle = scope.spawn(|| { + run_compare_view_task( + shared, + args.vrps_csv_out_path.as_deref(), + args.vaps_csv_out_path.as_deref(), + compare_view_trust_anchor, + ) }); + let report_result = match report_handle { + Some(handle) => handle + .join() + .map_err(|_| "report task panicked".to_string()) + .and_then(|result| result), + None => Ok(ReportTaskOutput::skipped()), + }; + let ccr_result = ccr_handle + .join() + .map_err(|_| "ccr task panicked".to_string()) + .and_then(|result| result); + let compare_view_result = compare_view_handle + .join() + .map_err(|_| "compare view task panicked".to_string()) + .and_then(|result| result); + (report_result, ccr_result, compare_view_result) + }); let report_output = report_result?; let ccr_output = ccr_result?; let compare_view_output = compare_view_result?; @@ -374,9 +375,7 @@ pub(crate) fn run_config(args: RunConfig) -> Result<(), String> { // the transport/validation stages rather than by the compact publication // point audit, so the final summary can use those authoritative counts. if let Some(path) = args.summary_out_path.as_deref() { - let analysis_counts = timing - .as_ref() - .map(|(_, handle)| handle.counts_snapshot()); + let analysis_counts = timing.as_ref().map(|(_, handle)| handle.counts_snapshot()); write_summary( path, &shared, @@ -458,9 +457,7 @@ fn write_summary( summary.synchronized_objects += point.objects.len(); match point.repo_sync_phase.as_deref() { Some(phase) if phase.contains("fallback") => summary.rrdp_snapshot_fallbacks += 1, - Some(phase) if phase.contains("snapshot") => { - summary.rrdp_snapshot_repositories += 1 - } + Some(phase) if phase.contains("snapshot") => summary.rrdp_snapshot_repositories += 1, Some(phase) if phase.contains("delta") => summary.rrdp_delta_repositories += 1, Some(phase) if phase.contains("noop") => summary.rrdp_noop_repositories += 1, _ => {} @@ -469,13 +466,25 @@ fn write_summary( let accepted = object.result == AuditObjectResult::Ok; match &object.kind { AuditObjectKind::Manifest => { - if accepted { summary.validated_manifests += 1; } else { summary.rejected_manifests += 1; } + if accepted { + summary.validated_manifests += 1; + } else { + summary.rejected_manifests += 1; + } } AuditObjectKind::Crl => { - if accepted { summary.validated_crls += 1; } else { summary.rejected_crls += 1; } + if accepted { + summary.validated_crls += 1; + } else { + summary.rejected_crls += 1; + } } AuditObjectKind::Roa => { - if accepted { summary.validated_roas += 1; } else { summary.rejected_roas += 1; } + if accepted { + summary.validated_roas += 1; + } else { + summary.rejected_roas += 1; + } } _ => {} } @@ -522,15 +531,17 @@ fn write_summary( .and_then(|counts| { let snapshot = counts.get("rrdp_snapshot_objects_applied_total").copied(); let delta = counts.get("rrdp_delta_ops_applied_total").copied(); - snapshot.zip(delta).map(|(snapshot, delta)| snapshot + delta) + snapshot + .zip(delta) + .map(|(snapshot, delta)| snapshot + delta) }) .unwrap_or_else(|| { shared - .download_stats - .by_kind - .values() - .filter_map(|stats| stats.objects_count_total) - .sum() + .download_stats + .by_kind + .values() + .filter_map(|stats| stats.objects_count_total) + .sum() }); if downloaded_objects > 0 { summary.synchronized_objects = downloaded_objects as usize; @@ -538,8 +549,12 @@ fn write_summary( summary.synchronized_repositories = notification_uris.len(); if let Some(parent) = path.parent() { - std::fs::create_dir_all(parent) - .map_err(|error| format!("create Panda RPKI summary parent {}: {error}", parent.display()))?; + std::fs::create_dir_all(parent).map_err(|error| { + format!( + "create Panda RPKI summary parent {}: {error}", + parent.display() + ) + })?; } let bytes = serde_json::to_vec_pretty(&summary) .map_err(|error| format!("serialize Panda RPKI summary: {error}"))?; diff --git a/src/scheduler/repo_runtime/phase1_runtime.rs b/src/scheduler/repo_runtime/phase1_runtime.rs index 9057b41..caea771 100644 --- a/src/scheduler/repo_runtime/phase1_runtime.rs +++ b/src/scheduler/repo_runtime/phase1_runtime.rs @@ -228,9 +228,9 @@ impl Phase1RepoSyncRuntime { .lock() .expect("retry short rsync scopes lock poisoned") .contains(rsync_scope_uri) - { - task.retry_short_timeout = true; - } + { + task.retry_short_timeout = true; + } crate::logging::progress::emit!( "phase1_repo_task_enqueued", serde_json::json!({ diff --git a/src/scheduler/repo_runtime/runtime_trait_impl.rs b/src/scheduler/repo_runtime/runtime_trait_impl.rs index a2de18f..58e748b 100644 --- a/src/scheduler/repo_runtime/runtime_trait_impl.rs +++ b/src/scheduler/repo_runtime/runtime_trait_impl.rs @@ -64,5 +64,4 @@ impl RepoSyncRuntime for Phase1RepoSyncRuntime { } Ok(()) } - } diff --git a/src/scheduler/repo_runtime/types_and_trait.rs b/src/scheduler/repo_runtime/types_and_trait.rs index 54d1a05..a556895 100644 --- a/src/scheduler/repo_runtime/types_and_trait.rs +++ b/src/scheduler/repo_runtime/types_and_trait.rs @@ -74,5 +74,4 @@ pub trait RepoSyncRuntime: Send + Sync { } fn reset_run_state(&self) -> Result<(), String>; - } diff --git a/src/scheduler/repo_scheduler/transport_state.rs b/src/scheduler/repo_scheduler/transport_state.rs index bd106e0..274316d 100644 --- a/src/scheduler/repo_scheduler/transport_state.rs +++ b/src/scheduler/repo_scheduler/transport_state.rs @@ -92,9 +92,10 @@ impl TransportStateTables { return match record.state { RepoRuntimeState::WaitingRrdp => { if let Some(key) = record.rrdp_notification_key.as_ref() - && let Some(entry) = self.rrdp_inflight.get_mut(key) { - entry.waiting_requesters.push(requester); - } + && let Some(entry) = self.rrdp_inflight.get_mut(key) + { + entry.waiting_requesters.push(requester); + } TransportRequestAction::Waiting { state: RepoRuntimeState::WaitingRrdp, } @@ -129,102 +130,58 @@ impl TransportStateTables { } if sync_preference == SyncPreference::RrdpThenRsync - && let Some(notification_uri) = identity.notification_uri.clone() { - if let Some(entry) = self.rrdp_inflight.get_mut(¬ification_uri) { - if let Some(result) = entry.last_result.clone() { - return match result.result { - RepoTransportResultKind::Success { .. } => { - self.runtime_records.insert( - identity.clone(), - RepoRuntimeRecord { - identity, - state: RepoRuntimeState::RrdpOk, - rrdp_notification_key: Some(notification_uri), - rsync_scope_key: rsync_scope_uri, - rsync_failure_scope_key: rsync_failure_scope_uri.clone(), - requesters: vec![requester], - validation_time, - priority, - last_success: Some(result.clone()), - terminal_failure: None, - }, - ); - TransportRequestAction::ReusedSuccess(result) - } - RepoTransportResultKind::Failed { .. } => { - self.runtime_records.insert( - identity.clone(), - RepoRuntimeRecord { - identity: identity.clone(), - state: RepoRuntimeState::RrdpFailedPendingRsync, - rrdp_notification_key: Some(notification_uri), - rsync_scope_key: rsync_scope_uri.clone(), - rsync_failure_scope_key: rsync_failure_scope_uri.clone(), - requesters: vec![requester.clone()], - validation_time, - priority, - last_success: None, - terminal_failure: None, - }, - ); - self.register_rsync_request( + && let Some(notification_uri) = identity.notification_uri.clone() + { + if let Some(entry) = self.rrdp_inflight.get_mut(¬ification_uri) { + if let Some(result) = entry.last_result.clone() { + return match result.result { + RepoTransportResultKind::Success { .. } => { + self.runtime_records.insert( + identity.clone(), + RepoRuntimeRecord { identity, - requester, + state: RepoRuntimeState::RrdpOk, + rrdp_notification_key: Some(notification_uri), + rsync_scope_key: rsync_scope_uri, + rsync_failure_scope_key: rsync_failure_scope_uri.clone(), + requesters: vec![requester], validation_time, priority, - rsync_scope_uri, - rsync_failure_scope_uri, - ) - } - }; - } - - entry.waiting_requesters.push(requester.clone()); - self.runtime_records.insert( - identity.clone(), - RepoRuntimeRecord { - identity, - state: RepoRuntimeState::WaitingRrdp, - rrdp_notification_key: Some(notification_uri), - rsync_scope_key: rsync_scope_uri, - rsync_failure_scope_key: rsync_failure_scope_uri.clone(), - requesters: vec![requester], - validation_time, - priority, - last_success: None, - terminal_failure: None, - }, - ); - return TransportRequestAction::Waiting { - state: RepoRuntimeState::WaitingRrdp, + last_success: Some(result.clone()), + terminal_failure: None, + }, + ); + TransportRequestAction::ReusedSuccess(result) + } + RepoTransportResultKind::Failed { .. } => { + self.runtime_records.insert( + identity.clone(), + RepoRuntimeRecord { + identity: identity.clone(), + state: RepoRuntimeState::RrdpFailedPendingRsync, + rrdp_notification_key: Some(notification_uri), + rsync_scope_key: rsync_scope_uri.clone(), + rsync_failure_scope_key: rsync_failure_scope_uri.clone(), + requesters: vec![requester.clone()], + validation_time, + priority, + last_success: None, + terminal_failure: None, + }, + ); + self.register_rsync_request( + identity, + requester, + validation_time, + priority, + rsync_scope_uri, + rsync_failure_scope_uri, + ) + } }; } - let task = RepoTransportTask { - dedup_key: RepoDedupKey::RrdpNotify { - notification_uri: notification_uri.clone(), - }, - rsync_failure_scope_uri: None, - repo_identity: identity.clone(), - mode: RepoTransportMode::Rrdp, - retry_short_timeout: false, - tal_id: requester.tal_id.clone(), - rir_id: requester.rir_id.clone(), - validation_time, - priority, - requesters: vec![requester.clone()], - }; - self.rrdp_inflight.insert( - notification_uri.clone(), - TransportInFlightEntry { - state: TransportTaskState::Pending, - task: task.clone(), - waiting_requesters: Vec::new(), - last_result: None, - started_at: None, - finished_at: None, - }, - ); + entry.waiting_requesters.push(requester.clone()); self.runtime_records.insert( identity.clone(), RepoRuntimeRecord { @@ -240,9 +197,54 @@ impl TransportStateTables { terminal_failure: None, }, ); - return TransportRequestAction::Enqueue(task); + return TransportRequestAction::Waiting { + state: RepoRuntimeState::WaitingRrdp, + }; } + let task = RepoTransportTask { + dedup_key: RepoDedupKey::RrdpNotify { + notification_uri: notification_uri.clone(), + }, + rsync_failure_scope_uri: None, + repo_identity: identity.clone(), + mode: RepoTransportMode::Rrdp, + retry_short_timeout: false, + tal_id: requester.tal_id.clone(), + rir_id: requester.rir_id.clone(), + validation_time, + priority, + requesters: vec![requester.clone()], + }; + self.rrdp_inflight.insert( + notification_uri.clone(), + TransportInFlightEntry { + state: TransportTaskState::Pending, + task: task.clone(), + waiting_requesters: Vec::new(), + last_result: None, + started_at: None, + finished_at: None, + }, + ); + self.runtime_records.insert( + identity.clone(), + RepoRuntimeRecord { + identity, + state: RepoRuntimeState::WaitingRrdp, + rrdp_notification_key: Some(notification_uri), + rsync_scope_key: rsync_scope_uri, + rsync_failure_scope_key: rsync_failure_scope_uri.clone(), + requesters: vec![requester], + validation_time, + priority, + last_success: None, + terminal_failure: None, + }, + ); + return TransportRequestAction::Enqueue(task); + } + self.register_rsync_request( identity, requester, @@ -351,26 +353,26 @@ impl TransportStateTables { && self .rsync_failure_probe_inflight .contains_key(failure_scope_uri) - { - self.runtime_records.insert( - identity.clone(), - RepoRuntimeRecord { - identity, - state: RepoRuntimeState::WaitingRsync, - rrdp_notification_key: None, - rsync_scope_key: rsync_scope_uri, - rsync_failure_scope_key: rsync_failure_scope_uri.clone(), - requesters: vec![requester], - validation_time, - priority, - last_success: None, - terminal_failure: None, - }, - ); - return TransportRequestAction::Waiting { + { + self.runtime_records.insert( + identity.clone(), + RepoRuntimeRecord { + identity, state: RepoRuntimeState::WaitingRsync, - }; - } + rrdp_notification_key: None, + rsync_scope_key: rsync_scope_uri, + rsync_failure_scope_key: rsync_failure_scope_uri.clone(), + requesters: vec![requester], + validation_time, + priority, + last_success: None, + terminal_failure: None, + }, + ); + return TransportRequestAction::Waiting { + state: RepoRuntimeState::WaitingRsync, + }; + } } let task = RepoTransportTask { @@ -402,10 +404,10 @@ impl TransportStateTables { && !self .rsync_failure_scope_reachable .contains(failure_scope_uri) - { - self.rsync_failure_probe_inflight - .insert(failure_scope_uri.clone(), rsync_scope_uri.clone()); - } + { + self.rsync_failure_probe_inflight + .insert(failure_scope_uri.clone(), rsync_scope_uri.clone()); + } self.runtime_records.insert( identity.clone(), RepoRuntimeRecord { @@ -496,9 +498,10 @@ impl TransportStateTables { }, ); if let Some(failure_scope_uri) = record.rsync_failure_scope_key.as_ref() - && !rsync_failure_scope_reachable.contains(failure_scope_uri) { - rsync_failure_probe_inflight.insert(failure_scope_uri.clone(), rsync_scope_uri); - } + && !rsync_failure_scope_reachable.contains(failure_scope_uri) + { + rsync_failure_probe_inflight.insert(failure_scope_uri.clone(), rsync_scope_uri); + } record.state = RepoRuntimeState::WaitingRsync; follow_up_tasks.push(task); } @@ -686,29 +689,30 @@ impl TransportStateTables { } } if let Some(failure_scope_uri) = result.rsync_failure_scope_uri.as_ref() - && reusable_failure_scope.as_deref() != Some(failure_scope_uri.as_str()) { - let mut follow_up_tasks = Vec::new(); - for record in self.runtime_records.values_mut() { - if record.rsync_scope_key != *rsync_scope_uri - && record.rsync_failure_scope_key.as_deref() - == Some(failure_scope_uri.as_str()) - && matches!(record.state, RepoRuntimeState::WaitingRsync) - { - Self::schedule_rsync_for_record( - record, - &mut self.rsync_inflight, - &self.rsync_failure_by_scope, - &mut self.rsync_failure_probe_inflight, - &self.rsync_failure_scope_reachable, - &mut follow_up_tasks, - ); - } + && reusable_failure_scope.as_deref() != Some(failure_scope_uri.as_str()) + { + let mut follow_up_tasks = Vec::new(); + for record in self.runtime_records.values_mut() { + if record.rsync_scope_key != *rsync_scope_uri + && record.rsync_failure_scope_key.as_deref() + == Some(failure_scope_uri.as_str()) + && matches!(record.state, RepoRuntimeState::WaitingRsync) + { + Self::schedule_rsync_for_record( + record, + &mut self.rsync_inflight, + &self.rsync_failure_by_scope, + &mut self.rsync_failure_probe_inflight, + &self.rsync_failure_scope_reachable, + &mut follow_up_tasks, + ); } - return Ok(TransportCompletion { - released_requesters, - follow_up_tasks, - }); } + return Ok(TransportCompletion { + released_requesters, + follow_up_tasks, + }); + } Ok(TransportCompletion { released_requesters, follow_up_tasks: Vec::new(), diff --git a/src/scheduler/repo_worker/executors.rs b/src/scheduler/repo_worker/executors.rs index 62c580d..91ed467 100644 --- a/src/scheduler/repo_worker/executors.rs +++ b/src/scheduler/repo_worker/executors.rs @@ -59,16 +59,19 @@ impl RepoTransportExecutor for LiveRrdpTransportExecutor RepoTransportExecutor for LiveRrdpTransportExecutor { - RepoTransportResultEnvelope { - dedup_key: task.dedup_key, - rsync_failure_scope_uri: task.rsync_failure_scope_uri, - repo_identity: task.repo_identity, - mode: RepoTransportMode::Rrdp, - tal_id: task.tal_id, - rir_id: task.rir_id, - timing_ms: started.elapsed().as_millis() as u64, - result: RepoTransportResultKind::Failed { - detail: err.to_string(), - warnings: Vec::new(), - error_class: RepoTransportErrorClass::Unknown, - }, - } - } + Err(err) => RepoTransportResultEnvelope { + dedup_key: task.dedup_key, + rsync_failure_scope_uri: task.rsync_failure_scope_uri, + repo_identity: task.repo_identity, + mode: RepoTransportMode::Rrdp, + tal_id: task.tal_id, + rir_id: task.rir_id, + timing_ms: started.elapsed().as_millis() as u64, + result: RepoTransportResultKind::Failed { + detail: err.to_string(), + warnings: Vec::new(), + error_class: RepoTransportErrorClass::Unknown, + }, + }, } } } @@ -187,22 +188,20 @@ impl RepoTransportExecutor for LiveRsyncTransportExec warnings: Vec::new(), }, }, - Err(err) => { - RepoTransportResultEnvelope { - dedup_key: task.dedup_key, - rsync_failure_scope_uri: task.rsync_failure_scope_uri, - repo_identity: task.repo_identity, - mode: RepoTransportMode::Rsync, - tal_id: task.tal_id, - rir_id: task.rir_id, - timing_ms: started.elapsed().as_millis() as u64, - result: RepoTransportResultKind::Failed { - detail: err.to_string(), - warnings: Vec::new(), - error_class: RepoTransportErrorClass::Unknown, - }, - } - } + Err(err) => RepoTransportResultEnvelope { + dedup_key: task.dedup_key, + rsync_failure_scope_uri: task.rsync_failure_scope_uri, + repo_identity: task.repo_identity, + mode: RepoTransportMode::Rsync, + tal_id: task.tal_id, + rir_id: task.rir_id, + timing_ms: started.elapsed().as_millis() as u64, + result: RepoTransportResultKind::Failed { + detail: err.to_string(), + warnings: Vec::new(), + error_class: RepoTransportErrorClass::Unknown, + }, + }, } } } diff --git a/src/scheduler/repo_worker/pools.rs b/src/scheduler/repo_worker/pools.rs index 499071f..79e009a 100644 --- a/src/scheduler/repo_worker/pools.rs +++ b/src/scheduler/repo_worker/pools.rs @@ -100,9 +100,10 @@ impl RepoWorkerPool { let mut first_err: Option = None; for handle in self.workers.drain(..) { if let Err(e) = handle.join() - && first_err.is_none() { - first_err = Some(format!("join repo worker failed: {e:?}")); - } + && first_err.is_none() + { + first_err = Some(format!("join repo worker failed: {e:?}")); + } } if let Some(err) = first_err { @@ -187,9 +188,10 @@ impl RepoTransportWorkerPool { let mut first_err: Option = None; for handle in self.workers.drain(..) { if let Err(e) = handle.join() - && first_err.is_none() { - first_err = Some(format!("join repo transport worker failed: {e:?}")); - } + && first_err.is_none() + { + first_err = Some(format!("join repo transport worker failed: {e:?}")); + } } if let Some(err) = first_err { return Err(err); diff --git a/src/ta_constraints/implementation.rs b/src/ta_constraints/implementation.rs index 0bdd192..30c7e4c 100644 --- a/src/ta_constraints/implementation.rs +++ b/src/ta_constraints/implementation.rs @@ -1,6 +1,5 @@ // Interval indexes and trust-anchor resource constraint evaluation. - use std::collections::{BTreeMap, BTreeSet}; use std::net::{IpAddr, Ipv4Addr, Ipv6Addr}; use std::ops::Deref; @@ -215,7 +214,6 @@ impl TaConstraintsByTal { }) .collect() } - } fn adjacent_constraints_path(source: &TalSource) -> Option { diff --git a/src/validation/ca_path/certificate_checks.rs b/src/validation/ca_path/certificate_checks.rs index 8eebb32..1965bad 100644 --- a/src/validation/ca_path/certificate_checks.rs +++ b/src/validation/ca_path/certificate_checks.rs @@ -24,10 +24,9 @@ fn verify_child_signature( child: &X509Certificate<'_>, issuer_spki: &SubjectPublicKeyInfo<'_>, ) -> Result<(), CaPathError> { - - child - .verify_signature(Some(issuer_spki)) - .map_err(|e| CaPathError::ChildSignatureInvalid(e.to_string())) + child + .verify_signature(Some(issuer_spki)) + .map_err(|e| CaPathError::ChildSignatureInvalid(e.to_string())) } fn validate_child_aki_matches_issuer_ski( diff --git a/src/validation/ca_path/ip_resources.rs b/src/validation/ca_path/ip_resources.rs index f8c449e..d28614f 100644 --- a/src/validation/ca_path/ip_resources.rs +++ b/src/validation/ca_path/ip_resources.rs @@ -3,10 +3,7 @@ fn ip_resources_by_afi_items( set: &IpResourceSet, ) -> Result< - std::collections::BTreeMap< - crate::model::rc::Afi, - Vec, - >, + std::collections::BTreeMap>, CaPathError, > { let mut m: std::collections::BTreeMap< @@ -96,9 +93,7 @@ fn ip_items_to_merged_intervals( let mut intervals = Vec::new(); for item in items { match item { - crate::model::rc::IpAddressOrRange::Prefix(p) => { - intervals.push(prefix_to_range(p)) - } + crate::model::rc::IpAddressOrRange::Prefix(p) => intervals.push(prefix_to_range(p)), crate::model::rc::IpAddressOrRange::Range(r) => { intervals.push((r.min.clone(), r.max.clone())) } diff --git a/src/validation/manifest/helpers.rs b/src/validation/manifest/helpers.rs index 6ad38fa..b7623ad 100644 --- a/src/validation/manifest/helpers.rs +++ b/src/validation/manifest/helpers.rs @@ -65,8 +65,8 @@ fn validate_manifest_embedded_ee_cert_path( if !rem.is_empty() { return Err(CertPathError::IssuerSpkiTrailingBytes(rem.len()).into()); } - let issuer_crl = crate::model::crl::RpkixCrl::decode_der(crl_bytes) - .map_err(CertPathError::from)?; + let issuer_crl = + crate::model::crl::RpkixCrl::decode_der(crl_bytes).map_err(CertPathError::from)?; let revoked_serials = issuer_crl .revoked_certs .iter() diff --git a/src/validation/manifest/models_and_process.rs b/src/validation/manifest/models_and_process.rs index 245d7d3..71b6675 100644 --- a/src/validation/manifest/models_and_process.rs +++ b/src/validation/manifest/models_and_process.rs @@ -502,9 +502,7 @@ pub(crate) fn try_build_fresh_publication_point_with_timing( // - If manifestNumber is higher, require thisUpdate to be more recent than the previously // validated thisUpdate. let anti_rollback_started = std::time::Instant::now(); - if let Some(old_meta) = store - .get_manifest_anti_rollback_meta(manifest_rsync_uri)? - { + if let Some(old_meta) = store.get_manifest_anti_rollback_meta(manifest_rsync_uri)? { timing.anti_rollback_meta_hit = true; if old_meta.manifest_rsync_uri == manifest_rsync_uri { let new_num = manifest.manifest.manifest_number.bytes_be.as_slice(); diff --git a/src/validation/objects/object_validation.rs b/src/validation/objects/object_validation.rs index 487abd0..b209a3a 100644 --- a/src/validation/objects/object_validation.rs +++ b/src/validation/objects/object_validation.rs @@ -16,17 +16,47 @@ fn process_roa_with_issuer( ta_constraints: Option<&crate::ta_constraints::TaConstraints>, ) -> Result, ObjectValidateError> { let roa = { - let _span = timing.as_ref().map(|t| t.span_phase("objects_roa_decode_and_validate_total")); - RoaObject::decode_der_with_strict_options(file.bytes().map_err(ObjectValidateError::BytesLoad)?, strict_cms_der, strict_name)? + let _span = timing + .as_ref() + .map(|t| t.span_phase("objects_roa_decode_and_validate_total")); + RoaObject::decode_der_with_strict_options( + file.bytes().map_err(ObjectValidateError::BytesLoad)?, + strict_cms_der, + strict_name, + )? }; roa.validate_embedded_ee_cert()?; roa.signed_object.verify()?; let ee = &roa.signed_object.signed_data.certificates[0]; - let crl_uri = choose_crl_uri_for_certificate(ee.resource_cert.tbs.extensions.crl_distribution_points_uris.as_ref(), crl_states)?; + let crl_uri = choose_crl_uri_for_certificate( + ee.resource_cert + .tbs + .extensions + .crl_distribution_points_uris + .as_ref(), + crl_states, + )?; let verified_crl = ensure_issuer_crl_verified(crl_uri, crl_states, issuer_ca_der)?; - validate_signed_object_ee_cert_path_fast(ee, issuer_ca, issuer_spki, &verified_crl.crl, &verified_crl.revoked_serials, issuer_ca_rsync_uri, Some(crl_uri), validation_time)?; - let ee_vrs = validate_ee_resources_for_mode(&ee.resource_cert, issuer_effective_ip, issuer_effective_as, issuer_resources_index, resource_validation_mode)?; - if let Some(constraints) = ta_constraints { constraints.validate_ee_certificate(&ee.resource_cert)?; } + validate_signed_object_ee_cert_path_fast( + ee, + issuer_ca, + issuer_spki, + &verified_crl.crl, + &verified_crl.revoked_serials, + issuer_ca_rsync_uri, + Some(crl_uri), + validation_time, + )?; + let ee_vrs = validate_ee_resources_for_mode( + &ee.resource_cert, + issuer_effective_ip, + issuer_effective_as, + issuer_resources_index, + resource_validation_mode, + )?; + if let Some(constraints) = ta_constraints { + constraints.validate_ee_certificate(&ee.resource_cert)?; + } roa_to_vrps_with_vrs(&roa, ee_vrs.ip.as_ref()) } @@ -48,21 +78,52 @@ fn process_roa_with_issuer_parallel( ta_constraints: Option<&crate::ta_constraints::TaConstraints>, ) -> Result, ObjectValidateError> { let roa = { - let _span = timing.as_ref().map(|t| t.span_phase("objects_roa_decode_and_validate_total")); - RoaObject::decode_der_with_strict_options(file.bytes().map_err(ObjectValidateError::BytesLoad)?, strict_cms_der, strict_name)? + let _span = timing + .as_ref() + .map(|t| t.span_phase("objects_roa_decode_and_validate_total")); + RoaObject::decode_der_with_strict_options( + file.bytes().map_err(ObjectValidateError::BytesLoad)?, + strict_cms_der, + strict_name, + )? }; roa.validate_embedded_ee_cert()?; roa.signed_object.verify()?; let ee = &roa.signed_object.signed_data.certificates[0]; let (crl_uri, verified_crl) = { let mut states = crl_states.lock().expect("parallel issuer CRL state lock"); - let uri = choose_crl_uri_for_certificate(ee.resource_cert.tbs.extensions.crl_distribution_points_uris.as_ref(), &states)?.to_string(); + let uri = choose_crl_uri_for_certificate( + ee.resource_cert + .tbs + .extensions + .crl_distribution_points_uris + .as_ref(), + &states, + )? + .to_string(); let value = ensure_issuer_crl_verified(&uri, &mut states, issuer_ca_der)?; (uri, value) }; - validate_signed_object_ee_cert_path_fast(ee, issuer_ca, issuer_spki, &verified_crl.crl, &verified_crl.revoked_serials, issuer_ca_rsync_uri, Some(crl_uri.as_str()), validation_time)?; - let ee_vrs = validate_ee_resources_for_mode(&ee.resource_cert, issuer_effective_ip, issuer_effective_as, issuer_resources_index, resource_validation_mode)?; - if let Some(constraints) = ta_constraints { constraints.validate_ee_certificate(&ee.resource_cert)?; } + validate_signed_object_ee_cert_path_fast( + ee, + issuer_ca, + issuer_spki, + &verified_crl.crl, + &verified_crl.revoked_serials, + issuer_ca_rsync_uri, + Some(crl_uri.as_str()), + validation_time, + )?; + let ee_vrs = validate_ee_resources_for_mode( + &ee.resource_cert, + issuer_effective_ip, + issuer_effective_as, + issuer_resources_index, + resource_validation_mode, + )?; + if let Some(constraints) = ta_constraints { + constraints.validate_ee_certificate(&ee.resource_cert)?; + } roa_to_vrps_with_vrs(&roa, ee_vrs.ip.as_ref()) } @@ -84,17 +145,50 @@ fn process_aspa_with_issuer( ta_constraints: Option<&crate::ta_constraints::TaConstraints>, ) -> Result { let aspa = { - let _span = timing.as_ref().map(|t| t.span_phase("objects_aspa_decode_and_validate_total")); - AspaObject::decode_der_with_strict_options(file.bytes().map_err(ObjectValidateError::BytesLoad)?, strict_cms_der, strict_name)? + let _span = timing + .as_ref() + .map(|t| t.span_phase("objects_aspa_decode_and_validate_total")); + AspaObject::decode_der_with_strict_options( + file.bytes().map_err(ObjectValidateError::BytesLoad)?, + strict_cms_der, + strict_name, + )? }; aspa.validate_embedded_ee_cert()?; aspa.signed_object.verify()?; let ee = &aspa.signed_object.signed_data.certificates[0]; - let crl_uri = choose_crl_uri_for_certificate(ee.resource_cert.tbs.extensions.crl_distribution_points_uris.as_ref(), crl_states)?; + let crl_uri = choose_crl_uri_for_certificate( + ee.resource_cert + .tbs + .extensions + .crl_distribution_points_uris + .as_ref(), + crl_states, + )?; let verified_crl = ensure_issuer_crl_verified(crl_uri, crl_states, issuer_ca_der)?; - validate_signed_object_ee_cert_path_fast(ee, issuer_ca, issuer_spki, &verified_crl.crl, &verified_crl.revoked_serials, issuer_ca_rsync_uri, Some(crl_uri), validation_time)?; - let ee_vrs = validate_ee_resources_for_mode(&ee.resource_cert, issuer_effective_ip, issuer_effective_as, issuer_resources_index, resource_validation_mode)?; - if let Some(constraints) = ta_constraints { constraints.validate_ee_certificate(&ee.resource_cert)?; } + validate_signed_object_ee_cert_path_fast( + ee, + issuer_ca, + issuer_spki, + &verified_crl.crl, + &verified_crl.revoked_serials, + issuer_ca_rsync_uri, + Some(crl_uri), + validation_time, + )?; + let ee_vrs = validate_ee_resources_for_mode( + &ee.resource_cert, + issuer_effective_ip, + issuer_effective_as, + issuer_resources_index, + resource_validation_mode, + )?; + if let Some(constraints) = ta_constraints { + constraints.validate_ee_certificate(&ee.resource_cert)?; + } validate_aspa_customer_in_vrs(&aspa, ee_vrs.asn.as_ref())?; - Ok(AspaAttestation { customer_as_id: aspa.aspa.customer_as_id, provider_as_ids: aspa.aspa.provider_as_ids.clone() }) + Ok(AspaAttestation { + customer_as_id: aspa.aspa.customer_as_id, + provider_as_ids: aspa.aspa.provider_as_ids.clone(), + }) } diff --git a/src/validation/objects/outputs.rs b/src/validation/objects/outputs.rs index 50f8c56..b250371 100644 --- a/src/validation/objects/outputs.rs +++ b/src/validation/objects/outputs.rs @@ -2,21 +2,33 @@ const RFC_NONE: &[RfcRef] = &[]; const RFC_CRLDP: &[RfcRef] = &[RfcRef("RFC 6487 §4.8.6")]; -const RFC_CRLDP_AND_LOCKED_PACK: &[RfcRef] = &[RfcRef("RFC 6487 §4.8.6"), RfcRef("RFC 9286 §4.2.1")]; +const RFC_CRLDP_AND_LOCKED_PACK: &[RfcRef] = + &[RfcRef("RFC 6487 §4.8.6"), RfcRef("RFC 9286 §4.2.1")]; fn ber_compatible_cms_warning(der: &[u8], rsync_uri: &str, object_kind: &str) -> Option { let strict_error = RpkiSignedObject::strict_cms_der_error(der)?; - Some(Warning::new(format!("accepted BER-compatible CMS encoding for {object_kind}: {rsync_uri}: {strict_error}")) + Some( + Warning::new(format!( + "accepted BER-compatible CMS encoding for {object_kind}: {rsync_uri}: {strict_error}" + )) .with_category(WarningCategory::BerCompatibleCmsEncoding) .with_rfc_refs(&[RfcRef("X.690 §10"), RfcRef("RFC 6488 §2")]) - .with_context(rsync_uri)) + .with_context(rsync_uri), + ) } fn ber_compatible_cms_warning_for_file(file: &PackFile, object_kind: &str) -> Option { ber_compatible_cms_warning(file.bytes().ok()?, &file.rsync_uri, object_kind) } -fn decode_resource_certificate_with_policy(der: &[u8], policy: &Policy) -> Result { - if policy.strict.name { ResourceCertificate::decode_der_with_strict_name(der) } else { ResourceCertificate::decode_der(der) } +fn decode_resource_certificate_with_policy( + der: &[u8], + policy: &Policy, +) -> Result { + if policy.strict.name { + ResourceCertificate::decode_der_with_strict_name(der) + } else { + ResourceCertificate::decode_der(der) + } } #[derive(Clone, Debug)] @@ -39,13 +51,24 @@ pub(crate) struct IssuerResourcesIndex { } fn extra_rfc_refs_for_crl_selection(error: &ObjectValidateError) -> &'static [RfcRef] { - match error { ObjectValidateError::MissingCrlDpUris => RFC_CRLDP, ObjectValidateError::CrlNotFound(_) => RFC_CRLDP_AND_LOCKED_PACK, _ => RFC_NONE } + match error { + ObjectValidateError::MissingCrlDpUris => RFC_CRLDP, + ObjectValidateError::CrlNotFound(_) => RFC_CRLDP_AND_LOCKED_PACK, + _ => RFC_NONE, + } } #[derive(Clone, Debug, PartialEq, Eq)] -pub struct Vrp { pub asn: u32, pub prefix: IpPrefix, pub max_length: u16 } +pub struct Vrp { + pub asn: u32, + pub prefix: IpPrefix, + pub max_length: u16, +} #[derive(Clone, Debug, PartialEq, Eq)] -pub struct AspaAttestation { pub customer_as_id: u32, pub provider_as_ids: Vec } +pub struct AspaAttestation { + pub customer_as_id: u32, + pub provider_as_ids: Vec, +} #[derive(Clone, Debug, PartialEq, Eq)] pub struct RouterKeyPayload { pub as_id: u32, @@ -76,7 +99,9 @@ pub struct ObjectsStats { } #[derive(Debug)] -pub(crate) struct RoaTaskOk { pub(crate) vrps: Vec } +pub(crate) struct RoaTaskOk { + pub(crate) vrps: Vec, +} #[derive(Debug)] pub(crate) struct RoaTaskResult { pub(crate) publication_point_id: u64, diff --git a/src/validation/objects/parallel_stage.rs b/src/validation/objects/parallel_stage.rs index ca34013..6444b7f 100644 --- a/src/validation/objects/parallel_stage.rs +++ b/src/validation/objects/parallel_stage.rs @@ -88,11 +88,9 @@ pub(crate) fn prepare_publication_point_for_parallel_roa_and_ta_constraints< ); return ParallelObjectsPrepare::Complete(empty(stats, warnings)); } - if let Some(warning) = ber_compatible_cms_warning( - publication_point.manifest_bytes(), - manifest_uri, - "manifest", - ) { + if let Some(warning) = + ber_compatible_cms_warning(publication_point.manifest_bytes(), manifest_uri, "manifest") + { warnings.push(warning); } let issuer_ca = match decode_resource_certificate_with_policy(issuer_ca_der, policy) { @@ -138,22 +136,17 @@ pub(crate) fn prepare_publication_point_for_parallel_roa_and_ta_constraints< .iter() .filter(|file| file.rsync_uri.ends_with(".crl")) .filter_map(|file| { - file.bytes_cloned().ok().map(|bytes| { - ( - file.rsync_uri.clone(), - IssuerCrlState::Pending { bytes }, - ) - }) + file.bytes_cloned() + .ok() + .map(|bytes| (file.rsync_uri.clone(), IssuerCrlState::Pending { bytes })) }) .collect::>(); if crl_states.is_empty() && (stats.roa_total > 0 || stats.aspa_total > 0) { stats.publication_point_dropped = true; warnings.push( - Warning::new( - "dropping publication point: no CRL files in validated publication point", - ) - .with_rfc_refs(&[RfcRef("RFC 6487 §4.8.6"), RfcRef("RFC 9286 §7")]) - .with_context(manifest_uri), + Warning::new("dropping publication point: no CRL files in validated publication point") + .with_rfc_refs(&[RfcRef("RFC 6487 §4.8.6"), RfcRef("RFC 9286 §7")]) + .with_context(manifest_uri), ); return ParallelObjectsPrepare::Complete(empty(stats, warnings)); } @@ -217,9 +210,7 @@ pub(crate) fn reduce_parallel_roa_stage( for (index, file) in shared.locked_files.iter().enumerate() { if file.rsync_uri.ends_with(".roa") { let result = match result_iter.peek() { - Some(result) if result.index == index => { - result_iter.next().expect("peeked result") - } + Some(result) if result.index == index => result_iter.next().expect("peeked result"), Some(result) => { return Err(format!( "unexpected ROA task result index {} while reducing {}", @@ -254,12 +245,9 @@ pub(crate) fn reduce_parallel_roa_stage( let mut refs = vec![RfcRef("RFC 6488 §3"), RfcRef("RFC 9582 §4-§5")]; refs.extend_from_slice(extra_rfc_refs_for_crl_selection(&error)); warnings.push( - Warning::new(format!( - "dropping invalid ROA: {}: {error}", - file.rsync_uri - )) - .with_rfc_refs(&refs) - .with_context(&file.rsync_uri), + Warning::new(format!("dropping invalid ROA: {}: {error}", file.rsync_uri)) + .with_rfc_refs(&refs) + .with_context(&file.rsync_uri), ); } } diff --git a/src/validation/objects/resource_validation.rs b/src/validation/objects/resource_validation.rs index 594944f..50e7861 100644 --- a/src/validation/objects/resource_validation.rs +++ b/src/validation/objects/resource_validation.rs @@ -37,8 +37,8 @@ fn ensure_issuer_crl_verified( IssuerCrlState::Verified(v) => Ok(Arc::clone(v)), IssuerCrlState::Pending { bytes } => { let der = std::mem::take(bytes); - let crl = crate::model::crl::RpkixCrl::decode_der(&der) - .map_err(CertPathError::CrlDecode)?; + let crl = + crate::model::crl::RpkixCrl::decode_der(&der).map_err(CertPathError::CrlDecode)?; crl.verify_signature_with_issuer_certificate_der(issuer_ca_der) .map_err(CertPathError::CrlVerify)?; @@ -227,10 +227,7 @@ fn validate_aspa_customer_in_vrs( Ok(()) } -fn as_resource_set_contains_asn( - resources: &crate::model::rc::AsResourceSet, - asn: u32, -) -> bool { +fn as_resource_set_contains_asn(resources: &crate::model::rc::AsResourceSet, asn: u32) -> bool { let Some(choice) = resources.asnum.as_ref() else { return false; }; @@ -605,13 +602,15 @@ fn build_issuer_resources_index( if let Some(asr) = issuer_effective_as { if let Some(choice) = asr.asnum.as_ref() - && !matches!(choice, AsIdentifierChoice::Inherit) { - idx.asnum = Some(as_choice_to_merged_intervals(choice)); - } + && !matches!(choice, AsIdentifierChoice::Inherit) + { + idx.asnum = Some(as_choice_to_merged_intervals(choice)); + } if let Some(choice) = asr.rdi.as_ref() - && !matches!(choice, AsIdentifierChoice::Inherit) { - idx.rdi = Some(as_choice_to_merged_intervals(choice)); - } + && !matches!(choice, AsIdentifierChoice::Inherit) + { + idx.rdi = Some(as_choice_to_merged_intervals(choice)); + } } idx diff --git a/src/validation/objects/serial_processing.rs b/src/validation/objects/serial_processing.rs index 4902f3f..66444b3 100644 --- a/src/validation/objects/serial_processing.rs +++ b/src/validation/objects/serial_processing.rs @@ -9,8 +9,15 @@ pub fn process_publication_point_for_issuer( timing: Option<&TimingHandle>, ) -> ObjectsOutput { process_publication_point_for_issuer_with_ta_constraints( - publication_point, policy, issuer_ca_der, issuer_ca_rsync_uri, - issuer_effective_ip, issuer_effective_as, validation_time, timing, None, + publication_point, + policy, + issuer_ca_der, + issuer_ca_rsync_uri, + issuer_effective_ip, + issuer_effective_as, + validation_time, + timing, + None, ) } @@ -28,30 +35,58 @@ pub fn process_publication_point_for_issuer_with_ta_constraints, audit: Vec| ObjectsOutput { - vrps: Vec::new(), aspas: Vec::new(), router_keys: Vec::new(), warnings, stats, audit, - }; + let empty = + |stats: ObjectsStats, warnings: Vec, audit: Vec| ObjectsOutput { + vrps: Vec::new(), + aspas: Vec::new(), + router_keys: Vec::new(), + warnings, + stats, + audit, + }; if let Err(error) = ManifestObject::decode_der_with_strict_options( - publication_point.manifest_bytes(), policy.strict.cms_der, policy.strict.name, + publication_point.manifest_bytes(), + policy.strict.cms_der, + policy.strict.name, ) { stats.publication_point_dropped = true; - warnings.push(Warning::new(format!("dropping publication point: manifest decode failed: {error}")) - .with_rfc_refs(&[RfcRef("RFC 9286 §4"), RfcRef("RFC 9286 §6.6")]).with_context(manifest_uri)); + warnings.push( + Warning::new(format!( + "dropping publication point: manifest decode failed: {error}" + )) + .with_rfc_refs(&[RfcRef("RFC 9286 §4"), RfcRef("RFC 9286 §6.6")]) + .with_context(manifest_uri), + ); return empty(stats, warnings, audit); } - if let Some(warning) = ber_compatible_cms_warning(publication_point.manifest_bytes(), manifest_uri, "manifest") { warnings.push(warning); } + if let Some(warning) = + ber_compatible_cms_warning(publication_point.manifest_bytes(), manifest_uri, "manifest") + { + warnings.push(warning); + } let issuer_ca = match decode_resource_certificate_with_policy(issuer_ca_der, policy) { Ok(value) => value, Err(error) => { stats.publication_point_dropped = true; - warnings.push(Warning::new(format!("dropping publication point: issuer CA decode failed: {error}")) - .with_rfc_refs(&[RfcRef("RFC 6487 §7.2"), RfcRef("RFC 5280 §6.1")]).with_context(manifest_uri)); + warnings.push( + Warning::new(format!( + "dropping publication point: issuer CA decode failed: {error}" + )) + .with_rfc_refs(&[RfcRef("RFC 6487 §7.2"), RfcRef("RFC 5280 §6.1")]) + .with_context(manifest_uri), + ); return empty(stats, warnings, audit); } }; @@ -59,59 +94,205 @@ pub fn process_publication_point_for_issuer_with_ta_constraints spki, Ok((remaining, _)) => { stats.publication_point_dropped = true; - warnings.push(Warning::new(format!("dropping publication point: issuer SPKI has {} trailing bytes", remaining.len())).with_context(manifest_uri)); + warnings.push( + Warning::new(format!( + "dropping publication point: issuer SPKI has {} trailing bytes", + remaining.len() + )) + .with_context(manifest_uri), + ); return empty(stats, warnings, audit); } Err(error) => { stats.publication_point_dropped = true; - warnings.push(Warning::new(format!("dropping publication point: issuer SPKI parse failed: {error}")).with_context(manifest_uri)); + warnings.push( + Warning::new(format!( + "dropping publication point: issuer SPKI parse failed: {error}" + )) + .with_context(manifest_uri), + ); return empty(stats, warnings, audit); } }; - let mut crl_states: std::collections::HashMap = match files.iter().filter(|f| f.rsync_uri.ends_with(".crl")).map(|file| { - Ok((file.rsync_uri.clone(), IssuerCrlState::Pending { bytes: file.bytes_cloned().map_err(|e| format!("CRL bytes load failed: {e}"))? })) - }).collect::>() { Ok(value) => value, Err(error) => { stats.publication_point_dropped = true; warnings.push(Warning::new(error).with_context(manifest_uri)); return empty(stats, warnings, audit); } }; - let issuer_resources_index = build_issuer_resources_index(issuer_effective_ip, issuer_effective_as); + let mut crl_states: std::collections::HashMap = match files + .iter() + .filter(|f| f.rsync_uri.ends_with(".crl")) + .map(|file| { + Ok(( + file.rsync_uri.clone(), + IssuerCrlState::Pending { + bytes: file + .bytes_cloned() + .map_err(|e| format!("CRL bytes load failed: {e}"))?, + }, + )) + }) + .collect::>() + { + Ok(value) => value, + Err(error) => { + stats.publication_point_dropped = true; + warnings.push(Warning::new(error).with_context(manifest_uri)); + return empty(stats, warnings, audit); + } + }; + let issuer_resources_index = + build_issuer_resources_index(issuer_effective_ip, issuer_effective_as); if crl_states.is_empty() && (stats.roa_total > 0 || stats.aspa_total > 0) { stats.publication_point_dropped = true; - warnings.push(Warning::new("dropping publication point: no CRL files in validated publication point").with_rfc_refs(&[RfcRef("RFC 6487 §4.8.6"), RfcRef("RFC 9286 §7")]).with_context(manifest_uri)); + warnings.push( + Warning::new("dropping publication point: no CRL files in validated publication point") + .with_rfc_refs(&[RfcRef("RFC 6487 §4.8.6"), RfcRef("RFC 9286 §7")]) + .with_context(manifest_uri), + ); return empty(stats, warnings, audit); } let mut vrps = Vec::new(); let mut aspas = Vec::new(); for (index, file) in files.iter().enumerate() { if file.rsync_uri.ends_with(".roa") { - let result = process_roa_with_issuer(file, issuer_ca_der, &issuer_ca, &issuer_spki, issuer_ca_rsync_uri, &mut crl_states, &issuer_resources_index, issuer_effective_ip, issuer_effective_as, validation_time, timing, policy.strict.cms_der, policy.strict.name, policy.resource_validation_mode, ta_constraints); + let result = process_roa_with_issuer( + file, + issuer_ca_der, + &issuer_ca, + &issuer_spki, + issuer_ca_rsync_uri, + &mut crl_states, + &issuer_resources_index, + issuer_effective_ip, + issuer_effective_as, + validation_time, + timing, + policy.strict.cms_der, + policy.strict.name, + policy.resource_validation_mode, + ta_constraints, + ); match result { Ok(mut values) => { stats.roa_ok += 1; vrps.append(&mut values); - audit.push(ObjectAuditEntry { rsync_uri: file.rsync_uri.clone(), sha256_hex: sha256_hex_from_32(&file.sha256), kind: AuditObjectKind::Roa, result: AuditObjectResult::Ok, detail: None }); - if let Some(warning) = ber_compatible_cms_warning_for_file(file, "ROA") { warnings.push(warning); } + audit.push(ObjectAuditEntry { + rsync_uri: file.rsync_uri.clone(), + sha256_hex: sha256_hex_from_32(&file.sha256), + kind: AuditObjectKind::Roa, + result: AuditObjectResult::Ok, + detail: None, + }); + if let Some(warning) = ber_compatible_cms_warning_for_file(file, "ROA") { + warnings.push(warning); + } } Err(error) => match policy.signed_object_failure_policy { SignedObjectFailurePolicy::DropObject => { - audit.push(ObjectAuditEntry { rsync_uri: file.rsync_uri.clone(), sha256_hex: sha256_hex_from_32(&file.sha256), kind: AuditObjectKind::Roa, result: AuditObjectResult::Error, detail: Some(error.to_string()) }); - let mut refs = vec![RfcRef("RFC 6488 §3"), RfcRef("RFC 9582 §4-§5")]; refs.extend_from_slice(extra_rfc_refs_for_crl_selection(&error)); - warnings.push(Warning::new(format!("dropping invalid ROA: {}: {error}", file.rsync_uri)).with_rfc_refs(&refs).with_context(&file.rsync_uri)); + audit.push(ObjectAuditEntry { + rsync_uri: file.rsync_uri.clone(), + sha256_hex: sha256_hex_from_32(&file.sha256), + kind: AuditObjectKind::Roa, + result: AuditObjectResult::Error, + detail: Some(error.to_string()), + }); + let mut refs = vec![RfcRef("RFC 6488 §3"), RfcRef("RFC 9582 §4-§5")]; + refs.extend_from_slice(extra_rfc_refs_for_crl_selection(&error)); + warnings.push( + Warning::new(format!( + "dropping invalid ROA: {}: {error}", + file.rsync_uri + )) + .with_rfc_refs(&refs) + .with_context(&file.rsync_uri), + ); } SignedObjectFailurePolicy::DropPublicationPoint => { stats.publication_point_dropped = true; - warnings.push(Warning::new(format!("dropping publication point due to invalid ROA: {}: {error}", file.rsync_uri)).with_context(manifest_uri)); - for later in files.iter().skip(index + 1) { if later.rsync_uri.ends_with(".roa") || later.rsync_uri.ends_with(".asa") { audit.push(ObjectAuditEntry { rsync_uri: later.rsync_uri.clone(), sha256_hex: sha256_hex_from_32(&later.sha256), kind: if later.rsync_uri.ends_with(".roa") { AuditObjectKind::Roa } else { AuditObjectKind::Aspa }, result: AuditObjectResult::Skipped, detail: Some("skipped due to signed_object_failure_policy=drop_publication_point".to_string()) }); } } + warnings.push( + Warning::new(format!( + "dropping publication point due to invalid ROA: {}: {error}", + file.rsync_uri + )) + .with_context(manifest_uri), + ); + for later in files.iter().skip(index + 1) { + if later.rsync_uri.ends_with(".roa") + || later.rsync_uri.ends_with(".asa") + { + audit.push(ObjectAuditEntry { rsync_uri: later.rsync_uri.clone(), sha256_hex: sha256_hex_from_32(&later.sha256), kind: if later.rsync_uri.ends_with(".roa") { AuditObjectKind::Roa } else { AuditObjectKind::Aspa }, result: AuditObjectResult::Skipped, detail: Some("skipped due to signed_object_failure_policy=drop_publication_point".to_string()) }); + } + } return empty(stats, warnings, audit); } }, } } else if file.rsync_uri.ends_with(".asa") { - match process_aspa_with_issuer(file, issuer_ca_der, &issuer_ca, &issuer_spki, issuer_ca_rsync_uri, &mut crl_states, &issuer_resources_index, issuer_effective_ip, issuer_effective_as, validation_time, timing, policy.strict.cms_der, policy.strict.name, policy.resource_validation_mode, ta_constraints) { - Ok(attestation) => { stats.aspa_ok += 1; aspas.push(attestation); audit.push(ObjectAuditEntry { rsync_uri: file.rsync_uri.clone(), sha256_hex: sha256_hex_from_32(&file.sha256), kind: AuditObjectKind::Aspa, result: AuditObjectResult::Ok, detail: None }); if let Some(warning) = ber_compatible_cms_warning_for_file(file, "ASPA") { warnings.push(warning); } } - Err(error) => match policy.signed_object_failure_policy { - SignedObjectFailurePolicy::DropObject => { audit.push(ObjectAuditEntry { rsync_uri: file.rsync_uri.clone(), sha256_hex: sha256_hex_from_32(&file.sha256), kind: AuditObjectKind::Aspa, result: AuditObjectResult::Error, detail: Some(error.to_string()) }); warnings.push(Warning::new(format!("dropping invalid ASPA: {}: {error}", file.rsync_uri)).with_context(&file.rsync_uri)); } - SignedObjectFailurePolicy::DropPublicationPoint => { stats.publication_point_dropped = true; warnings.push(Warning::new(format!("dropping publication point due to invalid ASPA: {}: {error}", file.rsync_uri)).with_context(manifest_uri)); return empty(stats, warnings, audit); } + match process_aspa_with_issuer( + file, + issuer_ca_der, + &issuer_ca, + &issuer_spki, + issuer_ca_rsync_uri, + &mut crl_states, + &issuer_resources_index, + issuer_effective_ip, + issuer_effective_as, + validation_time, + timing, + policy.strict.cms_der, + policy.strict.name, + policy.resource_validation_mode, + ta_constraints, + ) { + Ok(attestation) => { + stats.aspa_ok += 1; + aspas.push(attestation); + audit.push(ObjectAuditEntry { + rsync_uri: file.rsync_uri.clone(), + sha256_hex: sha256_hex_from_32(&file.sha256), + kind: AuditObjectKind::Aspa, + result: AuditObjectResult::Ok, + detail: None, + }); + if let Some(warning) = ber_compatible_cms_warning_for_file(file, "ASPA") { + warnings.push(warning); + } } + Err(error) => match policy.signed_object_failure_policy { + SignedObjectFailurePolicy::DropObject => { + audit.push(ObjectAuditEntry { + rsync_uri: file.rsync_uri.clone(), + sha256_hex: sha256_hex_from_32(&file.sha256), + kind: AuditObjectKind::Aspa, + result: AuditObjectResult::Error, + detail: Some(error.to_string()), + }); + warnings.push( + Warning::new(format!( + "dropping invalid ASPA: {}: {error}", + file.rsync_uri + )) + .with_context(&file.rsync_uri), + ); + } + SignedObjectFailurePolicy::DropPublicationPoint => { + stats.publication_point_dropped = true; + warnings.push( + Warning::new(format!( + "dropping publication point due to invalid ASPA: {}: {error}", + file.rsync_uri + )) + .with_context(manifest_uri), + ); + return empty(stats, warnings, audit); + } + }, } } } - ObjectsOutput { vrps, aspas, router_keys: Vec::new(), warnings, stats, audit } + ObjectsOutput { + vrps, + aspas, + router_keys: Vec::new(), + warnings, + stats, + audit, + } } diff --git a/src/validation/run_tree_from_tal/discovery.rs b/src/validation/run_tree_from_tal/discovery.rs index 3d98707..0283f27 100644 --- a/src/validation/run_tree_from_tal/discovery.rs +++ b/src/validation/run_tree_from_tal/discovery.rs @@ -187,8 +187,8 @@ fn root_discovery_from_tal_input( let tal_bytes = std::fs::read(path).map_err(|e| { FromTalError::TalFetch(format!("read TAL file failed: {}: {e}", path.display())) })?; - let tal = crate::model::tal::Tal::decode_bytes(&tal_bytes) - .map_err(FromTalError::from)?; + let tal = + crate::model::tal::Tal::decode_bytes(&tal_bytes).map_err(FromTalError::from)?; if strict_name { discover_root_ca_instance_from_tal_with_fetchers_strict_name( http_fetcher, diff --git a/src/validation/tree_parallel/ready_stage.rs b/src/validation/tree_parallel/ready_stage.rs index dc4fb12..128493b 100644 --- a/src/validation/tree_parallel/ready_stage.rs +++ b/src/validation/tree_parallel/ready_stage.rs @@ -243,9 +243,10 @@ fn compute_ready_publication_point_stage( .ta_constraints .shared_for_tal(&ready.node.handle.tal_id); if ta_constraints.is_some() - && let Some(timing) = runner.timing.as_ref() { - timing.record_count("ta_constraints_parallel_publication_points", 1); - } + && let Some(timing) = runner.timing.as_ref() + { + timing.record_count("ta_constraints_parallel_publication_points", 1); + } match prepare_publication_point_for_parallel_roa_and_ta_constraints( ready.node.id, &fresh_stage.fresh_point, diff --git a/src/validation/tree_parallel/state.rs b/src/validation/tree_parallel/state.rs index 50e37e4..139a9a7 100644 --- a/src/validation/tree_parallel/state.rs +++ b/src/validation/tree_parallel/state.rs @@ -102,7 +102,6 @@ enum StageOutcome { Fresh(Box), } - struct FreshErrorOutcome { ready: ReadyCaInstance, publication_point_started: Instant, diff --git a/src/validation/tree_runner/audit_projection.rs b/src/validation/tree_runner/audit_projection.rs index ac7b0fd..1f63b90 100644 --- a/src/validation/tree_runner/audit_projection.rs +++ b/src/validation/tree_runner/audit_projection.rs @@ -86,7 +86,10 @@ fn build_publication_point_audit_from_snapshot( })); } - let mut warnings = runner_warnings.iter().map(AuditWarning::from).collect::>(); + let mut warnings = runner_warnings + .iter() + .map(AuditWarning::from) + .collect::>(); warnings.extend(objects.warnings.iter().map(AuditWarning::from)); PublicationPointAudit { node_id: None, @@ -119,7 +122,10 @@ fn build_publication_point_audit_from_failed_fetch( runner_warnings: &[Warning], fresh_error: &ManifestFreshError, ) -> PublicationPointAudit { - let mut warnings = runner_warnings.iter().map(AuditWarning::from).collect::>(); + let mut warnings = runner_warnings + .iter() + .map(AuditWarning::from) + .collect::>(); warnings.push(AuditWarning::from( &Warning::new(fresh_error.to_string()).with_context(&ca.manifest_rsync_uri), )); diff --git a/src/validation/tree_runner/discovery.rs b/src/validation/tree_runner/discovery.rs index ca3e903..3bcbd6d 100644 --- a/src/validation/tree_runner/discovery.rs +++ b/src/validation/tree_runner/discovery.rs @@ -12,15 +12,25 @@ fn discover_children_from_fresh_snapshot_with_audit_with_issuer_der Result { let locked_files = publication_point.files(); let issuer_ca = match crate::model::rc::ResourceCertificate::decode_der(issuer_ca_der) { - Ok(ca) => match ca.validate_rfc6487_profile(crate::model::rc::ResourceCertificateRole::Ca) { - Ok(()) => Some(ca), - Err(error) => { - crate::logging::emit(crate::logging::Level::Debug, "issuer_ca_profile_error", || serde_json::json!({"error": error.to_string()})); - None + Ok(ca) => { + match ca.validate_rfc6487_profile(crate::model::rc::ResourceCertificateRole::Ca) { + Ok(()) => Some(ca), + Err(error) => { + crate::logging::emit( + crate::logging::Level::Debug, + "issuer_ca_profile_error", + || serde_json::json!({"error": error.to_string()}), + ); + None + } } - }, + } Err(error) => { - crate::logging::emit(crate::logging::Level::Debug, "issuer_ca_decode_error", || serde_json::json!({"error": error.to_string()})); + crate::logging::emit( + crate::logging::Level::Debug, + "issuer_ca_decode_error", + || serde_json::json!({"error": error.to_string()}), + ); None } }; @@ -36,10 +46,7 @@ fn discover_children_from_fresh_snapshot_with_audit_with_issuer_der>()?; @@ -63,7 +70,10 @@ fn discover_children_from_fresh_snapshot_with_audit_with_issuer_der BgpsecRouterCertificate::validate_path_with_prevalidated_issuer( - child_der, - issuer_ca, - issuer_spki, - &verified_crl.crl, - &verified_crl.revoked_serials, - issuer.ca_certificate_rsync_uri.as_deref(), - Some(issuer_crl_uri.as_str()), - validation_time, - ), + Ok(verified_crl) => { + BgpsecRouterCertificate::validate_path_with_prevalidated_issuer( + child_der, + issuer_ca, + issuer_spki, + &verified_crl.crl, + &verified_crl.revoked_serials, + issuer.ca_certificate_rsync_uri.as_deref(), + Some(issuer_crl_uri.as_str()), + validation_time, + ) + } Err(error) => { router_error = router_error.saturating_add(1); audits.push(ObjectAuditEntry { @@ -156,7 +169,9 @@ fn discover_children_from_fresh_snapshot_with_audit_with_issuer_der { if let Some(constraints) = policy.ta_constraints.for_tal(&issuer.tal_id) - && let Err(error) = constraints.validate_ee_certificate(&router.resource_cert) { - router_error = router_error.saturating_add(1); - audits.push(ObjectAuditEntry { - rsync_uri: file.rsync_uri.clone(), - sha256_hex: sha256_hex_from_32(&file.sha256), - kind: AuditObjectKind::RouterCertificate, - result: AuditObjectResult::Error, - detail: Some(format!("router certificate violates TA constraints: {error}")), - }); - continue; - } + && let Err(error) = + constraints.validate_ee_certificate(&router.resource_cert) + { + router_error = router_error.saturating_add(1); + audits.push(ObjectAuditEntry { + rsync_uri: file.rsync_uri.clone(), + sha256_hex: sha256_hex_from_32(&file.sha256), + kind: AuditObjectKind::RouterCertificate, + result: AuditObjectResult::Error, + detail: Some(format!( + "router certificate violates TA constraints: {error}" + )), + }); + continue; + } let asns = match router_asns_for_resource_mode( &router.asns, issuer.effective_as_resources.as_ref(), @@ -194,7 +213,9 @@ fn discover_children_from_fresh_snapshot_with_audit_with_issuer_der { @@ -221,7 +244,10 @@ fn discover_children_from_fresh_snapshot_with_audit_with_issuer_der { @@ -271,13 +297,14 @@ fn discover_children_from_fresh_snapshot_with_audit_with_issuer_der( file: &'a PackFile, elapsed_nanos: &mut u64, @@ -32,12 +31,8 @@ fn load_child_certificate_der_for_discovery<'a>( let bytes = file .bytes() .map_err(|e| format!("child certificate bytes load failed: {e}"))?; - *elapsed_nanos = elapsed_nanos.saturating_add( - started - .elapsed() - .as_nanos() - .min(u128::from(u64::MAX)) as u64, - ); + *elapsed_nanos = + elapsed_nanos.saturating_add(started.elapsed().as_nanos().min(u128::from(u64::MAX)) as u64); *count = count.saturating_add(1); Ok(bytes) } @@ -51,10 +46,7 @@ fn ca_certificate_der_for_validation<'a>( let was_lazy = ca.ca_certificate_sha256_hex().is_some(); let der = ca.ca_certificate_der(store)?; if was_lazy { - let elapsed = started - .elapsed() - .as_nanos() - .min(u128::from(u64::MAX)) as u64; + let elapsed = started.elapsed().as_nanos().min(u128::from(u64::MAX)) as u64; if let Some(timing) = timing { timing.record_count("ca_certificate_lazy_load_count", 1); timing.record_count("ca_certificate_lazy_load_bytes", der.len() as u64); @@ -86,7 +78,11 @@ fn select_issuer_crl_uri_for_child<'a>( } Err(format!( "CRL referenced by child certificate CRLDistributionPoints not found in publication point snapshot: {} (RFC 6487 §4.8.6; RFC 9286 §4.2.1)", - crldp_uris.iter().map(|uri| uri.as_str()).collect::>().join(", ") + crldp_uris + .iter() + .map(|uri| uri.as_str()) + .collect::>() + .join(", ") )) } @@ -104,7 +100,8 @@ fn ensure_issuer_crl_verified<'a>( let der = std::mem::take(bytes); let crl = crate::model::crl::RpkixCrl::decode_der(&der)?; crl.verify_signature_with_issuer_certificate_der(issuer_ca_der)?; - let mut revoked_serials = std::collections::HashSet::with_capacity(crl.revoked_certs.len()); + let mut revoked_serials = + std::collections::HashSet::with_capacity(crl.revoked_certs.len()); for revoked in &crl.revoked_certs { revoked_serials.insert(revoked.serial_number.bytes_be.clone()); } @@ -131,22 +128,40 @@ fn router_asns_for_resource_mode( let contains = |asn: u32| { resources.asnum.as_ref().is_some_and(|choice| match choice { crate::model::rc::AsIdentifierChoice::Inherit => false, - crate::model::rc::AsIdentifierChoice::AsIdsOrRanges(items) => items.iter().any(|item| match item { - crate::model::rc::AsIdOrRange::Id(id) => *id == asn, - crate::model::rc::AsIdOrRange::Range { min, max } => *min <= asn && asn <= *max, - }), + crate::model::rc::AsIdentifierChoice::AsIdsOrRanges(items) => { + items.iter().any(|item| match item { + crate::model::rc::AsIdOrRange::Id(id) => *id == asn, + crate::model::rc::AsIdOrRange::Range { min, max } => *min <= asn && asn <= *max, + }) + } }) }; match mode { ResourceValidationMode::Rfc6487 => { - let outside = router_asns.iter().copied().filter(|asn| !contains(*asn)).collect::>(); - if outside.is_empty() { Ok(router_asns.to_vec()) } else { - Err(format!("router AS resources are not a subset of issuer effective AS resources: {outside:?}")) + let outside = router_asns + .iter() + .copied() + .filter(|asn| !contains(*asn)) + .collect::>(); + if outside.is_empty() { + Ok(router_asns.to_vec()) + } else { + Err(format!( + "router AS resources are not a subset of issuer effective AS resources: {outside:?}" + )) } } ResourceValidationMode::ValidationUpdate03 => { - let filtered = router_asns.iter().copied().filter(|asn| contains(*asn)).collect::>(); - if filtered.is_empty() { Err("router AS resources have empty validated resource set".to_string()) } else { Ok(filtered) } + let filtered = router_asns + .iter() + .copied() + .filter(|asn| contains(*asn)) + .collect::>(); + if filtered.is_empty() { + Err("router AS resources have empty validated resource set".to_string()) + } else { + Ok(filtered) + } } } } diff --git a/src/validation/tree_runner/fresh_pipeline.rs b/src/validation/tree_runner/fresh_pipeline.rs index c1d1bc0..02a9fa0 100644 --- a/src/validation/tree_runner/fresh_pipeline.rs +++ b/src/validation/tree_runner/fresh_pipeline.rs @@ -232,16 +232,26 @@ impl<'a> Rpkiv1PublicationPointRunner<'a> { let mut ccr_append_ms = 0; if self.ccr_accumulator.is_some() { let ccr_projection_build_started = std::time::Instant::now(); - let subordinate_skis = discovered_children.iter().map(|child| { - if let Some(projection) = &child.child_entry_projection { - return hex::decode(&projection.child_ski).map_err(|error| error.to_string()); - } - let der = child.handle.ca_certificate_der(self.store)?; - let certificate = ResourceCertificate::decode_der(der.as_ref()).map_err(|error| error.to_string())?; - certificate.tbs.extensions.subject_key_identifier.clone() - .ok_or_else(|| "child certificate missing SubjectKeyIdentifier".to_string()) - }).collect::, String>>()?; - let ccr_manifest_projection = crate::ccr::projection::from_snapshot(ca, &pack, subordinate_skis)?; + let subordinate_skis = discovered_children + .iter() + .map(|child| { + if let Some(projection) = &child.child_entry_projection { + return hex::decode(&projection.child_ski) + .map_err(|error| error.to_string()); + } + let der = child.handle.ca_certificate_der(self.store)?; + let certificate = ResourceCertificate::decode_der(der.as_ref()) + .map_err(|error| error.to_string())?; + certificate + .tbs + .extensions + .subject_key_identifier + .clone() + .ok_or_else(|| "child certificate missing SubjectKeyIdentifier".to_string()) + }) + .collect::, String>>()?; + let ccr_manifest_projection = + crate::ccr::projection::from_snapshot(ca, &pack, subordinate_skis)?; ccr_projection_build_ms = ccr_projection_build_started.elapsed().as_millis() as u64; let ccr_append_started = std::time::Instant::now(); self.append_ccr_manifest_projection(&ccr_manifest_projection)?; @@ -296,15 +306,18 @@ impl<'a> Rpkiv1PublicationPointRunner<'a> { // Commit protocol freshness only after successful publication-point // validation/finalization, independently of optional CCR output. - self.store.put_manifest_anti_rollback_meta( - &crate::repository::storage::ManifestAntiRollbackMeta { - manifest_rsync_uri: pack.manifest_rsync_uri.clone(), - manifest_number_be: pack.manifest_number_be.clone(), - manifest_this_update: pack.this_update.clone(), - manifest_sha256: ::digest(&pack.manifest_bytes).to_vec(), - updated_at_validation_time: pack.verified_at.clone(), - }, - ).map_err(|error| format!("persist manifest freshness failed: {error}"))?; + self.store + .put_manifest_anti_rollback_meta( + &crate::repository::storage::ManifestAntiRollbackMeta { + manifest_rsync_uri: pack.manifest_rsync_uri.clone(), + manifest_number_be: pack.manifest_number_be.clone(), + manifest_this_update: pack.this_update.clone(), + manifest_sha256: ::digest(&pack.manifest_bytes) + .to_vec(), + updated_at_validation_time: pack.verified_at.clone(), + }, + ) + .map_err(|error| format!("persist manifest freshness failed: {error}"))?; Ok(FreshPublicationPointFinalizeOutput { result: PublicationPointRunResult { diff --git a/src/validation/tree_runner/publication_point_runner.rs b/src/validation/tree_runner/publication_point_runner.rs index 017a150..192f377 100644 --- a/src/validation/tree_runner/publication_point_runner.rs +++ b/src/validation/tree_runner/publication_point_runner.rs @@ -99,9 +99,12 @@ impl<'a> PublicationPointRunner for Rpkiv1PublicationPointRunner<'a> { Err(stage_error) => { let mut warnings = Vec::new(); warnings.push( - Warning::new(format!("publication point processing failed: {}", stage_error.error)) - .with_rfc_refs(&[RfcRef("RFC 9286 §6.6")]) - .with_context(&ca.manifest_rsync_uri), + Warning::new(format!( + "publication point processing failed: {}", + stage_error.error + )) + .with_rfc_refs(&[RfcRef("RFC 9286 §6.6")]) + .with_context(&ca.manifest_rsync_uri), ); let audit = build_publication_point_audit_from_failed_fetch( ca, @@ -186,8 +189,14 @@ impl<'a> PublicationPointRunner for Rpkiv1PublicationPointRunner<'a> { ) }; let object_ms = object_started.elapsed().as_millis() as u64; - self.record_publication_point_step_ms(&ca.manifest_rsync_uri, "objects_processing", object_ms); - objects.router_keys.extend(stage.discovered_router_keys.clone()); + self.record_publication_point_step_ms( + &ca.manifest_rsync_uri, + "objects_processing", + object_ms, + ); + objects + .router_keys + .extend(stage.discovered_router_keys.clone()); let finalized = self.finalize_fresh_publication_point_from_reducer( ca, diff --git a/src/validation/tree_runner/timing.rs b/src/validation/tree_runner/timing.rs index 5869710..1dfc9c6 100644 --- a/src/validation/tree_runner/timing.rs +++ b/src/validation/tree_runner/timing.rs @@ -19,5 +19,4 @@ impl<'a> Rpkiv1PublicationPointRunner<'a> { ); } } - } diff --git a/tests/cli_errors.rs b/tests/cli_errors.rs new file mode 100644 index 0000000..1da7e28 --- /dev/null +++ b/tests/cli_errors.rs @@ -0,0 +1,64 @@ +use std::process::Command; + +fn command() -> Command { + let mut cmd = Command::new(env!("CARGO_BIN_EXE_panda-rpki")); + cmd.env_remove("PANDA_RPKI_LOG_LEVEL") + .env_remove("PANDA_RPKI_LOG_FORMAT"); + cmd +} + +#[test] +fn help_is_successful_and_goes_to_stdout() { + for args in [vec![], vec!["--help"], vec!["-h"]] { + let result = command().args(args).output().unwrap(); + assert!(result.status.success()); + assert!(String::from_utf8_lossy(&result.stdout).contains("panda-rpki")); + assert!(result.stderr.is_empty()); + } +} + +#[test] +fn invalid_command_returns_two_without_stdout() { + let result = command().arg("not-a-command").output().unwrap(); + assert_eq!(result.status.code(), Some(2)); + assert!(result.stdout.is_empty()); + assert!(String::from_utf8_lossy(&result.stderr).contains("unknown command")); +} + +#[test] +fn malformed_validation_arguments_do_not_create_outputs() { + let directory = tempfile::tempdir().unwrap(); + for args in [ + vec!["validate", "--out", "output"], + vec!["validate", "--unknown", "value"], + vec!["validate", "--out"], + ] { + let result = command() + .current_dir(directory.path()) + .args(args) + .output() + .unwrap(); + assert_eq!(result.status.code(), Some(2)); + assert!(result.stdout.is_empty()); + assert_eq!(std::fs::read_dir(directory.path()).unwrap().count(), 0); + } +} + +#[test] +fn missing_daemon_state_root_fails_without_creating_state() { + let directory = tempfile::tempdir().unwrap(); + let result = command() + .current_dir(directory.path()) + .args([ + "daemon", + "--", + "--tal", + "missing.tal", + "--ta", + "missing.cer", + ]) + .output() + .unwrap(); + assert_eq!(result.status.code(), Some(2)); + assert_eq!(std::fs::read_dir(directory.path()).unwrap().count(), 0); +} diff --git a/tests/synthetic_docker_e2e.sh b/tests/synthetic_docker_e2e.sh index 6df5297..6436ec1 100755 --- a/tests/synthetic_docker_e2e.sh +++ b/tests/synthetic_docker_e2e.sh @@ -38,7 +38,12 @@ for _ in $(seq 1 50); do done rg -qx 'ready' "$work_root/server.log" -docker build -f "$repo_root/docker/Dockerfile" -t panda-rpki:synthetic-e2e "$repo_root" +image="${PANDA_RPKI_TEST_IMAGE:-panda-rpki:synthetic-e2e}" +if [[ -z "${PANDA_RPKI_TEST_IMAGE:-}" ]]; then + docker build -f "$repo_root/docker/Dockerfile" -t "$image" "$repo_root" +else + docker image inspect "$image" >/dev/null +fi run_validator() { local out="$1" local mode="$2" @@ -46,7 +51,7 @@ run_validator() { -v "$fixture:/fixture:ro" \ -v "$work_root/state:/state" \ -v "$out:/output" \ - panda-rpki:synthetic-e2e validate \ + "$image" validate \ --tal /fixture/tal/custom.tal --ta /fixture/ta/custom-ta.cer \ --http-root-cert /fixture/certs/rrdp-ca.pem \ --http-timeout-secs 5 \ @@ -94,7 +99,7 @@ daemon_container="panda-rpki-daemon-$(basename "$work_root")" docker run --rm --name "$daemon_container" --network host --read-only --tmpfs /tmp \ --user "$(id -u):$(id -g)" \ -v "$fixture:/fixture:ro" -v "$work_root/daemon:/data" \ - panda-rpki:synthetic-e2e daemon \ + "$image" daemon \ --state-root /data --max-runs 3 --interval-secs 2 --retain-runs 3 -- \ --tal /fixture/tal/custom.tal --ta /fixture/ta/custom-ta.cer \ --http-root-cert /fixture/certs/rrdp-ca.pem --log-format json \ diff --git a/tools/check_format.py b/tools/check_format.py new file mode 100644 index 0000000..780bb66 --- /dev/null +++ b/tools/check_format.py @@ -0,0 +1,9 @@ +"""Check every Rust source file, including include! fragments.""" +import subprocess +from pathlib import Path + +root = Path(__file__).resolve().parents[1] +files = sorted(p for folder in ("src", "tests") for p in (root / folder).rglob("*.rs")) +subprocess.run(["cargo", "fmt", "--all", "--check"], cwd=root, check=True) +subprocess.run(["rustfmt", "--check", "--edition", "2024", *map(str, files)], + cwd=root, check=True) diff --git a/tools/dependency_notices.py b/tools/dependency_notices.py new file mode 100644 index 0000000..ba37af0 --- /dev/null +++ b/tools/dependency_notices.py @@ -0,0 +1,86 @@ +"""Reproduce notices from locked Cargo archives and pinned upstream supplements.""" +import argparse +import hashlib +import json +import subprocess +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +HEADER = """# Third-party notices + +Panda RPKI is distributed under BSD-3-Clause; dependencies retain their own +licenses. The table below records the locked Cargo dependency graph, including +build-time and platform-specific packages that may not be linked on Linux. +Upstream license expressions are reproduced as declared by each package. + +The accompanying [license texts](docs/third-party-licenses.txt) retain notices +from package archives, including bundled native code. These files must accompany +binary distributions. Container OS package notices remain under +`/usr/share/doc/` and `/usr/share/common-licenses/` in the Debian image. + +Python cryptography and OpenSSL are external test tools, not bundled source. +Consult their installed license notices when distributing a test environment. +No production trust-anchor or repository data is included in this source tree. + +| Package | Locked version | Declared license | +| --- | --- | --- | +""" + + +def generate(): + metadata = json.loads(subprocess.check_output( + ["cargo", "metadata", "--locked", "--format-version", "1"], cwd=ROOT)) + supplements = json.loads((ROOT / "tools/license_supplements.json").read_text()) + rows, texts = [], [] + for package in sorted(metadata["packages"], key=lambda p: (p["name"], p["version"])): + if package["id"] in metadata["workspace_members"]: + continue + name, version = package["name"], package["version"] + source = Path(package["manifest_path"]).parent + candidates = sorted(p for p in source.rglob("*") if p.is_file() + and (p.name.upper().startswith(("LICENSE", "LICENCE", "COPYING", "COPYRIGHT", "NOTICE", "AUTHORS")) + or "LICENSES" in p.relative_to(source).parts) + and p.stat().st_size < 500_000) + contents = [] + for path in candidates: + try: + contents.append((str(path.relative_to(source)), path.read_text(encoding="utf-8"))) + except UnicodeDecodeError: + continue + if not contents: + entry = supplements.get(f"{name}@{version}") + if entry is None: + raise RuntimeError(f"Missing license texts: {name}@{version}; review upstream sources") + data = subprocess.check_output([ + "curl", "--fail", "--silent", "--show-error", "--location", + "--proto", "=https", "--proto-redir", "=https", "--connect-timeout", "15", + "--max-time", "120", entry["url"]]) + if hashlib.sha256(data).hexdigest() != entry["sha256"]: + raise RuntimeError(f"Supplement checksum mismatch: {name}@{version}") + contents.append(("LICENSE (upstream package revision)", data.decode("utf-8"))) + rows.append(f"| {name} | {version} | {package['license'] or 'See upstream license'} |") + for label, content in contents: + texts.append(f"\n{'=' * 72}\n{name} {version} — {label}\n{'=' * 72}\n{content}\n") + return { + ROOT / "THIRD_PARTY_NOTICES.md": HEADER + "\n".join(rows) + "\n", + ROOT / "docs/third-party-licenses.txt": + "Third-party license texts from locked package archives.\n" + "".join(texts), + } + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--check", action="store_true", help="Fail if tracked notices differ; do not write files") + args = parser.parse_args() + outputs = generate() # Resolve everything before writing any output. + for path, content in outputs.items(): + if args.check: + if not path.exists() or path.read_bytes() != content.encode("utf-8"): + raise SystemExit(f"Outdated notices: {path.name}; run tools/dependency_notices.py") + else: + path.write_bytes(content.encode("utf-8")) + print("Dependency notices are current." if args.check else "Dependency notices generated.") + + +if __name__ == "__main__": + main() diff --git a/tools/license_supplements.json b/tools/license_supplements.json new file mode 100644 index 0000000..300355d --- /dev/null +++ b/tools/license_supplements.json @@ -0,0 +1,14 @@ +{ + "alloc-stdlib@0.2.4": { + "url": "https://raw.githubusercontent.com/dropbox/rust-alloc-no-stdlib/ae42d22078b98549e987d2f03d12df7b984fde47/LICENSE", + "sha256": "c0c56f26d9c051cac4d200c34c84e7ae9aaa853e01a982a1df08b09931e518ae" + }, + "asn1-rs-impl@0.2.0": { + "url": "https://raw.githubusercontent.com/rusticata/asn1-rs/a20e5f7319c896737ad0f2557037817b91ad854f/LICENSE-MIT", + "sha256": "a5c61b93b6ee1d104af9920cf020ff3c7efe818e31fe562c72261847a728f513" + }, + "cookie-factory@0.3.3": { + "url": "https://raw.githubusercontent.com/rust-bakery/cookie-factory/d36b805dbd7dd65f2df947235c5bcc573afe2c76/LICENSES/MIT.txt", + "sha256": "d09216dc1ea5f273997667935a8d6514d80f00b89fb6954ecc3a43e0a6e360de" + } +} diff --git a/tools/test_dependency_notices.py b/tools/test_dependency_notices.py new file mode 100644 index 0000000..8c50a31 --- /dev/null +++ b/tools/test_dependency_notices.py @@ -0,0 +1,59 @@ +"""Offline regression checks for release-maintenance failure handling.""" +import json +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +import dependency_notices as notices + + +class NoticeTests(unittest.TestCase): + def setUp(self): + self.temp = tempfile.TemporaryDirectory() + self.addCleanup(self.temp.cleanup) + self.root = Path(self.temp.name) + (self.root / "tools").mkdir() + (self.root / "package").mkdir() + (self.root / "tools/license_supplements.json").write_text("{}") + self.metadata = json.dumps({"workspace_members": [], "packages": [{ + "id": "example", "name": "example", "version": "1.0.0", "license": "MIT", + "manifest_path": str(self.root / "package/Cargo.toml"), + }]}).encode() + self.patch = patch.object(notices, "ROOT", self.root) + self.patch.start() + self.addCleanup(self.patch.stop) + + def test_missing_license_fails(self): + with patch.object(notices.subprocess, "check_output", return_value=self.metadata): + with self.assertRaisesRegex(RuntimeError, "Missing license"): + notices.generate() + + def test_license_text_is_preserved(self): + original = "Copyright Example \n\nMIT license text\n" + (self.root / "package/LICENSE").write_text(original) + with patch.object(notices.subprocess, "check_output", return_value=self.metadata): + outputs = notices.generate() + self.assertIn(original, outputs[self.root / "docs/third-party-licenses.txt"]) + self.assertFalse((self.root / "THIRD_PARTY_NOTICES.md").exists()) + + def test_supplement_checksum_mismatch_fails(self): + (self.root / "tools/license_supplements.json").write_text(json.dumps({ + "example@1.0.0": {"url": "https://example.invalid/LICENSE", "sha256": "0" * 64} + })) + with patch.object(notices.subprocess, "check_output", side_effect=[self.metadata, b"bad"]): + with self.assertRaisesRegex(RuntimeError, "checksum mismatch"): + notices.generate() + + def test_check_does_not_overwrite_outdated_file(self): + path = self.root / "THIRD_PARTY_NOTICES.md" + path.write_text("old") + with patch.object(notices, "generate", return_value={path: "new"}): + with patch("sys.argv", ["dependency_notices.py", "--check"]): + with self.assertRaisesRegex(SystemExit, "Outdated notices"): + notices.main() + self.assertEqual(path.read_text(), "old") + + +if __name__ == "__main__": + unittest.main()