nordisk/holger
server/lib/src/grpc.rs
//! The tonic gRPC surfce — holger's only network API (no REST; the OCI/HTTP
//! gateway in `exposed/http.rs` is a separate shim). [`HolgerGrpc`] is the shared
//! state behind three services impl'd on `Arc<HolgerGrpc>`: `RepositoryService`
//! (fetch/list/put/stream artifacts), `ArchiveService` (list/stat the underlying
//! znippy archive), and `AdminService` (health, repo catalog, `ServerProfile`).
//! Every RPC resolves the target through [`FastRoutes`] to a
//! [`RepositoryBackendTrait`] and appends a best-effort audit line.
//!
//! Trust model, and the gotcha: READS ARE OPEN (anonymous), only writes go
//! through [`HolgerGrpc::authorize_write`] (Bearer/mTLS-CN authN, then RBAC
//! write-gate that fails closed once role policy is configured). Because reads
//! are unauthenticated, backend errors must be laundered through [`internal_err`]
//! — raw errors carry archive/file paths and never reach the client; the detail
//! lives only in the server log and audit sink. Because they are unauthenticated
//! they are also unmetered by anything else, so every read door charges the
//! process-wide in-flight read budget ([`DEFAULT_MAX_IN_FLIGHT_READ_BYTES`]) —
//! the backend trait has no streaming read, so a bounded byte budget is what
//! stands in for one. `ServerProfile` reports
//! server-side truth (read-only iff no repo is writable, unless pinned via
//! [`HolgerGrpc::with_read_only_profile`]). The generated proto is committed
//! (`generated/holger.v1.rs`, `cargo xtask gen-proto`), not built at compile time.
pub mod holger_proto {
// The proto Rust is generated AHEAD OF TIME by `cargo xtask gen-proto`
// (protox + tonic-build, no `protoc`, no shell-out) and committed under
// `src/generated/holger.v1.rs`. We `include!` the checked-in file instead of
// the old `tonic::include_proto!("holger.v1")`, which read from `OUT_DIR` and
// depended on a build.rs that shelled out to `protoc` at every build.
// Regenerate after editing `proto/holger.proto`: `cargo xtask gen-proto`.
include!("generated/holger.v1.rs");
}
/// **The paid-state contract — compiled ONLY by the metered build.**
///
/// `holger-metered.proto` is its own package (`holger.metered.v1`) precisely so
/// tonic-build emits it as its own file, and this `include!` is the only thing
/// that pulls it in. Without the feature, `plan`, `paid_until_unix_ms` and `caps`
/// are not in the binary at all — the open build does not decline to enforce a
/// plan, it has never heard of one (`tests/open_build_speaks_no_money.rs`).
#[cfg(feature = "metered")]
#[allow(clippy::all)] // generated by tonic-build: its doc comments are the .proto's
pub mod holger_metered_proto {
include!("generated/holger.metered.v1.rs");
}
/// **The estate over the wire** (`holger.estate.v1`, owner plan 2b, 2026-09-27): a
/// twin's or a recharge's manifest + fetch of the data root's side stores and config.
/// Its own package so holger.proto, and every downstream copy of it, does not move.
pub mod holger_estate_proto {
include!("generated/holger.estate.v1.rs");
}
/// **The data set grows without a restart** (`holger.grow.v1`, 2026-09-27): Start /
/// Status (no credential) / Resume / Flush, and a twin's position — `crate::grow`.
/// Its own package for the estate's reason.
pub mod holger_grow_proto {
include!("generated/holger.grow.v1.rs");
}
/// **Where more room is bought** (`holger.room.v1`, 2026-10-17) — `crate::room`.
pub mod holger_room_proto {
include!("generated/holger.room.v1.rs");
}
/// **The repositories a person has here** (`holger.account.v1`, 2026-10-21, lane P4) —
/// `crate::account`.
pub mod holger_account_proto {
include!("generated/holger.account.v1.rs");
}
/// **Repository tokens for package clients** (`holger.token.v1`, 2026-10-21) — `crate::repo_token`.
pub mod holger_token_proto {
include!("generated/holger.token.v1.rs");
}
/// **The thin client's two missing doors** (`holger.doors.v1`, 2026-10-26, lane B3): the
/// bundle export and the first admin key — `grpc/doors.rs`.
pub mod holger_doors_proto {
include!("generated/holger.doors.v1.rs");
}
use std::sync::Arc;
use std::time::Instant;
use tonic::{Request, Status};
use holger_proto::*;
use crate::audit::{AuditAction, AuditEvent, AuditLog, NoopAuditLog};
use crate::auth::{self, AuthConfig};
use crate::exposed::fast_routes::FastRoutes;
/// Default ingest body cap (1 GiB) — matches the HTTP gateway's default
/// (`lib.rs`), so both surfaces share one configurable knob (`ron_max_body_bytes`).
pub const DEFAULT_MAX_BODY_BYTES: usize = 1024 * 1024 * 1024;
/// Default **in-flight READ budget** (4 GiB): the ceiling on artifact bytes every
/// gRPC read door may hold resident *across all concurrent requests*.
///
/// Why a budget exists at all. [`traits::RepositoryBackendTrait::fetch`] is the
/// only read the backends expose and it hands back the WHOLE artifact as a
/// `Vec<u8>` — there is no streaming backend API and no size hint — so
/// `StreamArtifact`'s 64 KiB chunking buys no bound whatsoever: it materialises
/// the entire blob and *then* dribbles it out at whatever rate the client reads.
/// Reads are open by default (the airgap-mirror posture), and tonic here sets no
/// `concurrency_limit_per_connection`, so K anonymous clients asking for the
/// largest artifact in the archive used to multiply K × its size with nothing
/// anywhere to say no. This is the no.
///
/// Why 4 GiB. It is four times [`DEFAULT_MAX_BODY_BYTES`] — the largest single
/// artifact this server will accept — so the worst *legitimate* case, four
/// maximum-size artifacts in flight at once, still fits, while an ordinary mirror
/// workload (wheels, crates, jars: single-digit to low-hundreds of MiB) never
/// comes near the ceiling and pays nothing but an atomic. It is deliberately a
/// number a 16 GiB node survives with the page cache intact. Lower it on a small
/// node, raise it on a fat one, or set `0` to disable the accounting entirely, via
/// [`HolgerGrpc::with_max_in_flight_read_bytes`].
pub const DEFAULT_MAX_IN_FLIGHT_READ_BYTES: usize = 4 * 1024 * 1024 * 1024;
/// The accounting unit of the read budget: 1 MiB. Permits count whole MiB rather
/// than bytes for two reasons — `Semaphore::try_acquire_many_owned` takes a `u32`,
/// which a byte count overflows at 4 GiB; and a floor of one unit per read puts a
/// ceiling on request count as a side effect — but a loose one. At the default
/// budget that ceiling is 4096 simultaneous reads, which is a memory bound doing
/// a concurrency bound's job badly. [`DEFAULT_MAX_CONCURRENT_READS`] is the
/// explicit bound; this stays the memory one.
const READ_BUDGET_UNIT_BYTES: usize = 1024 * 1024;
/// **The maximum number of reads in flight at once, process-wide.**
///
/// The byte budget above bounds MEMORY; this bounds everything else a read costs
/// that is not bytes — a tokio task, an HTTP/2 stream, a file descriptor, and (on
/// the paths that still call a backend synchronously) a worker thread. The two do
/// not imply each other: with a 4 GiB budget and a 1 MiB floor per read, four
/// thousand concurrent index reads of a few hundred bytes each fit the byte budget
/// completely while being far more requests than this server should have open.
///
/// This is what the review asked tonic's `concurrency_limit_per_connection` for,
/// applied a level up. It is deliberately PROCESS-WIDE rather than per-connection,
/// because a per-connection limit multiplies by the number of connections and the
/// number of connections is the attacker's choice. holger's serve tail is the
/// shared `nornir_service::serve_with_tls` in edda, so this also does not require
/// changing a seam three other services depend on.
///
/// 512 is chosen to be far above any real fleet's concurrent reads (reads are
/// short; a CI fleet's index requests finish in milliseconds) and far below the
/// point where task and thread pressure matters. Tune with
/// [`HolgerGrpc::with_max_concurrent_reads`]; `0` disables the cap.
pub const DEFAULT_MAX_CONCURRENT_READS: usize = 512;
/// Units of the in-flight read budget that `n` artifact bytes cost — `n` rounded
/// UP to whole [`READ_BUDGET_UNIT_BYTES`], never zero (every read costs at least
/// one unit, which is what makes the budget a concurrency cap too).
fn read_budget_units(n: usize) -> u32 {
let units = n.div_ceil(READ_BUDGET_UNIT_BYTES).max(1);
u32::try_from(units).unwrap_or(u32::MAX)
}
/// A held slice of the process-wide in-flight read budget, released on drop.
///
/// The point of the type (rather than a bare permit) is that the read doors can
/// MOVE it: `StreamArtifact` hands it to the task that feeds the chunk channel, so
/// the budget stays charged for as long as the blob is actually resident — a slow
/// client holding a 900 MiB artifact open keeps paying for it, which is precisely
/// the case a per-request check cannot see.
///
/// `None` = the budget is disabled (`max_in_flight_read_bytes: 0`) and the guard
/// is a zero-cost nothing.
pub(crate) struct ReadBudget {
permit: Option<tokio::sync::OwnedSemaphorePermit>,
/// Units (of `READ_BUDGET_UNIT_BYTES`) this guard holds so far, so a
/// stream that charges AS THE BYTES GO OUT knows what it has already paid.
units: u32,
/// The read's SLOT, held for exactly as long as the byte permit. Bytes and
/// slots bound different things and neither implies the other — see
/// [`DEFAULT_MAX_CONCURRENT_READS`].
_slot: Option<tokio::sync::OwnedSemaphorePermit>,
}
/// **THE ONE WRITER of every coded gRPC `Status` holger returns.** Every refusal on the
/// network surface is built here, or it is not coded.
///
/// A remote client sees only the `Status`, so a registry that codes internal error enums
/// and leaves `Status::*` bare codes nothing that matters (that was SK178: 65 registry
/// codes, 0 of 38 `Status` sites carrying one). This puts the frozen
/// `holger-<subsystem>-<n>` on the wire in **both channels**, matching what nornir, korp
/// and dwarves already do:
///
/// 1. **in the message** — `[holger-server-1] Repository 'x' not found`, so a client that
/// only logs `status.message()` still records the code; and
/// 2. **in the metadata** — [`holger_errcode::WIRE_CODE_KEY`] (`x-holger-code`), so a
/// client reacts to the exact verdict structurally without parsing prose.
///
/// One helper rather than 38 hand-tagged sites, so a site cannot drift into a
/// slightly-different shape and the message/metadata halves can never disagree.
/// `errcode/tests/wiring_guard.rs` proves each code really is raised here, and
/// `server/lib/tests/grpc_error_codes_on_the_wire.rs` proves a real remote client
/// receives it.
pub(crate) fn coded(
ec: &'static holger_errcode::ErrCode,
grpc: tonic::Code,
detail: impl std::fmt::Display,
) -> Status {
stamped(ec, grpc, ec.tag(detail))
}
/// The ONE `Status::new` in holger: a message that already carries `ec`'s bracketed
/// code, with the code put on the metadata channel too. Private, so every caller goes
/// through [`coded`], [`internal_err`] or [`refusal`].
fn stamped(ec: &'static holger_errcode::ErrCode, grpc: tonic::Code, message: String) -> Status {
let mut st = Status::new(grpc, message);
// A registry code is ASCII `holger-<sub>-<n>` by construction (guarded by
// `code_strings_match_their_parts`), so this parse cannot realistically fail — but a
// silent drop would take the metadata channel with it, so shout instead.
match ec
.code
.parse::<tonic::metadata::MetadataValue<tonic::metadata::Ascii>>()
{
Ok(v) => {
st.metadata_mut().insert(holger_errcode::WIRE_CODE_KEY, v);
}
Err(e) => log::error!(
"holger grpc: error code {} is not a valid metadata value ({e}) — the \
{} channel was DROPPED for this status",
ec.code,
holger_errcode::WIRE_CODE_KEY,
),
}
st
}
/// Log a backend error server-side and return a GENERIC gRPC status. Read paths are
/// unauthenticated in this model, so the raw error (which can carry archive/file
/// paths + internal corruption detail) must never reach the client — operators get
/// the detail in the server log, clients get an opaque "internal error" — now with
/// `holger-server-4` on it, which leaks nothing and lets a client tell "the server
/// broke" from "you asked for the wrong thing".
pub(crate) fn internal_err(e: impl std::fmt::Display) -> Status {
log::warn!("holger grpc: backend error: {e}");
coded(
&holger_errcode::SERVER_INTERNAL,
tonic::Code::Internal,
"internal error",
)
}
/// **A refusal worded by the module that decided it** (`Result<_, String>` from
/// `twin`, `twin_tls`, `estate_wire`, the metered guard, …).
///
/// The rule is the registry's own: a reason that carries a registry code
/// (`ec.tag(..)` / `ec.bracketed()`) was worded for the client on purpose, so it goes
/// out verbatim with that code on the metadata channel as well. A reason that carries
/// NO code is backend text (an IO error, a path, a parse failure), so it is handled
/// exactly like [`internal_err`]: logged here, and the client gets
/// `holger-server-4` and nothing else.
pub(crate) fn refusal(grpc: tonic::Code, why: impl std::fmt::Display) -> Status {
let why = why.to_string();
match holger_errcode::parse(&why) {
Some(ec) => stamped(ec, grpc, why),
None => internal_err(why),
}
}
/// **Introspection / emit marker** — record one functional-status row for the
/// nornir test matrix. Wraps `nornir_testmatrix::functional_status` behind the
/// `testmatrix` feature (a compiled-out `#[inline]` no-op otherwise, with no
/// nornir dep in a release build). `component` is the reporting surface (e.g.
/// `"holger-grpc/fetch_artifact"`), `check` what it verified, `ok` the verdict,
/// `detail` a short human note. Every gRPC handler + the RBAC write gate emit one
/// so `nornir test --features testmatrix` SEES each live surface. Mirrors the
/// reference emitter in `korp-collectors/src/lib.rs`.
#[inline]
pub(crate) fn functional_status(component: &str, check: &str, ok: bool, detail: &str) {
#[cfg(feature = "testmatrix")]
nornir_testmatrix::functional_status(component, check, ok, detail);
#[cfg(not(feature = "testmatrix"))]
{
let _ = (component, check, ok, detail);
}
}
/// `MigrationService` — the four RPCs a UI drives a ONE-TIME registry migration
/// with. ONLINE only: it is the one service here that reaches out to a foreign
/// registry, so the zero-egress airgap tiers do not carry it.
#[cfg(feature = "migration")]
mod migration;
/// The three gRPC service impls (Repository / Archive / Admin) on `Arc<HolgerGrpc>`
/// — a child module so it reaches this module's private helpers + `HolgerGrpc`'s
/// private state, keeping the root focused on shared state + wiring.
mod services;
// The estate over the wire (twin / recharge), holger.estate.v1.
/// `holger.doors.v1`: the bundle export and the first admin key (lane B3).
pub mod doors;
mod estate;
/// Shared state for all gRPC services
/// **One row of the module census** `AdminService.ServerProfile` reports.
///
/// An appliance has no shell, so this is the only way an operator can ask what
/// is loaded and what it may do — and the only way a test can ask it of a
/// RUNNING box rather than of a build.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct ModuleCensusEntry {
pub name: String,
/// `linked`, or the module's file name.
pub origin: String,
/// `package-handler` | `meter`.
pub kind: String,
/// Hex blake3-256 of the module bytes. Empty for a linked guest, which has
/// none — and empty is distinguishable from a digest, which is the point.
pub digest: String,
/// The verified signer, when a trust policy was in force.
pub receipt: String,
pub granted: Vec<String>,
}
/// What this boot loaded, and what it may reach.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct ModuleCensus {
pub modules: Vec<ModuleCensusEntry>,
/// The union of every capability granted to any module on this boot.
pub granted_capabilities: Vec<String>,
/// The host functions actually defined on this boot's wasm linkers.
pub linker_defines: Vec<String>,
}
pub struct HolgerGrpc {
pub routes: FastRoutes,
pub start_time: Instant,
pub auth_config: Arc<AuthConfig>,
/// **The per-repository read scope** — which repositories demand a
/// credential of their own accord, on top of the server-wide read gate.
/// Built at boot from the repositories that declared `ron_require_auth`;
/// [`crate::auth::ReadGate::GLOBAL_ONLY`] (empty) unless one did, which is
/// every deployment that has not opted in. Install with
/// [`HolgerGrpc::with_read_gate`].
pub read_gate: Arc<crate::auth::ReadGate>,
/// Append-only audit sink. Defaults to [`NoopAuditLog`]; install a real
/// backend (e.g. `audit::default_audit_log(dir)`) via [`HolgerGrpc::with_audit`].
pub audit: Arc<dyn AuditLog>,
/// Maximum accepted `put_artifact` payload size in bytes. The streamed HTTP
/// gateway caps uploads (`http.rs`); without this the gRPC write path was
/// unbounded. Same default + config source as the gateway.
pub max_body_bytes: usize,
/// Ceiling on artifact bytes the READ doors may hold resident at once, across
/// every concurrent request. `0` disables the accounting. See
/// [`DEFAULT_MAX_IN_FLIGHT_READ_BYTES`] for why this exists and why it is 4 GiB;
/// set via [`HolgerGrpc::with_max_in_flight_read_bytes`].
pub max_in_flight_read_bytes: usize,
/// The live budget behind `max_in_flight_read_bytes`, counted in whole MiB
/// ([`READ_BUDGET_UNIT_BYTES`]). Private: a read door takes a [`ReadBudget`]
/// guard through [`HolgerGrpc::reserve_read_budget`] and never touches the
/// semaphore itself, so the reserve-then-charge protocol lives in ONE place.
read_budget: Arc<tokio::sync::Semaphore>,
/// How many reads may be in flight at once (0 = uncapped). See
/// [`DEFAULT_MAX_CONCURRENT_READS`].
pub max_concurrent_reads: usize,
/// The slots behind `max_concurrent_reads`. Private for the same reason
/// `read_budget` is: a read door takes a [`ReadBudget`] guard and never
/// touches this directly.
read_slots: Arc<tokio::sync::Semaphore>,
/// **Property keys only an ADMIN may set over the wire**, on top of the ones
/// derived from live state (the promotion engine's `required_property` key —
/// see [`HolgerGrpc::is_reserved_property`]).
///
/// Defaults to `{"keep"}`: the retention PIN (`RetentionPolicy::protect_property`,
/// documented and tested as `("keep", "true")`) is the third property that
/// decides policy rather than describing an artifact, and unlike the other two
/// it lives in a per-repository RON policy this plane never sees — so its key
/// is named here instead of derived. Reserved-by-default is the fail-closed
/// choice: a deployment that uses `keep` as an ordinary label and wants Writers
/// to set it clears the set with [`HolgerGrpc::with_reserved_properties`]; a
/// deployment that renamed the pin names the new key the same way.
pub reserved_properties: std::collections::BTreeSet<String>,
/// Explicit sealed/read-only profile override. `None` (default) ⇒ the
/// `ServerProfile` RPC *derives* the profile from the live route table (the
/// server is read-only iff no repository accepts writes — server truth).
/// `Some(true)` forces the static "rigged for silent running" profile even
/// when writable repos exist (a sealed static-holger declaring itself);
/// `Some(false)` forces dynamic.
pub read_only_profile: Option<bool>,
/// Immutable, content-addressed PROD store for the DEV→PROD promote workflow
/// (`PromotionService`). `None` (default) ⇒ the `PromoteArtifact` RPC returns
/// `failed_precondition` (promotion not configured on this server). Installed
/// via [`HolgerGrpc::with_promotion`].
#[cfg(feature = "intake")]
pub promotion: Option<Arc<crate::promote::PromotionEngine>>,
/// **The durable job queue** (`jobs` feature) — what makes
/// `PromoteArtifact`/`Replicate` able to answer with an id instead of
/// holding the caller for the whole run, and what `JobService` reads.
///
/// `None` ⇒ no queue is attached, and the async flags on those RPCs refuse
/// BY NAME rather than quietly running inline: a client that asked for a job
/// id and got a synchronous seal was told something untrue about what the
/// server did. Installed via [`HolgerGrpc::with_job_queue`].
#[cfg(feature = "jobs")]
pub job_queue: Option<Arc<crate::jobs::JobQueue>>,
/// **The door admission** (`crate::admission`), shared with both doors, read
/// by `AdmissionService` (`admit plan`, the usage book). `None` ⇒ the
/// service answers from a pass-through door: every plan `Full`.
#[cfg(feature = "admission")]
pub admission: Option<Arc<crate::admission::DoorAdmission>>,
/// Hosted-SBOM store for the `SbomService` (attach/fetch SBOM documents
/// keyed by artifact coordinate). `None` (default) ⇒ `AttachSbom` returns
/// `failed_precondition` and `FetchSbom` returns `not_found` (SBOM hosting
/// not configured on this server). Installed via [`HolgerGrpc::with_sboms`].
#[cfg(feature = "intake")]
pub sboms: Option<crate::sbom::SharedSbomStore>,
/// Custom-property store powering the `Search` property axis. `None` (default)
/// ⇒ a property filter on a `Search` matches nothing (fail-closed — an
/// unconfigured store never yields false hits). Installed via
/// [`HolgerGrpc::with_properties`].
#[cfg(feature = "intake")]
pub properties: Option<crate::properties::SharedPropertyStore>,
/// **The evidence plane** — [`crate::attest::AttestStore`], keyed by artifact
/// DIGEST.
///
/// `attest.rs` is 1 458 lines of DSSE + in-toto verification that, until
/// 2026-09-20, had no door, no constructor and no consumer anywhere in the
/// shipped tree. This field is the constructor call site: `Holger::serve`
/// builds the store from `<data_root>/holger-attestations` and every surface
/// reads it from here, so a gate and an operator see one store.
///
/// `None` ⇒ not configured; a gate that needs evidence then refuses and says
/// the plane is absent, rather than treating "no evidence" and "no store" as
/// the same answer.
#[cfg(feature = "intake")]
pub attestations: Option<Arc<crate::attest::AttestStore>>,
/// holger's own signing identity for the evidence a promotion emits.
/// `None` ⇒ no verification summary is written, and the promote handler
/// records that in the audit detail instead of staying quiet.
#[cfg(feature = "intake")]
pub attest_signer: Option<Arc<crate::attest::AttestSigner>>,
/// **The gates that run on a PUT**, before the bytes reach the backend.
///
/// Empty ⇒ ingest is ungated, which is what holger did until 2026-09-20.
/// With gates here a vulnerable artifact is refused at the door instead of
/// being stored, served from the DEV repository, and only stopped when
/// somebody tries to promote it.
#[cfg(feature = "intake")]
pub ingest_gates: crate::promote::GateSet,
/// **The intake chain** (`promotion.intake`), shared with the HTTP door.
/// `Some` ⇒ a `PutArtifact` into its incoming repository is INGESTED through
/// it, a write into its other repositories and a promotion touching any of
/// them is refused by name, and a read of one of them goes through
/// [`IntakeChain::read_view`](crate::promote::intake::IntakeChain::read_view).
#[cfg(feature = "intake")]
pub intake: Option<Arc<crate::promote::intake::IntakeChain>>,
/// **The box's import doors** (lane W4, [`crate::import_door`]):
/// `AdminService.ImportIntelBundle` / `ImportBundle`, shared with the HTTP
/// door and the inbox jobs. `None` ⇒ both RPCs refuse by name.
#[cfg(feature = "intake")]
pub imports: Option<Arc<crate::import_door::ImportDoor>>,
/// **The box's export door** (lane B3, [`crate::export_door`]):
/// `holger.doors.v1.BundleService.Export`. `None` ⇒ the RPC refuses by name.
#[cfg(feature = "intake")]
pub exports: Option<Arc<crate::export_door::ExportDoor>>,
/// Network mutation posture. `ReadOnly` refuses every write/admin RPC at the
/// auth choke point, **before** the request body is collected — enforcement,
/// not the display-only `read_only_profile` above. Installed via
/// [`HolgerGrpc::with_operation_mode`].
pub operation_mode: crate::OperationMode,
/// **What this box IS** — the declared role and the `twin >= primary` store
/// verdict, shared with the HTTP gateway so both doors answer one posture.
///
/// Default: a primary, which refuses nothing on account of its role. A
/// `Twin` refuses every write and admin mutation by name
/// (`holger-twin-1`), at the same choke points `OperationMode::ReadOnly`
/// sits on and for the same reason: before the body is collected and before
/// a credential is parsed, because no credential can buy a write on a
/// follower. Installed via [`HolgerGrpc::with_twin_posture`].
pub twin: Arc<crate::twin::TwinPosture>,
/// **The growth's write hold** — shared with the HTTP door and every held
/// backend (`crate::grow::Hold`). While a growth drains, every write and
/// admin mutation is refused here by name (`holger-grow-1`), after the twin
/// check and before the body. Default: a hold nothing pauses.
pub grow_hold: Arc<crate::grow::Hold>,
/// **The growth plane** behind `holger.grow.v1.GrowService`. `None` ⇒ the
/// door refuses by name (`holger-grow-3`). Installed via [`HolgerGrpc::with_grow`].
pub grow: Option<Arc<crate::grow::GrowPlane>>,
/// **The live follow** — what `BecomeTwin` starts the moment it has recorded
/// the election (stinky 14 §P3), and what `Recharge` pulls with. Installed
/// by `Holger::start` ([`HolgerGrpc::with_follower`]); `None` in a bare
/// `HolgerGrpc`, where `BecomeTwin` still records and switches the posture
/// and says the follow was not started.
pub follower: Option<crate::twin_follow::Follower>,
/// **The paid state, and the key that vouches for it** — the metered build's
/// half, absent from every open build (the field itself is `cfg`'d out, so an
/// open binary has no slot for it).
///
/// `None` (default) ⇒ `EntitlementService` answers `failed_precondition`: a
/// metered BUILD that was never configured with monetize's public key must
/// refuse to store a signed fact rather than store one it cannot check.
/// Installed via [`HolgerGrpc::with_entitlements`].
#[cfg(feature = "metered")]
pub entitlements: Option<Arc<crate::metered::EntitlementStore>>,
/// What this boot loaded and what it may reach, reported verbatim by
/// `ServerProfile`. Default EMPTY, which is the honest answer for a server
/// that was never handed one — not a claim that nothing is loaded.
pub module_census: ModuleCensus,
/// **The filesystem the product-scope storage meters are measured over** —
/// holger's own data root, the parent of the resolved audit directory.
///
/// `None` (the default) ⇒ `storage_total_bytes`, `storage_servable_bytes`
/// and `storage_used_bytes` are REFUSED BY NAME rather than answered zero.
/// That is the honest answer for a `HolgerGrpc` nobody handed a root to —
/// a fixture in a test has no filesystem it means to measure, and
/// measuring the test runner's disk instead would be a number about the
/// wrong thing. Installed by the serve path via
/// [`HolgerGrpc::with_data_root`].
pub data_root: Option<std::path::PathBuf>,
/// **The box's repository key** (lane HOL-SIGN-2, [`crate::repo_sign`]): the console's
/// apt and dnf setup name its fingerprint as the key a client trusts. `None`: the box
/// holds none (the setup says so). Installed via [`HolgerGrpc::with_repo_key`].
pub repo_key: Option<Arc<crate::repo_sign::RepoKey>>,
/// The declaration's graph the estate board draws (HBOARD-1; [`crate::topology::declared`]).
/// Empty by default; installed via [`HolgerGrpc::with_declared`].
pub board_declared: Arc<crate::topology::Declared>,
/// **The artifact flow's shape** ([`crate::flow::FlowGraph::from_config`], lane
/// HOL-SVG-RON). `None` ⇒ `AdminService.FlowSvg` answers NotFound (`flow: (enabled:
/// false)`, or a service nobody installed it on). Installed via [`HolgerGrpc::with_flow`].
pub flow_shape: Option<Arc<crate::flow::FlowGraph>>,
/// A paid repository's disk cap, for the flow's repository boxes (metered builds).
pub flow_quota: Option<crate::flow::QuotaOf>,
/// Where each repository's bytes are on disk (`disk_bytes`, owner 2026-09-27).
/// Empty ⇒ that meter is refused by name, never zero. Installed via
/// [`HolgerGrpc::with_storage_places`].
pub storage_places: Arc<crate::metering::StoragePlaces>,
/// **What serves a repository made at run time** (lane M4b): a paid signup's repository
/// goes into [`Self::routes`] at once. `None` ⇒ such a repository serves from the next
/// start. Installed via [`HolgerGrpc::with_live_routes`].
pub live_routes: Option<Arc<crate::live_routes::LiveRoutes>>,
/// Each repository's upstreams (empty = it holds its own bytes, not a
/// cache) — the admin view's cache line. Installed via
/// [`HolgerGrpc::with_upstreams`].
pub upstreams: Arc<std::collections::BTreeMap<String, Vec<String>>>,
/// **Running and recent MIGRATIONS**, by job id — the ledger that lets
/// `MigrationService.Start` return in milliseconds while the work runs for
/// hours, and lets `Watch` and `Cancel` find the job afterwards.
///
/// Always present (an empty ledger costs a `HashMap`), unlike the optional
/// stores above: there is no configuration to get wrong, and a server that
/// could accept `Start` but had nowhere to put the job would be a server
/// that loses migrations. ONLINE only, with the service itself.
#[cfg(feature = "migration")]
pub migration_jobs: Arc<crate::migration::MigrationJobs>,
/// **The meter guests that narrate this server's usage doors.**
///
/// Empty ⇒ `ReadUsage` reads `crate::metering::count` directly, which is
/// what holger has always done. Non-empty ⇒ the guests answer: the numbers
/// cross the plugin ABI into the module through its `usage.read` import and
/// come back out of its `read` export, and every reading is labelled
/// `meter:<module>` so nothing has to infer who answered.
///
/// Installed by the serve path from the plane `wire_holger` built; see
/// [`HolgerGrpc::with_meters`].
#[cfg(feature = "meter-guests")]
pub meters: Vec<Arc<dyn holger_plugin_host::MeterRuntime>>,
/// The configured name of the door this service answers on — what the
/// visitor tally counts gRPC requests against. `None` in a fixture.
pub door: Option<String>,
/// The per-door, per-repository visitor tally, shared with the HTTP door.
/// See [`crate::visitors`].
pub visitors: Arc<crate::visitors::Visitors>,
/// The endpoint's `ron_trusted_proxies` (SEC-REVIEW F1): from a peer in it the right-most
/// `X-Forwarded-For` entry is the client the visitor tally counts.
pub trusted_proxies: Option<Arc<crate::listen::AllowList>>,
/// When this server's console last called, shared with the HTTP door
/// ([`crate::flow::ConsoleSeen`]).
pub console: crate::flow::ConsoleSeen,
/// The paired holgers this box knows by key. See [`crate::peers`].
pub peers: Arc<crate::peers::PeerBoard>,
}
/// The property keys reserved to ADMIN when the config declares none.
///
/// `keep` is the DOCUMENTED default retention pin (`--protect-property keep=true`,
/// `RetentionPolicy::protect_property`). It is a default, not a derivation: the
/// retention policy is built in a SEPARATE process from a CLI flag, so a running
/// server cannot see which key this deployment actually pins on. A deployment
/// that renamed the pin must say so in its config's `reserved_properties`, and
/// the `retention` verb refuses to run against a key the server does not reserve
/// — otherwise any Writer sets the pin over the wire and retention never
/// reclaims what it was told to.
pub const DEFAULT_RESERVED_PROPERTY_KEYS: &[&str] = &["keep"];
impl HolgerGrpc {
pub fn new(routes: FastRoutes) -> Self {
Self::with_auth(routes, Arc::new(AuthConfig::default()))
}
pub fn with_auth(routes: FastRoutes, auth_config: Arc<AuthConfig>) -> Self {
Self {
routes,
start_time: Instant::now(),
auth_config,
read_gate: Arc::new(crate::auth::ReadGate::GLOBAL_ONLY),
audit: Arc::new(NoopAuditLog),
max_body_bytes: DEFAULT_MAX_BODY_BYTES,
max_in_flight_read_bytes: DEFAULT_MAX_IN_FLIGHT_READ_BYTES,
read_budget: Arc::new(tokio::sync::Semaphore::new(
DEFAULT_MAX_IN_FLIGHT_READ_BYTES / READ_BUDGET_UNIT_BYTES,
)),
max_concurrent_reads: DEFAULT_MAX_CONCURRENT_READS,
read_slots: Arc::new(tokio::sync::Semaphore::new(DEFAULT_MAX_CONCURRENT_READS)),
reserved_properties: DEFAULT_RESERVED_PROPERTY_KEYS
.iter()
.map(|k| k.to_string())
.collect(),
read_only_profile: None,
#[cfg(feature = "intake")]
promotion: None,
#[cfg(feature = "jobs")]
job_queue: None,
#[cfg(feature = "admission")]
admission: None,
#[cfg(feature = "intake")]
sboms: None,
#[cfg(feature = "intake")]
properties: None,
#[cfg(feature = "intake")]
attestations: None,
#[cfg(feature = "intake")]
attest_signer: None,
#[cfg(feature = "intake")]
ingest_gates: crate::promote::GateSet::empty(),
#[cfg(feature = "intake")]
intake: None,
#[cfg(feature = "intake")]
imports: None,
#[cfg(feature = "intake")]
exports: None,
operation_mode: crate::OperationMode::default(),
twin: Arc::new(crate::twin::TwinPosture::default()),
grow_hold: Arc::new(crate::grow::Hold::new()),
grow: None,
follower: None,
#[cfg(feature = "metered")]
entitlements: None,
module_census: ModuleCensus::default(),
data_root: None,
repo_key: None,
board_declared: Arc::new(Default::default()),
flow_shape: None,
flow_quota: None,
storage_places: Arc::new(Default::default()),
live_routes: None,
upstreams: Arc::new(Default::default()),
#[cfg(feature = "migration")]
migration_jobs: Arc::new(crate::migration::MigrationJobs::default()),
#[cfg(feature = "meter-guests")]
meters: Vec::new(),
door: None,
visitors: Arc::new(crate::visitors::Visitors::empty()),
trusted_proxies: None,
console: Default::default(),
peers: Arc::new(crate::peers::PeerBoard::default()),
}
}
/// **Install the door name and the visitor tally** (builder style): the
/// name of the door this service answers on, and the tally both transports
/// count into.
pub fn with_door(mut self, door: &str, visitors: Arc<crate::visitors::Visitors>) -> Self {
self.door = Some(door.to_owned());
self.visitors = visitors;
self
}
/// The endpoint's `ron_trusted_proxies` (builder style, SEC-REVIEW F1).
pub fn with_trusted_proxies(mut self, trusted: Option<Arc<crate::listen::AllowList>>) -> Self {
self.trusted_proxies = trusted;
self
}
/// **Install when this server's console last called** (builder style, HOL-SYNC-SKEW):
/// the SAME [`crate::flow::ConsoleSeen`] the HTTP door of this server marks, so the
/// flow `AdminService.FlowSvg` draws is this server's own.
pub fn with_console(mut self, console: crate::flow::ConsoleSeen) -> Self {
self.console = console;
self
}
/// **Install the peer board** (builder style): the paired downstreams this
/// box recognises by certificate, and the paired upstreams its mirror pins.
pub fn with_peers(mut self, peers: Arc<crate::peers::PeerBoard>) -> Self {
self.peers = peers;
self
}
/// **Who pulled: a paired downstream, or a visitor.**
///
/// A request whose mTLS leaf certificate matches a paired downstream's pin
/// is that downstream: it is counted on its own row, refused by name
/// (`holger-peer-2`) for a repository its pairing does not list, and kept
/// OUT of the anonymous visitor tally. Anything else is a visitor, counted as
/// before. `Ok(Some(idx))` names the peer.
pub(crate) fn count_visitor<T>(
&self,
request: &Request<T>,
repo: &str,
pull: bool,
) -> Result<Option<usize>, Status> {
let paired = request.peer_certs().and_then(|certs| {
certs
.first()
.and_then(|leaf| self.peers.downstream_by_cert(leaf.as_ref()))
});
if let Some(idx) = paired {
if !pull {
// A paired key writing is not a pull and not a visit: its
// write is authorised by the role map as always.
return Ok(None);
}
let rec = &self.peers.downstreams()[idx];
if !rec.repos.iter().any(|r| r == repo) {
return Err(coded(
&holger_errcode::PEER_REPO_NOT_PAIRED,
tonic::Code::PermissionDenied,
format!(
"downstream holger `{}` is paired for [{}], not for `{repo}`",
rec.name,
rec.repos.join(", ")
),
));
}
self.peers.record_pull(idx, 0, crate::visitors::now_unix());
return Ok(Some(idx));
}
let xff = request
.metadata()
.get_all("x-forwarded-for")
.iter()
.filter_map(|v| v.to_str().ok())
.rfind(|v| !v.trim().is_empty());
self.visitors.record(
crate::visitors::now_unix(),
self.door.as_deref(),
Some(repo),
crate::visitors::client_via(
request.remote_addr().map(|a| a.ip()),
xff,
self.trusted_proxies.as_deref(),
),
);
Ok(None)
}
/// The audit principal for a read: the paired downstream's name when the
/// pull came from one, else what the read gate resolved.
pub(crate) fn pull_principal(
&self,
peer: Option<usize>,
identity: &Option<auth::AuthIdentity>,
) -> String {
match peer.and_then(|i| self.peers.downstreams().get(i)) {
Some(d) => format!("{}{}", crate::peers::PEER_PRINCIPAL_PREFIX, d.name),
None => HolgerGrpc::read_principal(identity),
}
}
/// **Install the meter guests that narrate the usage doors** (builder style).
///
/// They are the ones `wire_holger` loaded and wired to the three host
/// services; this hands the SAME runtimes to the doors, so the plane the
/// census reports, the plane the gates run and the plane that answers
/// `ReadUsage` are one plane and not three.
#[cfg(feature = "meter-guests")]
pub fn with_meters(mut self, meters: Vec<Arc<dyn holger_plugin_host::MeterRuntime>>) -> Self {
self.meters = meters;
self
}
/// **Install the data root the storage meters measure** (builder style).
///
/// The same path `build_meter_plane` hands the guest seam, so the door and a
/// loaded meter guest report one filesystem and not two.
pub fn with_data_root(mut self, root: impl Into<std::path::PathBuf>) -> Self {
self.data_root = Some(root.into());
self
}
/// The box's repository key (HOL-SIGN-2), which the console's client setup names.
pub fn with_repo_key(mut self, key: Option<Arc<crate::repo_sign::RepoKey>>) -> Self {
self.repo_key = key;
self
}
/// The artifact flow's shape and the quota reader (builder style; HOL-SVG-RON).
pub fn with_flow_door(mut self, door: Option<crate::flow::Door>) -> Self {
self.flow_shape = door.as_ref().map(|d| Arc::clone(&d.shape));
self.flow_quota = door.and_then(|d| d.quota);
self
}
/// The declaration's graph the estate board draws (builder style; HBOARD-1).
pub fn with_declared(mut self, declared: Arc<crate::topology::Declared>) -> Self {
self.board_declared = declared;
self
}
/// What serves a repository made at run time (builder style) — [`Self::live_routes`].
pub fn with_live_routes(mut self, live: Option<Arc<crate::live_routes::LiveRoutes>>) -> Self {
self.live_routes = live;
self
}
/// Install where each repository's bytes are (builder style).
pub fn with_storage_places(mut self, places: crate::metering::StoragePlaces) -> Self {
self.storage_places = Arc::new(places);
self
}
/// Each repository's upstreams (builder style) — [`Self::upstreams`].
pub fn with_upstreams(
mut self,
upstreams: std::collections::BTreeMap<String, Vec<String>>,
) -> Self {
self.upstreams = Arc::new(upstreams);
self
}
/// Set the network mutation posture (builder style). `ReadOnly` fails every
/// write/admin RPC before the body is read.
pub fn with_operation_mode(mut self, mode: crate::OperationMode) -> Self {
self.operation_mode = mode;
self
}
/// **Declare this box's role** (builder style) — plan §3 W9.
///
/// A `Twin` posture makes every write and admin gate below refuse by name.
pub fn with_twin_posture(mut self, posture: Arc<crate::twin::TwinPosture>) -> Self {
self.twin = posture;
self
}
/// **The data-set growth** (builder style): the plane the `GrowService` door
/// drives and the hold every write gate asks. The hold must be the one the
/// repositories were wrapped with (`Holger::grow_hold`).
#[cfg(feature = "estate")]
pub fn with_grow(mut self, plane: Arc<crate::grow::GrowPlane>) -> Self {
self.grow_hold = Arc::clone(&plane.hold);
self.grow = Some(plane);
self
}
/// The growth plane when this build has one, else the process's write hold alone
/// (a build without `estate` has no plane: `GrowService` answers `holger-grow-3`).
pub fn with_grow_or_hold(
mut self,
plane: Option<Arc<crate::grow::GrowPlane>>,
hold: Arc<crate::grow::Hold>,
) -> Self {
match plane {
#[cfg(feature = "estate")]
Some(p) => self.with_grow(p),
#[cfg(not(feature = "estate"))]
Some(p) => match *p {},
None => {
self.grow_hold = hold;
self
}
}
}
/// Install the live follow (builder style) — see [`HolgerGrpc::follower`].
pub fn with_follower(mut self, follower: crate::twin_follow::Follower) -> Self {
self.follower = Some(follower);
self
}
/// Install the immutable PROD store for the promote workflow (builder style).
#[cfg(feature = "intake")]
pub fn with_promotion(mut self, engine: Arc<crate::promote::PromotionEngine>) -> Self {
self.promotion = Some(engine);
self
}
/// Install the door admission `AdmissionService` reads (builder style).
#[cfg(feature = "admission")]
pub fn with_admission(mut self, admission: Arc<crate::admission::DoorAdmission>) -> Self {
self.admission = Some(admission);
self
}
/// Install the durable job queue (builder style). With one attached, a
/// promotion or replication that asks for the async shape is ADMITTED and
/// answered with its job id; without one, asking is refused by name.
#[cfg(feature = "jobs")]
pub fn with_job_queue(mut self, queue: Arc<crate::jobs::JobQueue>) -> Self {
self.job_queue = Some(queue);
self
}
/// Install the hosted-SBOM store for the `SbomService` (builder style). The
/// server wires `Holger::sbom_store()` here so gRPC, HTTP, and the CLI share
/// one on-disk store.
#[cfg(feature = "intake")]
pub fn with_sboms(mut self, store: crate::sbom::SharedSbomStore) -> Self {
self.sboms = Some(store);
self
}
/// Install the custom-property store powering the `Search` property axis
/// (builder style). The server wires `Holger::property_store()` here so gRPC,
/// HTTP, and the CLI share one on-disk store.
#[cfg(feature = "intake")]
pub fn with_properties(mut self, store: crate::properties::SharedPropertyStore) -> Self {
self.properties = Some(store);
self
}
/// Mount the attestation store (builder style). Without this call
/// `attest.rs` has no consumer, which is exactly what was measured at
/// `542e296`.
#[cfg(feature = "intake")]
pub fn with_attestations(mut self, store: Option<Arc<crate::attest::AttestStore>>) -> Self {
self.attestations = store;
self
}
/// Mount holger's own signing identity (builder style). Without it a
/// promotion writes no verification summary, which the promote handler
/// records by name.
/// Emit ONE SLSA Verification Summary for a promotion, and return the audit
/// detail that says what happened.
///
/// Three outcomes, all of them recorded by name:
///
/// * no attestation store configured — nothing was attempted;
/// * a store but no signer — nothing could be signed, so nothing was
/// written;
/// * signed and filed, or refused by the store's own verification.
///
/// The last case matters: holger's statement goes through
/// [`crate::attest::AttestStore::accept`] like anyone else's, so if holger's
/// signer certificate does not chain to a configured trust root its own
/// evidence is refused. That is the right answer. A statement nobody can
/// verify is not evidence, and a store that files it anyway produces green
/// audit rows with nothing behind them.
///
/// This function never fails the promotion. The bytes are already sealed and
/// the seal is immutable; refusing here would leave a promoted artifact and
/// an error to the client, which is worse than a promoted artifact and an
/// audit row saying its evidence could not be written.
#[cfg(feature = "intake")]
pub fn emit_promotion_vsa(&self, label: &str, bytes: &[u8], gate_summary: &str) -> String {
let Some(store) = self.attestations.as_ref() else {
return "no verification summary: no attestation store is configured on this server \
(`attestation:` block absent)"
.to_string();
};
let Some(signer) = self.attest_signer.as_ref() else {
return "no verification summary: an attestation store is configured but no signer \
(`attestation.signer_key` / `signer_cert` absent), so nothing could be \
signed. The artifact is promoted and has NO evidence."
.to_string();
};
// Without `sign` there is no DSSE implementation linked, and
// `AttestSigner` is uninhabited, so the branch above has already
// returned. This arm exists so the tier that ships without `sign`
// compiles and says out loud why it emits nothing.
#[cfg(not(feature = "sign"))]
#[allow(clippy::needless_return)] // the tail only in a build without `sign`
{
let _ = (label, store, signer, bytes, gate_summary);
return "no verification summary: this build carries no `sign` feature, so it links \
no DSSE implementation and cannot emit one"
.to_string();
}
#[cfg(feature = "sign")]
{
let digest = crate::attest::ArtifactDigest::of_sha256(bytes);
// The estate's clock (`crate::clock`): an attestation's time is the
// front's signed time on a box of the estate, not the bare system clock.
let now = i64::try_from(crate::clock::now_unix_secs()).unwrap_or(0);
let payload = crate::attest::verification_summary_bytes(
&[(label.to_string(), digest.clone())],
"holger",
crate::attest::VerificationResult::Passed,
"holger://promotion-gates",
&[gate_summary.to_string()],
&crate::attest::rfc3339_utc(now),
);
match store.emit_signed(&payload, signer.as_ref(), Some(&digest)) {
Ok(a) => format!(
"verification summary written: envelope {} predicate {} over {}",
a.envelope_id, a.predicate_type, digest
),
Err(e) => format!(
"verification summary REFUSED by the attestation store: {e}. The artifact is \
promoted and has NO evidence for these bytes."
),
}
}
}
/// Mount the INGEST gates (builder style).
#[cfg(feature = "intake")]
pub fn with_ingest_gates(mut self, gates: crate::promote::GateSet) -> Self {
self.ingest_gates = gates;
self
}
/// Mount the intake chain (builder style).
#[cfg(feature = "intake")]
pub fn with_intake(mut self, chain: Option<Arc<crate::promote::intake::IntakeChain>>) -> Self {
self.intake = chain;
self
}
/// Mount the import doors (builder style).
#[cfg(feature = "intake")]
pub fn with_imports(mut self, door: Option<Arc<crate::import_door::ImportDoor>>) -> Self {
self.imports = door;
self
}
/// Mount the export door (builder style, lane B3).
#[cfg(feature = "intake")]
pub fn with_exports(mut self, door: Option<Arc<crate::export_door::ExportDoor>>) -> Self {
self.exports = door;
self
}
/// **The import RPCs' receiving half**: the holds every admin mutation
/// answers, the credential, the duty (Repository Admin or Security
/// Approver), then the streamed bytes into a staged file under the data
/// root, capped at the door's body limit. Answers the principal, the door,
/// and the staged bundle; a chunk that asks for a first import is refused (holger-import-10).
#[cfg(feature = "intake")]
pub(crate) async fn receive_import(
&self,
request: Request<tonic::Streaming<holger_proto::ImportBundleChunk>>,
kind: crate::import_door::ImportKind,
) -> Result<
(
String,
Arc<crate::import_door::ImportDoor>,
crate::import_door::StagedBundle,
),
Status,
> {
let identity = self.authorize_intake(&request, true).await?;
let Some(door) = self.imports.clone() else {
return Err(coded(
&holger_errcode::SERVER_CAPABILITY_NOT_CONFIGURED,
tonic::Code::FailedPrecondition,
"this server has no import door (no data root, or a build without the supply chain)",
));
};
let principal =
crate::import_door::ImportDoor::authorize(&self.auth_config, identity.as_ref(), kind)
.map_err(|e| {
// HOL-NCSC: an unauthorised importer leaves an audit row.
let who = crate::promote::duties::principal_of(identity.as_ref());
self.record_audit(
crate::audit::AuditEvent::new(
&who,
crate::audit::AuditAction::Other,
"",
kind.as_str(),
self.audit_peer(&request),
e.status,
0,
)
.with_detail(format!(
"refused: import authorization: {} import by {who}: {}",
kind.as_str(),
e.tagged()
)),
);
import_status(&e)
})?;
let mut staged = door.stage().map_err(|e| import_status(&e))?;
let mut stream = request.into_inner();
let mut total: usize = 0;
while let Some(chunk) = stream.message().await? {
// The caller's `first_import` is REMOVED (lane HOL-CDS, NCSC gap 2): the first
// import is an admin's authorization on this side.
if chunk.first_import {
return Err(import_status(
&crate::import_door::first_import_flag_refusal("a chunk's `first_import`"),
));
}
total += chunk.data.len();
if total > self.max_body_bytes {
return Err(coded(
&holger_errcode::IMPORT_STAGING_FAILED,
tonic::Code::ResourceExhausted,
format!(
"the bundle is larger than this door's limit of {} bytes",
self.max_body_bytes
),
));
}
staged.write(&chunk.data).map_err(|e| import_status(&e))?;
}
staged.finish().map_err(|e| import_status(&e))?;
Ok((principal, door, staged))
}
/// The backend a READ of `repo` must go through: the intake chain's view of
/// an intake repository, the backend itself otherwise.
pub(crate) fn read_backend(
&self,
repo: &str,
backend: Arc<dyn traits::RepositoryBackendTrait>,
) -> Arc<dyn traits::RepositoryBackendTrait> {
#[cfg(feature = "intake")]
if let Some(chain) = &self.intake {
return chain.read_view(repo, backend);
}
let _ = repo;
backend
}
#[cfg(feature = "intake")]
pub fn with_attest_signer(mut self, signer: Option<Arc<crate::attest::AttestSigner>>) -> Self {
self.attest_signer = signer;
self
}
/// Install the module census this boot built, reported verbatim by
/// . In EVERY build: what is loaded is not a paid question.
pub fn with_module_census(mut self, census: ModuleCensus) -> Self {
self.module_census = census;
self
}
/// Install the entitlement store — the metered build's paid state, verified
/// against monetize's public key (builder style).
///
/// Builder style rather than a constructor argument on purpose: `with_auth`
/// builds a struct literal that 26 call sites already spell, and a new required
/// field would break every one of them for a capability almost none of them has.
#[cfg(feature = "metered")]
pub fn with_entitlements(mut self, store: Arc<crate::metered::EntitlementStore>) -> Self {
self.entitlements = Some(store);
self
}
/// Install the per-repository read scope (builder style). The server wires
/// the repositories that declared `ron_require_auth` here.
///
/// Builder-style rather than a `with_auth` parameter on purpose: there are
/// two dozen `with_auth` call sites and every one of them means "the global
/// posture", which is exactly what an empty scope is. A door that never opts
/// in behaves exactly as it did.
pub fn with_read_gate(mut self, read_gate: Arc<crate::auth::ReadGate>) -> Self {
self.read_gate = read_gate;
self
}
/// Install an audit backend (builder style). The server wires the default
/// Arrow-IPC log here when an audit directory is configured.
pub fn with_audit(mut self, audit: Arc<dyn AuditLog>) -> Self {
self.audit = audit;
self
}
/// Override the ingest body cap (builder style). The server wires the
/// endpoint's `ron_max_body_bytes` here so gRPC and HTTP share one limit.
pub fn with_max_body_bytes(mut self, max_body_bytes: usize) -> Self {
self.max_body_bytes = max_body_bytes;
self
}
/// Override the in-flight READ budget (builder style); `0` disables the
/// accounting and restores the old unbounded behaviour. See
/// [`DEFAULT_MAX_IN_FLIGHT_READ_BYTES`] for the reasoning behind the default.
pub fn with_max_in_flight_read_bytes(mut self, max_bytes: usize) -> Self {
self.max_in_flight_read_bytes = max_bytes;
self.read_budget = Arc::new(tokio::sync::Semaphore::new(
max_bytes / READ_BUDGET_UNIT_BYTES,
));
self
}
/// Cap how many reads may be in flight at once, process-wide (`0` = uncapped).
/// See [`DEFAULT_MAX_CONCURRENT_READS`] for why this exists beside the byte
/// budget and why it is not per-connection.
pub fn with_max_concurrent_reads(mut self, max_reads: usize) -> Self {
self.max_concurrent_reads = max_reads;
self.read_slots = Arc::new(tokio::sync::Semaphore::new(max_reads));
self
}
/// Replace the set of property keys that require ADMIN to set (builder style).
/// Pass an empty iterator to reserve nothing beyond the key derived from live
/// state (the promotion requirement), which is never optional.
/// See [`HolgerGrpc::reserved_properties`].
pub fn with_reserved_properties<I, S>(mut self, keys: I) -> Self
where
I: IntoIterator<Item = S>,
S: Into<String>,
{
self.reserved_properties = keys.into_iter().map(Into::into).collect();
self
}
/// **Reserve the minimum slice of the in-flight read budget, BEFORE fetching.**
///
/// One unit — the floor every read pays — so a burst of requests cannot all get
/// as far as `fetch` and materialise their blobs simultaneously; the rest of the
/// cost is charged by [`charge_read_budget`](Self::charge_read_budget) once the
/// size is known. The two-step shape is forced by the backend contract: `fetch`
/// is the only read and it returns the finished `Vec<u8>`, so there is no size
/// to reserve against until the allocation has already happened (a streaming
/// backend API would let this be one exact reservation — see the module docs).
///
/// Non-blocking on purpose: a read that does not fit is REFUSED
/// (`resource_exhausted`) rather than queued, because a queue of waiters on an
/// open read door is itself an amplifier.
pub(crate) fn reserve_read_budget(&self, rpc: &'static str) -> Result<ReadBudget, Status> {
// The SLOT first: it is the cheaper refusal and the one that bounds how
// many requests can even reach the byte budget. Taken before the bytes so
// a flood of tiny reads is turned away at the door rather than each
// charging a 1 MiB floor against the memory budget on its way in.
let slot = if self.max_concurrent_reads == 0 {
None
} else {
match self.read_slots.clone().try_acquire_owned() {
Ok(slot) => Some(slot),
Err(_) => {
functional_status(rpc, "read_slot_available", false, "all read slots busy");
return Err(coded(
&holger_errcode::SERVER_TOO_MANY_CONCURRENT_READS,
tonic::Code::ResourceExhausted,
format!(
"server is already serving its maximum of {} concurrent reads; \
retry shortly",
self.max_concurrent_reads
),
));
}
}
};
if self.max_in_flight_read_bytes == 0 {
return Ok(ReadBudget {
permit: None,
units: 0,
_slot: slot,
});
}
match self.read_budget.clone().try_acquire_many_owned(1) {
Ok(permit) => Ok(ReadBudget {
permit: Some(permit),
units: 1,
_slot: slot,
}),
Err(_) => {
functional_status(rpc, "read_budget_available", false, "budget exhausted");
Err(coded(
&holger_errcode::SERVER_READ_BUDGET_EXHAUSTED,
tonic::Code::ResourceExhausted,
format!(
"server is at its in-flight read budget ({} bytes); retry shortly",
self.max_in_flight_read_bytes
),
))
}
}
}
/// **Charge the rest of an artifact's cost once its real size is known.**
///
/// Tops the held reservation up to `ceil(n / 1 MiB)` units. On refusal the
/// caller drops the bytes immediately and answers `resource_exhausted`, so the
/// blob is resident for the length of one handler frame instead of the length
/// of a slow client's download — which is the multiplication this bounds. An
/// artifact larger than the WHOLE budget can never be charged and is refused
/// with a message that says so, rather than looking like transient load.
pub(crate) fn charge_read_budget(
&self,
held: &mut ReadBudget,
n: usize,
rpc: &'static str,
) -> Result<(), Status> {
let Some(permit) = held.permit.as_mut() else {
return Ok(()); // budget disabled
};
let extra = read_budget_units(n).saturating_sub(1);
if extra == 0 {
return Ok(());
}
match self.read_budget.clone().try_acquire_many_owned(extra) {
Ok(more) => {
permit.merge(more);
held.units = held.units.saturating_add(extra);
Ok(())
}
Err(_) => {
functional_status(rpc, "read_budget_available", false, "artifact does not fit");
Err(coded(
&holger_errcode::SERVER_READ_BUDGET_EXHAUSTED,
tonic::Code::ResourceExhausted,
format!(
"an artifact of {n} bytes does not fit the server's in-flight read \
budget ({} bytes); retry shortly",
self.max_in_flight_read_bytes
),
))
}
}
}
/// Declare the server's profile explicitly (builder style). `true` pins the
/// static "rigged for silent running" / read-only profile, `false` pins
/// dynamic. Without this the `ServerProfile` RPC derives the profile from the
/// live route table. A sealed static-holger sets `true` here.
pub fn with_read_only_profile(mut self, read_only: bool) -> Self {
self.read_only_profile = Some(read_only);
self
}
/// The number of configured repositories that accept writes.
fn writable_repo_count(&self) -> usize {
self.routes
.all_repos()
.iter()
.filter(|(_, repo)| repo.is_writable())
.count()
}
/// The server's effective read-only status: the explicit override if set,
/// else derived from the route table (read-only iff *no* repo is writable).
/// An empty server (no repos) is read-only — it can serve nothing writable.
fn is_read_only(&self) -> bool {
self.read_only_profile
.unwrap_or_else(|| self.writable_repo_count() == 0)
}
/// Best-effort audit append. An audit-backend failure must NOT fail the
/// request it describes (availability over a single lost line), but it is
/// surfaced via `log::warn!` so a broken sink is never silent.
/// Charge the budget AS A STREAM GOES OUT: bring `held` up to what
/// `total_so_far` bytes cost, acquiring only the units not yet paid for.
/// The streaming door calls this per chunk, so a stream is charged for
/// what it has actually put in flight and never for a size it had to know
/// up front — which is what lets it stream without materialising first.
pub(crate) fn charge_read_budget_to(
&self,
held: &mut ReadBudget,
total_so_far: usize,
rpc: &'static str,
) -> Result<(), Status> {
let Some(permit) = held.permit.as_mut() else {
return Ok(()); // budget disabled
};
let want = read_budget_units(total_so_far);
if want <= held.units {
return Ok(());
}
let extra = want - held.units;
match self.read_budget.clone().try_acquire_many_owned(extra) {
Ok(more) => {
permit.merge(more);
held.units = want;
Ok(())
}
Err(_) => {
functional_status(
rpc,
"read_budget_available",
false,
"stream outgrew the budget",
);
Err(coded(
&holger_errcode::SERVER_READ_BUDGET_EXHAUSTED,
tonic::Code::ResourceExhausted,
format!(
"a stream of {total_so_far} bytes so far does not fit the server's \
in-flight read budget; retry shortly"
),
))
}
}
}
pub(crate) fn record_audit(&self, event: AuditEvent) {
// Count the request on the process-global `/metrics` registry BEFORE the
// (best-effort) audit append — this is the single gRPC-plane choke point
// every RPC exits through, so the whole gRPC surface is observable in
// `/metrics` without threading a handle into each handler. The HTTP door
// bumps at its own exit (`exposed::http`); the two never double-count
// because HTTP does not route through this method. `Other` actions aren't
// request traffic and map to `None` (not counted).
if let Some(verb) = crate::metrics::Verb::from_audit_action(event.action) {
crate::metrics::global().record_request(verb, event.status, event.bytes);
}
if let Err(e) = self.audit.record(event) {
log::warn!("audit record failed: {e}");
}
}
pub(crate) fn get_repo(
&self,
name: &str,
) -> Result<Arc<dyn traits::RepositoryBackendTrait>, Status> {
self.routes.lookup(name).ok_or_else(|| {
coded(
&holger_errcode::SERVER_REPO_UNKNOWN,
tonic::Code::NotFound,
format!("Repository '{}' not found", name),
)
})
}
/// Validate write-request credentials against the configured auth methods.
/// Accepts a Bearer token or an mTLS client-cert CN. An empty config means
/// open access. Since the streamed upload (OX-WIRED2) only the intake doors and the tests ask it.
#[cfg(any(feature = "intake", test))]
async fn authorize_write<T>(
&self,
request: &Request<T>,
repo: &str,
) -> Result<Option<auth::AuthIdentity>, Status> {
self.write_principal(request, repo)
.await
.map_err(|(status, _)| status)
}
/// **An upload's write check, audited when it refuses** (lane OX-WIRED2): what
/// [`authorize_write`](HolgerGrpc::authorize_write) decides, and a refusal is an
/// `Upload` audit row (who, when known; the repository; the artifact; 401 or 403), as
/// the HTTP door writes one. Before this a refused gRPC upload left no trace.
pub(crate) async fn authorize_upload<T>(
&self,
request: &Request<T>,
repo: &str,
artifact: &str,
source_ip: &str,
) -> Result<Option<auth::AuthIdentity>, Status> {
match self.write_principal(request, repo).await {
Ok(identity) => Ok(identity),
Err((status, who)) => {
let code = match status.code() {
tonic::Code::Unauthenticated => 401,
tonic::Code::Unavailable => 503,
_ => 403,
};
let who = who.unwrap_or_else(|| "anonymous".to_string());
self.record_audit(
AuditEvent::new(
&who,
AuditAction::Upload,
repo,
artifact,
source_ip,
code,
0,
)
.with_detail(format!("upload refused: {}", status.message())),
);
Err(status)
}
}
}
/// [`authorize_write`](HolgerGrpc::authorize_write)'s decision, a refusal with the
/// subject it was refused to when one was named.
async fn write_principal<T>(
&self,
request: &Request<T>,
repo: &str,
) -> Result<Option<auth::AuthIdentity>, (Status, Option<String>)> {
// A TWIN refuses every write, FIRST — before the read-only check, before
// the body, before a credential is parsed. It is asked first because it
// is the more specific answer: on a follower "this is a twin of X, write
// there" is an instruction, and "server is in read-only mode" is a
// riddle. No credential can buy a write here, so there is nothing to
// A box awaiting its recharge serves nothing, and says so before anything
// else (`holger-twin-15`).
if let Some(refusal) = self.twin.serve_refusal() {
return Err((
coded(
&holger_errcode::TWIN_AWAITING_RECHARGE,
tonic::Code::Unavailable,
refusal,
),
None,
));
}
// authenticate. Plan §3 W9; `holger-twin-1`.
if let Some(refusal) = self.twin.write_refusal() {
return Err((
coded(
&holger_errcode::TWIN_REFUSES_WRITES,
tonic::Code::PermissionDenied,
refusal,
),
None,
));
}
// A GROWTH holds every write while it drains and while the set is off
// (`crate::grow`, `holger-grow-1`): UNAVAILABLE, so a client retries after
// the growth instead of giving up — refused, never queued.
if let Some(held) = self.grow_hold.write_refusal() {
return Err((
coded(
&holger_errcode::GROW_WRITES_HELD,
tonic::Code::Unavailable,
held,
),
None,
));
}
// ReadOnly mode refuses every mutation BEFORE the body is collected.
if self.operation_mode == crate::OperationMode::ReadOnly {
return Err((
coded(
&holger_errcode::SERVER_READ_ONLY,
tonic::Code::PermissionDenied,
"server is in read-only mode",
),
None,
));
}
let bearer = request
.metadata()
.get("authorization")
.and_then(|v| v.to_str().ok())
.and_then(|v| v.strip_prefix("Bearer "));
// mTLS identity from the TLS handshake, if the server runs with a
// client CA configured.
let client_cn = request.peer_certs().and_then(|certs| {
let ders: Vec<rustls::pki_types::CertificateDer<'_>> = certs
.iter()
.map(|c| rustls::pki_types::CertificateDer::from(c.as_ref().to_vec()))
.collect();
crate::exposed::tls::leaf_common_name(&ders)
});
let identity =
match auth::validate_request(&self.auth_config, bearer, client_cn.as_deref()).await {
Ok(id) => id,
Err(_) => {
functional_status(
"holger-grpc/authorize_write",
"credentials_valid",
false,
"invalid or missing write credentials",
);
return Err((
coded(
&holger_errcode::AUTH_UNAUTHENTICATED,
tonic::Code::Unauthenticated,
"Valid credentials required for write access",
),
None,
));
}
};
// Success path: credentials accepted (or auth fully open). Emitted for
// every authorized write so the surface has a GREEN row, symmetric with
// the RED above — the RBAC leg below adds its own row when policy is on.
functional_status(
"holger-grpc/authorize_write",
"credentials_valid",
true,
"write credentials accepted",
);
// Authorization (RBAC). Inactive unless role policy is configured, in
// which case the writer's resolved role must permit writes; `None`
// identity here means auth is open (no methods) — leave writes open to
// preserve the authN-only behaviour, EXCEPT when role policy is set
// (fail closed: policy configured but nobody authenticated).
if self.auth_config.rbac_enabled() {
match &identity {
Some(id) => {
let role = self.auth_config.role_for_identity_in_repo(repo, id);
// The owner of an owner-only repository, and the people it granted read and
// write, write it whatever their role elsewhere (lane P1).
if !role.can_write() && !self.read_gate.owner_writes(repo, id) {
functional_status(
"holger-grpc/authorize_write",
"rbac_write_permitted",
false,
"identity lacks writer/admin role",
);
return Err((coded(
&holger_errcode::AUTH_PERMISSION_DENIED,
tonic::Code::PermissionDenied,
format!(
"identity '{}' has role {:?}; writer or admin required for write access",
id.subject, role
),
), Some(id.subject.clone())));
}
// The repository's own role floor (`ron_require_role`): a
// writer below it is refused here exactly as a reader is.
if !self.read_gate.clears_floor(repo, role) {
functional_status(
"holger-grpc/authorize_write",
"rbac_write_permitted",
false,
"below the repository's role floor",
);
return Err((coded(
&holger_errcode::AUTH_PERMISSION_DENIED,
tonic::Code::PermissionDenied,
format!(
"identity '{}' has role {:?} on repository '{repo}', which only the role {:?} may write",
id.subject,
role,
self.read_gate.floor(repo).unwrap_or(role)
),
), Some(id.subject.clone())));
}
// A repository a paid order made is its owner's (lane P1).
if !self.read_gate.write_admits(repo, id, role) {
functional_status(
"holger-grpc/authorize_write",
"rbac_write_permitted",
false,
"a private repository of another owner",
);
return Err((coded(
&holger_errcode::AUTH_PERMISSION_DENIED,
tonic::Code::PermissionDenied,
format!("repository '{repo}' is private: only its owner, the people its owner granted read and write, and the estate's admins write it, and '{}' is none of them", id.subject),
), Some(id.subject.clone())));
}
functional_status(
"holger-grpc/authorize_write",
"rbac_write_permitted",
true,
"writer/admin role granted",
);
}
None => {
functional_status(
"holger-grpc/authorize_write",
"rbac_write_permitted",
false,
"rbac configured but no authenticated identity",
);
return Err((
coded(
&holger_errcode::AUTH_UNAUTHENTICATED,
tonic::Code::Unauthenticated,
"authenticated identity required for write access",
),
None,
));
}
}
}
Ok(identity)
}
/// Authorize a READ (download / list / archive / search).
///
/// Reads are **open by default** — holger's airgap-mirror posture, where the
/// network boundary is the perimeter — and this is then a zero-cost `Ok(None)`
/// with no credential parsing at all. When the operator opts into the read gate
/// (`AuthConfig::require_auth_for_reads`) it mirrors
/// [`authorize_write`](HolgerGrpc::authorize_write) exactly, but against the
/// `Reader` floor (`Role::can_read`) and scoped to `repo`, so a per-repo
/// `Denied` carves a private repo out of an otherwise readable server.
///
/// Returns the resolved identity so the caller can audit the REAL subject
/// instead of `anonymous` — with the gate on, knowing *who* downloaded is the
/// point.
pub(crate) async fn authorize_read<T>(
&self,
request: &Request<T>,
repo: &str,
) -> Result<Option<auth::AuthIdentity>, Status> {
// The gate applies when the SERVER said so for everything, or when THIS
// repository said so for itself (`ron_require_auth`). Neither ⇒ the
// …except on a box awaiting its recharge, which serves NOTHING: half a
// box read as whole is the fault a recharge must never cause.
if let Some(refusal) = self.twin.serve_refusal() {
return Err(coded(
&holger_errcode::TWIN_AWAITING_RECHARGE,
tonic::Code::Unavailable,
refusal,
));
}
// open-read posture, and not one byte of credential parsing.
if !self.read_gate.applies(&self.auth_config, repo) {
return Ok(None);
}
let bearer = request
.metadata()
.get("authorization")
.and_then(|v| v.to_str().ok())
.and_then(|v| v.strip_prefix("Bearer "));
let client_cn = request.peer_certs().and_then(|certs| {
let ders: Vec<rustls::pki_types::CertificateDer<'_>> = certs
.iter()
.map(|c| rustls::pki_types::CertificateDer::from(c.as_ref().to_vec()))
.collect();
crate::exposed::tls::leaf_common_name(&ders)
});
let identity =
match auth::validate_request(&self.auth_config, bearer, client_cn.as_deref()).await {
Ok(id) => id,
Err(_) => {
functional_status(
"holger-grpc/authorize_read",
"rbac_read_permitted",
false,
"invalid or missing read credentials",
);
return Err(coded(
&holger_errcode::AUTH_UNAUTHENTICATED,
tonic::Code::Unauthenticated,
"Valid credentials required for read access",
));
}
};
match crate::auth::authorize_read_identity_scoped(
&self.auth_config,
&self.read_gate,
identity.as_ref(),
repo,
) {
Ok(()) => {
functional_status(
"holger-grpc/authorize_read",
"rbac_read_permitted",
true,
identity
.as_ref()
.map(|i| i.subject.as_str())
.unwrap_or("anonymous"),
);
Ok(identity)
}
Err((401, detail)) => {
functional_status(
"holger-grpc/authorize_read",
"rbac_read_permitted",
false,
detail,
);
Err(coded(
&holger_errcode::AUTH_UNAUTHENTICATED,
tonic::Code::Unauthenticated,
detail,
))
}
Err((_, detail)) => {
functional_status(
"holger-grpc/authorize_read",
"rbac_read_permitted",
false,
detail,
);
Err(coded(
&holger_errcode::AUTH_PERMISSION_DENIED,
tonic::Code::PermissionDenied,
detail,
))
}
}
}
/// **Who is asking for the CATALOG** — the cross-repository doors
/// (`ListRepositories`, `Search`), which then list each repository to the
/// callers that pass it ([`readable_repos`](Self::readable_repos)).
///
/// With the global gate on, or no per-repository gate at all, this is
/// [`authorize_read`](Self::authorize_read) against `""` exactly as before.
/// With per-repository gates only (`ron_require_auth` / `ron_require_role`
/// and no `require_auth_for_reads`), `""` is covered by no gate, so
/// `authorize_read` never resolved the caller — and every gated repository
/// was filtered out for EVERYBODY, an admin included. Here the caller is
/// resolved without being required: no credential is the anonymous caller
/// (who is listed the open repositories), a credential that does not
/// validate is refused like a gated read's.
pub(crate) async fn authorize_catalog<T>(
&self,
request: &Request<T>,
) -> Result<Option<auth::AuthIdentity>, Status> {
// HCON-5: public repositories on a gated box: the catalog resolves its caller too.
if (self.auth_config.require_auth_for_reads || self.read_gate.is_global_only())
&& !self.read_gate.has_public()
{
return self.authorize_read(request, "").await;
}
if let Some(refusal) = self.twin.serve_refusal() {
return Err(coded(
&holger_errcode::TWIN_AWAITING_RECHARGE,
tonic::Code::Unavailable,
refusal,
));
}
let bearer = request
.metadata()
.get("authorization")
.and_then(|v| v.to_str().ok())
.and_then(|v| v.strip_prefix("Bearer "));
let client_cn = request.peer_certs().and_then(|certs| {
let ders: Vec<rustls::pki_types::CertificateDer<'_>> = certs
.iter()
.map(|c| rustls::pki_types::CertificateDer::from(c.as_ref().to_vec()))
.collect();
crate::exposed::tls::leaf_common_name(&ders)
});
if bearer.is_none() && client_cn.is_none() {
return Ok(None);
}
auth::validate_request(&self.auth_config, bearer, client_cn.as_deref())
.await
.map_err(|_| {
coded(
&holger_errcode::AUTH_UNAUTHENTICATED,
tonic::Code::Unauthenticated,
"Valid credentials required for read access",
)
})
}
/// **May this ALREADY-RESOLVED identity read `repo`?**
///
/// The same predicate [`authorize_read`](Self::authorize_read) applies, minus
/// the credential parsing — for the doors that authorize ONCE and then have to
/// answer the question repeatedly, per repository. Gate off ⇒ always `true`,
/// so an open-read deployment is untouched.
pub(crate) fn may_read_repo(&self, identity: Option<&auth::AuthIdentity>, repo: &str) -> bool {
crate::auth::authorize_read_identity_scoped(
&self.auth_config,
&self.read_gate,
identity,
repo,
)
.is_ok()
}
/// **The repositories `identity` may read**, in route order.
///
/// The cross-repo doors (`Search`, `ListRepositories`) authorize once to
/// establish *who is asking* and then walk every configured backend. Authorizing
/// once is not enough: the gate they pass is the caller's GLOBAL role, and a
/// per-repo `Denied` — the whole point of `repo_roles`, "this identity may not
/// read this repository" — says nothing about the global role. So a caller who
/// is correctly refused a direct `FetchArtifact` on `classified` used to get its
/// coordinates, versions and archive paths back out of `Search`. Enumeration IS
/// disclosure; the ACL has to bite on the repo SET, not just on the doorway.
///
/// Filtering silently (rather than refusing the whole RPC) is deliberate: a 403
/// naming the repository would itself confirm that a repository the caller may
/// not read exists. With the read gate off — every deployment that has not opted
/// in — `authorize_read` is unconditionally `Ok` and this is `all_repos()`
/// verbatim.
pub(crate) fn readable_repos(
&self,
identity: Option<&auth::AuthIdentity>,
) -> Vec<(String, Arc<dyn traits::RepositoryBackendTrait>)> {
self.routes
.all_repos()
.into_iter()
.filter(|(name, _)| self.may_read_repo(identity, name))
.collect()
}
/// The read floor applied to a repository that is NOT the one the RPC was gated
/// on — the SOURCE leg of a two-ended admin RPC (`Replicate`, `PromoteArtifact`).
///
/// Those RPCs authorize `Admin` on the DESTINATION and then read the source with
/// a bare `get_repo`, which makes them a read primitive: an admin on one shelf
/// names any other repository as `from_repo` and the server copies its bytes
/// somewhere that admin can read — in one RPC, and for `Replicate` the WHOLE
/// repository at once.
///
/// Two legs, because the open read door cannot enforce everything this one can:
///
/// 1. **The read gate's own floor** ([`AuthConfig::authorize_read`]) — a no-op
/// when the gate is off, exactly as it is on every read door.
/// 2. **An explicit per-repo `Denied`, gate or no gate.** The read door is open
/// by default because it does not know *who* is asking — an anonymous pull is
/// the airgap-mirror posture. Here we DO know: the admin gate just resolved
/// an identity. `Denied` is documented as "no access at all — not even reads",
/// it is purely additive (nothing produces it implicitly), and an operator who
/// writes it has said the one thing this check needs to hear. Honouring it
/// only when the opt-in read gate happens to be on would leave the hole open
/// in the far commoner posture: locked-down writes, open reads.
///
/// Leg 2 is scoped to `rbac_enabled()` and to roles that fail `can_read` — i.e.
/// `Denied` and nothing else — so a deployment that never wrote `Denied` behaves
/// exactly as it did.
#[cfg_attr(not(feature = "intake"), allow(dead_code))]
pub(crate) fn authorize_read_repo(
&self,
identity: Option<&auth::AuthIdentity>,
repo: &str,
) -> Result<(), Status> {
match crate::auth::authorize_read_identity_scoped(
&self.auth_config,
&self.read_gate,
identity,
repo,
) {
Ok(()) => {}
Err((401, detail)) => {
return Err(coded(
&holger_errcode::AUTH_UNAUTHENTICATED,
tonic::Code::Unauthenticated,
detail,
))
}
Err((_, detail)) => {
return Err(coded(
&holger_errcode::AUTH_PERMISSION_DENIED,
tonic::Code::PermissionDenied,
detail,
))
}
}
if self.auth_config.rbac_enabled() {
if let Some(id) = identity {
let role = self.auth_config.role_for_identity_in_repo(repo, id);
if !role.can_read() {
return Err(coded(
&holger_errcode::AUTH_PERMISSION_DENIED,
tonic::Code::PermissionDenied,
format!(
"identity '{}' has role {role:?} on repository '{repo}' and may not \
read it as the source of this operation",
id.subject
),
));
}
}
}
Ok(())
}
/// **Is `key` a property that decides POLICY rather than describing an artifact?**
///
/// Two keys are: the promotion engine's `required_property` key
/// (`promotion.require_property` — setting it forges the `signed=true` the
/// promote gate trusts), and the retention pin
/// ([`reserved_properties`](Self::reserved_properties), `keep` by default —
/// setting it pins a blob past the GC that was meant to reclaim it).
///
/// The first is derived from live state, so it tracks whatever the operator
/// actually configured and cannot drift out of sync with the gate it protects.
/// A server with no promotion engine reserves only the configured keys.
/// (The serve-time quarantine property is retired, W6: a hold is the intake
/// chain's QUARANTINED state, which no property write can clear.)
#[cfg(feature = "intake")]
pub(crate) fn is_reserved_property(&self, key: &str) -> bool {
// ONE rule, shared with the HTTP `/-/properties` door — see
// [`crate::properties::is_reserved_property`] for why that matters.
crate::properties::is_reserved_property(
key,
self.promotion.as_ref().and_then(|e| e.required_property()),
&self.reserved_properties,
)
}
/// The ADMIN floor over a reserved property key, applied ON TOP of the write
/// gate the caller has already passed.
///
/// `SetProperty` was plain writer-gated, which handed every Writer the two
/// levers the server's own policy reads: clear the quarantine flag on a blob an
/// operator quarantined, or set the property the promotion gate demands and walk
/// an unsigned artifact into PROD. Writing a *label* is a write; writing the
/// server's *policy input* is an admin action.
///
/// Deliberately NOT [`authorize_admin`](Self::authorize_admin): the caller is
/// already authenticated (the write gate ran) and an admin-gate refusal here
/// would answer "admin required to promote", which is the wrong sentence. This
/// is the role leg only, with the true reason on the wire. With RBAC off the
/// server has no notion of admin at all and this is open — exactly as every
/// other admin action on an RBAC-less server is.
#[cfg_attr(not(feature = "intake"), allow(dead_code))]
pub(crate) fn authorize_reserved_property(
&self,
identity: Option<&auth::AuthIdentity>,
repo: &str,
key: &str,
) -> Result<(), Status> {
if !self.auth_config.rbac_enabled() {
return Ok(());
}
let Some(id) = identity else {
functional_status(
"holger-grpc/set_property",
"reserved_property_permitted",
false,
"rbac configured but no authenticated identity",
);
return Err(coded(
&holger_errcode::AUTH_UNAUTHENTICATED,
tonic::Code::Unauthenticated,
format!(
"property '{key}' is reserved; an authenticated admin is required to set it"
),
));
};
let role = self.auth_config.role_for_identity_in_repo(repo, id);
if !role.can_admin() {
functional_status(
"holger-grpc/set_property",
"reserved_property_permitted",
false,
"identity lacks admin role",
);
return Err(coded(
&holger_errcode::AUTH_PERMISSION_DENIED,
tonic::Code::PermissionDenied,
format!(
"property '{key}' is reserved by server policy (promotion \
requirement / retention pin); identity '{}' has role {role:?}, admin \
required to set it",
id.subject
),
));
}
functional_status(
"holger-grpc/set_property",
"reserved_property_permitted",
true,
key,
);
Ok(())
}
/// The principal to record for a read: the gated identity's subject, or
/// `anonymous` when reads are open (the default). One helper so every read
/// door audits the same way.
fn read_principal(identity: &Option<auth::AuthIdentity>) -> String {
identity
.as_ref()
.map(|i| i.subject.clone())
.unwrap_or_else(|| "anonymous".to_string())
}
/// Authorize a PRIVILEGED (admin) request — promotion/delete/manage. Mirrors
/// [`HolgerGrpc::authorize_write`] but requires the `Admin` role, and is
/// **fail-closed**: with RBAC on, the identity must resolve to `Admin`
/// (`can_admin`) or the request is denied; a configured-but-unauthenticated
/// request is denied. With no auth methods at all (fully open) it preserves
/// the dev/test open behaviour, exactly as writes do.
pub(crate) async fn authorize_admin<T>(
&self,
request: &Request<T>,
repo: &str,
) -> Result<Option<auth::AuthIdentity>, Status> {
self.mutation_hold()?;
self.authorize_admin_identity(request, repo, "promote")
.await
}
/// The holds every admin MUTATION answers before its credential is looked
/// at: a twin, a recharge, a growth, a read-only server.
pub(crate) fn mutation_hold(&self) -> Result<(), Status> {
// A TWIN refuses every admin MUTATION for the same reason writes are
// refused, and by the same name. `authorize_admin_read` — the audit
// query plane — is deliberately NOT gated: reading a follower's log
// changes nothing, and a twin whose history could not be asked about is
if let Some(refusal) = self.twin.serve_refusal() {
return Err(coded(
&holger_errcode::TWIN_AWAITING_RECHARGE,
tonic::Code::Unavailable,
refusal,
));
}
// exactly the un-provable box W10 exists to prevent.
if let Some(refusal) = self.twin.write_refusal() {
return Err(coded(
&holger_errcode::TWIN_REFUSES_WRITES,
tonic::Code::PermissionDenied,
refusal,
));
}
// A GROWTH holds every write while it drains and while the set is off
// (`crate::grow`, `holger-grow-1`): UNAVAILABLE, so a client retries after
// the growth instead of giving up — refused, never queued.
if let Some(held) = self.grow_hold.write_refusal() {
return Err(coded(
&holger_errcode::GROW_WRITES_HELD,
tonic::Code::Unavailable,
held,
));
}
// ReadOnly mode refuses every admin mutation BEFORE the body is collected.
if self.operation_mode == crate::OperationMode::ReadOnly {
return Err(coded(
&holger_errcode::SERVER_READ_ONLY,
tonic::Code::PermissionDenied,
"server is in read-only mode",
));
}
Ok(())
}
/// **The intake verbs' gate** (`PromotionService.AdvanceArtifact` /
/// `GetArtifactState`): the credential, and with RBAC on an authenticated
/// identity. What that identity may DO is its duties'
/// ([`crate::promote::duties`]), decided by the verb.
#[cfg(feature = "intake")]
pub(crate) async fn authorize_intake<T>(
&self,
request: &Request<T>,
mutating: bool,
) -> Result<Option<auth::AuthIdentity>, Status> {
if mutating {
self.mutation_hold()?;
}
let identity = self.authenticate(request, "use the intake chain").await?;
if self.auth_config.rbac_enabled() && identity.is_none() {
return Err(coded(
&holger_errcode::AUTH_UNAUTHENTICATED,
tonic::Code::Unauthenticated,
"an authenticated identity is required to use the intake chain",
));
}
Ok(identity)
}
/// Admin authorization for a privileged READ — the audit query plane.
///
/// Identical to [`authorize_admin`](Self::authorize_admin) except that it does
/// NOT consult `OperationMode::ReadOnly`. That gate exists to refuse admin
/// MUTATIONS on a sealed server; refusing an audit read there would make the
/// one surface that explains what a sealed appliance served unreachable on
/// exactly the appliance that is sealed. Reading the log changes nothing, and
/// `AuditService` has no RPC that writes.
/// Admin authorization for the GROWTH door (`crate::grow`): the admin role,
/// and NOTHING about the box's ROLE. A TWIN grows exactly like main
/// (gunnar's T21 ruling: never restarted), so the twin refusal the write
/// gates ask first would make the twin lane's `Start` impossible. A
/// READ-ONLY door refuses it like every admin mutation — which is also what
/// keeps the replication door (read-only by construction, where a twin
/// reports its position) from being a door that grows the primary.
pub(crate) async fn authorize_grow<T>(
&self,
request: &Request<T>,
purpose: &str,
) -> Result<Option<auth::AuthIdentity>, Status> {
if self.operation_mode == crate::OperationMode::ReadOnly {
return Err(coded(
&holger_errcode::SERVER_READ_ONLY,
tonic::Code::PermissionDenied,
format!("this door is read-only; it cannot {purpose}"),
));
}
self.authorize_admin_identity(request, "", purpose).await
}
/// The estate board's evidence on this service (lane E3).
pub fn topology_sources(&self) -> crate::topology::Sources {
crate::topology::Sources {
twin: Arc::clone(&self.twin),
grow: self.grow.clone(),
follower: self.follower.clone(),
peers: Arc::clone(&self.peers),
hold: Arc::clone(&self.grow_hold),
data_root: self.data_root.clone(),
declared: Arc::clone(&self.board_declared),
#[cfg(feature = "sync")]
sync: crate::topology::sync_rows(self.intake.as_ref()),
#[cfg(not(feature = "sync"))]
sync: None,
}
}
/// The artifact flow's evidence on this service (lane HOL-SVG-RON): the SAME inputs
/// the HTTP door's `GET /-/flow.svg` reads. `None` when no shape is installed.
pub fn flow_sources(&self) -> Option<crate::flow::Sources> {
Some(crate::flow::Sources {
shape: Arc::clone(self.flow_shape.as_ref()?),
repos: self.routes.all_repos(),
audit: Arc::clone(&self.audit),
visitors: Arc::clone(&self.visitors),
console: self.console.clone(),
board: Some(self.topology_sources()),
#[cfg(feature = "intake")]
intake: self.intake.clone(),
quota: self.flow_quota.clone(),
})
}
/// The Security page's evidence on this service (lane HOL-SEC).
pub fn security_sources(&self) -> crate::security_view::Sources {
crate::security_view::Sources {
#[cfg(feature = "intake")]
intake: self.intake.clone(),
declared: Arc::clone(&self.board_declared),
read_gate: Arc::clone(&self.read_gate),
gated: self.auth_config.require_auth_for_reads,
repos: self
.routes
.all_repos()
.into_iter()
.map(|(name, _)| name)
.collect(),
audit: Arc::clone(&self.audit),
}
}
pub(crate) async fn authorize_admin_read<T>(
&self,
request: &Request<T>,
repo: &str,
) -> Result<Option<auth::AuthIdentity>, Status> {
// An Auditor reads the audit and changes nothing (separation of duties).
self.authorize_admin_identity_for(request, repo, "read the audit log", true)
.await
}
/// The shared body of the two admin gates: validate the credential, then —
/// with RBAC on — require the `Admin` role for `repo` (pass `""` for a
/// server-wide surface, which falls through to the identity's global role).
///
/// Fail-closed: a configured-but-unauthenticated request is denied, and with
/// no auth methods at all it preserves the dev/test open behaviour, exactly as
/// writes do. `purpose` is the verb the refusal names, so an operator is told
/// which door refused him rather than a generic sentence.
pub(crate) async fn authorize_admin_identity<T>(
&self,
request: &Request<T>,
repo: &str,
purpose: &str,
) -> Result<Option<auth::AuthIdentity>, Status> {
self.authorize_admin_identity_for(request, repo, purpose, false)
.await
}
/// [`authorize_admin_identity`](Self::authorize_admin_identity), and with
/// `auditor_reads` the Auditor duty passes too (the audit query plane).
async fn authorize_admin_identity_for<T>(
&self,
request: &Request<T>,
repo: &str,
purpose: &str,
auditor_reads: bool,
) -> Result<Option<auth::AuthIdentity>, Status> {
let identity = self.authenticate(request, purpose).await?;
self.require_admin_role(identity, repo, purpose, auditor_reads)
}
/// Validate the request's credential (bearer or mTLS CN). `Ok(None)` when
/// no auth method is configured.
pub(crate) async fn authenticate<T>(
&self,
request: &Request<T>,
purpose: &str,
) -> Result<Option<auth::AuthIdentity>, Status> {
let bearer = request
.metadata()
.get("authorization")
.and_then(|v| v.to_str().ok())
.and_then(|v| v.strip_prefix("Bearer "));
let client_cn = request.peer_certs().and_then(|certs| {
let ders: Vec<rustls::pki_types::CertificateDer<'_>> = certs
.iter()
.map(|c| rustls::pki_types::CertificateDer::from(c.as_ref().to_vec()))
.collect();
crate::exposed::tls::leaf_common_name(&ders)
});
let identity =
match auth::validate_request(&self.auth_config, bearer, client_cn.as_deref()).await {
Ok(id) => id,
Err(_) => {
functional_status(
"holger-grpc/authorize_admin",
"credentials_valid",
false,
"invalid or missing admin credentials",
);
return Err(coded(
&holger_errcode::AUTH_UNAUTHENTICATED,
tonic::Code::Unauthenticated,
format!("valid credentials required to {purpose}"),
));
}
};
// Success path: admin credentials accepted (or auth fully open) — a GREEN
// row symmetric with the RED above; the RBAC leg below adds its own.
functional_status(
"holger-grpc/authorize_admin",
"credentials_valid",
true,
"admin credentials accepted",
);
Ok(identity)
}
fn require_admin_role(
&self,
identity: Option<auth::AuthIdentity>,
repo: &str,
purpose: &str,
auditor_reads: bool,
) -> Result<Option<auth::AuthIdentity>, Status> {
if self.auth_config.rbac_enabled() {
match &identity {
Some(id) => {
let role = self.auth_config.role_for_identity_in_repo(repo, id);
let auditor = auditor_reads
&& self
.auth_config
.duties_for_identity(id)
.contains(&auth::Duty::Auditor);
if !role.can_admin() && !auditor {
functional_status(
"holger-grpc/authorize_admin",
"rbac_admin_permitted",
false,
"identity lacks admin role",
);
return Err(coded(
&holger_errcode::AUTH_PERMISSION_DENIED,
tonic::Code::PermissionDenied,
format!(
"identity '{}' has role {:?}; admin required to {purpose}",
id.subject, role
),
));
}
functional_status(
"holger-grpc/authorize_admin",
"rbac_admin_permitted",
true,
"admin role granted",
);
}
None => {
functional_status(
"holger-grpc/authorize_admin",
"rbac_admin_permitted",
false,
"rbac configured but no authenticated identity",
);
return Err(coded(
&holger_errcode::AUTH_UNAUTHENTICATED,
tonic::Code::Unauthenticated,
format!("authenticated admin identity required to {purpose}"),
));
}
}
}
Ok(identity)
}
}
/// Client source address (`ip:port`) as the server sees it, or empty when the
/// transport exposes none.
/// An import refusal on the gRPC wire: its own code, a status by kind.
#[cfg(feature = "intake")]
pub(crate) fn import_status(e: &crate::import_door::ImportError) -> Status {
let code = match e.status {
401 => tonic::Code::Unauthenticated,
403 => tonic::Code::PermissionDenied,
409 | 410 | 412 => tonic::Code::FailedPrecondition,
400 | 422 => tonic::Code::InvalidArgument,
_ => tonic::Code::Internal,
};
coded(e.code, code, &e.message)
}
impl HolgerGrpc {
/// **The source a gRPC audit row records** (SEC-REVIEW 2026-11-01 F1, lane HOL-312-HARDEN):
/// from a peer in the endpoint's `ron_trusted_proxies` with a forwarded client, the right-most
/// `X-Forwarded-For` entry and the front it came through (`<client> via <front>`, the HTTP
/// door's form); else the TCP peer ([`peer_addr_string`]).
pub(crate) fn audit_peer<T>(&self, request: &Request<T>) -> String {
let peer = peer_addr_string(request);
let trusted = request
.remote_addr()
.zip(self.trusted_proxies.as_deref())
.is_some_and(|(a, t)| t.allows(a.ip()));
if !trusted {
return peer;
}
let client = request
.metadata()
.get_all("x-forwarded-for")
.iter()
.filter_map(|v| v.to_str().ok())
.rfind(|v| !v.trim().is_empty())
.and_then(|l| l.rsplit(',').next())
.map(str::trim)
.filter(|c| c.parse::<std::net::IpAddr>().is_ok());
match client {
Some(c) => format!("{c} via {peer}"),
None => peer,
}
}
}
pub(crate) fn peer_addr_string<T>(request: &Request<T>) -> String {
request
.remote_addr()
.map(|a| a.to_string())
.unwrap_or_default()
}
/// Human-readable artifact label for the audit log: `namespace/name@version`
/// (the namespace segment dropped when empty, the Rust-style no-namespace case).
fn artifact_label(id: &ArtifactId) -> String {
if id.namespace.is_empty() {
format!("{}@{}", id.name, id.version)
} else {
format!("{}/{}@{}", id.namespace, id.name, id.version)
}
}
// ─── RepositoryService ───────────────────────────────────────────────
#[cfg(test)]
mod tests {
use super::*;
use crate::audit::MemoryAuditLog;
use std::sync::Arc;
use traits::{ArtifactFormat, RepositoryBackendTrait};
// The service traits + AuditAction used to reach the tests via super's glob;
// the split moved the service impls out, so import them here directly.
use super::holger_proto::admin_service_server::AdminService;
use super::holger_proto::archive_service_server::ArchiveService;
use super::holger_proto::repository_service_server::RepositoryService;
use crate::audit::AuditAction;
// ── the read SLOT cap, distinct from the read BYTE budget ────────────────
fn bare_state() -> HolgerGrpc {
HolgerGrpc::new(FastRoutes::new(vec![]))
}
/// The gap this closes: a flood of TINY reads fits the byte budget completely
/// (each charges the 1 MiB floor, so 4 GiB admits 4096 of them) while being far
/// more simultaneous requests — tasks, streams, descriptors, worker threads —
/// than the server should hold. The slot cap refuses on count, before the bytes
/// are ever charged.
#[test]
fn concurrent_reads_are_capped_on_count_even_when_every_read_is_tiny() {
let state = bare_state().with_max_concurrent_reads(3);
// The byte budget is left at its 4 GiB default and is nowhere near spent:
// three reads at the 1 MiB floor is 3 MiB.
let held: Vec<ReadBudget> = (0..3)
.map(|_| match state.reserve_read_budget("test") {
Ok(b) => b,
Err(e) => panic!("within the cap, but refused: {e}"),
})
.collect();
let refused = match state.reserve_read_budget("test") {
Err(e) => e,
Ok(_) => panic!("the fourth read must be refused on count"),
};
assert_eq!(refused.code(), tonic::Code::ResourceExhausted);
assert!(
refused.message().contains("concurrent reads"),
"the refusal must say it was the COUNT, not the bytes: {}",
refused.message()
);
assert_eq!(
refused
.metadata()
.get(holger_errcode::WIRE_CODE_KEY)
.map(|v| v.to_str().unwrap()),
Some(holger_errcode::SERVER_TOO_MANY_CONCURRENT_READS.code),
"and must carry its own frozen code, not the byte budget's"
);
// A slot is returned when its guard drops, so the cap is a ceiling on
// SIMULTANEOUS reads, not a quota that runs out.
drop(held);
assert!(
state.reserve_read_budget("test").is_ok(),
"a finished read must return its slot"
);
}
/// `0` disables the cap, matching how the byte budget's `0` behaves — an
/// operator who wants the old unbounded behaviour can have it explicitly.
/// The BYTE budget is disabled here too, so the only thing under test is
/// whether the slot accounting still refuses.
#[test]
fn a_zero_cap_disables_the_slot_accounting() {
let state = bare_state()
.with_max_concurrent_reads(0)
.with_max_in_flight_read_bytes(0);
let held: Vec<ReadBudget> = (0..5000)
.filter_map(|_| state.reserve_read_budget("test").ok())
.collect();
assert_eq!(held.len(), 5000, "a zero cap must refuse nothing on count");
}
/// The two budgets are independent, and the SLOT is what a flood of tiny reads
/// hits first. With the byte budget at its 4 GiB default, the old implicit
/// ceiling was 4096 tiny reads; the slot cap now stops them at its own number,
/// well before that.
#[test]
fn the_slot_cap_binds_before_the_byte_budget_for_tiny_reads() {
let state = bare_state(); // both at their shipped defaults
let held: Vec<ReadBudget> = std::iter::repeat_with(|| state.reserve_read_budget("test"))
.take_while(|r| r.is_ok())
.filter_map(|r| r.ok())
.collect();
assert_eq!(
held.len(),
DEFAULT_MAX_CONCURRENT_READS,
"the slot cap, not the byte budget's implicit {} , is what stopped them",
DEFAULT_MAX_IN_FLIGHT_READ_BYTES / READ_BUDGET_UNIT_BYTES
);
}
/// The shipped default is a real cap, not `0`. RED-when-broken if anyone
/// "temporarily" disables it.
#[test]
#[allow(clippy::assertions_on_constants)] // the test IS about the shipped constants
fn the_shipped_default_caps_concurrent_reads() {
assert_eq!(
bare_state().max_concurrent_reads,
DEFAULT_MAX_CONCURRENT_READS
);
assert!(
DEFAULT_MAX_CONCURRENT_READS > 0,
"the default must actually cap"
);
assert!(
DEFAULT_MAX_CONCURRENT_READS
< DEFAULT_MAX_IN_FLIGHT_READ_BYTES / READ_BUDGET_UNIT_BYTES,
"a slot cap at or above the byte budget's implicit ceiling bounds nothing new"
);
}
/// Minimal in-memory backend: holds one named blob, reports a fixed archive
/// stat block. Read-only by default; `writable` flips `is_writable` so the
/// `ServerProfile` derivation (read-only iff no writable repo) is testable.
struct MockRepo {
name: String,
blob: Option<Vec<u8>>,
writable: bool,
}
impl RepositoryBackendTrait for MockRepo {
fn caps(&self) -> traits::Caps {
traits::Caps {
archive: true,
..traits::Caps::NONE
}
}
fn name(&self) -> &str {
&self.name
}
fn format(&self) -> ArtifactFormat {
ArtifactFormat::Rust
}
fn is_writable(&self) -> bool {
self.writable
}
fn fetch(&self, _id: &traits::ArtifactId) -> anyhow::Result<Option<traits::Bytes>> {
Ok(self.blob.clone().map(traits::Bytes::from))
}
fn put(&self, _id: &traits::ArtifactId, _data: &[u8]) -> anyhow::Result<()> {
anyhow::bail!("read-only")
}
fn archive(&self) -> anyhow::Result<traits::Archive> {
Ok(traits::Archive {
name: "mock".into(),
files: vec![("a.rs".into(), 49), ("b.rs".into(), 50)],
})
}
fn handle_http2_request(
&self,
_method: &str,
_suburl: &str,
_body: &[u8],
) -> anyhow::Result<(u16, Vec<(String, String)>, Vec<u8>)> {
Ok((200, vec![], Vec::new()))
}
}
fn grpc_with(blob: Option<Vec<u8>>) -> (Arc<HolgerGrpc>, Arc<MemoryAuditLog>) {
let backend: Arc<dyn RepositoryBackendTrait> = Arc::new(MockRepo {
name: "crates".into(),
blob,
writable: false,
});
let routes = FastRoutes::new(vec![("crates".to_string(), backend)]);
let audit = Arc::new(MemoryAuditLog::new());
let grpc = Arc::new(HolgerGrpc::new(routes).with_audit(audit.clone()));
(grpc, audit)
}
/// Build a grpc state over one repo whose writability is `writable`.
fn grpc_with_writable(writable: bool) -> Arc<HolgerGrpc> {
let backend: Arc<dyn RepositoryBackendTrait> = Arc::new(MockRepo {
name: "crates".into(),
blob: None,
writable,
});
let routes = FastRoutes::new(vec![("crates".to_string(), backend)]);
Arc::new(HolgerGrpc::new(routes))
}
fn artifact_id() -> ArtifactId {
ArtifactId {
namespace: String::new(),
name: "serde".into(),
version: "1.0".into(),
}
}
// ── R3: the opt-in READ gate, over the REAL gRPC read doors. ──────────────
//
// Reads are open by default (the airgap-mirror posture holger has always had).
// `AuthConfig::require_auth_for_reads` closes them behind the `Reader` floor.
// These drive the actual `RepositoryService`/`ArchiveService`/`AdminService`
// impls, not the predicate in isolation, so a door that forgot to call the gate
// shows up here.
fn read_token_cfg(require_reads: bool) -> Arc<crate::auth::AuthConfig> {
use crate::auth::{AuthMethodConfig, Role};
Arc::new(crate::auth::AuthConfig {
methods: vec![
AuthMethodConfig::StaticToken {
subject: "pull-bot".into(),
token: Some("pull-bot-token-of-sufficient-length".into()),
token_env: None,
token_file: None,
},
AuthMethodConfig::StaticToken {
subject: "banned-bot".into(),
token: Some("banned-bot-token-of-sufficient-len".into()),
token_env: None,
token_file: None,
},
],
roles: std::collections::HashMap::from([
("pull-bot".to_string(), Role::Reader),
("banned-bot".to_string(), Role::Denied),
]),
default_role: Some(Role::Reader),
require_auth_for_reads: require_reads,
..Default::default()
})
}
fn bearer_fetch(token: Option<&str>) -> Request<FetchArtifactRequest> {
let mut req = Request::new(FetchArtifactRequest {
repository: "crates".into(),
id: Some(artifact_id()),
});
if let Some(t) = token {
req.metadata_mut()
.insert("authorization", format!("Bearer {t}").parse().unwrap());
}
req
}
fn served_grpc(
cfg: Arc<crate::auth::AuthConfig>,
audit: Arc<MemoryAuditLog>,
) -> Arc<HolgerGrpc> {
let backend: Arc<dyn RepositoryBackendTrait> = Arc::new(MockRepo {
name: "crates".into(),
blob: Some(b"bytes".to_vec()),
writable: false,
});
let routes = FastRoutes::new(vec![("crates".to_string(), backend)]);
Arc::new(HolgerGrpc::with_auth(routes, cfg).with_audit(audit))
}
/// **The default is untouched.** With the gate off — every deployment that
/// exists — an anonymous `FetchArtifact` still serves bytes, even though auth
/// methods and a role policy ARE configured (i.e. writes are locked down).
/// RED-when-broken: if the gate ever became implicit, this 401s.
#[tokio::test]
async fn read_gate_off_serves_anonymous_reads_even_with_rbac_on() {
let audit = Arc::new(MemoryAuditLog::new());
let grpc = served_grpc(read_token_cfg(false), audit.clone());
let out = RepositoryService::fetch_artifact(&grpc, bearer_fetch(None))
.await
.expect("reads are open by default")
.into_inner();
assert_eq!(&out.data[..], &b"bytes"[..]);
assert_eq!(audit.events().last().unwrap().ident, "anonymous");
}
/// Gate ON: anonymous is refused, a `Reader` is served, a `Denied` identity is
/// permission-denied — and the served read audits the REAL subject, which is
/// the operational point of authenticating a download.
#[tokio::test]
async fn read_gate_on_refuses_anonymous_serves_reader_denies_denied() {
let audit = Arc::new(MemoryAuditLog::new());
let grpc = served_grpc(read_token_cfg(true), audit.clone());
let err = RepositoryService::fetch_artifact(&grpc, bearer_fetch(None))
.await
.expect_err("a gated read with no credentials must be refused");
assert_eq!(err.code(), tonic::Code::Unauthenticated);
let err = RepositoryService::fetch_artifact(
&grpc,
bearer_fetch(Some("not-a-real-token-at-all-x")),
)
.await
.expect_err("a bad bearer must be refused");
assert_eq!(err.code(), tonic::Code::Unauthenticated);
let ok = RepositoryService::fetch_artifact(
&grpc,
bearer_fetch(Some("pull-bot-token-of-sufficient-length")),
)
.await
.expect("a Reader must be served")
.into_inner();
assert_eq!(&ok.data[..], &b"bytes"[..]);
assert_eq!(
audit.events().last().unwrap().ident,
"pull-bot",
"a gated read must audit who took it, not 'anonymous'"
);
let err = RepositoryService::fetch_artifact(
&grpc,
bearer_fetch(Some("banned-bot-token-of-sufficient-len")),
)
.await
.expect_err("a Denied identity must not read");
assert_eq!(err.code(), tonic::Code::PermissionDenied);
}
/// Every read DOOR is gated, not just the unary fetch. `StreamArtifact` once
/// bypassed the quarantine predicate the same way; the gate must not repeat
/// that. Listings, archive stat/files, the catalog and search are all reads.
#[tokio::test]
async fn read_gate_covers_every_read_door_not_just_fetch() {
let grpc = served_grpc(read_token_cfg(true), Arc::new(MemoryAuditLog::new()));
let anon = tonic::Code::Unauthenticated;
let e = RepositoryService::stream_artifact(&grpc, bearer_fetch(None))
.await
.unwrap_err();
assert_eq!(e.code(), anon, "StreamArtifact must be gated");
let e = RepositoryService::list_artifacts(
&grpc,
Request::new(ListArtifactsRequest {
repository: "crates".into(),
name_filter: String::new(),
limit: 0,
page_token: String::new(),
}),
)
.await
.unwrap_err();
assert_eq!(e.code(), anon, "ListArtifacts must be gated");
let e = ArchiveService::list_archive_files(
&grpc,
Request::new(ListArchiveFilesRequest {
repository: "crates".into(),
prefix: String::new(),
}),
)
.await
.unwrap_err();
assert_eq!(e.code(), anon, "ListArchiveFiles must be gated");
let e = ArchiveService::archive_info(
&grpc,
Request::new(ArchiveInfoRequest {
repository: "crates".into(),
}),
)
.await
.unwrap_err();
assert_eq!(e.code(), anon, "ArchiveInfo must be gated");
let e = AdminService::list_repositories(&grpc, Request::new(ListRepositoriesRequest {}))
.await
.unwrap_err();
assert_eq!(
e.code(),
anon,
"ListRepositories leaks the catalog — must be gated"
);
// …and the same doors serve once a Reader authenticates (so the gate is
// the credential check, not a blanket outage).
let tok = "pull-bot-token-of-sufficient-length";
let mut req = Request::new(ListRepositoriesRequest {});
req.metadata_mut()
.insert("authorization", format!("Bearer {tok}").parse().unwrap());
assert!(AdminService::list_repositories(&grpc, req).await.is_ok());
}
/// Health is NOT a read of artifacts and must stay reachable with the gate on —
/// otherwise every load balancer and `readyz` probe breaks the moment an
/// operator opts in.
#[tokio::test]
async fn read_gate_leaves_health_open() {
let grpc = served_grpc(read_token_cfg(true), Arc::new(MemoryAuditLog::new()));
assert!(AdminService::health(&grpc, Request::new(HealthRequest {}))
.await
.is_ok());
}
#[tokio::test]
async fn stream_artifact_audits_download_hit_and_miss() {
// Hit: a present blob records a 200 download with the byte count.
let (grpc, audit) = grpc_with(Some(b"hello".to_vec()));
let req = Request::new(FetchArtifactRequest {
repository: "crates".into(),
id: Some(artifact_id()),
});
grpc.stream_artifact(req).await.expect("stream ok");
let ev = audit.events();
assert_eq!(ev.len(), 1);
assert_eq!(ev[0].action, AuditAction::Download);
assert_eq!(ev[0].ident, "anonymous");
assert_eq!(ev[0].repo, "crates");
assert_eq!(ev[0].artifact, "serde@1.0");
assert_eq!(ev[0].status, 200);
assert_eq!(ev[0].bytes, 5);
// Miss: absent blob records a 404 download.
let (grpc, audit) = grpc_with(None);
let req = Request::new(FetchArtifactRequest {
repository: "crates".into(),
id: Some(artifact_id()),
});
assert!(grpc.stream_artifact(req).await.is_err());
let ev = audit.events();
assert_eq!(ev.len(), 1);
assert_eq!(ev[0].action, AuditAction::Download);
assert_eq!(ev[0].status, 404);
}
/// ★ **A gRPC audit row records the forwarded client from a trusted front** (SEC-REVIEW
/// 2026-11-01 F1's remainder, lane HOL-312-HARDEN): the estate's front dials from its
/// WireGuard address, so every row said `10.44.0.1`. From a peer in `ron_trusted_proxies` the
/// row says `<client> via <front>` (the HTTP door's form, ba766195); from any other peer, or
/// with no forwarded client, the TCP peer as before.
#[tokio::test]
async fn grpc_audit_rows_record_the_forwarded_client_from_a_trusted_front() {
let trusted = crate::listen::AllowList::parse(&["10.44.0.0/24".to_string()]).unwrap();
let req = |peer: &str, xff: Option<&str>| {
let mut r = Request::new(FetchArtifactRequest {
repository: "crates".into(),
id: Some(artifact_id()),
});
r.extensions_mut()
.insert(tonic::transport::server::TcpConnectInfo {
local_addr: None,
remote_addr: Some(peer.parse().unwrap()),
});
if let Some(x) = xff {
r.metadata_mut()
.insert("x-forwarded-for", x.parse().unwrap());
}
r
};
for (peer, xff, want) in [
(
"10.44.0.1:40000",
Some("203.0.113.9, 198.51.100.7"),
"198.51.100.7 via 10.44.0.1:40000",
),
("10.44.0.1:40000", None, "10.44.0.1:40000"),
("192.0.2.5:1234", Some("198.51.100.7"), "192.0.2.5:1234"),
("10.44.0.1:40000", Some("not-an-address"), "10.44.0.1:40000"),
] {
let backend: Arc<dyn RepositoryBackendTrait> = Arc::new(MockRepo {
name: "crates".into(),
blob: Some(b"hello".to_vec()),
writable: false,
});
let routes = FastRoutes::new(vec![("crates".to_string(), backend)]);
let audit = Arc::new(MemoryAuditLog::new());
let grpc = Arc::new(
HolgerGrpc::new(routes)
.with_audit(audit.clone())
.with_trusted_proxies(Some(Arc::new(trusted.clone()))),
);
grpc.stream_artifact(req(peer, xff))
.await
.expect("stream ok");
let ev = audit.events();
assert_eq!(ev.len(), 1);
assert_eq!(ev[0].source_ip, want, "peer {peer}, forwarded {xff:?}");
}
}
#[tokio::test]
async fn list_artifacts_audits_list() {
let (grpc, audit) = grpc_with(None);
let req = Request::new(ListArtifactsRequest {
repository: "crates".into(),
name_filter: "ser".into(),
limit: 10,
page_token: String::new(),
});
RepositoryService::list_artifacts(&grpc, req)
.await
.expect("list ok");
let ev = audit.events();
assert_eq!(ev.len(), 1);
assert_eq!(ev[0].action, AuditAction::List);
assert_eq!(ev[0].repo, "crates");
assert_eq!(ev[0].artifact, "ser"); // filter recorded in the artifact column
assert_eq!(ev[0].status, 200);
}
#[tokio::test]
async fn list_artifacts_unknown_repo_audits_404() {
let (grpc, audit) = grpc_with(None);
let req = Request::new(ListArtifactsRequest {
repository: "ghost".into(),
name_filter: String::new(),
limit: 10,
page_token: String::new(),
});
assert!(RepositoryService::list_artifacts(&grpc, req).await.is_err());
let ev = audit.events();
assert_eq!(ev.len(), 1);
assert_eq!(ev[0].action, AuditAction::List);
assert_eq!(ev[0].repo, "ghost");
assert_eq!(ev[0].status, 404);
}
#[tokio::test]
async fn archive_endpoints_audit_list() {
let (grpc, audit) = grpc_with(None);
let req = Request::new(ListArchiveFilesRequest {
repository: "crates".into(),
prefix: "src/".into(),
});
grpc.list_archive_files(req)
.await
.expect("archive files ok");
let req = Request::new(ArchiveInfoRequest {
repository: "crates".into(),
});
grpc.archive_info(req).await.expect("archive info ok");
let ev = audit.events();
assert_eq!(ev.len(), 2);
assert!(ev
.iter()
.all(|e| e.action == AuditAction::List && e.status == 200));
assert_eq!(ev[0].artifact, "src/"); // prefix recorded
assert_eq!(ev[1].bytes, 2); // archive_info logs the file count
}
#[tokio::test]
async fn list_repositories_audits_list() {
let (grpc, audit) = grpc_with(None);
let req = Request::new(ListRepositoriesRequest {});
AdminService::list_repositories(&grpc, req)
.await
.expect("list repos ok");
let ev = audit.events();
assert_eq!(ev.len(), 1);
assert_eq!(ev[0].action, AuditAction::List);
assert_eq!(ev[0].bytes, 1); // one configured repo
}
#[tokio::test]
async fn server_profile_derived_static_when_all_repos_read_only() {
// No writable repo ⇒ the server reports the static "silent running"
// profile as read-only truth (derived from the route table, no override).
let grpc = grpc_with_writable(false);
let resp = AdminService::server_profile(&grpc, Request::new(ServerProfileRequest {}))
.await
.expect("profile ok")
.into_inner();
assert_eq!(resp.profile, "static");
assert!(resp.read_only);
assert_eq!(resp.writable_repo_count, 0);
assert!(
resp.label.contains("SILENT RUNNING"),
"label = {}",
resp.label
);
}
#[tokio::test]
async fn server_profile_derived_dynamic_when_a_repo_is_writable() {
// A writable repo ⇒ derived dynamic profile (writable count reflected).
let grpc = grpc_with_writable(true);
let resp = AdminService::server_profile(&grpc, Request::new(ServerProfileRequest {}))
.await
.expect("profile ok")
.into_inner();
assert_eq!(resp.profile, "dynamic");
assert!(!resp.read_only);
assert_eq!(resp.writable_repo_count, 1);
}
#[tokio::test]
async fn server_profile_override_forces_static_over_writable_repos() {
// A sealed static-holger declares itself read-only even with writable
// repos present — the explicit override wins over the derivation.
let backend: Arc<dyn RepositoryBackendTrait> = Arc::new(MockRepo {
name: "crates".into(),
blob: None,
writable: true,
});
let routes = FastRoutes::new(vec![("crates".to_string(), backend)]);
let grpc = Arc::new(HolgerGrpc::new(routes).with_read_only_profile(true));
let resp = AdminService::server_profile(&grpc, Request::new(ServerProfileRequest {}))
.await
.expect("profile ok")
.into_inner();
assert_eq!(resp.profile, "static");
assert!(resp.read_only);
// The override pins read-only, but the honest writable count still shows
// 1 so the UI can warn "sealed despite writable repos".
assert_eq!(resp.writable_repo_count, 1);
}
/// Writable in-memory backend: accepts any put. Lets us exercise the put
/// path (the read-only MockRepo short-circuits before the body cap).
struct WritableMock {
name: String,
}
impl RepositoryBackendTrait for WritableMock {
fn caps(&self) -> traits::Caps {
traits::Caps::NONE
}
fn name(&self) -> &str {
&self.name
}
fn format(&self) -> ArtifactFormat {
ArtifactFormat::Rust
}
fn is_writable(&self) -> bool {
true
}
fn fetch(&self, _id: &traits::ArtifactId) -> anyhow::Result<Option<traits::Bytes>> {
Ok(None)
}
fn put(&self, _id: &traits::ArtifactId, _data: &[u8]) -> anyhow::Result<()> {
Ok(())
}
fn handle_http2_request(
&self,
_method: &str,
_suburl: &str,
_body: &[u8],
) -> anyhow::Result<(u16, Vec<(String, String)>, Vec<u8>)> {
Ok((200, vec![], Vec::new()))
}
}
#[tokio::test]
async fn put_artifact_enforces_body_cap() {
let backend: Arc<dyn RepositoryBackendTrait> = Arc::new(WritableMock {
name: "crates".into(),
});
let routes = FastRoutes::new(vec![("crates".to_string(), backend)]);
let audit = Arc::new(MemoryAuditLog::new());
let grpc = Arc::new(
HolgerGrpc::new(routes)
.with_audit(audit.clone())
.with_max_body_bytes(16),
);
// Over the cap → rejected with ResourceExhausted, audited 413, put skipped.
let req = Request::new(PutArtifactRequest {
repository: "crates".into(),
id: Some(artifact_id()),
data: traits::Bytes::from(vec![0u8; 17]),
async_job: false,
});
let err = RepositoryService::put_artifact(&grpc, req)
.await
.expect_err("over-cap put must be rejected");
assert_eq!(err.code(), tonic::Code::ResourceExhausted);
assert_eq!(audit.events().last().unwrap().status, 413);
// Under the cap → accepted and audited 200.
let req = Request::new(PutArtifactRequest {
repository: "crates".into(),
id: Some(artifact_id()),
data: traits::Bytes::from(vec![0u8; 8]),
async_job: false,
});
RepositoryService::put_artifact(&grpc, req)
.await
.expect("under-cap put must succeed");
assert_eq!(audit.events().last().unwrap().status, 200);
}
// ─── PromotionService (DEV → PROD), admin-gated, offline OIDC auth ────────
//
// These exercise Part B's handler end-to-end AND reuse Part A's offline OIDC:
// the admin/writer identities come from a signed JWT verified offline against
// an inline test JWKS (no network), so the RBAC admin gate is real.
#[cfg(all(feature = "auth-oidc", feature = "intake"))]
use super::holger_proto::promotion_service_server::PromotionService;
#[cfg(feature = "auth-oidc")]
use crate::auth::AuthMethodConfig;
#[cfg(all(feature = "auth-oidc", feature = "intake"))]
use crate::auth::{AuthConfig, Role};
#[cfg(feature = "intake")]
#[cfg(feature = "auth-oidc")]
use crate::promote::PromotionEngine;
// The signed admin/writer JWTs these tests promote with come from the OIDC
// provider's shared test-support (throwaway RSA keypair + inline test JWKS),
// verified fully offline. Gated on `auth-oidc`: without the OIDC provider
// pinned there is no offline-JWT path to authenticate the token.
#[cfg(feature = "auth-oidc")]
use nornir_auth_oidc::test_support;
/// Build an offline-OIDC `AuthMethodConfig` from the shared test JWKS.
#[cfg(feature = "auth-oidc")]
fn oidc_offline_method(issuer: &str) -> AuthMethodConfig {
AuthMethodConfig::Oidc {
issuer_url: issuer.to_string(),
audience: Some("holger".into()),
jwks: Some(test_support::jwks_json()),
jwks_path: None,
jwks_uri: None,
}
}
/// Build a promotion-ready grpc state: a source repo `rust-dev` holding
/// `blob`, OIDC offline auth for `issuer` with `admin@corp`=Admin +
/// `bot@corp`=Writer, a fresh PROD store under `prod_root`, and a memory audit.
#[cfg(feature = "intake")]
#[cfg(feature = "auth-oidc")]
fn promo_grpc(
issuer: &str,
blob: Vec<u8>,
prod_root: std::path::PathBuf,
) -> (Arc<HolgerGrpc>, Arc<MemoryAuditLog>) {
let backend: Arc<dyn RepositoryBackendTrait> = Arc::new(MockRepo {
name: "rust-dev".into(),
blob: Some(blob),
writable: true,
});
let routes = FastRoutes::new(vec![("rust-dev".to_string(), backend)]);
let audit = Arc::new(MemoryAuditLog::new());
let mut roles = std::collections::HashMap::new();
roles.insert("admin@corp".to_string(), Role::Admin);
roles.insert("bot@corp".to_string(), Role::Writer);
let auth = AuthConfig {
methods: vec![oidc_offline_method(issuer)],
roles,
default_role: Some(Role::Reader),
..Default::default()
};
let grpc = Arc::new(
HolgerGrpc::with_auth(routes, Arc::new(auth))
.with_audit(audit.clone())
.with_promotion(Arc::new(PromotionEngine::new(prod_root))),
);
(grpc, audit)
}
#[cfg(feature = "auth-oidc")]
#[cfg(feature = "intake")]
fn promote_req(sub_token: &str) -> Request<PromoteArtifactRequest> {
let mut req = Request::new(PromoteArtifactRequest {
from_repo: "rust-dev".into(),
to_repo: "rust-prod".into(),
artifact: Some(ArtifactId {
namespace: String::new(),
name: "serde".into(),
version: "1.0.0".into(),
}),
async_job: false,
});
req.metadata_mut().insert(
"authorization",
format!("Bearer {sub_token}").parse().unwrap(),
);
req
}
/// An ADMIN promotes: the artifact transitions + seals into the PROD store
/// (content address returned), and a `Promote` audit row is written.
#[cfg(feature = "intake")]
#[cfg(feature = "auth-oidc")]
#[tokio::test]
async fn promote_seals_and_audits_for_admin() {
let dir = tempfile::tempdir().unwrap();
let issuer = "https://idp.example/promote-admin";
let bytes = b"the-crate-bytes-to-promote".to_vec();
let (grpc, audit) = promo_grpc(issuer, bytes.clone(), dir.path().to_path_buf());
let token = test_support::token_for(issuer, "holger", "admin@corp", 3600);
let resp = PromotionService::promote_artifact(&grpc, promote_req(&token))
.await
.expect("admin promote ok")
.into_inner();
assert!(resp.success);
assert_eq!(resp.promoted_by, "admin@corp");
assert_eq!(
resp.content_address,
blake3::hash(&bytes).to_hex().to_string()
);
let ev = audit.events();
let last = ev.last().unwrap();
assert_eq!(last.action, AuditAction::Promote);
assert_eq!(last.ident, "admin@corp");
assert_eq!(last.repo, "rust-prod");
assert_eq!(last.artifact, "serde@1.0.0");
assert_eq!(last.status, 200);
}
/// A NON-admin (writer) is refused — fail-closed at the admin gate, before any
/// store access.
#[cfg(feature = "intake")]
#[cfg(feature = "auth-oidc")]
#[tokio::test]
async fn promote_refused_for_non_admin() {
let dir = tempfile::tempdir().unwrap();
let issuer = "https://idp.example/promote-writer";
let (grpc, _audit) = promo_grpc(issuer, b"bytes".to_vec(), dir.path().to_path_buf());
let token = test_support::token_for(issuer, "holger", "bot@corp", 3600);
let err = PromotionService::promote_artifact(&grpc, promote_req(&token))
.await
.expect_err("writer must be refused");
assert_eq!(err.code(), tonic::Code::PermissionDenied);
}
/// A promoted artifact is IMMUTABLE: a second admin promote of the same
/// coordinate is refused (`already_exists`) and audited 409.
#[cfg(feature = "intake")]
#[cfg(feature = "auth-oidc")]
#[tokio::test]
async fn promote_is_immutable_on_repromote() {
let dir = tempfile::tempdir().unwrap();
let issuer = "https://idp.example/promote-immut";
let (grpc, audit) = promo_grpc(issuer, b"v1-bytes".to_vec(), dir.path().to_path_buf());
let token = test_support::token_for(issuer, "holger", "admin@corp", 3600);
PromotionService::promote_artifact(&grpc, promote_req(&token))
.await
.expect("first promote ok");
let err = PromotionService::promote_artifact(&grpc, promote_req(&token))
.await
.expect_err("re-promote must be refused");
assert_eq!(err.code(), tonic::Code::AlreadyExists);
assert_eq!(audit.events().last().unwrap().status, 409);
}
/// With no promotion store configured, the RPC fails cleanly (still admin-gated
/// first).
#[cfg(feature = "intake")]
#[cfg(feature = "auth-oidc")]
#[tokio::test]
async fn promote_without_store_fails_precondition() {
let issuer = "https://idp.example/promote-nostore";
let backend: Arc<dyn RepositoryBackendTrait> = Arc::new(MockRepo {
name: "rust-dev".into(),
blob: Some(b"x".to_vec()),
writable: true,
});
let routes = FastRoutes::new(vec![("rust-dev".to_string(), backend)]);
let mut roles = std::collections::HashMap::new();
roles.insert("admin@corp".to_string(), Role::Admin);
let auth = AuthConfig {
methods: vec![oidc_offline_method(issuer)],
roles,
default_role: None,
..Default::default()
};
// No .with_promotion(...)
let grpc = Arc::new(HolgerGrpc::with_auth(routes, Arc::new(auth)));
let token = test_support::token_for(issuer, "holger", "admin@corp", 3600);
let err = PromotionService::promote_artifact(&grpc, promote_req(&token))
.await
.expect_err("no store → error");
assert_eq!(err.code(), tonic::Code::FailedPrecondition);
}
/// Property-requirement gate (property-policy §): with a `signed=true`
/// requirement, an UNSIGNED source artifact is refused promotion
/// (RED-when-broken — a broken gate would seal it), and the SAME artifact
/// promotes once the property is set on the source coordinate.
#[cfg(feature = "intake")]
#[cfg(feature = "auth-oidc")]
#[tokio::test]
async fn promote_requires_source_property() {
use crate::properties::{PropertyStore, SharedPropertyStore};
let dir = tempfile::tempdir().unwrap();
let issuer = "https://idp.example/promote-prop";
let bytes = b"signed-build-bytes".to_vec();
let backend: Arc<dyn RepositoryBackendTrait> = Arc::new(MockRepo {
name: "rust-dev".into(),
blob: Some(bytes),
writable: true,
});
let routes = FastRoutes::new(vec![("rust-dev".to_string(), backend)]);
let mut roles = std::collections::HashMap::new();
roles.insert("admin@corp".to_string(), Role::Admin);
let auth = AuthConfig {
methods: vec![oidc_offline_method(issuer)],
roles,
default_role: None,
..Default::default()
};
let props: SharedPropertyStore = Arc::new(PropertyStore::new(dir.path().join("props")));
let engine = PromotionEngine::new(dir.path().join("prod"))
.with_required_property("signed".into(), "true".into());
let audit = Arc::new(MemoryAuditLog::new());
let grpc = Arc::new(
HolgerGrpc::with_auth(routes, Arc::new(auth))
.with_audit(audit.clone())
.with_promotion(Arc::new(engine))
.with_properties(props.clone()),
);
let token = test_support::token_for(issuer, "holger", "admin@corp", 3600);
// Unsigned source → refused (FailedPrecondition), audited 412, nothing sealed.
let err = PromotionService::promote_artifact(&grpc, promote_req(&token))
.await
.expect_err("an unsigned artifact must be refused promotion");
assert_eq!(err.code(), tonic::Code::FailedPrecondition);
assert_eq!(audit.events().last().unwrap().status, 412);
// Tag the SOURCE coordinate signed=true → the same promote now succeeds.
let id = traits::ArtifactId {
namespace: None,
name: "serde".into(),
version: "1.0.0".into(),
};
props
.set("rust-dev", &id, "signed", vec!["true".into()])
.unwrap();
let resp = PromotionService::promote_artifact(&grpc, promote_req(&token))
.await
.expect("a signed artifact promotes")
.into_inner();
assert!(resp.success, "signed=true satisfies the requirement");
}
/// Sanity: the `AuthMethodConfig` import is exercised (offline method built).
#[cfg(feature = "auth-oidc")]
#[test]
fn offline_method_is_oidc() {
assert!(matches!(
oidc_offline_method("https://x"),
AuthMethodConfig::Oidc { .. }
));
}
// ─── SearchService (cross-repo discovery) ────────────────────────────────
use super::holger_proto::search_service_server::SearchService;
use traits::{ArtifactEntry, ArtifactId as TraitArtifactId};
/// A list-capable backend for the search RPC test: holds a fixed entry set
/// (name → size) so `list()` returns real artifacts to match against.
struct SearchMock {
name: String,
entries: Vec<(String, i64)>,
}
impl RepositoryBackendTrait for SearchMock {
fn caps(&self) -> traits::Caps {
traits::Caps {
listing: true,
..traits::Caps::NONE
}
}
fn name(&self) -> &str {
&self.name
}
fn format(&self) -> ArtifactFormat {
ArtifactFormat::Rust
}
fn is_writable(&self) -> bool {
false
}
fn fetch(&self, _id: &traits::ArtifactId) -> anyhow::Result<Option<traits::Bytes>> {
Ok(None)
}
fn put(&self, _id: &traits::ArtifactId, _data: &[u8]) -> anyhow::Result<()> {
anyhow::bail!("read-only")
}
fn list(&self, _f: Option<&str>, _l: usize) -> anyhow::Result<Vec<ArtifactEntry>> {
Ok(self
.entries
.iter()
.map(|(n, sz)| ArtifactEntry {
id: TraitArtifactId {
namespace: None,
name: n.clone(),
version: "1.0.0".into(),
},
size_bytes: *sz,
content_type: "application/octet-stream".into(),
// A test double knows neither, and says so.
..ArtifactEntry::UNKNOWN_PROVENANCE
})
.collect())
}
fn handle_http2_request(
&self,
_m: &str,
_s: &str,
_b: &[u8],
) -> anyhow::Result<(u16, Vec<(String, String)>, Vec<u8>)> {
Ok((404, vec![], Vec::new()))
}
}
/// The Search RPC finds the matching artifact across repos, EXCLUDES the
/// non-matches, tags each hit with its repo, and audits a List. The exclusion
/// assertion is the RED-when-broken guard.
#[tokio::test]
async fn search_rpc_finds_hits_excludes_non_matches_and_audits() {
let a: Arc<dyn RepositoryBackendTrait> = Arc::new(SearchMock {
name: "rust-dev".into(),
entries: vec![("serde".into(), 10), ("tokio".into(), 20)],
});
let b: Arc<dyn RepositoryBackendTrait> = Arc::new(SearchMock {
name: "extra".into(),
entries: vec![("serde_json".into(), 30)],
});
let routes = FastRoutes::new(vec![("rust-dev".into(), a), ("extra".into(), b)]);
let audit = Arc::new(MemoryAuditLog::new());
let grpc = Arc::new(HolgerGrpc::new(routes).with_audit(audit.clone()));
let resp = SearchService::search(
&grpc,
Request::new(SearchRequest {
name: "serde".into(),
limit: 0,
..Default::default()
}),
)
.await
.expect("search ok")
.into_inner();
// serde (rust-dev) + serde_json (extra) match; tokio must be EXCLUDED.
assert_eq!(resp.artifacts.len(), 2, "both serde* names match");
assert!(
!resp
.artifacts
.iter()
.any(|h| h.id.as_ref().unwrap().name == "tokio"),
"tokio must be EXCLUDED from a serde search"
);
// Hits carry the repo they were found in.
let repos: std::collections::HashSet<_> = resp
.artifacts
.iter()
.map(|h| h.repository.clone())
.collect();
assert!(repos.contains("rust-dev") && repos.contains("extra"));
// Audited as a List with the hit count.
let ev = audit.events();
assert_eq!(ev.len(), 1);
assert_eq!(ev[0].action, AuditAction::List);
assert_eq!(ev[0].bytes, 2, "audit records the hit count");
}
/// The Search RPC's `property` axis (§8): with a property store installed
/// (`with_properties`), a `PropertyMatch` filter returns ONLY the tagged
/// coordinate and EXCLUDES the untagged one (RED-when-broken). Without the
/// store the same filter matches nothing (fail-closed).
#[cfg(feature = "intake")]
#[tokio::test]
async fn search_rpc_property_axis_filters_and_fails_closed() {
use super::holger_proto::PropertyMatch;
let a: Arc<dyn RepositoryBackendTrait> = Arc::new(SearchMock {
name: "rust-dev".into(),
entries: vec![("serde".into(), 10), ("tokio".into(), 20)],
});
let routes = FastRoutes::new(vec![("rust-dev".into(), a)]);
// Tag only serde 1.0.0 with env=prod.
let tmp = tempfile::tempdir().unwrap();
let store: crate::properties::SharedPropertyStore =
Arc::new(crate::properties::PropertyStore::new(tmp.path()));
let serde_id = traits::ArtifactId {
namespace: None,
name: "serde".into(),
version: "1.0.0".into(),
};
store
.set("rust-dev", &serde_id, "env", vec!["prod".into()])
.unwrap();
let req = || SearchRequest {
properties: vec![PropertyMatch {
key: "env".into(),
value: "prod".into(),
}],
limit: 0,
..Default::default()
};
// With the store: only serde matches; tokio (untagged) is excluded.
let grpc = Arc::new(HolgerGrpc::new(routes.clone()).with_properties(store));
let resp = SearchService::search(&grpc, Request::new(req()))
.await
.unwrap()
.into_inner();
assert_eq!(resp.artifacts.len(), 1, "only the env=prod artifact");
assert_eq!(resp.artifacts[0].id.as_ref().unwrap().name, "serde");
// WITHOUT the store: the property filter fails closed (no hits).
let grpc_nostore = Arc::new(HolgerGrpc::new(routes));
let resp2 = SearchService::search(&grpc_nostore, Request::new(req()))
.await
.unwrap()
.into_inner();
assert_eq!(
resp2.artifacts.len(),
0,
"no property store ⇒ property filter matches nothing"
);
}
// ── Enumeration is disclosure: the cross-repo doors honour a per-repo ACL ──
//
// `Denied` on a repository has to mean denied, not "denied through the front
// door". `Search` and `ListRepositories` authorize the caller's GLOBAL role
// (they have no single repo to scope to) and then walked EVERY configured
// backend, so an identity correctly refused a direct read of `classified` got
// its coordinates, versions, sizes and archive paths out of the search index
// instead — and the catalog confirmed the repo exists at all.
const CAROL_TOKEN: &str = "carol-token-of-quite-sufficient-length";
/// Read gate ON, one identity: `carol` is a global `Reader` who is `Denied` on
/// `classified` by a per-repo ACL — the exact configuration the finding names.
fn carol_denied_on_classified_cfg() -> Arc<crate::auth::AuthConfig> {
use crate::auth::{AuthMethodConfig, Role};
Arc::new(crate::auth::AuthConfig {
methods: vec![AuthMethodConfig::StaticToken {
subject: "carol".into(),
token: Some(CAROL_TOKEN.into()),
token_env: None,
token_file: None,
}],
repo_roles: std::collections::HashMap::from([(
"classified".to_string(),
std::collections::HashMap::from([("carol".to_string(), Role::Denied)]),
)]),
default_role: Some(Role::Reader),
require_auth_for_reads: true,
..Default::default()
})
}
/// Two repos: a `public` one carol may read and a `classified` one she may not.
fn public_and_classified_routes() -> FastRoutes {
let public: Arc<dyn RepositoryBackendTrait> = Arc::new(SearchMock {
name: "public".into(),
entries: vec![("serde".into(), 10)],
});
let classified: Arc<dyn RepositoryBackendTrait> = Arc::new(SearchMock {
name: "classified".into(),
entries: vec![("nuke-codes".into(), 1)],
});
FastRoutes::new(vec![
("public".into(), public),
("classified".into(), classified),
])
}
fn with_carol<T>(mut req: Request<T>) -> Request<T> {
req.metadata_mut().insert(
"authorization",
format!("Bearer {CAROL_TOKEN}").parse().unwrap(),
);
req
}
/// **RED-when-broken:** an identity `Denied` on `classified` must not see
/// `nuke-codes` in a search that scans every repo. Before the repo-set filter
/// this returned BOTH hits — the direct-read refusal was a formality.
#[tokio::test]
async fn search_excludes_repositories_the_identity_may_not_read() {
let grpc = Arc::new(HolgerGrpc::with_auth(
public_and_classified_routes(),
carol_denied_on_classified_cfg(),
));
// A bare query lists everything the caller is allowed to see.
let resp = SearchService::search(
&grpc,
with_carol(Request::new(SearchRequest {
limit: 0,
..Default::default()
})),
)
.await
.expect("a Reader searches")
.into_inner();
let names: Vec<&str> = resp
.artifacts
.iter()
.map(|h| h.id.as_ref().unwrap().name.as_str())
.collect();
assert!(
!names.contains(&"nuke-codes"),
"a repo carol is Denied on must not leak a coordinate through Search, got {names:?}"
);
assert!(
!resp.artifacts.iter().any(|h| h.repository == "classified"),
"no hit may name the denied repository"
);
// …and the repo she MAY read still answers, so this is the ACL biting and
// not a blanket outage.
assert_eq!(names, vec!["serde"], "the readable repo is still searched");
}
/// **RED-when-broken:** the catalog must not name a repository the caller may
/// not read — the cheapest enumeration surface on the server.
#[tokio::test]
async fn list_repositories_excludes_repositories_the_identity_may_not_read() {
let grpc = Arc::new(HolgerGrpc::with_auth(
public_and_classified_routes(),
carol_denied_on_classified_cfg(),
));
let resp = AdminService::list_repositories(
&grpc,
with_carol(Request::new(ListRepositoriesRequest {})),
)
.await
.expect("a Reader lists")
.into_inner();
let names: Vec<&str> = resp.repositories.iter().map(|r| r.name.as_str()).collect();
assert_eq!(
names,
vec!["public"],
"the denied repo must not appear in the catalog"
);
}
/// ★ **Per-repository gates only (no global gate): the catalog is listed per
/// CALLER.** `ListRepositories` and `Search` authorized against `""`, which no
/// per-repository gate covers, so the identity was never resolved and every
/// gated repository was filtered out for EVERYBODY — the console's admin
/// bearer included (the estate's `monetize-backup`, measured 2026-10-04).
/// Now: an admin sees the admin-floored repository, a reader sees the
/// auth-gated one but not the admin-floored one, an anonymous caller sees
/// only the open one, and a credential that does not validate is refused.
#[tokio::test]
async fn per_repository_gates_list_each_repository_to_those_who_pass_it() {
use crate::auth::{AuthMethodConfig, Role};
let mk = |name: &str| -> Arc<dyn RepositoryBackendTrait> {
Arc::new(SearchMock {
name: name.into(),
entries: vec![(format!("{name}-pkg"), 1)],
})
};
let routes = FastRoutes::new(vec![
("open".into(), mk("open")),
("members".into(), mk("members")),
("backup".into(), mk("backup")),
]);
let token = |subject: &str, t: &str| AuthMethodConfig::StaticToken {
subject: subject.into(),
token: Some(t.into()),
token_env: None,
token_file: None,
};
let cfg = crate::auth::AuthConfig {
methods: vec![
token("console", "console-token"),
token("vera", "vera-token"),
],
roles: std::collections::HashMap::from([
("console".to_string(), Role::Admin),
("vera".to_string(), Role::Reader),
]),
require_auth_for_reads: false,
..Default::default()
};
let gate = crate::auth::ReadGate::from_repos(["members".to_string()])
.with_floors([("backup".to_string(), Role::Admin)]);
let grpc =
Arc::new(HolgerGrpc::with_auth(routes, Arc::new(cfg)).with_read_gate(Arc::new(gate)));
let as_who = |t: Option<&str>| {
let mut r = Request::new(ListRepositoriesRequest {});
if let Some(t) = t {
r.metadata_mut()
.insert("authorization", format!("Bearer {t}").parse().unwrap());
}
r
};
let names = |r: ListRepositoriesResponse| -> Vec<String> {
let mut v: Vec<String> = r.repositories.into_iter().map(|r| r.name).collect();
v.sort();
v
};
let admin = AdminService::list_repositories(&grpc, as_who(Some("console-token")))
.await
.expect("admin lists")
.into_inner();
assert_eq!(
names(admin),
["backup", "members", "open"],
"an admin passes every floor"
);
let reader = AdminService::list_repositories(&grpc, as_who(Some("vera-token")))
.await
.expect("reader lists")
.into_inner();
assert_eq!(
names(reader),
["members", "open"],
"a reader passes the auth gate, not the admin floor"
);
let anon = AdminService::list_repositories(&grpc, as_who(None))
.await
.expect("anonymous lists")
.into_inner();
assert_eq!(
names(anon),
["open"],
"anonymous: the open repositories only"
);
let bad = AdminService::list_repositories(&grpc, as_who(Some("forged")))
.await
.expect_err("a forged credential");
assert_eq!(bad.code(), tonic::Code::Unauthenticated);
let mut sr = Request::new(SearchRequest {
limit: 0,
..Default::default()
});
sr.metadata_mut()
.insert("authorization", "Bearer console-token".parse().unwrap());
let hits = SearchService::search(&grpc, sr)
.await
.expect("admin searches")
.into_inner();
let mut repos: Vec<String> = hits.artifacts.into_iter().map(|h| h.repository).collect();
repos.sort();
assert_eq!(
repos,
["backup", "members", "open"],
"the search walks what the caller may read"
);
}
/// **The default is untouched.** With the read gate OFF — every deployment that
/// has not opted in — the filter is `all_repos()` verbatim: an anonymous search
/// and catalog still see every repository, per-repo ACL present or not.
#[tokio::test]
async fn the_repo_filter_is_a_no_op_when_the_read_gate_is_off() {
let mut cfg = (*carol_denied_on_classified_cfg()).clone();
cfg.require_auth_for_reads = false;
let grpc = Arc::new(HolgerGrpc::with_auth(
public_and_classified_routes(),
Arc::new(cfg),
));
let resp = SearchService::search(
&grpc,
Request::new(SearchRequest {
limit: 0,
..Default::default()
}),
)
.await
.expect("open search")
.into_inner();
assert_eq!(
resp.artifacts.len(),
2,
"gate off ⇒ every repo is searched, as before"
);
let cat = AdminService::list_repositories(&grpc, Request::new(ListRepositoriesRequest {}))
.await
.expect("open catalog")
.into_inner();
assert_eq!(
cat.repositories.len(),
2,
"gate off ⇒ the whole catalog, as before"
);
}
// ── the in-flight READ budget (findings 4) ───────────────────────────────
/// Build a server whose only repo serves `blob` for every coordinate, with the
/// in-flight read budget pinned to `budget` bytes.
fn budgeted_grpc(blob: Vec<u8>, budget: usize) -> Arc<HolgerGrpc> {
let backend: Arc<dyn RepositoryBackendTrait> = Arc::new(MockRepo {
name: "crates".into(),
blob: Some(blob),
writable: false,
});
let routes = FastRoutes::new(vec![("crates".to_string(), backend)]);
Arc::new(HolgerGrpc::new(routes).with_max_in_flight_read_bytes(budget))
}
/// **RED-when-broken:** an artifact that does not fit the in-flight read budget
/// is REFUSED (`resource_exhausted`) instead of being materialised anyway — and
/// the same artifact on the same server serves fine once the budget can hold it,
/// so this is the budget and not a broken read path.
#[tokio::test]
async fn a_read_larger_than_the_budget_is_refused_not_served() {
const MIB: usize = 1024 * 1024;
let blob = vec![0u8; 2 * MIB];
let tight = budgeted_grpc(blob.clone(), MIB);
let err = RepositoryService::fetch_artifact(&tight, bearer_fetch(None))
.await
.expect_err("2 MiB cannot be served from a 1 MiB budget");
assert_eq!(err.code(), tonic::Code::ResourceExhausted);
let roomy = budgeted_grpc(blob.clone(), 8 * MIB);
let ok = RepositoryService::fetch_artifact(&roomy, bearer_fetch(None))
.await
.expect("the same read fits an 8 MiB budget")
.into_inner();
assert_eq!(ok.data.len(), blob.len());
// …and `0` disables the accounting entirely (the documented escape hatch).
let unmetered = budgeted_grpc(blob.clone(), 0);
assert!(
RepositoryService::fetch_artifact(&unmetered, bearer_fetch(None))
.await
.is_ok()
);
}
/// **RED-when-broken, and the reason the budget exists:** `StreamArtifact`'s
/// chunking is wire framing, not a memory bound — it materialises the whole
/// artifact and then feeds it out at the client's pace. A stream that is opened
/// and not drained must therefore KEEP holding its slice of the budget, so the
/// next concurrent stream is refused rather than multiplying the resident bytes.
#[tokio::test]
async fn an_undrained_stream_holds_its_budget_against_the_next_reader() {
const MIB: usize = 1024 * 1024;
// Exactly enough for one 2 MiB artifact and nothing more.
let grpc = budgeted_grpc(vec![0u8; 2 * MIB], 2 * MIB);
let first = RepositoryService::stream_artifact(&grpc, bearer_fetch(None))
.await
.expect("the first stream fits the budget");
// Not drained: the feeding task still owns the buffer (and the permit).
let err = RepositoryService::stream_artifact(&grpc, bearer_fetch(None))
.await
.expect_err("a second concurrent stream must not multiply the resident bytes");
assert_eq!(err.code(), tonic::Code::ResourceExhausted);
// Drop the first stream ⇒ the feeder's send fails, the task ends, the budget
// comes back and the second reader is served. (Yield rather than sleep: the
// feeder is a task on this same runtime.)
drop(first);
let mut served = false;
for _ in 0..100 {
tokio::task::yield_now().await;
if RepositoryService::stream_artifact(&grpc, bearer_fetch(None))
.await
.is_ok()
{
served = true;
break;
}
}
assert!(
served,
"the budget must be RELEASED when a stream is dropped, not leaked"
);
}
// ── blocking backend I/O must not run on a reactor worker (finding 5) ────
/// A backend that records the thread each synchronous call ran on. In a
/// `#[tokio::test]` (current-thread runtime) the async handler runs on the test's
/// own thread and `spawn_blocking` runs on a pool thread, so "same thread as the
/// test" is exactly "this ran inline on the reactor".
struct ThreadRecordingRepo {
name: String,
calls: Arc<std::sync::Mutex<Vec<(&'static str, std::thread::ThreadId)>>>,
}
impl RepositoryBackendTrait for ThreadRecordingRepo {
fn caps(&self) -> traits::Caps {
traits::Caps::NONE
}
fn name(&self) -> &str {
&self.name
}
fn format(&self) -> ArtifactFormat {
ArtifactFormat::Rust
}
fn is_writable(&self) -> bool {
true
}
fn fetch(&self, _id: &traits::ArtifactId) -> anyhow::Result<Option<traits::Bytes>> {
self.calls
.lock()
.unwrap()
.push(("fetch", std::thread::current().id()));
Ok(Some(traits::Bytes::from_static(b"bytes")))
}
fn put(&self, _id: &traits::ArtifactId, _data: &[u8]) -> anyhow::Result<()> {
self.calls
.lock()
.unwrap()
.push(("put", std::thread::current().id()));
Ok(())
}
fn handle_http2_request(
&self,
_m: &str,
_s: &str,
_b: &[u8],
) -> anyhow::Result<(u16, Vec<(String, String)>, Vec<u8>)> {
Ok((404, vec![], Vec::new()))
}
}
/// **RED-when-broken:** `fetch` / `put` on a backend are synchronous disk,
/// znippy-decompression and — for a proxy repo — outbound upstream HTTP. Called
/// inline from an async handler they pin a reactor worker for the whole
/// round-trip; the HTTP door has always dispatched the identical call through
/// `spawn_blocking`. Every one of these must land off this thread.
#[tokio::test]
async fn backend_reads_and_writes_run_off_the_reactor_thread() {
let calls: Arc<std::sync::Mutex<Vec<(&'static str, std::thread::ThreadId)>>> =
Arc::new(std::sync::Mutex::new(Vec::new()));
let backend: Arc<dyn RepositoryBackendTrait> = Arc::new(ThreadRecordingRepo {
name: "crates".into(),
calls: calls.clone(),
});
let grpc = Arc::new(HolgerGrpc::new(FastRoutes::new(vec![(
"crates".to_string(),
backend,
)])));
RepositoryService::fetch_artifact(&grpc, bearer_fetch(None))
.await
.expect("fetch");
RepositoryService::stream_artifact(&grpc, bearer_fetch(None))
.await
.expect("stream");
RepositoryService::put_artifact(
&grpc,
Request::new(PutArtifactRequest {
repository: "crates".into(),
id: Some(artifact_id()),
data: traits::Bytes::from_static(b"payload"),
async_job: false,
}),
)
.await
.expect("put");
let here = std::thread::current().id();
let recorded = calls.lock().unwrap().clone();
assert_eq!(
recorded.len(),
3,
"fetch + stream's fetch + put all reached the backend"
);
for (what, tid) in recorded {
assert_ne!(
tid, here,
"{what} ran INLINE on the reactor thread — it must go through spawn_blocking"
);
}
}
// ── W2 / read-only posture: the gRPC WRITE + ADMIN gates ────────────────
//
// The HTTP twin (`exposed::http::authorize_http_write`) is unit-tested arm by
// arm. These drive the gRPC side of the SAME policy through the real
// `authorize_write` / `authorize_admin`, so the two doors cannot drift apart
// silently.
/// Three static-token identities with the three interesting roles.
fn rbac_token_cfg() -> Arc<crate::auth::AuthConfig> {
use crate::auth::{AuthMethodConfig, Role};
let method = |s: &str| AuthMethodConfig::StaticToken {
subject: s.to_string(),
token: Some(rbac_token(s)),
token_env: None,
token_file: None,
};
Arc::new(crate::auth::AuthConfig {
The file goes on; the console shows all of it.