server/lib/src/admission_grpc.rs
//! **The gRPC door's admission** — a tower layer around `RepositoryService`,
//! the one service that moves artifact bytes (`FetchArtifact`,
//! `StreamArtifact`, `PutArtifact`). The same [`DoorAdmission`] the HTTP door
//! asks, so one consumer has ONE share across both transports.
//!
//! The consumer is resolved the way the handlers resolve it: a paired
//! downstream by its certificate pin ([`crate::peers`]), else the principal the
//! credential validates to (bearer or mTLS CN, `auth::validate_request`), else
//! the anonymous day token of the client address. A credential that does not
//! validate is NOT refused here — the handler owns that answer — it is simply
//! not a principal, so it is admitted as the anonymous client it is.
//!
//! Admitted: the [`Ticket`] rides in the request's extensions; the handler
//! charges the bytes it moved to it and the share is released when the
//! response future completes. Refused: `UNAVAILABLE` with a `retry-after`
//! metadata entry, at once, before the handler runs.
use std::convert::Infallible;
use std::sync::Arc;
use std::task::{Context, Poll};
use tonic::codegen::http;
use tonic::codegen::{BoxFuture, Service};
use tonic::server::NamedService;
use tonic::transport::server::{TcpConnectInfo, TlsConnectInfo};
use crate::admission::{DoorAdmission, Ticket, Who, WorkKind};
/// What the layer needs to name a consumer.
pub struct GrpcGate {
pub door: Arc<DoorAdmission>,
pub visitors: Arc<crate::visitors::Visitors>,
pub peers: Arc<crate::peers::PeerBoard>,
pub auth: Arc<crate::auth::AuthConfig>,
/// The endpoint's `ron_trusted_proxies` (SEC-REVIEW F1): a trusted front's forwarded
/// client is the one admitted and charged.
pub trusted_proxies: Option<Arc<crate::listen::AllowList>>,
}
/// The work an RPC path is, or `None` for an RPC that moves no artifact bytes
/// (a listing) — those are not admitted, as before.
pub fn work_of(path: &str) -> Option<WorkKind> {
let method = path.rsplit('/').next().unwrap_or("");
match method {
"FetchArtifact" | "StreamArtifact" => Some(WorkKind::Download),
"PutArtifact" => Some(WorkKind::Upload),
_ => None,
}
}
/// What the layer reads off a request before any await — owned, so the
/// request body type need not be `Sync`.
struct Parts {
peer: Option<String>,
bearer: Option<String>,
client_cn: Option<String>,
token: edda_admit::DayToken,
}
impl GrpcGate {
fn parts<B>(&self, req: &http::Request<B>) -> Parts {
let tls = req.extensions().get::<TlsConnectInfo<TcpConnectInfo>>();
let certs = tls.and_then(|t| t.peer_certs());
let peer = certs
.as_ref()
.and_then(|c| {
c.first()
.and_then(|leaf| self.peers.downstream_by_cert(leaf.as_ref()))
})
.and_then(|idx| self.peers.downstreams().get(idx).map(|d| d.name.clone()));
let bearer = req
.headers()
.get("authorization")
.and_then(|v| v.to_str().ok())
.and_then(|v| v.strip_prefix("Bearer "))
.map(str::to_string);
let client_cn = certs.as_ref().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 addr = tls.and_then(|t| t.get_ref().remote_addr()).or_else(|| {
req.extensions()
.get::<TcpConnectInfo>()
.and_then(|t| t.remote_addr())
});
let xff = req
.headers()
.get_all("x-forwarded-for")
.iter()
.filter_map(|v| v.to_str().ok())
.rfind(|v| !v.trim().is_empty());
let ip_text = addr
.map(|a| a.ip().to_string())
.unwrap_or_else(|| "unknown".to_string());
let client =
crate::visitors::client_via(addr.map(|a| a.ip()), xff, self.trusted_proxies.as_deref())
.unwrap_or(crate::visitors::Client::Raw(&ip_text));
let token = self.door.day_token(client);
Parts {
peer,
bearer,
client_cn,
token,
}
}
async fn who(&self, p: Parts) -> Who {
if let Some(name) = p.peer {
return Who::Peer(name);
}
let principal = if p.bearer.is_some() || p.client_cn.is_some() {
crate::auth::validate_request(&self.auth, p.bearer.as_deref(), p.client_cn.as_deref())
.await
.ok()
.flatten()
.map(|i| i.subject)
} else {
None
};
Who::resolve(None, principal.as_deref(), p.token)
}
}
/// The layer: wrap a generated service server.
pub fn admitted<S>(inner: S, gate: Arc<GrpcGate>) -> Admitted<S> {
Admitted { inner, gate }
}
#[derive(Clone)]
pub struct Admitted<S> {
inner: S,
gate: Arc<GrpcGate>,
}
impl<S: NamedService> NamedService for Admitted<S> {
const NAME: &'static str = S::NAME;
}
/// `UNAVAILABLE` + `retry-after`, as an HTTP response tonic hands back.
fn refuse<B: Default>(retry_after_secs: u64, why: &str) -> http::Response<B> {
let mut status = crate::grpc::coded(
&holger_errcode::ADMISSION_REFUSED,
tonic::Code::Unavailable,
format!("admission: {why}"),
);
if let Ok(v) = retry_after_secs.max(1).to_string().parse() {
status.metadata_mut().insert("retry-after", v);
}
status.into_http()
}
impl<S, B, R> Service<http::Request<B>> for Admitted<S>
where
S: Service<http::Request<B>, Response = http::Response<R>, Error = Infallible>
+ Clone
+ Send
+ 'static,
S::Future: Send + 'static,
B: Send + 'static,
R: Default + Send + 'static,
{
type Response = http::Response<R>;
type Error = Infallible;
type Future = BoxFuture<Self::Response, Self::Error>;
fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
self.inner.poll_ready(cx)
}
fn call(&mut self, mut req: http::Request<B>) -> Self::Future {
// The clone-and-swap tower asks for: the service that was polled ready
// is the one that is called.
let clone = self.inner.clone();
let mut inner = std::mem::replace(&mut self.inner, clone);
let gate = Arc::clone(&self.gate);
Box::pin(async move {
let Some(work) = work_of(req.uri().path()) else {
return inner.call(req).await;
};
let parts = gate.parts(&req);
let who = gate.who(parts).await;
match gate.door.enter(&who, work, 0) {
Ok(ticket) => {
req.extensions_mut().insert(Arc::new(ticket));
inner.call(req).await
}
Err(busy) => {
crate::grpc::functional_status(
"holger-grpc/admission",
"admitted",
false,
&busy.why,
);
Ok(refuse(busy.retry_after_secs, &busy.why))
}
}
})
}
}
/// The ticket the layer put on this request, if any (a handler's accessor).
pub fn ticket_of<T>(req: &tonic::Request<T>) -> Option<Arc<Ticket>> {
req.extensions().get::<Arc<Ticket>>().cloned()
}
// ─────────────────────────────────────────────────────────────────────────────
// AdmissionService — the admin view's read of the door
// ─────────────────────────────────────────────────────────────────────────────
use crate::admission::{CostTally, UsageView};
use crate::grpc::holger_proto::admission_service_server::AdmissionService;
use crate::grpc::holger_proto::{
AdmitPlanRequest, AdmitPlanResponse, CostRow, EstimateRow, ListUsageRequest, ListUsageResponse,
UsageRow, WorkUsageRow,
};
use crate::grpc::HolgerGrpc;
use tonic::{Request, Response, Status};
fn cost_row(c: &CostTally) -> CostRow {
CostRow {
cpu_ms: c.cpu_ms,
disk_read: c.disk_read,
disk_write: c.disk_write,
net_in: c.net_in,
net_out: c.net_out,
upstream_fetch: c.upstream_fetch,
}
}
fn class_word(c: crate::admission::ClassWord) -> String {
match c {
crate::admission::ClassWord::Full => "full",
crate::admission::ClassWord::Reduced => "reduced",
crate::admission::ClassWord::Floor => "floor",
crate::admission::ClassWord::Queued => "queued",
}
.to_string()
}
fn usage_row(v: UsageView) -> UsageRow {
UsageRow {
consumer: v.consumer,
window: Some(cost_row(&v.window)),
today: Some(cost_row(&v.today)),
by_work: v
.by_work
.into_iter()
.map(|(work, t)| WorkUsageRow {
work,
requests: t.requests,
refused: t.refused,
estimated: Some(cost_row(&t.estimated)),
actual: Some(cost_row(&t.actual)),
})
.collect(),
class: v.class.map(class_word).unwrap_or_default(),
why: v.why,
in_flight: v.in_flight,
}
}
fn parse_work(s: &str) -> Result<WorkKind, Status> {
Ok(match s {
"" | "download" => WorkKind::Download,
"upload" => WorkKind::Upload,
"pull_through" => WorkKind::PullThrough,
"mirror_asset" => WorkKind::MirrorAsset,
other => {
return Err(crate::grpc::coded(
&holger_errcode::SERVER_BAD_REQUEST,
tonic::Code::InvalidArgument,
format!(
"work {other:?} is not one of: download, upload, pull_through, mirror_asset"
),
))
}
})
}
impl HolgerGrpc {
fn door(&self) -> Arc<DoorAdmission> {
self.admission
.clone()
.unwrap_or_else(DoorAdmission::pass_through)
}
}
#[tonic::async_trait]
impl AdmissionService for Arc<HolgerGrpc> {
async fn admit_plan(
&self,
request: Request<AdmitPlanRequest>,
) -> Result<Response<AdmitPlanResponse>, Status> {
let _identity = self.authorize_admin_read(&request, "").await?;
let req = request.into_inner();
let work = parse_work(&req.work)?;
let who = if req.consumer == "anonymous" {
// A probe token for today: what a stranger would be planned.
let probe = crate::visitors::Client::Raw("admission-plan-probe");
Who::Anonymous(self.door().day_token(probe))
} else {
Who::from_key(&req.consumer).ok_or_else(|| {
crate::grpc::coded(
&holger_errcode::SERVER_BAD_REQUEST,
tonic::Code::InvalidArgument,
format!(
"consumer {:?} is not `principal:<sub>`, `peer:<name>` or `anonymous`",
req.consumer
),
)
})?
};
let door = self.door();
let plan = door.plan(&who, work, req.size_hint);
let usage = crate::admission::book().view(&plan.consumer).map(usage_row);
crate::grpc::functional_status("holger-grpc/admit_plan", "planned", true, &plan.consumer);
Ok(Response::new(AdmitPlanResponse {
consumer: plan.consumer,
work: plan.work,
class: class_word(plan.class),
estimate: Some(cost_row(&plan.estimate)),
why: plan.why,
budget_left: plan.budget_left,
share: plan.share,
policy: plan.policy,
estimates: door
.estimates()
.into_iter()
.map(|e| EstimateRow {
kind: e.kind,
samples: e.samples,
hinted: e.hinted,
mean: Some(cost_row(&e.mean)),
per_gb: e.per_gb.as_ref().map(cost_row),
})
.collect(),
usage,
}))
}
async fn list_usage(
&self,
request: Request<ListUsageRequest>,
) -> Result<Response<ListUsageResponse>, Status> {
let _identity = self.authorize_admin_read(&request, "").await?;
let limit = request.into_inner().limit.max(0) as usize;
let door = self.door();
let cfg = door.config();
#[cfg(feature = "jobs")]
let queued = door.pull_through().is_some();
#[cfg(not(feature = "jobs"))]
let queued = false;
let consumers: Vec<UsageRow> = crate::admission::book()
.views(limit)
.into_iter()
.map(usage_row)
.collect();
crate::grpc::functional_status(
"holger-grpc/list_usage",
"listed",
true,
&format!("{} consumers", consumers.len()),
);
Ok(Response::new(ListUsageResponse {
policy: door.policy().to_string(),
consumers,
in_flight_total: door.in_flight_total(),
max_in_flight: cfg.map(|c| c.ron_max_in_flight),
share_full: cfg.map(|c| c.ron_share_full),
share_reduced: cfg.map(|c| c.ron_share_reduced),
share_floor: cfg.map(|c| c.ron_share_floor),
pull_through_queue: queued,
}))
}
}