nordisk/holger
server/lib/src/grpc/services.rs
//! The three tonic gRPC service impls on `Arc<HolgerGrpc>` — `RepositoryService`
//! (fetch/list/put/stream), `ArchiveService` (list/stat the znippy archive), and
//! `AdminService` (health/catalog/profile). Split out of the `grpc` module root to
//! keep it focused on the shared state + helpers; as a child module this still
//! reaches `HolgerGrpc`'s private fields/methods and the parent's free helpers
//! (`internal_err` / `peer_addr_string` / `artifact_label`).
use std::sync::Arc;
use tokio_stream::wrappers::ReceiverStream;
use tonic::{Request, Response, Status};
use super::holger_proto::admin_service_server::AdminService;
use super::holger_proto::archive_service_server::ArchiveService;
#[cfg(feature = "intake")]
use super::holger_proto::promotion_service_server::PromotionService;
use super::holger_proto::repository_service_server::RepositoryService;
use super::holger_proto::search_service_server::SearchService;
use super::holger_proto::*;
use super::{artifact_label, coded, internal_err, refusal, HolgerGrpc};
use crate::audit::{AuditAction, AuditEvent};
#[cfg(feature = "intake")]
use crate::promote::PromoteError;
use crate::search;
use holger_errcode as ec;
/// `StreamArtifact` wire framing: one `ArtifactChunk` per 64 KiB.
const CHUNK_SIZE: usize = 64 * 1024;
/// How much a backend is asked to READ per piece on the streaming door —
/// 16 wire chunks per blocking hop, so a 1 GiB file costs 1 024 hops and never
/// more than one piece of resident memory per stream.
const READ_PIECE_BYTES: usize = 1024 * 1024;
/// Turn an empty proto string into `None` (proto3 has no optional string here),
/// anything else into `Some`. The search axes all use "" = "not set".
fn opt(s: String) -> Option<String> {
if s.is_empty() {
None
} else {
Some(s)
}
}
/// Build the engine's [`search::SearchQuery`] from the proto request.
pub(crate) fn search_query_from_proto(req: SearchRequest) -> search::SearchQuery {
search::SearchQuery {
name: opt(req.name),
namespace: opt(req.namespace),
version: opt(req.version),
checksum: opt(req.checksum),
path: opt(req.path),
properties: req
.properties
.into_iter()
.filter_map(|p| {
if p.key.is_empty() {
None
} else {
Some((p.key, p.value))
}
})
.collect(),
repositories: req
.repositories
.into_iter()
.filter(|r| !r.is_empty())
.collect(),
limit: req.limit.max(0) as usize,
}
}
/// Convert a proto [`ArtifactId`] into the internal [`traits::ArtifactId`] (an
/// empty namespace string ⇒ `None`). Shared by the SBOM RPCs.
#[cfg_attr(not(feature = "intake"), allow(dead_code))]
pub(crate) fn trait_id_from_proto(p: &ArtifactId) -> traits::ArtifactId {
traits::ArtifactId {
namespace: if p.namespace.is_empty() {
None
} else {
Some(p.namespace.clone())
},
name: p.name.clone(),
version: p.version.clone(),
}
}
/// Map engine [`search::SearchResults`] onto the proto response.
pub(crate) fn search_results_to_proto(r: search::SearchResults) -> SearchResponse {
SearchResponse {
artifacts: r
.artifacts
.into_iter()
.map(|h| ArtifactHit {
repository: h.repository,
id: Some(ArtifactId {
namespace: h.id.namespace.unwrap_or_default(),
name: h.id.name,
version: h.id.version,
}),
size_bytes: h.size_bytes,
content_type: h.content_type,
checksum: h.checksum.unwrap_or_default(),
})
.collect(),
paths: r
.paths
.into_iter()
.map(|p| PathHit {
repository: p.repository,
path: p.path,
})
.collect(),
truncated: r.truncated,
}
}
/// How many pieces of a streamed gRPC upload may wait for the backend: the
/// whole of what the door holds besides the backend's own buffer.
const STREAM_UPLOAD_DEPTH: usize = 8;
/// The reader a streamed gRPC upload hands its backend: pieces off a bounded
/// channel, on a blocking thread. `Err(())` in the channel is the door refusing
/// the upload (over the cap, a broken stream); the backend sees it as a read
/// error and must leave nothing behind.
struct PiecesReader {
rx: tokio::sync::mpsc::Receiver<Result<traits::Bytes, ()>>,
current: traits::Bytes,
}
impl std::io::Read for PiecesReader {
fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
while self.current.is_empty() {
match self.rx.blocking_recv() {
Some(Ok(piece)) => self.current = piece,
Some(Err(())) => {
return Err(std::io::Error::other(
"the streamed upload was refused at the door",
));
}
None => return Ok(0),
}
}
let n = buf.len().min(self.current.len());
buf[..n].copy_from_slice(&self.current[..n]);
let _ = self.current.split_to(n);
Ok(n)
}
}
impl HolgerGrpc {
fn stream_over_cap(&self, nbytes: u64) -> Status {
coded(
&ec::SERVER_PAYLOAD_TOO_LARGE,
tonic::Code::ResourceExhausted,
format!(
"Payload too large: at least {nbytes} bytes exceeds limit of {} bytes",
self.max_body_bytes
),
)
}
}
#[tonic::async_trait]
impl RepositoryService for Arc<HolgerGrpc> {
async fn fetch_artifact(
&self,
request: Request<FetchArtifactRequest>,
) -> Result<Response<FetchArtifactResponse>, Status> {
// The visitor tally: one request on this door and this repository.
let peer = self.count_visitor(&request, &request.get_ref().repository, true)?;
// Reads are OPEN by default (the airgap-mirror posture) — `authorize_read`
// is then a no-op returning `None`. With the opt-in read gate on
// (`AuthConfig::require_auth_for_reads`) it authenticates the caller and
// authorizes against the `Reader` floor scoped to this repository, and the
// audit records the REAL subject instead of `anonymous`.
let gate_repo = request.get_ref().repository.clone();
let identity = self.authorize_read(&request, &gate_repo).await?;
let who = self.pull_principal(peer, &identity);
let source_ip = self.audit_peer(&request);
// The door admission's ticket (`crate::admission_grpc`), when the layer
// admitted this call: the bytes are charged to it.
#[cfg(feature = "admission")]
let ticket = crate::admission_grpc::ticket_of(&request);
#[cfg(not(feature = "admission"))]
let ticket: Option<Arc<crate::admission::Ticket>> = None;
let req = request.into_inner();
let repo_name = req.repository.clone();
let repo = self.get_repo(&repo_name)?;
// An intake repository is read through the chain: only an AVAILABLE
// artifact of production, from the sealed store (`promote::intake`).
let repo = self.read_backend(&repo_name, repo);
let id = req.id.ok_or_else(|| {
coded(
&ec::SERVER_BAD_REQUEST,
tonic::Code::InvalidArgument,
"Missing artifact ID",
)
})?;
let artifact = artifact_label(&id);
let trait_id = traits::ArtifactId {
namespace: if id.namespace.is_empty() {
None
} else {
Some(id.namespace)
},
name: id.name,
version: id.version,
};
// In-flight read budget: reserve the floor BEFORE the backend materialises
// the blob, so a burst cannot all reach `fetch` at once (the backend hands
// back a whole `Vec<u8>` — there is no streaming read to bound instead).
let mut budget = match self.reserve_read_budget("holger-grpc/fetch_artifact") {
Ok(b) => b,
Err(status) => {
self.record_audit(
AuditEvent::new(
&who,
AuditAction::Download,
&repo_name,
&artifact,
&source_ip,
503,
0,
)
.with_detail("in-flight read budget exhausted"),
);
return Err(status);
}
};
// The backend read is SYNCHRONOUS — disk, znippy decompression, and for a
// proxy repo an outbound upstream HTTP round-trip — so it runs on the
// blocking pool, never on a reactor worker. The HTTP door has always
// dispatched its identical backend call this way (`exposed/http.rs`).
let walk_ticket = ticket.clone();
let fetched = match tokio::task::spawn_blocking(move || match &walk_ticket {
Some(t) => t.walk(|| repo.fetch(&trait_id)),
None => repo.fetch(&trait_id),
})
.await
{
Ok(r) => r,
Err(e) => {
super::functional_status(
"holger-grpc/fetch_artifact",
"artifact_resolved",
false,
"dispatch task failed",
);
return Err(internal_err(format!("fetch task failed: {e}")));
}
};
match fetched {
Ok(Some(data)) => {
let n = data.len() as u64;
if let Some(i) = peer {
self.peers.delivered(i, n);
}
if let Some(t) = &ticket {
t.add_net_out(n);
}
// Now the size is known: charge the rest of the budget, or drop the
// bytes here rather than let them ride out a slow download.
if let Err(status) =
self.charge_read_budget(&mut budget, data.len(), "holger-grpc/fetch_artifact")
{
drop(data);
self.record_audit(
AuditEvent::new(
&who,
AuditAction::Download,
&repo_name,
&artifact,
&source_ip,
503,
0,
)
.with_detail("artifact exceeds the in-flight read budget"),
);
return Err(status);
}
self.record_audit(AuditEvent::new(
&who,
AuditAction::Download,
&repo_name,
&artifact,
&source_ip,
200,
n,
));
super::functional_status(
"holger-grpc/fetch_artifact",
"artifact_resolved",
true,
&repo_name,
);
// The budget is released as this frame ends — the unary door cannot
// hold it across tonic's encode, because the bytes leave with the
// response. It therefore bounds the FETCH-time pile-up (K backends
// each materialising a blob at once), which is the peak; `StreamArtifact`
// below, where the handler keeps the buffer for the whole download,
// holds its slice for the life of the stream.
Ok(Response::new(FetchArtifactResponse {
size_bytes: data.len() as i64,
content_type: "application/octet-stream".into(),
data,
}))
}
Ok(None) => {
self.record_audit(AuditEvent::new(
&who,
AuditAction::Download,
&repo_name,
&artifact,
&source_ip,
404,
0,
));
super::functional_status(
"holger-grpc/fetch_artifact",
"artifact_resolved",
false,
"not found",
);
Err(coded(
&ec::SERVER_ARTIFACT_UNKNOWN,
tonic::Code::NotFound,
"Artifact not found",
))
}
// The queued pull-through (`crate::pull_through`): UNAVAILABLE with a
// `retry-after`, the gRPC spelling of 503 — the fetch is under way.
#[cfg(feature = "admission")]
Err(e) if crate::pull_through::queued_of(&e).is_some() => {
let q = crate::pull_through::queued_of(&e).expect("checked");
self.record_audit(
AuditEvent::new(
&who,
AuditAction::Download,
&repo_name,
&artifact,
&source_ip,
503,
0,
)
.with_detail(q.to_string()),
);
let mut status = coded(&ec::ADMISSION_PULL_QUEUED, tonic::Code::Unavailable, q);
if let Ok(v) = q.retry_after_secs.to_string().parse() {
status.metadata_mut().insert("retry-after", v);
}
if let Ok(v) = q.job.parse() {
status.metadata_mut().insert("x-holger-job", v);
}
Err(status)
}
Err(e) => {
self.record_audit(
AuditEvent::new(
&who,
AuditAction::Download,
&repo_name,
&artifact,
&source_ip,
500,
0,
)
.with_detail(e.to_string()),
);
super::functional_status(
"holger-grpc/fetch_artifact",
"artifact_resolved",
false,
"backend error",
);
Err(internal_err(&e))
}
}
}
async fn list_artifacts(
&self,
request: Request<ListArtifactsRequest>,
) -> Result<Response<ListArtifactsResponse>, Status> {
// The visitor tally: one request on this door and this repository.
let peer = self.count_visitor(&request, &request.get_ref().repository, true)?;
// Reads are OPEN by default (the airgap-mirror posture) — `authorize_read`
// is then a no-op returning `None`. With the opt-in read gate on
// (`AuthConfig::require_auth_for_reads`) it authenticates the caller and
// authorizes against the `Reader` floor scoped to this repository, and the
// audit records the REAL subject instead of `anonymous`.
let gate_repo = request.get_ref().repository.clone();
let identity = self.authorize_read(&request, &gate_repo).await?;
let who = self.pull_principal(peer, &identity);
let source_ip = self.audit_peer(&request);
let req = request.into_inner();
let repo_name = req.repository.clone();
let filter_label = req.name_filter.clone();
let repo = match self.get_repo(&repo_name) {
Ok(r) => r,
Err(status) => {
self.record_audit(
AuditEvent::new(
&who,
AuditAction::List,
&repo_name,
&filter_label,
&source_ip,
404,
0,
)
.with_detail("repository not found"),
);
super::functional_status(
"holger-grpc/list_artifacts",
"listing_served",
false,
"repository not found",
);
return Err(status);
}
};
// Empty proto string means "no filter" (proto3 has no optional string
// here); anything else is a real substring filter.
let name_filter = if req.name_filter.is_empty() {
None
} else {
Some(req.name_filter.as_str())
};
// proto `limit` is int32 (the page size); the backend takes a usize.
let limit = req.limit.max(0) as usize;
// `page_token` is the opaque offset cursor; empty = first page.
let page_token = if req.page_token.is_empty() {
None
} else {
Some(req.page_token.as_str())
};
let (entries, next_page_token) = match crate::object::paginate_artifacts(
repo.as_ref(),
name_filter,
limit,
page_token,
) {
Ok(v) => v,
Err(e) => {
self.record_audit(
AuditEvent::new(
&who,
AuditAction::List,
&repo_name,
&filter_label,
&source_ip,
500,
0,
)
.with_detail(e.to_string()),
);
super::functional_status(
"holger-grpc/list_artifacts",
"listing_served",
false,
"backend error",
);
return Err(internal_err(&e));
}
};
self.record_audit(AuditEvent::new(
&who,
AuditAction::List,
&repo_name,
&filter_label,
&source_ip,
200,
entries.len() as u64,
));
super::functional_status(
"holger-grpc/list_artifacts",
"listing_served",
true,
&repo_name,
);
let artifacts = entries
.into_iter()
.map(|e| ArtifactEntry {
id: Some(ArtifactId {
// None namespace (Rust-style, no namespace) → empty proto
// string, the wire convention this server uses elsewhere.
namespace: e.id.namespace.unwrap_or_default(),
name: e.id.name,
version: e.id.version,
}),
size_bytes: e.size_bytes,
content_type: e.content_type,
// ── Publishing the two optional fields ──
//
// The backend's `Option` goes on the wire as proto3 explicit
// presence: `None` is sent as ABSENT, not as `0` / `""`. That
// is the whole reason both fields are `optional` in the proto —
// a zero timestamp sorts as 1970 and an empty digest compares
// equal to another empty digest, so a scalar default would be a
// fabricated answer from a backend that has none.
//
// Normalized on the way out as well as on the way in, so this
// server cannot be the one that publishes a malformed digest
// even if a backend hands it one.
published_unix_ms: traits::normalize_published_unix_ms(e.published_unix_ms),
sha256: e.sha256.as_deref().and_then(traits::normalize_sha256),
// The third optional field, same rule. A znippy-backed backend
// answers this one and only this one -- it reads the archive
// index, never an artifact byte.
change_token: e
.change_token
.as_deref()
.and_then(traits::normalize_change_token),
})
.collect();
Ok(Response::new(ListArtifactsResponse {
artifacts,
next_page_token,
}))
}
async fn put_artifact(
&self,
request: Request<PutArtifactRequest>,
) -> Result<Response<PutArtifactResponse>, Status> {
// The visitor tally: one request on this door and this repository.
let _ = self.count_visitor(&request, &request.get_ref().repository, false);
let source_ip = self.audit_peer(&request);
// Target repo is known from the request body — authorize the write SCOPED
// to it, so a per-repo ACL (`repo_roles`) can grant/deny per repository.
let repo_name = request.get_ref().repository.clone();
// The resolved write principal (OIDC sub / mTLS CN), or anonymous when
// auth is open. Captured for the audit trail.
let named = request
.get_ref()
.id
.as_ref()
.map(artifact_label)
.unwrap_or_default();
let identity = self
.authorize_upload(&request, &repo_name, &named, &source_ip)
.await?;
let ident = identity
.as_ref()
.map(|id| id.subject.clone())
.unwrap_or_else(|| "anonymous".to_string());
let req = request.into_inner();
let repo = self.get_repo(&repo_name)?;
if !repo.is_writable() {
self.record_audit(
AuditEvent::new(
&ident,
AuditAction::Upload,
&repo_name,
"",
&source_ip,
403,
0,
)
.with_detail("repository is read-only"),
);
super::functional_status(
"holger-grpc/put_artifact",
"artifact_stored",
false,
"repository is read-only",
);
return Err(coded(
&ec::SERVER_REPO_READ_ONLY,
tonic::Code::PermissionDenied,
"Repository is read-only",
));
}
// Ingest body cap (matches the HTTP gateway). An oversized gRPC upload
// must be rejected rather than buffered/stored unbounded.
if req.data.len() > self.max_body_bytes {
self.record_audit(
AuditEvent::new(
&ident,
AuditAction::Upload,
&repo_name,
"",
&source_ip,
413,
req.data.len() as u64,
)
.with_detail("payload too large"),
);
super::functional_status(
"holger-grpc/put_artifact",
"artifact_stored",
false,
"payload too large",
);
return Err(coded(
&ec::SERVER_PAYLOAD_TOO_LARGE,
tonic::Code::ResourceExhausted,
format!(
"Payload too large: {} bytes exceeds limit of {} bytes",
req.data.len(),
self.max_body_bytes
),
));
}
let id = req.id.ok_or_else(|| {
coded(
&ec::SERVER_BAD_REQUEST,
tonic::Code::InvalidArgument,
"Missing artifact ID",
)
})?;
let artifact = artifact_label(&id);
let trait_id = traits::ArtifactId {
namespace: if id.namespace.is_empty() {
None
} else {
Some(id.namespace)
},
name: id.name,
version: id.version,
};
let nbytes = req.data.len() as u64;
// ── THE INTAKE CHAIN (`promote::intake`) ────────────────────────────
// An upload into the chain's incoming repository is UNTRUSTED → INGESTED
// through the chain: its ingest gates, its fail-closed audit record, this
// backend's write, its record. A write into the chain's other three
// repositories is refused by name. Asked before the plain ingest gates
// below, which the chain itself runs.
#[cfg(feature = "intake")]
if let Some(chain) = self.intake.clone() {
// SEC-REVIEW 2026-11-01 P3: a raw write (this door does not speak docker, npm or
// cargo) is also refused where only a door's own protocol walks the chain.
if let Some(refusal) = chain.raw_write_refusal(&repo_name) {
self.record_audit(
AuditEvent::new(
&ident,
AuditAction::Upload,
&repo_name,
&artifact,
&source_ip,
403,
nbytes,
)
.with_detail(refusal.tagged()),
);
return Err(coded(
refusal.code(),
tonic::Code::PermissionDenied,
&refusal,
));
}
if chain.role_of(&repo_name).is_some() {
// Only an Importer brings an artifact in (lane W7).
if let Err(e) =
crate::promote::duties::may_ingest(&self.auth_config, identity.as_ref())
{
self.record_audit(
AuditEvent::new(
&ident,
AuditAction::Upload,
&repo_name,
&artifact,
&source_ip,
403,
nbytes,
)
.with_detail(e.to_string()),
);
return Err(coded(e.code(), tonic::Code::PermissionDenied, &e));
}
let (who, ip, rn, data) = (
ident.clone(),
source_ip.clone(),
repo_name.clone(),
req.data,
);
let id = trait_id.clone();
let ingested = tokio::task::spawn_blocking(move || {
chain.ingest(&who, &ip, &rn, &id, &data, || repo.put(&id, &data))
})
.await
.unwrap_or_else(|_| {
Err(crate::promote::intake::IntakeError::Store(
"dispatch task failed".into(),
))
});
return match ingested {
Ok(entry) => {
super::functional_status(
"holger-grpc/put_artifact",
"artifact_ingested",
true,
&repo_name,
);
Ok(Response::new(PutArtifactResponse {
success: true,
message: format!("INGESTED {} sha256 {}", entry.artifact, entry.sha256),
job_id: String::new(),
}))
}
Err(e) => {
super::functional_status(
"holger-grpc/put_artifact",
"artifact_ingested",
false,
&e.to_string(),
);
let code = match e.status() {
403 => tonic::Code::PermissionDenied,
400 => tonic::Code::InvalidArgument,
409 | 412 => tonic::Code::FailedPrecondition,
_ => tonic::Code::Internal,
};
Err(coded(e.code(), code, &e))
}
};
}
}
// ── THE INGEST GATE ─────────────────────────────────────────────────
// Run BEFORE the bytes reach the backend.
//
// Promotion is not the first boundary an artifact crosses; upload is.
// Until 2026-09-20 holger gated only the promotion hop, which means a
// vulnerable artifact was written to the DEV repository and served from
// it, and the gate's job was stopping it from MOVING rather than
// stopping it from ARRIVING.
//
// Same trait, same catalog, same verdict vocabulary as the promotion
// boundary. The only difference is the context: no source repository
// (the bytes came off the wire) and no rung.
#[cfg(feature = "intake")]
if !self.ingest_gates.is_empty() {
let ecosystem = crate::scanner::ecosystem_for_format(repo.format());
let digest = crate::attest::ArtifactDigest::of_sha256(&req.data);
let attestations: Vec<crate::attest::Statement> = match &self.attestations {
Some(store) => store
.list(&digest)
.map(|v| v.into_iter().map(|a| a.statement).collect())
.unwrap_or_default(),
None => Vec::new(),
};
let set = [crate::promote::PromotedArtifact {
id: &trait_id,
ecosystem: &ecosystem,
digest,
bytes: &req.data,
attestations: &attestations,
}];
let ctx = crate::promote::PromotionContext::ingest(&ident, &repo_name, &set);
let decision = self.ingest_gates.evaluate(&ctx);
if !decision.is_allowed()
&& self.ingest_gates.enforcement() == crate::promote::Enforcement::Block
{
let detail = format!("ingest gates REFUSED: {}", decision.summary());
self.record_audit(
AuditEvent::new(
&ident,
AuditAction::Upload,
&repo_name,
&artifact,
&source_ip,
412,
nbytes,
)
.with_detail(detail.clone()),
);
super::functional_status(
"holger-grpc/put_artifact",
"artifact_stored",
false,
"blocked by the ingest gates",
);
return Err(coded(
&ec::PROMOTE_POLICY_BLOCKED,
tonic::Code::FailedPrecondition,
detail,
));
}
self.record_audit(
AuditEvent::new(
&ident,
AuditAction::Upload,
&repo_name,
&artifact,
&source_ip,
200,
0,
)
.with_detail(format!("ingest gates: {}", decision.summary())),
);
}
// ── THE GATED FINALIZE HALF, AS A JOB (stinky 01 §1.3) ──────────────
//
// Reached only after the ingest gates above have run and allowed, so the
// async shape moves the WRITE off the request and nothing else. The body
// is staged under the queue's own staging directory and the job owns it
// from there — the worker writes it into the backend and removes it.
//
// No digest is claimed here on purpose: the bytes never left this
// process, so re-hashing them on the worker would be ceremony, and
// claiming a digest nobody computed would be worse.
if req.async_job {
#[cfg(feature = "jobs")]
{
let Some(queue) = self.job_queue.clone() else {
return Err(coded(
&ec::SERVER_CAPABILITY_NOT_CONFIGURED,
tonic::Code::FailedPrecondition,
"this server has no job queue configured; retry without `async_job`",
));
};
let staged = queue.staging_dir().join(format!(
"{}-{}.blob",
crate::audit::now_nanos(),
artifact.replace(['/', ':', '\\'], "_")
));
if let Err(e) = std::fs::write(&staged, &req.data) {
return Err(internal_err(format!(
"could not stage the upload at {}: {e}",
staged.display()
)));
}
let job = crate::jobs::Job::Upload {
repo: repo_name.clone(),
artifact: crate::jobs::JobArtifact {
namespace: trait_id.namespace.clone().unwrap_or_default(),
name: trait_id.name.clone(),
version: trait_id.version.clone(),
},
staged: staged.clone(),
expect_sha256: String::new(),
requested_by: ident.clone(),
};
let job_id = match queue.enqueue(job) {
Ok(id) => id,
Err(e) => {
let _ = std::fs::remove_file(&staged);
return Err(internal_err(format!("could not admit the upload: {e:#}")));
}
};
self.record_audit(
AuditEvent::new(
&ident,
AuditAction::Upload,
&repo_name,
&artifact,
&source_ip,
202,
nbytes,
)
.with_detail(format!("upload finalize admitted as job {job_id}")),
);
super::functional_status(
"holger-grpc/put_artifact",
"upload_admitted_as_job",
true,
job_id.as_str(),
);
return Ok(Response::new(PutArtifactResponse {
success: true,
message: format!("upload finalize admitted as job {job_id}"),
job_id: job_id.0,
}));
}
#[cfg(not(feature = "jobs"))]
{
return Err(coded(
&ec::SERVER_CAPABILITY_NOT_CONFIGURED,
tonic::Code::FailedPrecondition,
"this binary was built without the `jobs` feature: there is no job \
queue to admit an asynchronous upload to. Retry without `async_job`, \
or rebuild with `--features jobs`",
));
}
}
// The backend write is SYNCHRONOUS (disk, archive append, and for a proxy
// repo an outbound upstream HTTP round-trip), so it runs on the blocking
// pool rather than pinning a reactor worker for the duration — the same
// dispatch the HTTP door has always used (`exposed/http.rs`).
let data = req.data;
let stored = match tokio::task::spawn_blocking(move || repo.put(&trait_id, &data)).await {
Ok(r) => r,
Err(e) => {
self.record_audit(
AuditEvent::new(
&ident,
AuditAction::Upload,
&repo_name,
&artifact,
&source_ip,
500,
0,
)
.with_detail("dispatch task failed"),
);
super::functional_status(
"holger-grpc/put_artifact",
"artifact_stored",
false,
"dispatch task failed",
);
return Err(internal_err(format!("put task failed: {e}")));
}
};
match stored {
Ok(()) => {
self.record_audit(AuditEvent::new(
&ident,
AuditAction::Upload,
&repo_name,
&artifact,
&source_ip,
200,
nbytes,
));
super::functional_status(
"holger-grpc/put_artifact",
"artifact_stored",
true,
&repo_name,
);
Ok(Response::new(PutArtifactResponse {
success: true,
message: "Artifact stored".into(),
job_id: String::new(),
}))
}
Err(e) => {
self.record_audit(
AuditEvent::new(
&ident,
AuditAction::Upload,
&repo_name,
&artifact,
&source_ip,
500,
0,
)
.with_detail(e.to_string()),
);
// The paid limit (metered build: the repository's backend is a
// `metered::QuotaGuard`, which measured on this blocking thread)
// answers by its own code, not laundered into `holger-server-4`.
let said = e.to_string();
if ec::parse(&said) == Some(&ec::METERED_OVER_CAP) {
super::functional_status(
"holger-grpc/put_artifact",
"artifact_stored",
false,
"over its disk_bytes limit",
);
return Err(refusal(tonic::Code::ResourceExhausted, said));
}
super::functional_status(
"holger-grpc/put_artifact",
"artifact_stored",
false,
"backend error",
);
Err(internal_err(&e))
}
}
}
/// **The streamed write door** (stinky 14). The pieces go to the backend's
/// `write_from_reader` AS THEY ARRIVE, through a bounded channel into a
/// reader on a blocking thread: a backend with `Caps::streaming_uploads`
/// writes them to disk as it reads, any other collects them for its `put`
/// (the whole-buffer convenience). What this process holds is the channel's
/// depth in pieces plus the backend's own buffer.
///
/// An upload a GATE must read whole — the intake chain's repositories, the
/// ingest gates, or the `async_job` stage — is collected (bounded by the
/// body cap) and handed to [`put_artifact`](Self::put_artifact), so the
/// streamed door is never a way around a gate.
async fn put_artifact_stream(
&self,
request: Request<tonic::Streaming<PutArtifactChunk>>,
) -> Result<Response<PutArtifactResponse>, Status> {
let source_ip = self.audit_peer(&request);
let (metadata, extensions, mut pieces) = request.into_parts();
let first = pieces.message().await?.ok_or_else(|| {
coded(
&ec::SERVER_BAD_REQUEST,
tonic::Code::InvalidArgument,
"the upload stream is empty: it names no artifact",
)
})?;
let Some(id) = first.id.clone() else {
return Err(coded(
&ec::SERVER_BAD_REQUEST,
tonic::Code::InvalidArgument,
"Missing artifact ID",
));
};
let repo_name = first.repository.clone();
#[cfg(feature = "intake")]
let gated = !self.ingest_gates.is_empty()
|| self
.intake
.as_ref()
.is_some_and(|c| c.role_of(&repo_name).is_some());
#[cfg(not(feature = "intake"))]
let gated = false;
if gated || first.async_job {
// Collected up to one byte past the cap, so `put_artifact` refuses it
// by its own 413 rather than this door buffering an unbounded body.
let mut data: Vec<u8> = first.data.to_vec();
while data.len() <= self.max_body_bytes {
match pieces.message().await? {
Some(m) => data.extend_from_slice(&m.data),
None => break,
}
}
let whole = PutArtifactRequest {
repository: repo_name,
id: Some(id),
data: data.into(),
async_job: first.async_job,
};
return self
.put_artifact(Request::from_parts(metadata, extensions, whole))
.await;
}
let probe = Request::from_parts(metadata, extensions, ());
let _ = self.count_visitor(&probe, &repo_name, false);
let identity = self
.authorize_upload(&probe, &repo_name, &artifact_label(&id), &source_ip)
.await?;
let ident = identity
.as_ref()
.map(|id| id.subject.clone())
.unwrap_or_else(|| "anonymous".to_string());
let repo = self.get_repo(&repo_name)?;
if !repo.is_writable() {
self.record_audit(
AuditEvent::new(
&ident,
AuditAction::Upload,
&repo_name,
"",
&source_ip,
403,
0,
)
.with_detail("repository is read-only"),
);
return Err(coded(
&ec::SERVER_REPO_READ_ONLY,
tonic::Code::PermissionDenied,
"Repository is read-only",
));
}
let artifact = artifact_label(&id);
// ★ SEC-REVIEW 2026-11-01 P2: the streamed door asks the chain the same question the
// whole-body door and HTTP ask, before a piece reaches the backend. `gated` above is
// true only with ingest gates or a chain role, so with neither (the default) a side
// door used to take the stream unrefused.
#[cfg(feature = "intake")]
if let Some(refusal) = self
.intake
.as_ref()
.and_then(|chain| chain.raw_write_refusal(&repo_name))
{
self.record_audit(
AuditEvent::new(
&ident,
AuditAction::Upload,
&repo_name,
&artifact,
&source_ip,
403,
0,
)
.with_detail(refusal.tagged()),
);
return Err(coded(
refusal.code(),
tonic::Code::PermissionDenied,
&refusal,
));
}
let trait_id = traits::ArtifactId {
namespace: if id.namespace.is_empty() {
None
} else {
Some(id.namespace)
},
name: id.name,
version: id.version,
};
let (tx, rx) = tokio::sync::mpsc::channel::<Result<traits::Bytes, ()>>(STREAM_UPLOAD_DEPTH);
// A32-14, stays a task: one upload's writer; its handle is awaited by the stream.
let dispatch = tokio::task::spawn_blocking(move || {
let mut body = PiecesReader {
rx,
current: traits::Bytes::new(),
};
repo.write_from_reader(&trait_id, &mut body, None)
});
let cap = self.max_body_bytes as u64;
let mut nbytes = first.data.len() as u64;
let mut refused: Option<Status> = None;
if nbytes > cap {
refused = Some(self.stream_over_cap(nbytes));
let _ = tx.send(Err(())).await;
} else if tx.send(Ok(first.data)).await.is_ok() {
loop {
match pieces.message().await {
Ok(Some(m)) => {
nbytes += m.data.len() as u64;
if nbytes > cap {
refused = Some(self.stream_over_cap(nbytes));
let _ = tx.send(Err(())).await;
break;
}
if tx.send(Ok(m.data)).await.is_err() {
break; // the backend stopped reading: it has an answer
}
}
Ok(None) => break,
Err(status) => {
let _ = tx.send(Err(())).await;
refused = Some(status);
break;
}
}
}
}
drop(tx);
let stored = dispatch.await;
if let Some(status) = refused {
let code = if status.code() == tonic::Code::ResourceExhausted {
413
} else {
400
};
self.record_audit(
AuditEvent::new(
&ident,
AuditAction::Upload,
&repo_name,
&artifact,
&source_ip,
code,
0,
)
.with_detail(format!("streamed upload refused: {}", status.message())),
);
super::functional_status(
"holger-grpc/put_artifact_stream",
"artifact_stored",
false,
status.message(),
);
return Err(status);
}
match stored {
Ok(Ok(())) => {
self.record_audit(AuditEvent::new(
&ident,
AuditAction::Upload,
&repo_name,
&artifact,
&source_ip,
200,
nbytes,
));
super::functional_status(
"holger-grpc/put_artifact_stream",
"artifact_stored",
true,
&repo_name,
);
Ok(Response::new(PutArtifactResponse {
success: true,
message: format!("Artifact stored ({nbytes} bytes, streamed)"),
job_id: String::new(),
}))
}
Ok(Err(e)) => {
self.record_audit(
AuditEvent::new(
&ident,
AuditAction::Upload,
&repo_name,
&artifact,
&source_ip,
500,
0,
)
.with_detail(e.to_string()),
);
let said = e.to_string();
if ec::parse(&said) == Some(&ec::METERED_OVER_CAP) {
return Err(refusal(tonic::Code::ResourceExhausted, said));
}
super::functional_status(
"holger-grpc/put_artifact_stream",
"artifact_stored",
false,
"backend error",
);
Err(internal_err(&e))
}
Err(e) => Err(internal_err(format!("put task failed: {e}"))),
}
}
type StreamArtifactStream = ReceiverStream<Result<ArtifactChunk, Status>>;
async fn stream_artifact(
&self,
request: Request<FetchArtifactRequest>,
) -> Result<Response<Self::StreamArtifactStream>, Status> {
// The visitor tally: one request on this door and this repository.
let peer = self.count_visitor(&request, &request.get_ref().repository, true)?;
// Reads are OPEN by default (the airgap-mirror posture) — `authorize_read`
// is then a no-op returning `None`. With the opt-in read gate on
// (`AuthConfig::require_auth_for_reads`) it authenticates the caller and
// authorizes against the `Reader` floor scoped to this repository, and the
// audit records the REAL subject instead of `anonymous`.
let gate_repo = request.get_ref().repository.clone();
let identity = self.authorize_read(&request, &gate_repo).await?;
let who = self.pull_principal(peer, &identity);
let source_ip = self.audit_peer(&request);
let req = request.into_inner();
let repo_name = req.repository.clone();
let repo = self.get_repo(&repo_name)?;
// An intake repository is read through the chain: only an AVAILABLE
// artifact of production, from the sealed store (`promote::intake`).
let repo = self.read_backend(&repo_name, repo);
let id = req.id.ok_or_else(|| {
coded(
&ec::SERVER_BAD_REQUEST,
tonic::Code::InvalidArgument,
"Missing artifact ID",
)
})?;
let artifact = artifact_label(&id);
let trait_id = traits::ArtifactId {
namespace: if id.namespace.is_empty() {
None
} else {
Some(id.namespace)
},
name: id.name,
version: id.version,
};
// In-flight read budget. This door is the reason the budget exists: the
// chunking below is a WIRE-FRAMING detail, not a memory bound — the whole
// artifact is materialised here first and then handed out 64 KiB at a time
// at whatever rate the client reads, so a handful of slow clients on a large
// artifact pinned an unbounded multiple of it. The permit is held by the
// feeding task until the last chunk leaves (or the stream is dropped), so
// the charge lasts exactly as long as the bytes do.
let mut budget = match self.reserve_read_budget("holger-grpc/stream_artifact") {
Ok(b) => b,
Err(status) => {
self.record_audit(
AuditEvent::new(
&who,
AuditAction::Download,
&repo_name,
&artifact,
&source_ip,
503,
0,
)
.with_detail("in-flight read budget exhausted"),
);
return Err(status);
}
};
// The backend is asked for the artifact IN PIECES (`fetch_chunks`, W12,
// stinky-01 section 2.3): a file store opens the file and reads a chunk
// at a time; a znippy store slices the `Bytes` it already holds. Either
// way the first frame is on the wire before the last byte has been read
// off the store, and what this process holds per stream is the channel's
// depth in chunks, not the artifact. The budget is charged AS THE BYTES
// GO OUT, per chunk, and the audit row carries what was actually sent.
// Opened off the reactor, walked off the reactor.
let opened = match tokio::task::spawn_blocking(move || {
repo.read_chunks(&trait_id, READ_PIECE_BYTES)
})
.await
{
Ok(r) => r,
Err(e) => {
super::functional_status(
"holger-grpc/stream_artifact",
"stream_opened",
false,
"dispatch task failed",
);
return Err(internal_err(format!("fetch task failed: {e}")));
}
};
let chunks = match opened {
Ok(Some(chunks)) => chunks,
Ok(None) => {
self.record_audit(AuditEvent::new(
&who,
AuditAction::Download,
&repo_name,
&artifact,
&source_ip,
404,
0,
));
super::functional_status(
"holger-grpc/stream_artifact",
"stream_opened",
false,
"not found",
);
return Err(coded(
&ec::SERVER_ARTIFACT_UNKNOWN,
tonic::Code::NotFound,
"Artifact not found",
));
}
Err(e) => {
self.record_audit(
AuditEvent::new(
&who,
AuditAction::Download,
&repo_name,
&artifact,
&source_ip,
500,
0,
)
.with_detail(e.to_string()),
);
super::functional_status(
"holger-grpc/stream_artifact",
"stream_opened",
false,
"backend error",
);
return Err(internal_err(&e));
}
};
// A backend that knows the artifact's length before the first piece
// (a file's size, a `Bytes` in hand) is charged for ALL of it now, so a
// stream that is opened and not drained keeps holding its slice of the
// budget against the next reader — `an_undrained_stream_holds_its_budget_
// against_the_next_reader` is the ruling. The audit row is written now
// too, with that length, as it always was. Only a backend that cannot
// say (a true pipe) is charged piece by piece and audited at the end.
let known_len = chunks.len.map(|n| usize::try_from(n).unwrap_or(usize::MAX));
if let Some(n) = known_len {
if let Err(status) =
self.charge_read_budget(&mut budget, n, "holger-grpc/stream_artifact")
{
self.record_audit(
AuditEvent::new(
&who,
AuditAction::Download,
&repo_name,
&artifact,
&source_ip,
503,
0,
)
.with_detail("artifact exceeds the in-flight read budget"),
);
return Err(status);
}
if let Some(i) = peer {
self.peers.delivered(i, n as u64);
}
self.record_audit(AuditEvent::new(
&who,
AuditAction::Download,
&repo_name,
&artifact,
&source_ip,
200,
n as u64,
));
}
super::functional_status(
"holger-grpc/stream_artifact",
"stream_opened",
true,
&repo_name,
);
let (tx, rx) = tokio::sync::mpsc::channel(16);
// The walk: each `next()` may block (a file read), so it runs on a
// blocking thread, one hop per READ piece; each read piece is then
// framed as 64 KiB `Bytes::slice`s — refcount bumps, no per-frame copy —
// from this async task, whose `send` fails the moment the client goes
// away. The budget guard rides along and is released when this task
// ends, i.e. when the last chunk has been accepted or the stream was
// dropped: a client that stalls mid-download keeps paying for the bytes
// it is still holding open.
// A32-14, stays a task: one task per download stream, ended with its stream.
let this = Arc::clone(self);
tokio::spawn(async move {
let mut budget = budget;
let mut chunks = chunks;
let mut total = 0usize;
let mut status = 200u16;
let mut detail = String::new();
'walk: loop {
// A client that went away releases the budget NOW, not after
// the read in flight returns: the closed channel wins the race.
// A32-14, stays a task: one chunk read of a download stream, bounded by the client (a gone client ends it).
let hop = tokio::task::spawn_blocking(move || {
let n = chunks.next();
(chunks, n)
});
let (back, next) = match tokio::select! {
biased;
_ = tx.closed() => None,
r = hop => Some(r),
} {
None => break,
Some(Ok(pair)) => pair,
Some(Err(e)) => {
status = 500;
detail = format!("read task failed: {e}");
let _ = tx.send(Err(internal_err(&detail))).await;
break;
}
};
chunks = back;
let piece = match next {
None => break,
Some(Ok(p)) => p,
Some(Err(e)) => {
status = 500;
detail = e.to_string();
let _ = tx.send(Err(internal_err(&e))).await;
break;
}
};
total += piece.len();
if known_len.is_some() {
// Paid in full at open.
} else if let Err(refused) =
this.charge_read_budget_to(&mut budget, total, "holger-grpc/stream_artifact")
{
status = 503;
detail = "artifact exceeds the in-flight read budget".to_string();
let _ = tx.send(Err(refused)).await;
break;
}
let mut offset = 0;
while offset < piece.len() {
let end = (offset + CHUNK_SIZE).min(piece.len());
if tx
.send(Ok(ArtifactChunk {
data: piece.slice(offset..end),
}))
.await
.is_err()
{
break 'walk;
}
offset = end;
}
}
// A pipe of unknown length is one package delivered once it ends whole.
if known_len.is_none() && status == 200 {
if let Some(i) = peer {
this.peers.delivered(i, total as u64);
}
}
if known_len.is_none() || status != 200 {
let mut ev = AuditEvent::new(
&who,
AuditAction::Download,
&repo_name,
&artifact,
&source_ip,
status,
total as u64,
);
if !detail.is_empty() {
ev = ev.with_detail(detail);
}
this.record_audit(ev);
}
drop(budget);
});
Ok(Response::new(ReceiverStream::new(rx)))
}
}
// ─── ArchiveService ──────────────────────────────────────────────────
#[tonic::async_trait]
impl ArchiveService for Arc<HolgerGrpc> {
async fn list_archive_files(
&self,
request: Request<ListArchiveFilesRequest>,
) -> Result<Response<ListArchiveFilesResponse>, Status> {
// Reads are OPEN by default (the airgap-mirror posture) — `authorize_read`
// is then a no-op returning `None`. With the opt-in read gate on
// (`AuthConfig::require_auth_for_reads`) it authenticates the caller and
// authorizes against the `Reader` floor scoped to this repository, and the
// audit records the REAL subject instead of `anonymous`.
let gate_repo = request.get_ref().repository.clone();
let identity = self.authorize_read(&request, &gate_repo).await?;
let who = HolgerGrpc::read_principal(&identity);
let source_ip = self.audit_peer(&request);
let req = request.into_inner();
let repo_name = req.repository.clone();
let prefix_label = req.prefix.clone();
let repo = match self.get_repo(&repo_name) {
Ok(r) => r,
Err(status) => {
self.record_audit(
AuditEvent::new(
&who,
AuditAction::List,
&repo_name,
&prefix_label,
&source_ip,
404,
0,
)
.with_detail("repository not found"),
);
super::functional_status(
"holger-grpc/list_archive_files",
"archive_listed",
false,
"repository not found",
);
return Err(status);
}
};
// Empty proto string means "no prefix filter" (proto3 has no optional
// string here); anything else is a real path-prefix filter.
let prefix = if req.prefix.is_empty() {
None
} else {
Some(req.prefix.as_str())
};
// A repository with no archive has no paths; the catalog's
// `has_archive` tells that apart from an empty archive.
let listed = if repo.caps().archive {
repo.archive().map(|a| a.paths(prefix))
} else {
Ok(Vec::new())
};
let paths = match listed {
Ok(p) => p,
Err(e) => {
self.record_audit(
AuditEvent::new(
&who,
AuditAction::List,
&repo_name,
&prefix_label,
&source_ip,
500,
0,
)
.with_detail(e.to_string()),
);
super::functional_status(
"holger-grpc/list_archive_files",
"archive_listed",
false,
"backend error",
);
return Err(internal_err(&e));
}
};
self.record_audit(AuditEvent::new(
&who,
AuditAction::List,
&repo_name,
&prefix_label,
&source_ip,
200,
paths.len() as u64,
));
super::functional_status(
"holger-grpc/list_archive_files",
"archive_listed",
true,
&repo_name,
);
Ok(Response::new(ListArchiveFilesResponse {
// proto `total_files` is int64.
total_files: paths.len() as i64,
paths,
}))
}
async fn archive_info(
&self,
request: Request<ArchiveInfoRequest>,
) -> Result<Response<ArchiveInfoResponse>, Status> {
// Reads are OPEN by default (the airgap-mirror posture) — `authorize_read`
// is then a no-op returning `None`. With the opt-in read gate on
// (`AuthConfig::require_auth_for_reads`) it authenticates the caller and
// authorizes against the `Reader` floor scoped to this repository, and the
// audit records the REAL subject instead of `anonymous`.
let gate_repo = request.get_ref().repository.clone();
let identity = self.authorize_read(&request, &gate_repo).await?;
let who = HolgerGrpc::read_principal(&identity);
let source_ip = self.audit_peer(&request);
let req = request.into_inner();
let repo_name = req.repository.clone();
let repo = match self.get_repo(&repo_name) {
Ok(r) => r,
Err(status) => {
self.record_audit(
AuditEvent::new(&who, AuditAction::List, &repo_name, "", &source_ip, 404, 0)
.with_detail("repository not found"),
);
super::functional_status(
"holger-grpc/archive_info",
"archive_stat_served",
false,
"repository not found",
);
return Err(status);
}
};
let stat = if repo.caps().archive {
repo.archive().map(|a| a.info())
} else {
Ok(traits::ArchiveInfo {
archive_path: repo_name.clone(),
..Default::default()
})
};
let info = match stat {
Ok(i) => i,
Err(e) => {
self.record_audit(
AuditEvent::new(&who, AuditAction::List, &repo_name, "", &source_ip, 500, 0)
.with_detail(e.to_string()),
);
super::functional_status(
"holger-grpc/archive_info",
"archive_stat_served",
false,
"backend error",
);
return Err(internal_err(&e));
}
};
self.record_audit(AuditEvent::new(
&who,
AuditAction::List,
&repo_name,
"",
&source_ip,
200,
info.file_count,
));
super::functional_status(
"holger-grpc/archive_info",
"archive_stat_served",
true,
&repo_name,
);
Ok(Response::new(ArchiveInfoResponse {
// proto counters are int64; the trait carries u64.
file_count: info.file_count as i64,
total_uncompressed_bytes: info.total_uncompressed_bytes as i64,
archive_path: info.archive_path,
}))
}
}
// ─── AdminService ────────────────────────────────────────────────────
#[tonic::async_trait]
impl AdminService for Arc<HolgerGrpc> {
async fn health(
&self,
_request: Request<HealthRequest>,
) -> Result<Response<HealthResponse>, Status> {
super::functional_status(
"holger-grpc/health",
"health_reported",
true,
crate::VERSION_LINE,
);
Ok(Response::new(HealthResponse {
status: "ok".into(),
// Single-source identity: `holger v<ver> (<git-hash>)` — matches the
// server bins' `--version` and the in-process health surface exactly.
version: crate::VERSION_LINE.into(),
uptime_seconds: self.start_time.elapsed().as_secs() as i64,
}))
}
async fn list_repositories(
&self,
request: Request<ListRepositoriesRequest>,
) -> Result<Response<ListRepositoriesResponse>, Status> {
// Two steps, and the second one is the security-relevant half.
//
// 1. Authorize against the empty repo name to ESTABLISH THE IDENTITY (and
// apply the global floor: a caller the policy denies outright, or an
// unauthenticated one under the read gate, never gets this far).
// 2. Then serve only the repositories that identity may actually read. The
// catalog is the cheapest enumeration surface there is — names, formats,
// writability — and step 1 alone checks the caller's GLOBAL role, which a
// per-repo `Denied` does not lower. Without the filter, an identity
// correctly refused a read of `classified` still learned it exists.
// `authorize_catalog`: with per-repository gates only, the caller is
// resolved (not required) so each gated repository is listed to those
// who pass it — `""` alone is covered by no gate.
let identity = self.authorize_catalog(&request).await?;
let who = HolgerGrpc::read_principal(&identity);
let source_ip = self.audit_peer(&request);
// With the read gate off (the default) this is `all_repos()` verbatim.
let repos = self.readable_repos(identity.as_ref());
self.record_audit(AuditEvent::new(
&who,
AuditAction::List,
"",
"",
&source_ip,
200,
repos.len() as u64,
));
super::functional_status(
"holger-grpc/list_repositories",
"catalog_served",
true,
&format!("{} repos", repos.len()),
);
// A SYSTEM repository (Admin-floored, the sealed `monetize-backup`) says so: the
// guest's `tenant_owners` leaves it out, as the pulled report does.
let infos = repos
.iter()
.map(|(name, repo)| RepositoryInfo {
system: self.read_gate.is_system(name),
..crate::capacity::repository_info(name, repo.as_ref())
})
.collect();
Ok(Response::new(ListRepositoriesResponse {
repositories: infos,
}))
}
/// Report the server's operating profile — server-reported truth so the
/// operator UI reflects the server, not a local toggle. Read-only is the
/// explicit sealed-mode override if set, else derived from the route table
/// (read-only iff no repository accepts writes). Maps to the demo's static
/// "rigged for silent running" chrome.
async fn server_profile(
&self,
_request: Request<ServerProfileRequest>,
) -> Result<Response<ServerProfileResponse>, Status> {
let read_only = self.is_read_only();
let writable = self.writable_repo_count() as i32;
let (profile, label) = if read_only {
("static", "RIGGED FOR SILENT RUNNING / READ-ONLY")
} else {
("dynamic", "DYNAMIC / WRITABLE")
};
super::functional_status(
"holger-grpc/server_profile",
"profile_reported",
true,
profile,
);
// The census, verbatim from what this boot built. Nothing is derived
// here and nothing is filtered: a census that edits itself on the way
// out is not a census.
let census = &self.module_census;
Ok(Response::new(ServerProfileResponse {
profile: profile.into(),
read_only,
label: label.into(),
writable_repo_count: writable,
modules: census
.modules
.iter()
.map(|m| crate::grpc::holger_proto::LoadedModule {
name: m.name.clone(),
origin: m.origin.clone(),
kind: m.kind.clone(),
digest: m.digest.clone(),
receipt: m.receipt.clone(),
granted: m.granted.clone(),
})
.collect(),
granted_capabilities: census.granted_capabilities.clone(),
linker_defines: census.linker_defines.clone(),
}))
}
/// Who does the server think the caller is? The presented credential goes
/// through the SAME `auth::validate_request` every gated call uses — one
/// validator, no twin — and the answer is identity + effective GLOBAL role.
/// Introspection only: no secrets leave (the credential is never echoed),
/// nothing is authorized here, and an invalid credential is reported as
/// anonymous rather than an error so a UI can always render a session
/// state. Mirrors `server_profile`'s shape: no audit event (no artifact is
/// touched), one functional-status row.
async fn who_am_i(
&self,
request: Request<WhoAmIRequest>,
) -> Result<Response<WhoAmIResponse>, 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 =
// Invalid credential ⇒ the caller IS anonymous for role
// purposes; the gates will refuse them the same way.
crate::auth::validate_request(&self.auth_config, bearer, client_cn.as_deref())
.await
.unwrap_or_default();
let read_gate = self.auth_config.require_auth_for_reads;
// HCON-5: what is open here, in the gate's own words (read only on the console).
let access_words = crate::posture::words(&crate::posture::Posture {
gated: read_gate,
front_open: self.read_gate.front_open(),
public: self.read_gate.public_names(),
why: String::new(),
});
// The per-repository answer, when one was asked for: the SAME
// resolution `authorize_write` / `authorize_admin` apply
// (`role_for_identity_in_repo`), so a console that asks it is told the
// gates' own verdict rather than computing one.
let repository = request.get_ref().repository.trim().to_owned();
let repository_role = match (&identity, repository.is_empty()) {
(_, true) => String::new(),
(Some(id), false) => format!(
"{:?}",
self.auth_config.role_for_identity_in_repo(&repository, id)
)
.to_lowercase(),
(None, false) => self
.auth_config
.default_role
.map(|r| format!("{r:?}").to_lowercase())
.unwrap_or_else(|| "reader".into()),
};
let resp = match identity {
Some(id) => WhoAmIResponse {
authenticated: true,
// The role the gates would actually resolve for this identity, groups
// included — not `role_for(subject)`, which ignores them and would
// report a privilege level the server does not grant.
global_role: format!("{:?}", self.auth_config.role_for_identity(&id))
.to_lowercase(),
subject: id.subject,
method: id.method,
read_gate,
repository_role,
access_words: access_words.clone(),
},
None => WhoAmIResponse {
authenticated: false,
subject: String::new(),
method: String::new(),
global_role: self
.auth_config
.default_role
.map(|r| format!("{r:?}").to_lowercase())
.unwrap_or_else(|| "reader".into()),
read_gate,
repository_role,
access_words,
},
};
super::functional_status(
"holger-grpc/who_am_i",
"identity_reported",
true,
if resp.authenticated {
"authenticated"
} else {
"anonymous"
},
);
Ok(Response::new(resp))
}
/// **How much storage there is** — one `statvfs(3)` over the data root plus
/// the archive bytes of every repository that owns storage (groups are views
/// and are skipped). See [`crate::capacity`].
///
/// Admin READ: it describes the host filesystem, which is not catalog
/// information. Gated like the audit plane (`authorize_admin_read`): the
/// `Admin` role under RBAC, open with no auth configured, and NOT refused on
/// a read-only server — measuring a sealed appliance changes nothing on it.
///
/// A `HolgerGrpc` nobody handed a data root refuses BY NAME
/// (`FailedPrecondition`, [`crate::metering::NO_FILESYSTEM_SOURCE`]); a
/// root the kernel will not describe, or an archive that cannot be read,
/// is `Unavailable` with the path and errno. Never a capacity of zero.
/// ★★ **The bootstrap door: elect this box a twin.**
///
/// Owner rulings 2026-09-21 — *"the weakest machine must become the twin.
/// PERIOD"* and *"IT must not CHANGE role later ! so its a bootstrap thing
/// from xtask"*. Both boxes install from ONE identical medium with no role
/// parameter; `xtask` measures them (this service's `Capacity` is half the
/// measurement) and tells the weaker one what it is, exactly once.
///
/// ADMIN-gated. Every refusal is by name, and none of them is a retry:
/// `holger-twin-4` (already elected), `holger-twin-5` (not a fresh box).
async fn become_twin(
&self,
request: Request<BecomeTwinRequest>,
) -> Result<Response<BecomeTwinReply>, Status> {
over_the_tunnel(&request, "BecomeTwin")?;
let identity = self.authorize_admin(&request, "").await?;
let req = request.into_inner();
let primary = req.primary.trim().to_string();
if primary.is_empty() {
return Err(coded(
&ec::SERVER_BAD_REQUEST,
tonic::Code::InvalidArgument,
"BecomeTwin names no primary: a twin that names no primary could neither \
refuse a write by name nor follow anything",
));
}
let Some(root) = self.data_root.clone() else {
return Err(no_data_root());
};
let role = crate::twin::NodeRole::Twin {
primary: primary.clone(),
poll_seconds: if req.poll_seconds == 0 {
crate::twin::DEFAULT_POLL_SECONDS
} else {
req.poll_seconds
},
};
let now = crate::clock::now_unix_secs();
let who = identity
.as_ref()
.map(|id| id.subject.clone())
.unwrap_or_else(|| "anonymous".to_string());
let who_for_log = who.clone();
let reason = req.reason.clone();
// The election touches the filesystem: off the executor, as capacity is.
let elected = tokio::task::spawn_blocking(move || {
crate::twin::elect(&root, role, &who, &reason, now)
})
.await
.map_err(|e| internal_err(format!("election task failed: {e}")))?;
match elected {
Ok(record) => {
// ★ LIVE (stinky 14 §2 correction 3): the record is on disk, so the
// posture switches NOW — every door that asks `write_refusal`
// refuses by name from this line on — and the follow starts. The
// record first, the posture second: a crash between the two is a
// twin at the next start, never a posture the disk does not hold.
self.twin.become_twin(record.role.clone());
let followed = match &self.follower {
Some(f) => f
.start(&record.role)
.map(|n| format!("follow started ({n} tasks)"))
.unwrap_or_else(|e| format!("follow NOT started: {e}")),
None => "no follower in this process; the follow starts at the next start"
.to_string(),
};
log::warn!("BecomeTwin: this box is now a TWIN of `{primary}`, live — every write refused by name ({}); {followed}. Its public doors keep their bind until the next start", ec::TWIN_REFUSES_WRITES.code);
super::functional_status(
"holger-grpc/become-twin",
"twin_elected",
true,
&format!(
"twin of {primary}, elected by {who_for_log}: {}; {followed}",
record.reason
),
);
Ok(Response::new(BecomeTwinReply {
elected_at: record.elected_at,
elected_by: record.elected_by,
// The loopback bind of the public doors waits for the next
// start: a running server does not rebind them underneath
// its clients. Everything else is live.
restart_required: true,
applied_live: true,
}))
}
Err(refusal_why) => {
super::functional_status(
"holger-grpc/become-twin",
"twin_election_refused",
false,
&refusal_why,
);
Err(refusal(tonic::Code::FailedPrecondition, refusal_why))
}
}
}
/// ★ **`TwinCsr`** — this box's replication request, the key minted HERE on
/// the first call (`crate::twin_tls`). Admin; asked of a twin too — pairing
/// changes no artifact, and the twin's certification comes after its role.
async fn twin_csr(
&self,
request: Request<TwinCsrRequest>,
) -> Result<Response<TwinCsrReply>, Status> {
over_the_tunnel(&request, "TwinCsr")?;
self.authorize_admin_identity(&request, "", "pair a replication reader")
.await?;
let root = self.data_root.clone().ok_or_else(no_data_root)?;
let purpose = crate::twin_tls::Purpose::parse(&request.get_ref().purpose)
.map_err(|e| refusal(tonic::Code::InvalidArgument, e))?;
let is_twin = self.twin.is_twin();
let (csr_pem, principal) =
tokio::task::spawn_blocking(move || crate::twin_tls::request(&root, purpose, is_twin))
.await
.map_err(|e| internal_err(format!("the request task: {e}")))?
.map_err(|e| refusal(tonic::Code::FailedPrecondition, e))?;
Ok(Response::new(TwinCsrReply { csr_pem, principal }))
}
/// ★ **`IssueTwinCertificate`** — this box's OWN replication issuer signs a
/// reader's request. Refused by name without a door (`holger-twin-7`) or for
/// a principal it does not certify (`holger-twin-8`). Admin.
async fn issue_twin_certificate(
&self,
request: Request<IssueTwinCertificateRequest>,
) -> Result<Response<IssueTwinCertificateReply>, Status> {
over_the_tunnel(&request, "IssueTwinCertificate")?;
let who = self
.authorize_admin_identity(&request, "", "certify a replication reader")
.await?;
let root = self.data_root.clone().ok_or_else(no_data_root)?;
let req = request.into_inner();
let is_twin = self.twin.is_twin();
let principal = req.principal.clone();
let issued = tokio::task::spawn_blocking(move || {
crate::twin_tls::issue(&root, &req.principal, &req.csr_pem, is_twin)
})
.await
.map_err(|e| internal_err(format!("the issue task: {e}")))?
.map_err(|e| refusal(tonic::Code::FailedPrecondition, e))?;
log::warn!(
"replication issuer: certified `{principal}` for the door on {}:{} until {} (asked by {})",
issued.hosts.join(","),
issued.port,
issued.not_after,
who.as_ref().map(|i| i.subject.as_str()).unwrap_or("anonymous")
);
Ok(Response::new(IssueTwinCertificateReply {
cert_pem: issued.cert_pem,
issuer_pem: issued.issuer_pem,
port: u32::from(issued.port),
hosts: issued.hosts,
not_after: issued.not_after,
}))
}
/// ★ **`InstallTwinCertificate`** — the four checks, the CA pinned. A twin
/// follows over it from its next cycle. Admin. With `repoint` (purpose
/// `twin`, on a twin): the DELIBERATE move to a main recharged from this twin
/// after its primary died — every check but the pin, the role record names
/// `primary`, the posture switches live, and the running follow dials the
/// new door from its next cycle.
async fn install_twin_certificate(
&self,
request: Request<InstallTwinCertificateRequest>,
) -> Result<Response<InstallTwinCertificateReply>, Status> {
over_the_tunnel(&request, "InstallTwinCertificate")?;
let who = self
.authorize_admin_identity(&request, "", "install a replication credential")
.await?;
let root = self.data_root.clone().ok_or_else(no_data_root)?;
let req = request.into_inner();
let purpose = crate::twin_tls::Purpose::parse(&req.purpose)
.map_err(|e| refusal(tonic::Code::InvalidArgument, e))?;
let is_twin = self.twin.is_twin();
if req.repoint {
if purpose != crate::twin_tls::Purpose::Twin {
return Err(coded(
&ec::TWIN_PAIRING_REFUSED,
tonic::Code::InvalidArgument,
"only a twin credential is re-pointed",
));
}
let now = crate::clock::now_unix_secs();
let (got, record) = tokio::task::spawn_blocking(move || {
crate::twin_tls::repoint(
&root,
is_twin,
&req.primary,
&req.cert_pem,
&req.issuer_pem,
req.port,
&req.hosts,
now,
)
})
.await
.map_err(|e| internal_err(format!("the re-point task: {e}")))?
.map_err(|e| refusal(tonic::Code::FailedPrecondition, e))?;
log::warn!(
"RE-POINTED: this twin now follows `{}` and pins its issuer (asked by {}); {}",
record.role.primary().unwrap_or("?"),
who.as_ref()
.map(|i| i.subject.as_str())
.unwrap_or("anonymous"),
record.reason
);
self.twin.become_twin(record.role);
self.twin.wake_follow();
return Ok(Response::new(InstallTwinCertificateReply {
days_left: got.days_left,
renewal: false,
}));
}
let got = tokio::task::spawn_blocking(move || {
crate::twin_tls::install(
&root,
purpose,
is_twin,
&req.cert_pem,
&req.issuer_pem,
req.port,
&req.hosts,
)
})
.await
.map_err(|e| internal_err(format!("the install task: {e}")))?
.map_err(|e| refusal(tonic::Code::FailedPrecondition, e))?;
// The follow that waited for this credential (`holger-twin-13`) runs now, not at its
// next tick: the pairing's `follow` step waits on its first whole cycle.
self.twin.wake_follow();
Ok(Response::new(InstallTwinCertificateReply {
days_left: got.days_left,
renewal: got.renewal,
}))
}
/// ★ **`Recharge`** — gunnar's `Bootstrap.Restore`, pull-shaped: this main
/// pulls its twin's estate and every repository over the twin's replication
/// door, with the `recharge` credential the lane certified. Refused unless the
/// box is FRESH (`holger-twin-16`; a re-image makes it so) or a recharge is
/// already under way here (it RESUMES — the estate pull and the mirror are
/// content-id idempotent). From the first byte the box serves nothing
/// (`holger-recharging`, `holger-twin-15`) until the pull is whole; then
/// `holger-recharged` is written once and it serves. A partial pull stays
/// refusing and resumable. Admin; a twin is never recharged.
async fn recharge(
&self,
request: Request<RechargeRequest>,
) -> Result<Response<RechargeReply>, Status> {
over_the_tunnel(&request, "Recharge")?;
self.authorize_admin_identity(&request, "", "recharge this box")
.await?;
if self.twin.is_twin() {
return Err(coded(
&ec::TWIN_PAIRING_REFUSED,
tonic::Code::FailedPrecondition,
"this box is a twin: a twin is never recharged, it recharges others",
));
}
let Some(follower) = self.follower.clone() else {
return Err(coded(&ec::SERVER_CAPABILITY_NOT_CONFIGURED, tonic::Code::FailedPrecondition, "this process holds no follower (a bare service), so it has nothing to recharge into"));
};
// ONE recharge at a time on this box (lane G11): the ONE box's pulled order and the
// front's autofill may both reach a re-laid main. The second waits for the first and
// then reads the marker it left: recharged once, refused by name (the lane reads the
// main again and calls that AlreadyRecharged).
static ONE_AT_A_TIME: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(());
let _one = ONE_AT_A_TIME.lock().await;
let root = follower.root().to_path_buf();
if root.join(crate::twin_follow::RECHARGED).exists() {
return Err(coded(
&ec::TWIN_RECHARGE_REFUSED,
tonic::Code::FailedPrecondition,
"this box was already recharged (holger-recharged); a recharge happens once",
));
}
let restoring = root.join(crate::twin_follow::RECHARGING).exists();
if !restoring {
let f = follower.clone();
if let Some(held) = tokio::task::spawn_blocking(move || f.recharge_blocker())
.await
.map_err(|e| internal_err(format!("the freshness walk: {e}")))?
{
return Err(coded(&ec::TWIN_RECHARGE_REFUSED, tonic::Code::FailedPrecondition, format!(
"this box is not fresh — it holds state of its own ({}), which a recharge would bury under its twin's. Making a box fresh is a RE-IMAGE",
held.display()
)));
}
}
let cred = crate::twin_tls::load(&root, crate::twin_tls::Purpose::Recharge)
.map_err(|e| refusal(tonic::Code::FailedPrecondition, e))?;
let Some(cred) = cred else {
return Err(coded(&ec::TWIN_NOT_CERTIFIED, tonic::Code::FailedPrecondition, "no `recharge` credential is installed: TwinCsr(recharge) here, IssueTwinCertificate(holger-recharge) on the twin, InstallTwinCertificate(recharge) here — then Recharge"));
};
let from = cred.doors().first().cloned().unwrap_or_default();
// From here the box serves nothing until the pull is whole — across a
// restart too (the marker), so a box that died mid-recharge never comes
// up serving half of itself.
std::fs::write(
root.join(crate::twin_follow::RECHARGING),
format!("recharging from {from}\n"),
)
.map_err(|e| internal_err(format!("{}: {e}", crate::twin_follow::RECHARGING)))?;
// The box keeps naming the source it was TOLD (`recharge_from`, the
// twin's control plane) when it was told one; else the door it dials.
if self.twin.awaiting_from().is_none() {
self.twin.await_recharge(&from);
}
let r = follower.recharge(&from).await.map_err(internal_err)?;
if r.failed.is_empty() {
crate::twin_follow::mark_recharged(&root, &from, &r).map_err(internal_err)?;
let _ = std::fs::remove_file(root.join(crate::twin_follow::RECHARGING));
// The boot hold is released by the recharge alone (Gunnar's pattern).
crate::twin_follow::release(&root);
self.twin.recharged();
log::warn!(
"RECHARGED from the twin's door {from}: {} — this box serves",
r.estate
);
} else {
log::warn!("recharge from {from} is NOT whole ({} failed); this box keeps serving nothing and the next Recharge resumes: {}", r.failed.len(), r.failed.join("; "));
}
Ok(Response::new(RechargeReply {
estate: r.estate,
artifacts: r.artifacts,
failed: r.failed,
}))
}
/// ★ **`RechargeStatus`** — gunnar's `Bootstrap.Status`: what the lane asks
/// the MAIN before a byte moves. Admin-read; answered while serving nothing.
async fn recharge_status(
&self,
request: Request<RechargeStatusRequest>,
) -> Result<Response<RechargeStatusReply>, Status> {
self.authorize_admin_read(&request, "").await?;
let Some(follower) = self.follower.clone() else {
return Err(coded(
&ec::SERVER_CAPABILITY_NOT_CONFIGURED,
tonic::Code::FailedPrecondition,
"this process holds no follower (a bare service)",
));
};
let root = follower.root().to_path_buf();
let recharged = root.join(crate::twin_follow::RECHARGED).exists();
let restoring = root.join(crate::twin_follow::RECHARGING).exists();
let f = follower.clone();
let held = tokio::task::spawn_blocking(move || f.recharge_blocker())
.await
.map_err(|e| internal_err(format!("the freshness walk: {e}")))?;
let fresh = held.is_none() && !recharged && !self.twin.is_twin();
let state = if recharged {
"recharged"
} else if restoring {
"restoring"
} else if fresh {
"fresh"
} else {
"holding"
};
Ok(Response::new(RechargeStatusReply {
state: state.into(),
fresh,
restoring,
recharged,
serving: self.twin.serving(),
held: held.map(|p| p.display().to_string()).unwrap_or_default(),
awaiting_from: self.twin.awaiting_from().unwrap_or_default(),
control: follower.control().to_string(),
}))
}
/// ★ **`FencePrimary`** — the old primary is told, by the box that runs a pulled
/// `fence_primary`, to refuse every write from now on (Gunnar's fence pattern). Admin
/// by identity: a box already fenced answers a second fence. Refused on a twin.
async fn fence_primary(
&self,
request: Request<FencePrimaryRequest>,
) -> Result<Response<FencePrimaryReply>, Status> {
over_the_tunnel(&request, "FencePrimary")?;
self.authorize_admin_identity(&request, "", "fence this primary")
.await?;
if self.twin.is_twin() {
return Err(coded(
&ec::TWIN_REFUSES_WRITES,
tonic::Code::FailedPrecondition,
"this box is a twin: a twin takes no writes and is never fenced",
));
}
let epoch = request.into_inner().epoch;
self.twin.fence(epoch);
let at = self.twin.fenced().unwrap_or(epoch);
log::warn!("FencePrimary: this primary is FENCED at epoch {at} by a peer's pulled order: every door refuses writes");
Ok(Response::new(FencePrimaryReply {
fenced: true,
epoch: at,
refusal: self.twin.write_refusal().unwrap_or_default(),
}))
}
/// ★ **`TwinStatus`** — what the lane asks the TWIN before a byte moves: what
/// it holds, and whether its last cycle that reached its primary was whole.
/// Admin-read.
async fn twin_status(
&self,
request: Request<TwinStatusRequest>,
) -> Result<Response<TwinStatusReply>, Status> {
self.authorize_admin_read(&request, "").await?;
let Some(follower) = self.follower.clone() else {
return Err(coded(
&ec::SERVER_CAPABILITY_NOT_CONFIGURED,
tonic::Code::FailedPrecondition,
"this process holds no follower (a bare service)",
));
};
let root = follower.root().to_path_buf();
let f = follower.clone();
let (estate_files, artifacts) = tokio::task::spawn_blocking(move || {
// The job queue and ledger every START lays down are not holdings.
let files = crate::estate_wire::manifest(&root)
.map(|m| {
m.files
.iter()
.filter(|f| {
!crate::twin::START_SCAFFOLDING
.iter()
.any(|s| f.path.starts_with(&format!("{s}/")))
})
.count() as u64
})
.unwrap_or(0);
(files, f.artifact_count())
})
.await
.map_err(|e| internal_err(format!("the holdings walk: {e}")))?;
let role = self.twin.role();
let (complete, why, last_whole) = if role.is_twin() {
self.twin.completeness()
} else {
(false, "this box is not a twin".to_string(), 0)
};
let heard = self.twin.hearing(crate::clock::now_unix_ms());
Ok(Response::new(TwinStatusReply {
role: if role.is_twin() { "twin" } else { "primary" }.into(),
primary: role.primary().unwrap_or_default().into(),
estate_files,
artifacts,
complete,
why,
last_whole_unix: last_whole,
follow: self.twin.follow(),
primary_answer_unix_ms: heard.last_answer_unix_ms,
primary_answer_ago_ms: heard.answer_ago_ms,
primary_head: heard.primary_head,
follow_state: heard.state,
}))
}
/// ★ **`AdminSnapshot`** — the admin view (`crate::admin_view`): every disk
/// and volume, the twin, the growth and its journal, the clock, plugins and
/// monetize, the audit log, the installer's log door. An ADMIN's read, and
/// nobody else's: `authorize_admin_identity` first — a caller whose
/// credential does not validate is `Unauthenticated`, a validated caller
/// below Admin is `PermissionDenied`, both BY NAME and before one byte of
/// the machine is measured. Not refused on a twin, a growing box or a
/// read-only server: measuring changes nothing, and those are exactly the
/// boxes an administrator needs to look at.
async fn admin_snapshot(
&self,
request: Request<AdminSnapshotRequest>,
) -> Result<Response<AdminSnapshotReply>, Status> {
let who = self
.authorize_admin_identity(&request, "", "read the admin view")
.await?;
let lines = crate::admin_view::log_lines(request.get_ref().log_lines);
let now_ms = crate::clock::now_unix_ms();
let volumes = crate::admin_view::volumes(self.data_root.as_deref(), &self.storage_places);
let (data_root, posture, grow, audit) = (
self.data_root.clone(),
Arc::clone(&self.twin),
self.grow.clone(),
Arc::clone(&self.audit),
);
let plugins = crate::admin_view::plugins(self);
let (places, upstreams, plugins_for_tenants) = (
Arc::clone(&self.storage_places),
Arc::clone(&self.upstreams),
plugins.clone(),
);
#[cfg(feature = "jobs")]
let jobs = self.job_queue.clone();
// statvfs, sysfs, the credential files, the capacity ladder, the audit
// directory, the meter's walk and the job ledger all block: off the
// executor, as `capacity` is.
let (disks, twin, growth, audit, tenants, queue) = tokio::task::spawn_blocking(move || {
#[cfg(feature = "jobs")]
let queue = crate::admin_view::queue(jobs.as_deref());
#[cfg(not(feature = "jobs"))]
let queue = crate::admin_view::queue_absent();
(
crate::admin_view::disks(&volumes),
crate::admin_view::twin(
&posture,
data_root.as_deref(),
grow.as_deref(),
now_ms.div_euclid(1000),
),
crate::admin_view::growth(grow.as_deref(), data_root.as_deref()),
crate::admin_view::audit(audit.as_ref()),
crate::admin_view::tenants(&places, &upstreams, &plugins_for_tenants),
queue,
)
})
.await
.map_err(|e| internal_err(format!("the admin view's measuring task: {e}")))?;
super::functional_status(
"holger-grpc/admin_snapshot",
"admin_view_served",
true,
&format!(
"{} mounts, {} block devices, for {}",
disks.mounts.len(),
disks.block_devices.len(),
who.as_ref()
.map(|i| i.subject.as_str())
.unwrap_or("anonymous")
),
);
Ok(Response::new(AdminSnapshotReply {
taken_unix_ms: now_ms,
host: crate::admin_view::host(),
version: crate::VERSION_LINE.into(),
uptime_seconds: self.start_time.elapsed().as_secs(),
role: twin.role.clone(),
disks: Some(disks),
twin: Some(twin),
growth: Some(growth),
clock: Some(crate::admin_view::clock()),
plugins: Some(plugins),
audit: Some(audit),
logs: Some(crate::admin_view::logs(lines)),
payments: crate::admin_view::payments(),
disk_speed: Some(crate::admin_view::disk_speed()),
tenants,
queue: Some(queue),
downstreams: crate::admin_view::downstreams(self.peers.rows(
self.grow_hold.boot(),
self.grow_hold.head(),
now_ms,
)),
}))
}
/// ★ **`Topology`** — the estate board (lane E3, Gunnar's `Admin.Topology`
/// pattern): this box first, then every box it holds evidence of, each with its
/// state, the twin's lag and the WireGuard overlay address, and the board drawn
/// in the same answer (`crate::topology`). Admin-read, Auditor too: the rows name
/// addresses, never a secret. Answered by a twin and by a held main alike.
async fn topology(
&self,
request: Request<TopologyRequest>,
) -> Result<Response<TopologyReply>, Status> {
self.authorize_admin_identity_for(&request, "", "read the estate board", true)
.await?;
let sources = self.topology_sources();
let facts = tokio::task::spawn_blocking(move || sources.gather())
.await
.map_err(|e| internal_err(format!("the estate board's gathering task: {e}")))?;
let reply = crate::topology::board(&facts);
super::functional_status(
"holger-grpc/topology",
"estate_board_served",
true,
&format!("{} boxes, {} links", reply.boxes.len(), reply.links.len()),
);
Ok(Response::new(reply))
}
/// ★ **`FlowSvg`** — the artifact flow (lane HOL-SVG-RON): the RON's graph, the estate
/// and the counters, drawn by [`crate::flow::live`], the one function `GET /-/flow.svg`
/// calls too. Admin-read, like the estate board.
async fn flow_svg(
&self,
request: Request<FlowSvgRequest>,
) -> Result<Response<FlowSvgReply>, Status> {
self.authorize_admin_identity_for(&request, "", "read the artifact flow", true)
.await?;
let Some(sources) = self.flow_sources() else {
return Err(coded(
&ec::SERVER_CAPABILITY_NOT_CONFIGURED,
tonic::Code::NotFound,
"the artifact flow is off on this holger (`flow: (enabled: false)`)",
));
};
// The zoom, as `?open=` names it: empty is level 1, `all` is the full RON.
let open = crate::flow::Open::parse(&request.get_ref().open);
let graph = tokio::task::spawn_blocking(move || crate::flow::live(&sources).zoom(&open))
.await
.map_err(|e| internal_err(format!("the artifact flow's gathering task: {e}")))?;
super::functional_status(
"holger-grpc/flow_svg",
"flow_served",
true,
&format!("{} nodes, {} edges", graph.nodes.len(), graph.edges.len()),
);
let (svg, links) = graph.render_with_links();
Ok(Response::new(FlowSvgReply {
svg,
graph_json: graph.to_json(),
links: links
.into_iter()
.map(|l| super::holger_proto::FlowLink {
href: l.href,
label: l.label,
left: l.left,
top: l.top,
width: l.width,
height: l.height,
})
.collect(),
}))
}
/// ★ **`SecuritySummary`** — the Security page (lane HOL-SEC): the intake chain, the
/// sources and scanners, what waits for approval, what is held or refused and why, who
/// may read, the sync signatures and the last security events, every word the server's
/// (`crate::security_view`). An Admin, a Repository Admin, a Security Approver or an
/// Auditor reads it.
async fn security_summary(
&self,
request: Request<SecuritySummaryRequest>,
) -> Result<Response<SecuritySummaryReply>, Status> {
let identity = self
.authenticate(&request, "read the Security page")
.await?;
crate::security_view::authorize_read(&self.auth_config, identity.as_ref()).map_err(
|m| {
coded(
&ec::AUTH_PERMISSION_DENIED,
tonic::Code::PermissionDenied,
m,
)
},
)?;
let sources = self.security_sources();
let facts = tokio::task::spawn_blocking(move || sources.gather())
.await
.map_err(|e| internal_err(format!("the Security page's gathering task: {e}")))?;
let reply = crate::security_view::summary(&facts);
super::functional_status(
"holger-grpc/security_summary",
"security_summary_served",
true,
&format!("{} sections", reply.sections.len()),
);
Ok(Response::new(reply))
}
/// ★ Import a streamed intel bundle (lane W4, [`crate::import_door`]):
/// Repository Admin or Security Approver; verify, floor, audit first, then
/// the live advisory database the next scan reads.
async fn import_intel_bundle(
&self,
request: Request<tonic::Streaming<ImportBundleChunk>>,
) -> Result<Response<ImportIntelBundleReply>, Status> {
#[cfg(feature = "intake")]
{
use crate::import_door::ImportKind;
let source = self.audit_peer(&request);
let (principal, door, staged) = self.receive_import(request, ImportKind::Intel).await?;
let r = tokio::task::spawn_blocking(move || {
let r = door.import_intel(&staged.path, None, &principal, &source);
drop(staged);
r
})
.await
.map_err(|e| internal_err(format!("the import task: {e}")))?
.map_err(|e| super::import_status(&e))?;
let message = format!(
"IMPORTED ({}) {} seq {} dated {}: {} advisories",
r.outcome, r.bundle_id, r.sequence, r.snapshot_at, r.advisories
);
super::functional_status(
"holger-grpc/import_intel_bundle",
"intel_imported",
true,
&message,
);
Ok(Response::new(ImportIntelBundleReply {
outcome: r.outcome,
bundle_id: r.bundle_id,
sequence: r.sequence,
snapshot_at: r.snapshot_at,
advisories: r.advisories,
message,
}))
}
#[cfg(not(feature = "intake"))]
{
let _ = request;
Err(coded(
&ec::SERVER_CAPABILITY_NOT_CONFIGURED,
tonic::Code::Unimplemented,
"this build carries no intel import (no supply chain)",
))
}
}
/// ★ Import a streamed update or airgap bundle (lane W4): every artifact
/// lands AVAILABLE through the intake chain, never into a raw directory.
async fn import_bundle(
&self,
request: Request<tonic::Streaming<ImportBundleChunk>>,
) -> Result<Response<ImportBundleReply>, Status> {
#[cfg(feature = "intake")]
{
use crate::import_door::ImportKind;
let source = self.audit_peer(&request);
let (principal, door, staged) =
self.receive_import(request, ImportKind::Bundle).await?;
let r = tokio::task::spawn_blocking(move || {
let r = door.import_bundle(
&staged.path,
None,
&Default::default(),
&principal,
&source,
);
drop(staged);
r
})
.await
.map_err(|e| internal_err(format!("the import task: {e}")))?
.map_err(|e| super::import_status(&e))?;
let message = format!(
"IMPORTED ({}) {} seq {}: {} artifact(s) AVAILABLE",
r.outcome,
r.bundle_id,
r.sequence,
r.artifacts.len()
);
super::functional_status(
"holger-grpc/import_bundle",
"bundle_imported",
true,
&message,
);
Ok(Response::new(ImportBundleReply {
outcome: r.outcome,
bundle_id: r.bundle_id,
sequence: r.sequence,
kind: r.kind,
artifacts: r
.artifacts
.into_iter()
.map(|a| ImportedArtifactRow {
artifact: a.artifact,
sha256: a.sha256,
state: a.state,
repository: a.repository,
outcome: a.outcome,
})
.collect(),
message,
}))
}
#[cfg(not(feature = "intake"))]
{
let _ = request;
Err(coded(
&ec::SERVER_CAPABILITY_NOT_CONFIGURED,
tonic::Code::Unimplemented,
"this build carries no bundle import (no supply chain)",
))
}
}
async fn capacity(
&self,
request: Request<CapacityRequest>,
) -> Result<Response<CapacityReply>, Status> {
let _identity = self.authorize_admin_read(&request, "").await?;
let Some(root) = self.data_root.clone() else {
super::functional_status(
"holger-grpc/capacity",
"capacity_refused",
false,
crate::metering::NO_FILESYSTEM_SOURCE,
);
return Err(no_data_root());
};
// `statvfs` is a syscall and `archive_info` reads archive indexes: both
// block, so they run off the executor, as the meters do.
let repos = self.routes.all_repos();
// The growth verdict rides every answer (edda appliance-grow's ladder over
// this machine's facts): "yes, and this is the last capacity" is a
// different sale from "yes, and it grows".
let measured = tokio::task::spawn_blocking(move || {
crate::capacity::measure(&root, &repos).map(|cap| {
let growth = crate::grow::capacity_growth(&cap.path);
(cap, growth)
})
})
.await
.map_err(|e| internal_err(format!("capacity task failed: {e}")))?;
match measured {
Ok((cap, (growth, growth_reason, data_set, hotplug))) => {
super::functional_status(
"holger-grpc/capacity",
"capacity_measured",
true,
&format!(
"{} total, {} servable, {} used by holger under {}",
cap.total_bytes,
cap.available_bytes,
cap.used_by_holger_bytes,
cap.path.display()
),
);
Ok(Response::new(CapacityReply {
total_bytes: cap.total_bytes,
available_bytes: cap.available_bytes,
used_by_holger_bytes: cap.used_by_holger_bytes,
path: cap.path.display().to_string(),
growth,
growth_reason,
data_set,
hotplug,
cpu_cores: cap.cpu_cores,
ram_bytes: cap.ram_bytes,
}))
}
Err(reason) => {
super::functional_status(
"holger-grpc/capacity",
"capacity_refused",
false,
&reason,
);
Err(coded(
&ec::SERVER_UNAVAILABLE,
tonic::Code::Unavailable,
reason,
))
}
}
}
}
// ─── PromotionService (DEV → PROD) ───────────────────────────────────
//
// The one write that is ADMIN-gated (fail-closed) rather than writer-gated:
// promotion moves an artifact from a mutable DEV repo into the immutable,
// content-addressed PROD store. Bytes are read from the source backend,
// content-addressed (blake3), and SEALED into the store — where a promoted
// coordinate is immutable (a re-promote is refused with `already_exists`). Every
// promotion (success or refusal) is audited under `AuditAction::Promote`.
#[cfg(feature = "intake")]
#[tonic::async_trait]
impl PromotionService for Arc<HolgerGrpc> {
async fn promote_artifact(
&self,
request: Request<PromoteArtifactRequest>,
) -> Result<Response<PromoteArtifactResponse>, Status> {
let source_ip = self.audit_peer(&request);
// ADMIN gate, fail-closed, SCOPED to the promotion DESTINATION repo (a
// per-repo ACL can name who may promote INTO a given PROD shelf). A
// non-admin (or unauthenticated, when policy is set) caller is refused
// here, before any store access.
let to_repo = request.get_ref().to_repo.clone();
let identity = self.authorize_admin(&request, &to_repo).await?;
let ident = identity
.as_ref()
.map(|id| id.subject.clone())
.unwrap_or_else(|| "anonymous".to_string());
// …and the SOURCE end, against the same identity. Promotion reads bytes out
// of `from_repo` and writes them somewhere the caller controls, so admin on
// the destination alone made this an exfiltration primitive: one RPC naming
// any repository as the source, and its bytes land on a shelf the caller can
// read. Both ends are authorized, or the ACL is decoration.
let from_repo = request.get_ref().from_repo.clone();
if let Err(status) = self.authorize_read_repo(identity.as_ref(), &from_repo) {
self.record_audit(
AuditEvent::new(
&ident,
AuditAction::Promote,
&to_repo,
"",
&source_ip,
403,
0,
)
.with_detail(format!(
"denied read access to promotion source repo '{from_repo}'"
)),
);
super::functional_status(
"holger-grpc/promote_artifact",
"artifact_promoted",
false,
"denied on the source repository",
);
return Err(status);
}
// The intake chain's repositories are moved only by its transitions: a
// promotion naming one would seal past every state before SEALED. Nor does a
// promotion write where a raw write is refused (SEC-REVIEW 2026-11-01 P9).
#[cfg(feature = "intake")]
if let Some(refusal) = self
.intake
.as_ref()
.and_then(|c| c.promotion_refusal(&from_repo, &to_repo))
{
self.record_audit(
AuditEvent::new(
&ident,
AuditAction::Promote,
&to_repo,
"",
&source_ip,
403,
0,
)
.with_detail(refusal.tagged()),
);
return Err(coded(
refusal.code(),
tonic::Code::FailedPrecondition,
&refusal,
));
}
let engine = match &self.promotion {
Some(e) => e.clone(),
None => {
return Err(coded(
&ec::SERVER_CAPABILITY_NOT_CONFIGURED,
tonic::Code::FailedPrecondition,
"promotion is not configured on this server",
));
}
};
let req = request.into_inner();
let id_proto = req.artifact.ok_or_else(|| {
coded(
&ec::SERVER_BAD_REQUEST,
tonic::Code::InvalidArgument,
"Missing artifact ID",
)
})?;
let label = artifact_label(&id_proto);
let trait_id = traits::ArtifactId {
namespace: if id_proto.namespace.is_empty() {
None
} else {
Some(id_proto.namespace.clone())
},
name: id_proto.name.clone(),
version: id_proto.version.clone(),
};
// ── THE ASYNC SHAPE (stinky 01 §1.3) ────────────────────────────────
//
// `async_job` unset is the ORIGINAL path below, unchanged: fetch, gate,
// seal, answer with the content address. Set, the promotion is ADMITTED
// to the durable queue and this RPC answers at once with the job id —
// which is the whole point, because a 1 GiB promote used to hold the
// caller for the entire fetch+blake3+write.
//
// The gates are NOT skipped: they run on the worker, before the seal
// (`jobs::ServerJobRunner`). What runs HERE, before admission, is the
// property requirement — it needs the property store, costs no bytes,
// and a request that can never be promoted should be refused at the door
// rather than admitted and failed.
if req.async_job {
#[cfg(feature = "jobs")]
{
let Some(queue) = self.job_queue.clone() else {
return Err(coded(
&ec::SERVER_CAPABILITY_NOT_CONFIGURED,
tonic::Code::FailedPrecondition,
"this server has no job queue configured; retry without `async_job`",
));
};
if let Some((key, value)) = engine.required_property() {
let props = match &self.properties {
Some(store) => store.get(&req.from_repo, &trait_id),
None => crate::properties::PropertyMap::new(),
};
if !engine.property_requirement_met(&props) {
let detail = format!(
"promotion requires property '{key}{}' on the source artifact",
if value.is_empty() {
String::new()
} else {
format!("={value}")
}
);
return Err(coded(
&ec::PROMOTE_PROPERTY_REQUIRED,
tonic::Code::FailedPrecondition,
detail,
));
}
}
let job = crate::jobs::Job::Promote {
from_repo: req.from_repo.clone(),
to_repo: req.to_repo.clone(),
artifact: crate::jobs::JobArtifact {
namespace: id_proto.namespace.clone(),
name: id_proto.name.clone(),
version: id_proto.version.clone(),
},
requested_by: ident.clone(),
};
let job_id = queue
.enqueue(job)
.map_err(|e| internal_err(format!("could not admit the promotion: {e:#}")))?;
self.record_audit(
AuditEvent::new(
&ident,
AuditAction::Promote,
&req.to_repo,
&label,
&source_ip,
202,
0,
)
.with_detail(format!("promotion admitted as job {job_id}")),
);
super::functional_status(
"holger-grpc/promote_artifact",
"promotion_admitted_as_job",
true,
job_id.as_str(),
);
return Ok(Response::new(PromoteArtifactResponse {
success: true,
message: format!("promotion admitted as job {job_id}"),
content_address: String::new(),
promoted_by: ident.clone(),
job_id: job_id.0,
}));
}
#[cfg(not(feature = "jobs"))]
{
return Err(coded(
&ec::SERVER_CAPABILITY_NOT_CONFIGURED,
tonic::Code::FailedPrecondition,
"this binary was built without the `jobs` feature: there is no job \
queue to admit an asynchronous promotion to. Retry without \
`async_job`, or rebuild with `--features jobs`",
));
}
}
// Resolve the source (DEV) backend and read the bytes to promote. The read
// is synchronous (disk / znippy / upstream HTTP), so it runs on the blocking
// pool like every other backend call — the reactor must not carry it.
let source = self.get_repo(&req.from_repo)?;
let fetched = {
let source = source.clone();
let id = trait_id.clone();
tokio::task::spawn_blocking(move || source.fetch(&id)).await
};
let fetched = match fetched {
Ok(r) => r,
Err(e) => {
super::functional_status(
"holger-grpc/promote_artifact",
"artifact_promoted",
false,
"dispatch task failed",
);
return Err(internal_err(format!("promotion source fetch failed: {e}")));
}
};
let data = match fetched {
Ok(Some(d)) => d,
Ok(None) => {
self.record_audit(
AuditEvent::new(
&ident,
AuditAction::Promote,
&req.to_repo,
&label,
&source_ip,
404,
0,
)
.with_detail(format!(
"artifact not found in source repo '{}'",
req.from_repo
)),
);
// Same registry row as the local `holger-server promote` verb raises —
// ONE code for one verdict across both surfaces, not a per-surface twin.
return Err(coded(
&ec::PROMOTE_SOURCE_NOT_FOUND,
tonic::Code::NotFound,
format!(
"artifact '{}' not found in source repo '{}'",
label, req.from_repo
),
));
}
Err(e) => {
self.record_audit(
AuditEvent::new(
&ident,
AuditAction::Promote,
&req.to_repo,
&label,
&source_ip,
500,
0,
)
.with_detail(e.to_string()),
);
return Err(internal_err(&e));
}
};
// THE GATES (design §4) — run every configured gate over the promoted
// set at the boundary, BEFORE sealing.
//
// Three corrections against what stood here until 2026-09-20:
//
// 1. It called `gate_check`, not `gate_check_hop`, so the rung scoping
// the ladder declares was bypassed on the served path. It now goes
// through the hop, which is where `Stage::gates` is read.
// 2. It passed a coordinate and no bytes. `data` has been in scope since
// the source fetch above, so the artifact's DIGEST and its bytes are
// available and are now required fields of the context. modgunn's
// `Verdict::cache_key()` is `(blob_sha256, policy_version,
// advisory_db_rev)`; with an empty digest every artifact collapsed
// onto one cache entry.
// 3. The decision is audited whether it refused or allowed, and the
// detail says which gates ran. An allow with no gates configured is
// recorded as "none configured", not as a pass.
// What the gates decided, carried into the verification summary below so
// the evidence says which checks ran rather than only that something did.
let mut gate_summary = String::from("no gates configured");
if engine.has_gate() {
let ecosystem = crate::scanner::ecosystem_for_format(source.format());
let digest = crate::attest::ArtifactDigest::of_sha256(&data);
let attestations: Vec<crate::attest::Statement> = match &self.attestations {
Some(store) => store
.list(&digest)
.map(|v| v.into_iter().map(|a| a.statement).collect())
.unwrap_or_default(),
None => Vec::new(),
};
let set = [crate::promote::PromotedArtifact {
id: &trait_id,
ecosystem: &ecosystem,
digest,
bytes: &data,
attestations: &attestations,
}];
match engine.gate_check_hop(&req.from_repo, &req.to_repo, &set, &ident) {
Ok(decision) => {
gate_summary = decision.summary();
self.record_audit(
AuditEvent::new(
&ident,
AuditAction::Promote,
&req.to_repo,
&label,
&source_ip,
200,
0,
)
.with_detail(format!("promotion gates: {}", decision.summary())),
);
}
Err(e) => {
let detail = e.to_string();
self.record_audit(
AuditEvent::new(
&ident,
AuditAction::Promote,
&req.to_repo,
&label,
&source_ip,
412,
0,
)
.with_detail(detail.clone()),
);
super::functional_status(
"holger-grpc/promote_artifact",
"artifact_promoted",
false,
"blocked by policy gate",
);
return Err(coded(
&ec::PROMOTE_POLICY_BLOCKED,
tonic::Code::FailedPrecondition,
detail,
));
}
}
}
// Property-requirement gate (property-policy §): only promote when the
// SOURCE coordinate carries the required custom property (e.g. signed=true).
// Fail-closed — a requirement configured but no property store means we
// cannot verify it, so the promotion is refused rather than waved through.
if let Some((key, value)) = engine.required_property() {
let props = match &self.properties {
Some(store) => store.get(&req.from_repo, &trait_id),
None => crate::properties::PropertyMap::new(),
};
if !engine.property_requirement_met(&props) {
let detail = format!(
"promotion requires property '{key}{}' on the source artifact",
if value.is_empty() {
String::new()
} else {
format!("={value}")
}
);
self.record_audit(
AuditEvent::new(
&ident,
AuditAction::Promote,
&req.to_repo,
&label,
&source_ip,
412,
0,
)
.with_detail(detail.clone()),
);
super::functional_status(
"holger-grpc/promote_artifact",
"artifact_promoted",
false,
"blocked by property requirement",
);
return Err(coded(
&ec::PROMOTE_PROPERTY_REQUIRED,
tonic::Code::FailedPrecondition,
detail,
));
}
}
// Timestamp from the lib's existing wall-clock source (audit already uses
// it); seconds since the epoch for the record.
let promoted_at = crate::audit::now_nanos() / 1_000_000_000;
match engine.seal(
&req.from_repo,
&req.to_repo,
&trait_id,
&data,
&ident,
promoted_at,
) {
Ok(record) => {
// The debug-info cache follows the new bytes the same minute.
self.record_audit(
AuditEvent::new(
&ident,
AuditAction::Promote,
&req.to_repo,
&label,
&source_ip,
200,
record.size,
)
.with_detail(format!(
"promoted from '{}' — content {}",
req.from_repo, record.content_address
)),
);
// ── EVIDENCE ────────────────────────────────────────────────
// One SLSA Verification Summary per promotion, over the whole
// promoted set, signed and filed through `AttestStore::accept`
// — the same door any other party's evidence goes through, so
// holger's own statement is subject-bound, enrolment-checked
// and cryptographically verified before it is written.
//
// This is what `attest.rs` was for. MEASURED at 542e296 it was
// 1458 lines with no constructor and no consumer anywhere in
// the shipped tree.
//
// Three outcomes, and all three are RECORDED. Writing nothing
// and saying nothing is the one thing that must not happen: an
// artifact with no evidence and an artifact whose evidence was
// never attempted are different facts.
let vsa_detail = self.emit_promotion_vsa(&label, &data, &gate_summary);
self.record_audit(
AuditEvent::new(
&ident,
AuditAction::Promote,
&req.to_repo,
&label,
&source_ip,
200,
0,
)
.with_detail(vsa_detail),
);
super::functional_status(
"holger-grpc/promote_artifact",
"artifact_promoted",
true,
&req.to_repo,
);
Ok(Response::new(PromoteArtifactResponse {
success: true,
message: format!("promoted '{}' to '{}'", label, req.to_repo),
content_address: record.content_address,
promoted_by: ident,
// Empty on the synchronous path: no job was admitted, and a
// fabricated id would name a row that does not exist.
job_id: String::new(),
}))
}
// Immutable: the coordinate is already sealed — refuse the overwrite.
Err(PromoteError::AlreadyPromoted) => {
self.record_audit(
AuditEvent::new(
&ident,
AuditAction::Promote,
&req.to_repo,
&label,
&source_ip,
409,
0,
)
.with_detail("already promoted — immutable, overwrite refused"),
);
super::functional_status(
"holger-grpc/promote_artifact",
"artifact_promoted",
false,
"already promoted (immutable)",
);
Err(coded(
&ec::PROMOTE_ALREADY_PROMOTED,
tonic::Code::AlreadyExists,
format!(
"'{}' is already promoted to '{}' (immutable)",
label, req.to_repo
),
))
}
// The TARGET does not exist. Named, not a generic 500: the caller asked
// for a repository this server does not serve, and the old code answered
// `success: true` because `to_repo` was only ever sanitized into a
// directory name, never resolved.
Err(e @ PromoteError::UnknownRepository { .. }) => {
self.record_audit(
AuditEvent::new(
&ident,
AuditAction::Promote,
&req.to_repo,
&label,
&source_ip,
404,
0,
)
.with_detail(e.to_string()),
);
super::functional_status(
"holger-grpc/promote_artifact",
"artifact_promoted",
false,
"target repository is not served",
);
Err(coded(
&ec::PROMOTE_UNKNOWN_REPOSITORY,
tonic::Code::NotFound,
e.to_string(),
))
}
Err(e) => {
self.record_audit(
AuditEvent::new(
&ident,
AuditAction::Promote,
&req.to_repo,
&label,
&source_ip,
500,
0,
)
.with_detail(e.to_string()),
);
Err(internal_err(&e))
}
}
}
/// **The intake chain's wire verb** (`promote::duties`): move one artifact
/// to its next state, or quarantine, release or block it, as the caller's
/// duties allow.
async fn advance_artifact(
&self,
request: Request<AdvanceArtifactRequest>,
) -> Result<Response<AdvanceArtifactResponse>, Status> {
use crate::promote::duties::{self, Action, AdvanceAsk};
let source = self.audit_peer(&request);
let identity = self.authorize_intake(&request, true).await?;
let Some(chain) = self.intake.clone() else {
return Err(coded(
&ec::SERVER_CAPABILITY_NOT_CONFIGURED,
tonic::Code::FailedPrecondition,
"this server declares no intake chain (`promotion.intake`)",
));
};
let req = request.into_inner();
let id = intake_id(req.artifact)?;
let action = match IntakeVerb::try_from(req.verb).unwrap_or(IntakeVerb::Advance) {
IntakeVerb::Advance => Action::Advance,
IntakeVerb::Quarantine => Action::Quarantine,
IntakeVerb::Release => Action::Release,
IntakeVerb::Block => Action::Block,
};
let to = if req.to_state.trim().is_empty() {
None
} else {
Some(
crate::promote::intake::ArtifactState::parse(&req.to_state)
.map_err(|e| intake_status(&e))?,
)
};
let evidence = req
.evidence
.into_iter()
.map(evidence_from_wire)
.collect::<Result<Vec<_>, Status>>()?;
let ask = AdvanceAsk {
id,
action,
to,
reason: req.reason,
evidence,
source,
};
let auth = Arc::clone(&self.auth_config);
let answer = tokio::task::spawn_blocking(move || {
duties::advance_as(&chain, &auth, identity.as_ref(), ask)
})
.await
.map_err(|e| internal_err(format!("the intake task: {e}")))?
.map_err(|e| intake_status(&e))?;
super::functional_status(
"holger-grpc/advance_artifact",
"artifact_advanced",
true,
&answer.message,
);
Ok(Response::new(AdvanceArtifactResponse {
from_state: answer.from.name().into(),
state: answer.state.name().into(),
approvals: answer.approvals,
approvals_required: answer.approvals_required,
evidence: answer.evidence.iter().map(evidence_to_wire).collect(),
message: answer.message,
sha256: answer.sha256,
actor: answer.actor,
}))
}
/// The state, history and evidence of one intake artifact. A Developer sees
/// only an AVAILABLE one.
async fn get_artifact_state(
&self,
request: Request<GetArtifactStateRequest>,
) -> Result<Response<GetArtifactStateResponse>, Status> {
let identity = self.authorize_intake(&request, false).await?;
let Some(chain) = self.intake.clone() else {
return Err(coded(
&ec::SERVER_CAPABILITY_NOT_CONFIGURED,
tonic::Code::FailedPrecondition,
"this server declares no intake chain (`promotion.intake`)",
));
};
let id = intake_id(request.into_inner().artifact)?;
let view =
crate::promote::duties::read_as(&chain, &self.auth_config, identity.as_ref(), &id)
.map_err(|e| intake_status(&e))?;
let last = view.history.last();
let source = last.map(|e| e.origin.clone()).unwrap_or_default();
let effective_policy = last
.and_then(|e| e.effective_policy.as_ref())
.and_then(|p| serde_json::to_string(p).ok())
.unwrap_or_default();
Ok(Response::new(GetArtifactStateResponse {
source,
effective_policy,
state: view.state.name().into(),
sha256: view.sha256,
importer: view.importer,
approvals: view.pending.len() as u32,
approvals_required: view.approvals_required,
history: view
.history
.iter()
.map(|e| IntakeTransition {
from_state: e.from.name().into(),
to_state: e.to.name().into(),
actor: e.actor.clone(),
at: e.at,
evidence: e.evidence.iter().map(evidence_to_wire).collect(),
policy: e.policy.clone(),
reason: e.reason.clone(),
})
.collect(),
pending_approvals: view
.pending
.iter()
.map(|v| evidence_to_wire(&v.evidence()))
.collect(),
}))
}
/// **The security dashboard's queues** (`promote::duties::list_as`): the
/// count of every queue and the rows of the one asked for. The read duty of
/// GetArtifactState decides each row.
async fn list_intake_artifacts(
&self,
request: Request<ListIntakeArtifactsRequest>,
) -> Result<Response<ListIntakeArtifactsResponse>, Status> {
use crate::promote::intake;
let identity = self.authorize_intake(&request, false).await?;
let Some(chain) = self.intake.clone() else {
return Err(coded(
&ec::SERVER_CAPABILITY_NOT_CONFIGURED,
tonic::Code::FailedPrecondition,
"this server declares no intake chain (`promotion.intake`)",
));
};
let req = request.into_inner();
let queue = match req.queue.trim() {
"" => None,
name => Some(intake::queue_named(name).ok_or_else(|| {
let known: Vec<&str> = intake::QUEUES.iter().map(|q| q.name).collect();
coded(
&ec::SERVER_BAD_REQUEST,
tonic::Code::InvalidArgument,
format!(
"'{name}' is not an intake queue (known: {})",
known.join(", ")
),
)
})?),
};
let limit = match req.limit {
0 => 200,
n => n.min(1_000) as usize,
};
let auth = Arc::clone(&self.auth_config);
let view = tokio::task::spawn_blocking(move || {
crate::promote::duties::list_as(&chain, &auth, identity.as_ref(), queue, limit)
})
.await
.map_err(|e| internal_err(format!("the intake task: {e}")))?;
Ok(Response::new(ListIntakeArtifactsResponse {
queues: view
.counts
.iter()
.map(|(q, n)| IntakeQueue {
name: q.name.into(),
title: q.title.into(),
states: q.states.iter().map(|s| s.name().into()).collect(),
count: *n,
})
.collect(),
artifacts: view
.rows
.iter()
.map(|r| IntakeArtifactRow {
artifact: Some(ArtifactId {
namespace: r.id.namespace.clone().unwrap_or_default(),
name: r.id.name.clone(),
version: r.id.version.clone(),
}),
state: r.status.state.name().into(),
sha256: r.status.sha256.clone(),
repository: r.status.last.repo.clone(),
actor: r.status.last.actor.clone(),
at: r.status.last.at,
held_from: r
.status
.held_from
.map(|s| s.name().to_string())
.unwrap_or_default(),
importer: r.importer.clone(),
approvals: r.approvals,
approvals_required: view.approvals_required,
policy: r.status.last.policy.clone(),
reason: r.status.last.reason.clone(),
})
.collect(),
truncated: view.truncated,
}))
}
/// **The scan evidence of one intake artifact** (`promote::duties::evidence_as`):
/// the latest record of the evidence store for its bytes.
async fn get_artifact_evidence(
&self,
request: Request<GetArtifactEvidenceRequest>,
) -> Result<Response<GetArtifactEvidenceResponse>, Status> {
let identity = self.authorize_intake(&request, false).await?;
let Some(chain) = self.intake.clone() else {
return Err(coded(
&ec::SERVER_CAPABILITY_NOT_CONFIGURED,
tonic::Code::FailedPrecondition,
"this server declares no intake chain (`promotion.intake`)",
));
};
let id = intake_id(request.into_inner().artifact)?;
let auth = Arc::clone(&self.auth_config);
let view = tokio::task::spawn_blocking(move || {
crate::promote::duties::evidence_as(&chain, &auth, identity.as_ref(), &id)
})
.await
.map_err(|e| internal_err(format!("the intake task: {e}")))?
.map_err(|e| intake_status(&e))?;
let hash = |h: &crate::scanner::InputHash| EvidenceHash {
alg: h.alg.clone(),
hex: h.hex.clone(),
};
let mut out = GetArtifactEvidenceResponse {
state: view.state.name().into(),
sha256: view.sha256,
policy: view.policy,
..Default::default()
};
if let Some(ev) = view.evidence {
out.scanned = true;
out.file_name = ev.subject.file_name.clone();
out.ecosystem = ev.subject.ecosystem.clone();
out.size = ev.subject.size;
out.hashes = ev.subject.inputs.iter().map(hash).collect();
out.verdict = ev.verdict.as_str().into();
out.reasons = ev.reasons.clone();
out.scanned_at = ev.scanned_at;
out.records = ev
.records
.iter()
.map(|r| EvidenceRecord {
scanner_id: r.scanner_id.clone(),
scanner_version: r.scanner_version.clone(),
kind: r.kind.as_str().into(),
verdict: r.verdict.as_str().into(),
findings: r
.findings
.iter()
.map(|f| EvidenceFinding {
id: f.id.clone(),
severity: serde_json::to_value(f.severity)
.ok()
.and_then(|v| v.as_str().map(str::to_string))
.unwrap_or_default(),
summary: f.summary.clone(),
})
.collect(),
inputs: r.inputs.iter().map(hash).collect(),
detail: r.detail.clone(),
required: r.required,
not_configured: r.is_not_configured(),
document_json: r
.document
.as_ref()
.map(|d| d.to_string())
.unwrap_or_default(),
})
.collect();
}
// W5: every stored record, signed, with why it is stale, and the SBOM link.
out.stored = view
.stored
.iter()
.map(|r| ArtifactEvidenceRecord {
id: r.id.clone(),
scanned_at: r.evidence.scanned_at,
verdict: r.evidence.verdict.as_str().into(),
reasons: r.evidence.reasons.clone(),
evidence_json: r.bytes.clone(),
envelope: r.envelope.clone().unwrap_or_default(),
stale: r.stale.clone().unwrap_or_default(),
})
.collect();
out.signer_cert = view.signer_cert;
out.sbom_link = view.sbom_link.unwrap_or_default();
Ok(Response::new(out))
}
/// **The dependency sets the caller may read** (`pull_intake::depset`, the code of
/// `GET /-/intake/sets`), newest first.
async fn list_dependency_sets(
&self,
request: Request<ListDependencySetsRequest>,
) -> Result<Response<ListDependencySetsResponse>, Status> {
let identity = self.authorize_intake(&request, false).await?;
let status = request.into_inner().status;
match self
.depsets(identity, String::new(), DepsetOp::List(status))
.await?
{
DepsetOut::List(sets, approvals_required) => {
Ok(Response::new(ListDependencySetsResponse {
sets,
approvals_required,
}))
}
_ => Err(internal_err("a dependency-set list answered as one set")),
}
}
/// **One dependency set** (the code of `GET /-/intake/set?id=`).
async fn get_dependency_set(
&self,
request: Request<GetDependencySetRequest>,
) -> Result<Response<DependencySet>, Status> {
let identity = self.authorize_intake(&request, false).await?;
let id = request.into_inner().id;
match self
.depsets(identity, String::new(), DepsetOp::Get(id))
.await?
{
DepsetOut::One(set) => Ok(Response::new(set)),
_ => Err(internal_err("a dependency set answered as a list")),
}
}
/// **Approve a dependency set as a whole** (the code of `POST /-/intake/set/approve`):
/// the caller's vote on every member waiting at POLICY_CHECKED.
async fn approve_dependency_set(
&self,
request: Request<JudgeDependencySetRequest>,
) -> Result<Response<JudgeDependencySetResponse>, Status> {
let identity = self.authorize_intake(&request, true).await?;
let peer = self.audit_peer(&request);
let r = request.into_inner();
match self
.depsets(
identity,
peer,
DepsetOp::Approve(r.id, r.revision, r.reason),
)
.await?
{
DepsetOut::Judged(a) => Ok(Response::new(a)),
_ => Err(internal_err("a set verb answered as a read")),
}
}
/// **Refuse a dependency set** (the code of `POST /-/intake/set/refuse`): every waiting
/// member BLOCKED with the reason.
async fn refuse_dependency_set(
&self,
request: Request<JudgeDependencySetRequest>,
) -> Result<Response<JudgeDependencySetResponse>, Status> {
let identity = self.authorize_intake(&request, true).await?;
let peer = self.audit_peer(&request);
let r = request.into_inner();
match self
.depsets(identity, peer, DepsetOp::Refuse(r.id, r.revision, r.reason))
.await?
{
DepsetOut::Judged(a) => Ok(Response::new(a)),
_ => Err(internal_err("a set verb answered as a read")),
}
}
}
/// One call on the dependency sets (lane HOL-ENOSPC-SETS): `(id, revision, reason)` for the verbs.
#[cfg(feature = "intake")]
// A build without `jobs` answers every set call Unimplemented and reads no field.
#[cfg_attr(not(feature = "jobs"), allow(dead_code))]
enum DepsetOp {
List(String),
Get(String),
Approve(String, String, String),
Refuse(String, String, String),
}
#[cfg(feature = "intake")]
// A build without `jobs` answers every set call Unimplemented and reads no field.
#[cfg_attr(not(feature = "jobs"), allow(dead_code))]
enum DepsetOut {
List(Vec<DependencySet>, u32),
One(DependencySet),
Judged(JudgeDependencySetResponse),
}
#[cfg(feature = "intake")]
impl HolgerGrpc {
/// The dependency-set calls, run where the HTTP doors run them (`pull_intake::depset`),
/// off the async runtime.
#[cfg(feature = "jobs")]
async fn depsets(
&self,
identity: Option<crate::auth::AuthIdentity>,
peer: String,
op: DepsetOp,
) -> Result<DepsetOut, Status> {
let Some(chain) = self.intake.clone() else {
return Err(coded(
&ec::SERVER_CAPABILITY_NOT_CONFIGURED,
tonic::Code::FailedPrecondition,
"this server declares no intake chain (`promotion.intake`)",
));
};
if chain.pull().is_none() {
return Err(coded(
&ec::SERVER_CAPABILITY_NOT_CONFIGURED,
tonic::Code::FailedPrecondition,
"this server pulls nothing through its intake chain (`pull_through`), so it has no dependency sets",
));
}
let auth = Arc::clone(&self.auth_config);
tokio::task::spawn_blocking(move || {
let Some(pull) = chain.pull() else {
return Err(internal_err("the pull-through went away"));
};
let who = identity.as_ref();
let judged = |r: Result<
crate::pull_intake::depset::SetAnswer,
crate::pull_intake::depset::SetError,
>| {
r.map(|a| {
DepsetOut::Judged(JudgeDependencySetResponse {
id: a.id,
status: a.status.name().into(),
moved: a.moved as u32,
message: a.message,
set: Some(depset_to_wire(&a.set)),
})
})
.map_err(|e| depset_status(&e))
};
match op {
DepsetOp::List(status) => Ok(DepsetOut::List(
pull.sets_as(&chain, &auth, who, Some(status.as_str()))
.iter()
.map(depset_to_wire)
.collect(),
chain.approvals_required(),
)),
DepsetOp::Get(id) => pull
.set_as(&chain, &auth, who, &id)
.map(|v| DepsetOut::One(depset_to_wire(&v)))
.map_err(|e| depset_status(&e)),
DepsetOp::Approve(id, revision, reason) => {
judged(pull.approve_set(&chain, &auth, who, &id, &revision, &reason, &peer))
}
DepsetOp::Refuse(id, revision, reason) => {
judged(pull.refuse_set(&chain, &auth, who, &id, &revision, &reason, &peer))
}
}
})
.await
.map_err(|e| internal_err(format!("the dependency-set task: {e}")))?
}
/// A build without the job queue pulls nothing through, so it has no dependency sets.
#[cfg(not(feature = "jobs"))]
async fn depsets(
&self,
_identity: Option<crate::auth::AuthIdentity>,
_peer: String,
_op: DepsetOp,
) -> Result<DepsetOut, Status> {
Err(coded(
&ec::SERVER_CAPABILITY_NOT_CONFIGURED,
tonic::Code::Unimplemented,
"this build runs no pull-through (the `jobs` feature), so it has no dependency sets",
))
}
}
/// A set verb's refusal: its status as a gRPC code, its tagged words as they are.
#[cfg(all(feature = "intake", feature = "jobs"))]
fn depset_status(e: &crate::pull_intake::depset::SetError) -> Status {
let code = match e.status {
400 => tonic::Code::InvalidArgument,
401 => tonic::Code::Unauthenticated,
403 => tonic::Code::PermissionDenied,
404 => tonic::Code::NotFound,
409 | 412 => tonic::Code::FailedPrecondition,
_ => tonic::Code::Internal,
};
refusal(code, &e.message)
}
#[cfg(all(feature = "intake", feature = "jobs"))]
fn depset_to_wire(v: &crate::pull_intake::depset::SetView) -> DependencySet {
let reasons = |l: &[(String, String)]| {
l.iter()
.map(|(name, why)| DependencySetReason {
name: name.clone(),
why: why.clone(),
})
.collect()
};
DependencySet {
id: v.id.clone(),
revision: v.revision.clone(),
source: v.source.clone(),
ecosystem: v.ecosystem.clone(),
root: v.root.clone(),
at: v.at,
status: v.status.name().into(),
summary: v.summary.clone(),
members: v
.members
.iter()
.map(|m| DependencySetMember {
name: m.name.clone(),
role: m.role.clone(),
via: m.via.clone(),
state: m.state.clone(),
sha256: m.sha256.clone(),
approvals: m.approvals,
approvers: m.approvers.clone(),
reason: m.reason.clone(),
warnings: m.warnings.clone(),
})
.collect(),
blockers: reasons(&v.blockers),
holds: reasons(&v.holds),
approvals_required: v.approvals_required,
approvers: v.approvers.clone(),
awaiting: v.awaiting as u32,
available: v.available as u32,
notes: v.notes.clone(),
}
}
#[cfg(feature = "intake")]
fn intake_id(id: Option<ArtifactId>) -> Result<traits::ArtifactId, Status> {
let id = id.ok_or_else(|| {
coded(
&ec::SERVER_BAD_REQUEST,
tonic::Code::InvalidArgument,
"Missing artifact ID",
)
})?;
Ok(traits::ArtifactId {
namespace: if id.namespace.is_empty() {
None
} else {
Some(id.namespace)
},
name: id.name,
version: id.version,
})
}
/// An intake refusal on the wire: its own code, the status its HTTP twin uses.
#[cfg(feature = "intake")]
fn intake_status(e: &crate::promote::intake::IntakeError) -> Status {
let code = match e.status() {
400 => tonic::Code::InvalidArgument,
403 => tonic::Code::PermissionDenied,
404 => tonic::Code::NotFound,
409 => tonic::Code::AlreadyExists,
412 => tonic::Code::FailedPrecondition,
_ => tonic::Code::Internal,
};
coded(e.code(), code, e)
}
#[cfg(feature = "intake")]
fn evidence_to_wire(e: &crate::promote::intake::Evidence) -> IntakeEvidence {
IntakeEvidence {
kind: e.kind.name().into(),
source: e.source.clone(),
verdict: match e.verdict {
crate::promote::intake::EvidenceVerdict::Pass => "pass",
crate::promote::intake::EvidenceVerdict::Warn => "warn",
crate::promote::intake::EvidenceVerdict::Fail => "fail",
}
.into(),
reference: e.reference.clone(),
}
}
#[cfg(feature = "intake")]
fn evidence_from_wire(e: IntakeEvidence) -> Result<crate::promote::intake::Evidence, Status> {
crate::promote::intake::Evidence::from_wire(&e.kind, &e.source, &e.verdict, &e.reference)
.map_err(|why| coded(&ec::SERVER_BAD_REQUEST, tonic::Code::InvalidArgument, why))
}
// ─── SearchService (cross-repo artifact discovery) ───────────────────
//
// A read-only aggregate over every configured backend's `list()` +
// `archive_files()`. Reads are open (anonymous) like the other read RPCs; the
// pure `search::search_repos` engine does the matching, so the same behaviour
// backs the HTTP `/-/search` door and the `holger-server search` CLI verb.
#[tonic::async_trait]
impl SearchService for Arc<HolgerGrpc> {
async fn search(
&self,
request: Request<SearchRequest>,
) -> Result<Response<SearchResponse>, Status> {
// Search is cross-repo, so it authorizes in TWO steps — and the second is
// the one that matters. Step 1 (below) authorizes the empty repo name, which
// establishes the identity and applies the caller's GLOBAL floor. Step 2
// (further down) restricts the repository SET to the ones that identity may
// read.
//
// Step 1 alone is what the door used to do, and it is not enough: a per-repo
// `Denied` — the entire purpose of `repo_roles` — says nothing about the
// global role, so a caller correctly refused a direct `FetchArtifact` on
// `classified` got its coordinates, versions, sizes and archive PATHS back
// from Search instead. A search that reaches further than a read is a read.
// Resolved per `authorize_catalog`, as `ListRepositories` is.
let identity = self.authorize_catalog(&request).await?;
let who = HolgerGrpc::read_principal(&identity);
let source_ip = self.audit_peer(&request);
let req = request.into_inner();
let query = search_query_from_proto(req);
let query_label = format!(
"name={} ns={} ver={} sum={} path={} props={}",
query.name.as_deref().unwrap_or("-"),
query.namespace.as_deref().unwrap_or("-"),
query.version.as_deref().unwrap_or("-"),
if query.checksum.is_some() { "yes" } else { "-" },
query.path.as_deref().unwrap_or("-"),
query.properties.len(),
);
// Step 2: the engine only ever sees repositories this identity may read, so
// a denied repo cannot contribute a hit, a path, or a hit COUNT. Filtering
// rather than refusing the whole RPC is deliberate — a 403 naming the repo
// would itself confirm that a repo the caller may not read exists. With the
// read gate off (the default) this is `all_repos()` verbatim.
//
// The engine is synchronous + may fetch bytes for checksum search, so run
// it off the async reactor. The property store (when configured) powers the
// `property` axis; without one, a property filter fails closed.
let repos = self.readable_repos(identity.as_ref());
#[cfg(feature = "intake")]
let props = self.properties.clone();
let results = tokio::task::spawn_blocking(move || {
#[cfg(feature = "intake")]
let lookup = props.as_deref().map(|p| p as &dyn search::PropertyLookup);
#[cfg(not(feature = "intake"))]
let lookup: Option<&dyn search::PropertyLookup> = None;
The file goes on; the console shows all of it.