vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Activity

responder checks the logs or metrics permission per surface

Browse source
This commit is contained in:
atlas 2026-11-01 17:40:09 +01:00
commit 3b1d118f25
5 changed files with 324 additions and 37 deletions
Download patch file Download diff file Expand all files Collapse all files

1
swarm-agent-verify/Cargo.toml
View file

@ -5,6 +5,7 @@ edition.workspace = true
[dependencies]
anyhow.workspace = true
serde.workspace = true
# Comparing an agent's presented secret with the stored one.
subtle.workspace = true
# The read of the stored credential the token is checked against.

24
swarm-agent-verify/src/lib.rs
View file

@ -52,11 +52,17 @@ impl Store {
}
}
impl CredentialSource for Store {
/// Logs in per lookup, so no token is left to expire inside a long-lived
/// process.
async fn lookup(&self, agent: &str) -> anyhow::Result<Option<AgentCredential>> {
let path = queue::agent_queue_path(agent)?;
impl Store {
/// The object at `path`, `None` when nothing is stored there. Logs in per
/// read, so no token is left to expire inside a long-lived process.
///
/// # Errors
///
/// When the login or the read fails, or the object does not decode.
pub async fn read_optional<T: serde::de::DeserializeOwned>(
&self,
path: &str,
) -> anyhow::Result<Option<T>> {
let store = swarm_secret_client::SecretStore::connect(
&self.settings,
&self.cert_role,
@ -65,12 +71,18 @@ impl CredentialSource for Store {
.await
.context("logging in to the secret store")?;
store
.read_optional(&path)
.read_optional(path)
.await
.with_context(|| format!("reading {path}"))
}
}
impl CredentialSource for Store {
async fn lookup(&self, agent: &str) -> anyhow::Result<Option<AgentCredential>> {
self.read_optional(&queue::agent_queue_path(agent)?).await
}
}
/// Whether `token`'s secret is the one stored for the agent it names.
///
/// The lookup is bounded by `timeout`, after which this denies.

273
swarm-logs-auth/src/main.rs
View file

@ -3,21 +3,29 @@
//! nginx sends it every `/select/logsql/` request (`swarm-victorialogs.nix`),
//! and every metrics read API request (`swarm-victoriametrics.nix`, as
//! `swarm-metrics-agent-auth`), whose `Authorization` is
//! `Bearer swarm-agent.<agent>.<secret>`. It answers 200 when the secret is
//! the one stored at `swarm/agents/<agent>/queue`, and 401 otherwise, store
//! failures included: `auth_request` turns any other status into a 500 for
//! the caller.
//! `Bearer swarm-agent.<agent>.<secret>`. `--surface` says which store this
//! instance guards. It answers 200 when the secret is the one stored at
//! `swarm/agents/<agent>/queue` and the agent holds the permission for that
//! surface (`logs` or `metrics`), 401 when the secret does not verify, and 403
//! when it does but the permission is not held. A grant it cannot read is not
//! held. Any other status becomes a 500 for the caller in `auth_request`.
//!
//! The permission is read from the leaf swarm-controller mirrors
//! (`swarm_secret_client::observability`), because this responder has no queue
//! access.
//!
//! It logs in to the store as the cert-auth role `--store-cert-role` names,
//! whose policy reaches every agent's `queue` secret. It serves a unix socket
//! whose directory only the gateway's nginx may enter.
//! whose policy reaches every agent's `queue` secret and grant leaf. It serves
//! a unix socket whose directory only the gateway's nginx may enter.
//!
//! A grant is remembered for `--verdict-ttl` seconds ([`verdicts`]), so an
//! agent's queries within that window cost no store lookup, and a secret that
//! stops verifying is still accepted for up to that long.
//! stops verifying, or a permission that is revoked, is still honoured for up
//! to that long. Only an allow is remembered; a denial is looked up again.
mod verdicts;
use std::future::Future;
use std::os::unix::fs::PermissionsExt as _;
use std::path::PathBuf;
use std::sync::{Arc, Mutex};
@ -26,9 +34,10 @@ use std::time::{Duration, Instant};
use anyhow::Context as _;
use axum::extract::State;
use axum::http::{HeaderMap, StatusCode, header};
use clap::Parser;
use clap::{Parser, ValueEnum};
use swarm_agent_verify::CredentialSource;
use swarm_agent_verify::agent_token::parse_agent_token;
use swarm_secret_client::observability::{Grant, grant_path};
use verdicts::Verdicts;
/// How long one store lookup may take before the request is denied. Well
@ -49,20 +58,57 @@ struct Args {
/// Cert-auth role this responder logs in to the store as.
#[arg(long, default_value = "swarm-logs-auth")]
store_cert_role: String,
/// Which store this instance guards, and so which permission it checks.
#[arg(long, value_enum)]
surface: Surface,
/// Seconds a verified bearer is granted again without a store lookup;
/// 0 looks up every request.
#[arg(long, default_value_t = 60)]
verdict_ttl: u64,
}
/// The store a gateway fronts.
#[derive(Clone, Copy, Debug, PartialEq, Eq, ValueEnum)]
enum Surface {
/// The log store, behind the `logs` permission.
Logs,
/// The metrics store, behind the `metrics` permission.
Metrics,
}
impl Surface {
fn allowed_by(self, grant: Grant) -> bool {
match self {
Self::Logs => grant.logs,
Self::Metrics => grant.metrics,
}
}
}
/// Where an agent's mirrored [`Grant`] comes from.
trait GrantSource {
/// The grant mirrored for `agent`, `None` when nothing is stored, or an
/// error when the store could not answer.
fn grant(&self, agent: &str) -> impl Future<Output = anyhow::Result<Option<Grant>>> + Send;
}
impl GrantSource for swarm_agent_verify::Store {
async fn grant(&self, agent: &str) -> anyhow::Result<Option<Grant>> {
self.read_optional(&grant_path(agent)?).await
}
}
struct Responder<S> {
source: S,
surface: Surface,
/// How long one store lookup may take before the request is denied.
lookup_timeout: Duration,
verdicts: Mutex<Verdicts>,
}
/// The answer for one subrequest's `Authorization` header, at `now`.
async fn decide(
responder: &Responder<impl CredentialSource>,
responder: &Responder<impl CredentialSource + GrantSource>,
authorization: Option<&str>,
now: Instant,
) -> StatusCode {
@ -85,13 +131,48 @@ async fn decide(
tracing::debug!(agent = token.agent, "verified recently");
return StatusCode::OK;
}
if swarm_agent_verify::verify(&responder.source, &token, LOOKUP_TIMEOUT).await {
tracing::info!(agent = token.agent, "verified");
lock(&responder.verdicts).grant(bearer, now);
StatusCode::OK
} else {
StatusCode::UNAUTHORIZED
if !swarm_agent_verify::verify(&responder.source, &token, responder.lookup_timeout).await {
return StatusCode::UNAUTHORIZED;
}
if !holds_permission(responder, token.agent).await {
return StatusCode::FORBIDDEN;
}
tracing::info!(agent = token.agent, surface = ?responder.surface, "verified");
lock(&responder.verdicts).grant(bearer, now);
StatusCode::OK
}
/// Whether `agent` holds the permission for the surface. A grant that is
/// absent, unreadable or slow is not held.
async fn holds_permission(responder: &Responder<impl GrantSource>, agent: &str) -> bool {
let grant =
match tokio::time::timeout(responder.lookup_timeout, responder.source.grant(agent)).await {
Ok(Ok(Some(grant))) => grant,
Ok(Ok(None)) => {
tracing::warn!(
agent,
"no permission grant is mirrored for this agent; denying"
);
return false;
}
Ok(Err(e)) => {
tracing::warn!(
agent,
error = format!("{e:#}"),
"grant lookup failed; denying"
);
return false;
}
Err(_) => {
tracing::warn!(agent, "grant lookup timed out; denying");
return false;
}
};
let held = responder.surface.allowed_by(grant);
if !held {
tracing::info!(agent, surface = ?responder.surface, "agent lacks the permission; denying");
}
held
}
/// The cache holds no invariant a panicking holder could break mid-way, so a
@ -129,6 +210,8 @@ async fn main() -> anyhow::Result<()> {
.context("reading the secret store's address and this responder's identity")?;
let responder = Arc::new(Responder {
source: swarm_agent_verify::Store::new(settings, args.store_cert_role),
surface: args.surface,
lookup_timeout: LOOKUP_TIMEOUT,
verdicts: Mutex::new(Verdicts::new(
Duration::from_secs(args.verdict_ttl),
VERDICT_CAP,
@ -161,18 +244,51 @@ mod tests {
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use swarm_secret_client::queue::AgentCredential;
/// A store holding atlas's credential until `revoke`, counting lookups.
#[derive(Default)]
/// What the mirrored grant leaf answers.
#[derive(Clone, Copy)]
enum Leaf {
Holds(Grant),
Absent,
Unreadable,
/// Accepts the read and never answers.
Hangs,
}
/// A store holding atlas's credential until `revoke` and her grant leaf,
/// counting lookups of each.
struct Atlas {
lookups: AtomicUsize,
revoked: AtomicBool,
leaf: Mutex<Leaf>,
grant_lookups: AtomicUsize,
}
impl Atlas {
fn with(leaf: Leaf) -> Self {
Self {
lookups: AtomicUsize::new(0),
revoked: AtomicBool::new(false),
leaf: Mutex::new(leaf),
grant_lookups: AtomicUsize::new(0),
}
}
/// Atlas holds `logs` and `metrics`.
fn both() -> Self {
Self::with(Leaf::Holds(Grant {
logs: true,
metrics: true,
}))
}
fn revoke(&self) {
self.revoked.store(true, Ordering::SeqCst);
}
fn set_leaf(&self, leaf: Leaf) {
*self.leaf.lock().unwrap() = leaf;
}
fn lookups(&self) -> usize {
self.lookups.load(Ordering::SeqCst)
}
@ -193,30 +309,97 @@ mod tests {
}
}
impl GrantSource for &Atlas {
async fn grant(&self, agent: &str) -> anyhow::Result<Option<Grant>> {
self.grant_lookups.fetch_add(1, Ordering::SeqCst);
let leaf = *self.leaf.lock().unwrap();
match (agent, leaf) {
("atlas", Leaf::Holds(grant)) => Ok(Some(grant)),
("atlas", Leaf::Unreadable) => anyhow::bail!("store unreachable"),
("atlas", Leaf::Hangs) => std::future::pending().await,
_ => Ok(None),
}
}
}
const TTL: Duration = Duration::from_mins(1);
const GOOD: &str = "Bearer swarm-agent.atlas.Ab9_-zSECRET";
const WRONG: &str = "Bearer swarm-agent.atlas.Ab9_-zSECREX";
fn responder(store: &Atlas) -> Responder<&Atlas> {
fn responder(store: &Atlas, surface: Surface) -> Responder<&Atlas> {
Responder {
source: store,
surface,
lookup_timeout: Duration::from_millis(100),
verdicts: Mutex::new(Verdicts::new(TTL, 16)),
}
}
#[tokio::test]
async fn the_stored_secret_is_200() {
let store = Atlas::default();
async fn the_stored_secret_with_the_grant_is_200() {
let store = Atlas::both();
for surface in [Surface::Logs, Surface::Metrics] {
assert_eq!(
decide(&responder(&store, surface), Some(GOOD), Instant::now()).await,
StatusCode::OK,
"{surface:?}"
);
}
}
/// Each surface checks its own permission: a `logs` grant opens the log
/// store and not the metrics store, and the reverse.
#[tokio::test]
async fn each_surface_needs_its_own_permission() {
async fn at(store: &Atlas, surface: Surface) -> StatusCode {
decide(&responder(store, surface), Some(GOOD), Instant::now()).await
}
let logs_only = Atlas::with(Leaf::Holds(Grant {
logs: true,
metrics: false,
}));
let metrics_only = Atlas::with(Leaf::Holds(Grant {
logs: false,
metrics: true,
}));
assert_eq!(at(&logs_only, Surface::Logs).await, StatusCode::OK);
assert_eq!(
decide(&responder(&store), Some(GOOD), Instant::now()).await,
StatusCode::OK
at(&logs_only, Surface::Metrics).await,
StatusCode::FORBIDDEN
);
assert_eq!(at(&metrics_only, Surface::Metrics).await, StatusCode::OK);
assert_eq!(
at(&metrics_only, Surface::Logs).await,
StatusCode::FORBIDDEN
);
}
/// No grant, an empty grant and a grant that cannot be read all deny.
#[tokio::test]
async fn without_a_readable_grant_the_answer_is_403() {
for leaf in [
Leaf::Absent,
Leaf::Unreadable,
Leaf::Hangs,
Leaf::Holds(Grant::default()),
] {
let store = Atlas::with(leaf);
assert_eq!(
decide(
&responder(&store, Surface::Logs),
Some(GOOD),
Instant::now()
)
.await,
StatusCode::FORBIDDEN
);
}
}
#[tokio::test]
async fn everything_else_is_401() {
let store = Atlas::default();
let r = responder(&store);
let store = Atlas::both();
let r = responder(&store, Surface::Logs);
for authorization in [
None,
Some("swarm-agent.atlas.Ab9_-zSECRET"),
@ -232,17 +415,21 @@ mod tests {
"{authorization:?}"
);
}
// A wrong secret is refused before the grant is read.
assert_eq!(store.grant_lookups.load(Ordering::SeqCst), 0);
}
/// A grant is reused within the TTL — a revocation the store already
/// holds goes unseen — and checked against the store again after it.
/// An allow is reused within the TTL — a revocation the store already
/// holds, of the secret or of the permission, goes unseen — and checked
/// against the store again after it.
#[tokio::test]
async fn a_grant_spares_the_store_until_the_ttl() {
let store = Atlas::default();
let r = responder(&store);
async fn an_allow_spares_the_store_until_the_ttl() {
let store = Atlas::both();
let r = responder(&store, Surface::Logs);
let t0 = Instant::now();
assert_eq!(decide(&r, Some(GOOD), t0).await, StatusCode::OK);
store.revoke();
store.set_leaf(Leaf::Holds(Grant::default()));
let within = t0 + Duration::from_secs(59);
assert_eq!(decide(&r, Some(GOOD), within).await, StatusCode::OK);
assert_eq!(store.lookups(), 1);
@ -253,10 +440,34 @@ mod tests {
assert_eq!(store.lookups(), 2);
}
/// A revoked permission is denied once the allow expires; a denial is not
/// remembered, so a permission granted later is honoured at once.
#[tokio::test]
async fn a_revoked_or_added_permission_takes_effect_by_the_ttl() {
let store = Atlas::both();
let r = responder(&store, Surface::Logs);
let t0 = Instant::now();
assert_eq!(decide(&r, Some(GOOD), t0).await, StatusCode::OK);
store.set_leaf(Leaf::Holds(Grant::default()));
assert_eq!(
decide(&r, Some(GOOD), t0 + TTL).await,
StatusCode::FORBIDDEN
);
assert_eq!(
decide(&r, Some(GOOD), t0 + TTL).await,
StatusCode::FORBIDDEN
);
store.set_leaf(Leaf::Holds(Grant {
logs: true,
metrics: false,
}));
assert_eq!(decide(&r, Some(GOOD), t0 + TTL).await, StatusCode::OK);
}
#[tokio::test]
async fn a_denial_is_looked_up_every_time() {
let store = Atlas::default();
let r = responder(&store);
let store = Atlas::both();
let r = responder(&store, Surface::Logs);
let t0 = Instant::now();
for _ in 0..3 {
assert_eq!(decide(&r, Some(WRONG), t0).await, StatusCode::UNAUTHORIZED);

1
swarm-secret-client/src/lib.rs
View file

@ -45,6 +45,7 @@ pub mod github;
pub mod grafana;
pub mod matrix;
pub mod mtls;
pub mod observability;
pub mod path;
pub mod policy;
pub mod proxy;

62
swarm-secret-client/src/observability.rs Normal file
View file

@ -0,0 +1,62 @@
//! The log and metric read grants: where swarm-controller mirrors an agent's
//! `logs` and `metrics` permissions in the store, and what the object holds.
//!
//! The permissions themselves live in the queue's `agent-permissions` bucket.
//! The store gateways' `swarm-logs-auth` has no queue access, so it reads this
//! mirror instead. A leaf that is absent or unreadable is no grant.
//!
//! Under the agent's own prefix, so the agent can read its own grant and no
//! agent can write it: the shared agent policy grants `create` on one other
//! leaf only.
use serde::{Deserialize, Serialize};
use crate::{
Error,
path::{Kind, principal_prefix},
};
const LEAF: &str = "observability";
/// The path holding `agent`'s [`Grant`].
///
/// # Errors
/// [`Error::PathSegment`] when `agent` contains anything but `[A-Za-z0-9_-]`,
/// which is what keeps one agent's name from addressing another agent's grant.
pub fn grant_path(agent: &str) -> Result<String, Error> {
let prefix = principal_prefix(Kind::Agent, agent)?;
Ok(format!("{prefix}/{LEAF}"))
}
/// What a [`grant_path`] holds: which stores the agent may read.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct Grant {
/// The agent holds the `logs` permission.
#[serde(default)]
pub logs: bool,
/// The agent holds the `metrics` permission.
#[serde(default)]
pub metrics: bool,
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn the_path_is_under_the_agents_own_prefix() {
assert_eq!(
grant_path("atlas").unwrap(),
"swarm/agents/atlas/observability"
);
assert!(grant_path("../argus").is_err());
}
#[test]
fn a_field_left_out_is_not_granted() {
let grant: Grant = serde_json::from_str("{}").unwrap();
assert_eq!(grant, Grant::default());
let grant: Grant = serde_json::from_str(r#"{"logs":true}"#).unwrap();
assert_eq!((grant.logs, grant.metrics), (true, false));
}
}