diff --git a/CHANGELOG.md b/CHANGELOG.md index e569ad36c..5c6e6805b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -87,6 +87,10 @@ Changelog tracking starts with 0.2.0. Prior versions were not tracked. ### Fixed +- **Invite commands accept every token the runtime can issue.** `invite redeem` + and `invite revoke` now treat a leading hyphen in an opaque base64url token as + token data instead of misparsing it as an unknown option. + - **Security Audit now uses a reproducible cargo-audit installation.** CI installs cargo-audit 0.22.2 with its published lockfile before invoking the pinned RustSec action, keeping the advisory scan compatible with the repository toolchain. @@ -140,6 +144,16 @@ Changelog tracking starts with 0.2.0. Prior versions were not tracked. identities fail closed. `capsule:list` controls global describe visibility, while exact `capsule:access:any` controls unrestricted execution. Closes #1239. +- **Audit views now use the live kernel authority evaluator.** Historical and + streaming firehose access carries the authenticated device scope; revoked, + disabled, malformed, unknown, or narrowed credentials fall back to the + caller's own records. Historical continuation cursors bind the immutable + audit-entry identity and visibility scope, preventing same-second appends + from shifting the page boundary and rejecting unsafe widening. Co-located + callers that need firehose visibility must now pass the kernel evaluator via + `astrid_gateway::run_with_capability_probe(...)` or + `routes::build_with_capability_probe(...)`; the legacy default runner and + router remain self-only. Closes #1241. - **Device attenuation now applies to every kernel authority view.** Capsule, agent, and group inventory checks use the authenticating device scope, and a diff --git a/crates/astrid-capabilities/src/policy.rs b/crates/astrid-capabilities/src/policy.rs index b6eeab63d..6efd70c59 100644 --- a/crates/astrid-capabilities/src/policy.rs +++ b/crates/astrid-capabilities/src/policy.rs @@ -41,6 +41,8 @@ //! caching. The caller is expected to have resolved the profile and //! the group config beforehand. +use std::borrow::Cow; + use astrid_core::{ CapabilityPattern, DeviceScope, GroupConfig, PrincipalId, PrincipalProfile, ValidatedGroupConfig, ValidatedProfileFields, capability_matches, @@ -110,7 +112,7 @@ pub enum PermissionError { pub struct CapabilityCheck<'a> { profile: Result, String>, groups: Result, String>, - principal: PrincipalId, + principal: Cow<'a, PrincipalId>, /// Optional per-device attenuation floor. `None` (the default) means the /// check is unattenuated — the principal's full effective capability set /// applies, which is the behaviour for every full-scope / un-paired @@ -136,7 +138,29 @@ impl<'a> CapabilityCheck<'a> { Self { profile, groups, - principal, + principal: Cow::Owned(principal), + device_scope: None, + } + } + + /// Build a new check while borrowing the resolved principal identifier. + /// + /// This is equivalent to [`Self::new`] but avoids cloning an identifier + /// for checks that only need the boolean result from [`Self::has`]. A + /// denied [`Self::require`] still returns an owned identifier in its + /// [`PermissionError`]. + #[must_use] + pub fn new_borrowed( + profile: &'a PrincipalProfile, + groups: &'a GroupConfig, + principal: &'a PrincipalId, + ) -> Self { + let profile = profile.typed_fields().map_err(|e| e.to_string()); + let groups = groups.typed().map_err(|e| e.to_string()); + Self { + profile, + groups, + principal: Cow::Borrowed(principal), device_scope: None, } } @@ -167,7 +191,7 @@ impl<'a> CapabilityCheck<'a> { Err(e) => { warn!( security_event = true, - principal = %self.principal, + principal = %self.principal.as_ref(), error = %e, "Principal profile contains invalid typed fields — no capabilities inherited" ); @@ -197,19 +221,19 @@ impl<'a> CapabilityCheck<'a> { Err(e) => { warn!( security_event = true, - principal = %self.principal, + principal = %self.principal.as_ref(), error = %e, "Principal profile contains invalid typed fields — denying capability" ); return Err(PermissionError::MissingCapability { - principal: self.principal.clone(), + principal: self.principal.as_ref().clone(), required: cap.to_string(), }); }, }; if let Some(revoke) = Self::first_matching_revoke(profile, cap) { return Err(PermissionError::RevokedCapability { - principal: self.principal.clone(), + principal: self.principal.as_ref().clone(), required: cap.to_string(), revoke_pattern: revoke.to_string(), }); @@ -219,7 +243,7 @@ impl<'a> CapabilityCheck<'a> { || self.holds_via_groups(profile, cap); if !principal_holds { return Err(PermissionError::MissingCapability { - principal: self.principal.clone(), + principal: self.principal.as_ref().clone(), required: cap.to_string(), }); } @@ -227,7 +251,7 @@ impl<'a> CapabilityCheck<'a> { // a scoped device's narrower grant can deny without ever widening. if !self.device_scope_admits(cap) { return Err(PermissionError::DeviceScopeDenied { - principal: self.principal.clone(), + principal: self.principal.as_ref().clone(), required: cap.to_string(), }); } @@ -277,7 +301,7 @@ impl<'a> CapabilityCheck<'a> { Err(e) => { warn!( security_event = true, - principal = %self.principal, + principal = %self.principal.as_ref(), error = %e, "Group config contains invalid typed fields — no group capabilities inherited" ); @@ -288,7 +312,7 @@ impl<'a> CapabilityCheck<'a> { let Some(group) = groups.get(name) else { warn!( security_event = true, - principal = %self.principal, + principal = %self.principal.as_ref(), group = %name, "Principal profile references unknown group — no capabilities inherited" ); @@ -348,7 +372,7 @@ pub fn device_scope_within( for pattern in allow { if !issuer_check.has(pattern) { return Err(PermissionError::MissingCapability { - principal: issuer_check.principal.clone(), + principal: issuer_check.principal.as_ref().clone(), required: pattern.clone(), }); } @@ -379,6 +403,35 @@ mod tests { PrincipalId::new("alice").unwrap() } + #[test] + fn borrowed_principal_check_preserves_allow_semantics() { + let profile = profile_in(&["agent"]); + let groups = gc(); + let principal = pid(); + let check = CapabilityCheck::new_borrowed(&profile, &groups, &principal); + + assert!(check.has("self:capsule:install")); + assert!(!check.has("system:shutdown")); + assert_eq!(principal, pid()); + } + + #[test] + fn borrowed_principal_is_owned_in_permission_errors() { + let profile = profile_in(&["restricted"]); + let groups = gc(); + let principal = pid(); + let check = CapabilityCheck::new_borrowed(&profile, &groups, &principal); + + assert_eq!( + check.require("system:shutdown"), + Err(PermissionError::MissingCapability { + principal: principal.clone(), + required: "system:shutdown".to_owned(), + }) + ); + assert_eq!(principal, pid()); + } + #[test] fn admin_has_universal() { let p = profile_in(&["admin"]); diff --git a/crates/astrid-cli/src/cli.rs b/crates/astrid-cli/src/cli.rs index 9ae24858c..be15e5736 100644 --- a/crates/astrid-cli/src/cli.rs +++ b/crates/astrid-cli/src/cli.rs @@ -563,7 +563,7 @@ pub(crate) enum DistroCommands { #[cfg(test)] mod tests { - use super::{CapsuleCommands, Cli, Commands}; + use super::{CapsuleCommands, Cli, Commands, InviteCommand}; use clap::{CommandFactory, Parser, Subcommand}; use std::collections::BTreeSet; @@ -654,6 +654,36 @@ mod tests { assert_eq!(cli.principal.as_deref(), Some("operator-1")); } + #[test] + fn opaque_invite_tokens_may_start_with_a_hyphen() { + let redeem = Cli::try_parse_from([ + "astrid", + "invite", + "redeem", + "-opaque-token", + "--public-key", + "ed25519:0000000000000000000000000000000000000000000000000000000000000000", + ]) + .expect("an issued base64url token may begin with a hyphen"); + assert!(matches!( + redeem.command, + Some(Commands::Invite { + command: InviteCommand::Redeem(ref args), + }) if args.token == "-opaque-token" + && args.public_key.as_deref() + == Some("ed25519:0000000000000000000000000000000000000000000000000000000000000000") + )); + + let revoke = Cli::try_parse_from(["astrid", "invite", "revoke", "-opaque-token"]) + .expect("the same issued token must be accepted by revoke"); + assert!(matches!( + revoke.command, + Some(Commands::Invite { + command: InviteCommand::Revoke(ref args), + }) if args.token_or_fingerprint == "-opaque-token" + )); + } + #[test] fn grant_capsules_parsing_stays_open_for_embedding_composition() { let cli = Cli::try_parse_from(["astrid", "init", "--grant-capsules"]) diff --git a/crates/astrid-cli/src/commands/invite.rs b/crates/astrid-cli/src/commands/invite.rs index 1c47bbaa2..c75181bcf 100644 --- a/crates/astrid-cli/src/commands/invite.rs +++ b/crates/astrid-cli/src/commands/invite.rs @@ -55,6 +55,7 @@ pub(crate) struct IssueArgs { #[derive(Args, Debug, Clone)] pub(crate) struct RedeemArgs { /// The opaque token returned by a prior `astrid invite issue`. + #[arg(allow_hyphen_values = true)] pub token: String, /// Hex-encoded ed25519 public key. Accepts bare 64 hex chars or /// the self-describing `ed25519:` form. The new principal's @@ -89,6 +90,7 @@ pub(crate) struct ListArgs { #[derive(Args, Debug, Clone)] pub(crate) struct RevokeArgs { /// Either the raw token or its hex fingerprint (from `invite list`). + #[arg(allow_hyphen_values = true)] pub token_or_fingerprint: String, } diff --git a/crates/astrid-daemon/src/lib.rs b/crates/astrid-daemon/src/lib.rs index ed5f815f4..0999d87e9 100644 --- a/crates/astrid-daemon/src/lib.rs +++ b/crates/astrid-daemon/src/lib.rs @@ -360,6 +360,7 @@ fn spawn_gateway( let bus = std::sync::Arc::clone(&kernel.event_bus); let audit_log = std::sync::Arc::clone(&kernel.audit_log); let session_id = kernel.session_id.clone(); + let capability_kernel = std::sync::Arc::clone(kernel); let readiness_probe = kernel.agent_readiness_probe(); let topic_probe = kernel.capsule_topic_probe_with_warm(); let state = astrid_gateway::GatewayState::new( @@ -377,7 +378,14 @@ fn spawn_gateway( let shutdown = async move { notify_for_task.notified().await; }; - if let Err(e) = astrid_gateway::run(state, shutdown).await { + let capability_probe = move |principal: &astrid_core::PrincipalId, + device_key_id: Option<&str>, + capability: &str| { + capability_kernel.runtime_capability_allows(principal, device_key_id, capability) + }; + if let Err(e) = + astrid_gateway::run_with_capability_probe(state, shutdown, capability_probe).await + { tracing::error!(error = %e, "astrid-gateway exited with error"); } }); diff --git a/crates/astrid-gateway/src/lib.rs b/crates/astrid-gateway/src/lib.rs index 462e11a23..e9341a69f 100644 --- a/crates/astrid-gateway/src/lib.rs +++ b/crates/astrid-gateway/src/lib.rs @@ -71,6 +71,8 @@ pub use state::GatewayState; /// router (see [`tls::serve_https`]). The default posture remains /// "TLS upstream"; native TLS is an opt-in feature for single-box /// installs that don't want to run a reverse proxy. +/// Audit views remain per-principal; co-located runtimes can use +/// [`run_with_capability_probe`] to enable exact firehose policy. /// /// # Errors /// Returns an error if the listener cannot bind, the rustls config @@ -79,6 +81,65 @@ pub use state::GatewayState; pub async fn run( state: Arc, shutdown: impl Future + Send + 'static, +) -> Result<()> { + run_inner(state, shutdown, routes::events::CapabilityProbe::deny_all()).await +} + +/// Run the gateway with a borrowed in-process capability evaluator. +/// +/// # Errors +/// Returns the same startup and server errors as [`run`]. +pub async fn run_with_capability_probe( + state: Arc, + shutdown: impl Future + Send + 'static, + capability_probe: F, +) -> Result<()> +where + F: Fn(&astrid_core::PrincipalId, Option<&str>, &str) -> bool + Send + Sync + 'static, +{ + run_inner( + state, + shutdown, + routes::events::CapabilityProbe::new(capability_probe), + ) + .await +} + +/// Run the gateway on an already-bound plain-HTTP listener. +/// +/// # Errors +/// Returns an error when native TLS is configured, the listener address is +/// unavailable, or the HTTP server fails. +#[doc(hidden)] +pub async fn run_with_capability_probe_on_listener( + state: Arc, + listener: TcpListener, + shutdown: impl Future + Send + 'static, + capability_probe: F, +) -> Result<()> +where + F: Fn(&astrid_core::PrincipalId, Option<&str>, &str) -> bool + Send + Sync + 'static, +{ + if state.config.tls.is_some() { + anyhow::bail!("an already-bound plain-HTTP listener cannot serve native TLS"); + } + let addr = listener + .local_addr() + .context("failed to read pre-bound gateway listener address")?; + warn_if_plaintext_non_loopback(addr); + serve_http_listener( + state, + listener, + shutdown, + routes::events::CapabilityProbe::new(capability_probe), + ) + .await +} + +async fn run_inner( + state: Arc, + shutdown: impl Future + Send + 'static, + capability_probe: routes::events::CapabilityProbe, ) -> Result<()> { let addr: SocketAddr = state.config.listen.parse().with_context(|| { format!( @@ -93,27 +154,36 @@ pub async fn run( // doesn't apply here — axum-server opens its own listener. info!(addr = %addr, scheme = "https", "astrid-gateway listening (TLS)"); let rustls = tls::load_rustls_config(tls_cfg).await?; - let router = tls::apply_hsts(routes::build(state)); + let router = tls::apply_hsts(routes::build_with_probe(state, capability_probe)); return tls::serve_https(addr, router, rustls, shutdown).await; } - // Plain HTTP path — unchanged behaviour from v0.7.0. Warn loudly - // when the operator binds beyond loopback without enabling TLS, - // since that's almost always a misconfig: either the gateway is - // about to serve unencrypted traffic on the LAN/public, or - // there's a reverse proxy upstream that the operator should - // confirm is actually fronting plain TCP correctly. + warn_if_plaintext_non_loopback(addr); + + let listener = TcpListener::bind(addr) + .await + .with_context(|| format!("failed to bind gateway listener on {addr}"))?; + serve_http_listener(state, listener, shutdown, capability_probe).await +} + +fn warn_if_plaintext_non_loopback(addr: SocketAddr) { if !addr.ip().is_loopback() { tracing::warn!( addr = %addr, "gateway is binding a non-loopback address without TLS; ensure a TLS-terminating reverse proxy fronts this listener, or enable [tls] in gateway-http.toml" ); } +} - let listener = TcpListener::bind(addr) - .await - .with_context(|| format!("failed to bind gateway listener on {addr}"))?; - let bound = listener.local_addr().unwrap_or(addr); +async fn serve_http_listener( + state: Arc, + listener: TcpListener, + shutdown: impl Future + Send + 'static, + capability_probe: routes::events::CapabilityProbe, +) -> Result<()> { + let bound = listener + .local_addr() + .context("failed to read bound gateway listener address")?; info!(addr = %bound, scheme = "http", "astrid-gateway listening"); // Spawn the revocation watcher as soon as we know we have a live @@ -130,7 +200,7 @@ pub async fn run( ); } - let router = routes::build(state); + let router = routes::build_with_probe(state, capability_probe); // `into_make_service_with_connect_info::()` is what // populates the `ConnectInfo` request extension that // `routes::auth::post_redeem` extracts for per-IP rate limiting. diff --git a/crates/astrid-gateway/src/routes/audit.rs b/crates/astrid-gateway/src/routes/audit.rs index 68c219d47..2fc30e3b1 100644 --- a/crates/astrid-gateway/src/routes/audit.rs +++ b/crates/astrid-gateway/src/routes/audit.rs @@ -38,6 +38,11 @@ const DEFAULT_LIMIT: usize = 100; /// paginate; the cap exists so a malicious bearer can't request a /// 10-million-entry page and OOM the gateway. const MAX_LIMIT: usize = 1000; +const CURSOR_SCOPE_ALL: &str = "all"; +const CURSOR_SCOPE_PRINCIPAL_PREFIX: char = 'p'; +const CURSOR_SCOPE_CHANGED: &str = "audit visibility widened or changed principal during pagination; restart pagination without a cursor"; +const CURSOR_ANCHOR_MISSING: &str = + "audit cursor anchor is no longer available; restart pagination without a cursor"; /// Query parameters for `GET /api/sys/audit`. All optional — /// the default behaviour is "the last [`DEFAULT_LIMIT`] entries @@ -109,6 +114,68 @@ pub struct AuditQueryResponse { pub next_cursor: Option, } +#[derive(Debug, Clone, PartialEq, Eq)] +enum CursorScope { + All, + Principal(PrincipalId), +} + +#[derive(Debug, Clone, Default, PartialEq, Eq)] +struct AuditCursor { + timestamp: Option, + same_second_offset: usize, + scope: Option, + anchor_entry_id: Option, +} + +impl CursorScope { + fn encode(&self) -> String { + match self { + Self::All => CURSOR_SCOPE_ALL.into(), + Self::Principal(principal) => format!( + "{CURSOR_SCOPE_PRINCIPAL_PREFIX}{}", + hex::encode(principal.as_str()) + ), + } + } + + fn decode(raw: &str) -> GatewayResult { + if raw == CURSOR_SCOPE_ALL { + return Ok(Self::All); + } + let encoded = raw + .strip_prefix(CURSOR_SCOPE_PRINCIPAL_PREFIX) + .ok_or_else(|| { + GatewayError::BadRequest( + "cursor scope must be \"all\" or \"p\"".into(), + ) + })?; + let principal = String::from_utf8(hex::decode(encoded).map_err(|_| { + GatewayError::BadRequest("cursor scope principal must be valid hex".into()) + })?) + .map_err(|_| GatewayError::BadRequest("cursor scope principal must be UTF-8".into()))?; + let principal = PrincipalId::new(&principal) + .map_err(|e| GatewayError::BadRequest(format!("invalid cursor principal: {e}")))?; + Ok(Self::Principal(principal)) + } + + fn principal_filter(&self) -> Option<&PrincipalId> { + match self { + Self::All => None, + Self::Principal(principal) => Some(principal), + } + } + + // A raw-offset cursor may safely continue only when the next page + // sees the same principal scope or a narrower one. If visibility + // widens (for example self-only → firehose or principal A → + // principal B), records skipped above the raw boundary could + // become newly visible and would otherwise be lost silently. + fn accepts_continuation_from(&self, previous: &Self) -> bool { + matches!(previous, Self::All) || previous == self + } +} + /// `GET /api/sys/audit` handler. #[utoipa::path( get, @@ -135,6 +202,11 @@ pub async fn get_audit( req: Request, ) -> GatewayResult> { let caller = caller_from(&req)?; + let capability_probe = req + .extensions() + .get::() + .cloned() + .unwrap_or_else(super::events::CapabilityProbe::deny_all); let caller_principal = caller.principal.clone(); let (audit_log, session_id) = match (state.audit_log.as_ref(), state.session_id.as_ref()) { @@ -156,17 +228,6 @@ pub async fn get_audit( Some(l) => l, }; - // Cap-gate: admin / `audit:read_all` callers get the firehose; - // everyone else is silently scoped to their own principal, - // matching the SSE handler's posture. - let firehose = super::events::caller_holds( - &state, - &caller_principal, - caller.device_key_id.as_deref(), - AUDIT_FIREHOSE_CAP, - ) - .await; - // Pull the full session slice from the audit log. The audit log // doesn't expose an "after cursor" query primitive today, so we // fetch + filter + paginate in-process. The persistent log on @@ -189,10 +250,16 @@ pub async fn get_audit( let mut entries: Vec = all; entries.reverse(); - // Effective principal filter: admins can use the `principal=` - // query param; non-admins are pinned to their own principal - // regardless of what they ask for. - let principal_filter: Option = if firehose { + // Resolve an explicit filter only under post-read authority. Pagination + // applies the current policy to each record, so narrowing takes effect at + // the next record boundary. + let firehose_at_query_boundary = super::events::caller_holds( + &capability_probe, + &caller_principal, + caller.device_key_id.as_deref(), + AUDIT_FIREHOSE_CAP, + ); + let requested_principal: Option = if firehose_at_query_boundary { match query.principal.as_deref() { Some(s) => Some(PrincipalId::new(s).map_err(|e| { GatewayError::BadRequest(format!("invalid `principal` query value: {e}")) @@ -200,18 +267,24 @@ pub async fn get_audit( None => None, } } else { - Some(caller_principal) + None + }; + let access = AuditAccess { + capability_probe: &capability_probe, + caller_principal: &caller_principal, + device_key_id: caller.device_key_id.as_deref(), + requested_principal: requested_principal.as_ref(), }; - // Cursor is `"_"` — `offset` is the number of - // entries with the same `ts_epoch` we've already returned from - // previous pages. Encoding both means same-second batches - // (timer ticks, scripted ops) can't silently lose or duplicate - // entries across the page boundary. The plain `""` - // shape is still accepted for compatibility with v1 cursors. - let cursor = parse_cursor(query.cursor.as_deref())?; - let (page, next_cursor) = - paginate_page(entries, &query, principal_filter.as_ref(), cursor, limit); + // New cursors bind the last visible entry's immutable ID and effective + // principal scope. The ID remains stable when another same-second entry is + // appended between page requests, while the scope forces a restart when + // visibility widens or changes principal. Legacy timestamp-only and raw + // same-second offset cursors remain accepted only for the unambiguous + // default self view. + let mut cursor = parse_cursor(query.cursor.as_deref())?; + cursor.scope = validate_cursor_scope(cursor.timestamp, cursor.scope.as_ref(), &query, &access)?; + let (page, next_cursor) = paginate_page(entries, &query, &access, cursor, limit)?; Ok(Json(AuditQueryResponse { entries: page, @@ -219,24 +292,115 @@ pub async fn get_audit( })) } +struct AuditAccess<'a> { + capability_probe: &'a super::events::CapabilityProbe, + caller_principal: &'a PrincipalId, + device_key_id: Option<&'a str>, + requested_principal: Option<&'a PrincipalId>, +} + +impl AuditAccess<'_> { + fn current_principal_filter(&self) -> Option<&PrincipalId> { + if super::events::caller_holds( + self.capability_probe, + self.caller_principal, + self.device_key_id, + AUDIT_FIREHOSE_CAP, + ) { + self.requested_principal + } else { + Some(self.caller_principal) + } + } + + fn current_cursor_scope(&self) -> CursorScope { + match self.current_principal_filter() { + Some(principal) => CursorScope::Principal(principal.clone()), + None => CursorScope::All, + } + } +} + +fn validate_cursor_scope( + cursor_ts: Option, + cursor_scope: Option<&CursorScope>, + query: &AuditQuery, + access: &AuditAccess<'_>, +) -> GatewayResult> { + let current_scope = access.current_cursor_scope(); + match cursor_scope { + Some(previous_scope) => { + validate_scope_continuation(previous_scope, ¤t_scope)?; + Ok(Some(previous_scope.clone())) + }, + None if cursor_ts.is_some() => { + validate_legacy_cursor_scope(query, access, ¤t_scope)?; + Ok(Some(current_scope)) + }, + None => Ok(None), + } +} + +fn validate_legacy_cursor_scope( + query: &AuditQuery, + access: &AuditAccess<'_>, + current_scope: &CursorScope, +) -> GatewayResult<()> { + match current_scope { + CursorScope::Principal(principal) + if query.principal.is_none() && principal == access.caller_principal => + { + Ok(()) + }, + _ => Err(GatewayError::BadRequest(CURSOR_SCOPE_CHANGED.into())), + } +} + +fn validate_scope_continuation( + previous_scope: &CursorScope, + current_scope: &CursorScope, +) -> GatewayResult<()> { + if current_scope.accepts_continuation_from(previous_scope) { + Ok(()) + } else { + Err(GatewayError::BadRequest(CURSOR_SCOPE_CHANGED.into())) + } +} + /// Walk the entries (newest first) and assemble one page worth of -/// rendered views, honouring every filter + the cursor offset. -/// Returns the page plus the next-page cursor (or `None` if the -/// result is the last page). Pulled out of [`get_audit`] so the -/// handler stays inside the function-length budget and the cursor -/// arithmetic lives in one place. +/// rendered views, honouring every stable filter plus the live +/// principal restriction. Returns the page plus the next-page cursor +/// (or `None` if the result is the last page), and rejects a live +/// scope widening that would invalidate the raw cursor boundary. Pulled out of +/// [`get_audit`] so the handler stays inside the function-length +/// budget and the cursor arithmetic lives in one place. fn paginate_page( entries: Vec, query: &AuditQuery, - principal_filter: Option<&PrincipalId>, - cursor: (Option, usize), + access: &AuditAccess<'_>, + cursor: AuditCursor, limit: usize, -) -> (Vec, Option) { - let (cursor_ts, cursor_offset) = cursor; +) -> GatewayResult<(Vec, Option)> { + let AuditCursor { + timestamp: cursor_ts, + same_second_offset: cursor_offset, + scope: cursor_scope, + anchor_entry_id, + } = cursor; let mut page: Vec = Vec::with_capacity(limit); - let mut equal_ts_skipped: usize = 0; - let mut equal_ts_count_in_page: usize = 0; - let mut last_ts: Option = None; + let mut raw_last_ts: Option = None; + let mut raw_ts_position: usize = 0; + let mut last_visible_ts: Option = None; + let mut last_visible_ts_position: usize = 0; + let mut last_visible_scope: Option = None; + let mut last_visible_entry_id: Option = None; + let mut effective_scope = access.current_cursor_scope(); + let mut anchor_found = anchor_entry_id.is_none(); + let mut has_more = false; + let mut skipped_by_scope_before_limit = false; + if let Some(cursor_scope) = cursor_scope.as_ref() { + validate_scope_continuation(cursor_scope, &effective_scope)?; + } for entry in entries { let Some(view) = render_entry(&entry) else { continue; @@ -257,74 +421,126 @@ fn paginate_page( { continue; } - if let Some(p) = principal_filter - && view.principal.as_deref() != Some(p.as_str()) - { + + raw_ts_position = match raw_last_ts { + Some(t) if t == view.ts_epoch => raw_ts_position.saturating_add(1), + _ => 1, + }; + raw_last_ts = Some(view.ts_epoch); + + let current_scope = access.current_cursor_scope(); + validate_scope_continuation(&effective_scope, ¤t_scope)?; + effective_scope = current_scope; + + // Identity-anchored cursors skip through the exact entry returned at + // the prior page boundary. Legacy cursors retain timestamp/offset + // positioning for compatibility. + if !anchor_found { + if anchor_entry_id.as_ref() == Some(&entry.id.0) { + anchor_found = true; + } continue; } - - // Cursor positioning: drop everything strictly newer than - // the cursor's `ts`, then skip the first `cursor_offset` - // entries that share `ts` (those were on the prior page). - if let Some(c_ts) = cursor_ts { + if anchor_entry_id.is_none() + && let Some(c_ts) = cursor_ts + { if view.ts_epoch > c_ts { continue; } - if view.ts_epoch == c_ts && equal_ts_skipped < cursor_offset { - equal_ts_skipped = equal_ts_skipped.saturating_add(1); + if view.ts_epoch == c_ts && raw_ts_position <= cursor_offset { continue; } } - equal_ts_count_in_page = match last_ts { - Some(t) if t == view.ts_epoch => equal_ts_count_in_page.saturating_add(1), - _ => 1, - }; - last_ts = Some(view.ts_epoch); - page.push(view); if page.len() >= limit { + has_more = true; break; } + + if let Some(p) = effective_scope.principal_filter() + && view.principal.as_deref() != Some(p.as_str()) + { + skipped_by_scope_before_limit = true; + continue; + } + + last_visible_ts = Some(view.ts_epoch); + last_visible_ts_position = raw_ts_position; + last_visible_scope = Some(effective_scope.clone()); + last_visible_entry_id = Some(entry.id.0); + page.push(view); + } + + if !anchor_found { + return Err(GatewayError::BadRequest(CURSOR_ANCHOR_MISSING.into())); } - let next_cursor = if page.len() == limit { - last_ts.map(|t| { - let offset = if cursor_ts == Some(t) { - cursor_offset.saturating_add(equal_ts_count_in_page) - } else { - equal_ts_count_in_page - }; - format!("{t}_{offset}") - }) + let next_cursor = if page.len() == limit && (has_more || skipped_by_scope_before_limit) { + last_visible_ts + .zip(last_visible_scope) + .zip(last_visible_entry_id) + .map(|((t, scope), entry_id)| { + format!( + "{t}_{last_visible_ts_position}_{}_{entry_id}", + scope.encode() + ) + }) } else { None }; - (page, next_cursor) + Ok((page, next_cursor)) } -/// Parse an opaque cursor into `(ts_epoch, equal_ts_offset)`. -/// Supports both the v2 shape (`"_"`) and the legacy -/// v1 plain-`""` shape so dashboards holding a v1 cursor across -/// the upgrade don't fail their next paginated fetch. -fn parse_cursor(cursor: Option<&str>) -> GatewayResult<(Option, usize)> { +/// Parse an opaque cursor. New cursors carry a visibility scope and immutable +/// audit-entry anchor. Legacy timestamp/offset cursors remain accepted only for +/// the caller's effective self scope because they cannot safely resume a +/// broader scope. +fn parse_cursor(cursor: Option<&str>) -> GatewayResult { let Some(raw) = cursor else { - return Ok((None, 0)); + return Ok(AuditCursor::default()); }; - if let Some((ts_str, off_str)) = raw.split_once('_') { - let ts = ts_str - .parse::() - .map_err(|_| GatewayError::BadRequest("cursor timestamp must be an integer".into()))?; - let off = off_str.parse::().map_err(|_| { - GatewayError::BadRequest("cursor offset must be a non-negative integer".into()) - })?; - Ok((Some(ts), off)) - } else { + let mut parts = raw.split('_'); + let ts_str = parts.next().unwrap_or_default(); + let Some(off_str) = parts.next() else { let ts = raw.parse::().map_err(|_| { GatewayError::BadRequest("cursor must be \"\" or \"_\"".into()) })?; - Ok((Some(ts), 0)) + return Ok(AuditCursor { + timestamp: Some(ts), + ..AuditCursor::default() + }); + }; + let ts = ts_str + .parse::() + .map_err(|_| GatewayError::BadRequest("cursor timestamp must be an integer".into()))?; + let off = off_str.parse::().map_err(|_| { + GatewayError::BadRequest("cursor offset must be a non-negative integer".into()) + })?; + let Some(scope_str) = parts.next() else { + return Ok(AuditCursor { + timestamp: Some(ts), + same_second_offset: off, + ..AuditCursor::default() + }); + }; + let entry_id_str = parts.next().ok_or_else(|| { + GatewayError::BadRequest("scoped audit cursor must include an entry ID".into()) + })?; + if parts.next().is_some() { + return Err(GatewayError::BadRequest( + "audit cursor contains too many fields".into(), + )); } + let scope = CursorScope::decode(scope_str)?; + let anchor_entry_id = uuid::Uuid::parse_str(entry_id_str) + .map_err(|_| GatewayError::BadRequest("audit cursor entry ID must be a UUID".into()))?; + Ok(AuditCursor { + timestamp: Some(ts), + same_second_offset: off, + scope: Some(scope), + anchor_entry_id: Some(anchor_entry_id), + }) } /// Map an `AuditEntry` into the flat JSON shape we ship over the @@ -360,99 +576,4 @@ fn render_entry(entry: &AuditEntry) -> Option { } #[cfg(test)] -mod tests { - use super::*; - - use astrid_audit::{AuditLog, AuthorizationProof}; - use astrid_core::SessionId; - use astrid_crypto::KeyPair; - - fn admin_action(method: &str, target: Option<&str>) -> AuditAction { - AuditAction::AdminRequest { - method: method.into(), - required_capability: "*".into(), - target_principal: target.map(|s| PrincipalId::new(s).unwrap()), - params: None, - device_key_id: None, - } - } - - #[tokio::test] - async fn render_drops_non_admin_actions() { - // Non-admin entries (MCP tool calls, capsule events) belong - // to a different audit feed; they must not surface in the - // historical-admin view. - let log = AuditLog::in_memory(KeyPair::generate()); - let session = SessionId::from_uuid(uuid::Uuid::nil()); - log.append( - session.clone(), - AuditAction::McpToolCall { - server: "x".into(), - tool: "y".into(), - args_hash: astrid_crypto::ContentHash::from_bytes([0u8; 32]), - }, - AuthorizationProof::System { - reason: "test".into(), - }, - AuditOutcome::Success { details: None }, - ) - .await - .expect("append"); - let entries = log.get_session_entries(&session).await.expect("read"); - assert_eq!(entries.len(), 1); - assert!( - render_entry(&entries[0]).is_none(), - "McpToolCall must not render into the admin-history view" - ); - } - - #[tokio::test] - async fn render_admin_request_round_trips() { - let log = AuditLog::in_memory(KeyPair::generate()); - let session = SessionId::from_uuid(uuid::Uuid::nil()); - log.append_with_principal( - session.clone(), - PrincipalId::new("admin").unwrap(), - admin_action("AgentDelete", Some("alice")), - AuthorizationProof::System { - reason: "test".into(), - }, - AuditOutcome::Success { details: None }, - ) - .await - .expect("append"); - let entries = log.get_session_entries(&session).await.expect("read"); - let view = render_entry(&entries[0]).expect("admin entry must render"); - assert_eq!(view.method.as_deref(), Some("AgentDelete")); - assert_eq!(view.principal.as_deref(), Some("admin")); - assert_eq!(view.target_principal.as_deref(), Some("alice")); - assert_eq!(view.outcome, "success"); - } - - #[test] - fn parse_cursor_handles_v1_and_v2_shapes() { - // v1 (legacy): bare integer, no underscore — offset - // defaults to 0. We accept this shape so v0.7.0 cursors - // already in flight don't fail the next paginated fetch. - let (ts, off) = parse_cursor(Some("1700000000")).expect("bare ts parses"); - assert_eq!(ts, Some(1_700_000_000)); - assert_eq!(off, 0); - - // v2: `_` — same-second batches resume cleanly - // without losing or duplicating entries across the page - // boundary. - let (ts, off) = parse_cursor(Some("1700000000_3")).expect("v2 cursor parses"); - assert_eq!(ts, Some(1_700_000_000)); - assert_eq!(off, 3); - - // None: no cursor → no positioning, start from newest. - let (ts, off) = parse_cursor(None).expect("None passes"); - assert_eq!(ts, None); - assert_eq!(off, 0); - - // Garbage rejected with `BadRequest`. - assert!(parse_cursor(Some("not-a-number")).is_err()); - assert!(parse_cursor(Some("123_not-a-number")).is_err()); - assert!(parse_cursor(Some("not-a-number_4")).is_err()); - } -} +mod tests; diff --git a/crates/astrid-gateway/src/routes/audit/tests.rs b/crates/astrid-gateway/src/routes/audit/tests.rs new file mode 100644 index 000000000..55c3d02a1 --- /dev/null +++ b/crates/astrid-gateway/src/routes/audit/tests.rs @@ -0,0 +1,867 @@ +use super::*; + +use std::sync::atomic::{AtomicUsize, Ordering}; + +use astrid_audit::{AuditLog, AuthorizationProof}; +use astrid_core::{SessionId, Timestamp}; +use astrid_crypto::KeyPair; +use chrono::TimeZone; + +fn admin_action(method: &str, target: Option<&str>) -> AuditAction { + AuditAction::AdminRequest { + method: method.into(), + required_capability: "*".into(), + target_principal: target.map(|s| PrincipalId::new(s).unwrap()), + params: None, + device_key_id: None, + } +} + +#[tokio::test] +async fn render_drops_non_admin_actions() { + // Non-admin entries (MCP tool calls, capsule events) belong + // to a different audit feed; they must not surface in the + // historical-admin view. + let log = AuditLog::in_memory(KeyPair::generate()); + let session = SessionId::from_uuid(uuid::Uuid::nil()); + log.append( + session.clone(), + AuditAction::McpToolCall { + server: "x".into(), + tool: "y".into(), + args_hash: astrid_crypto::ContentHash::from_bytes([0u8; 32]), + }, + AuthorizationProof::System { + reason: "test".into(), + }, + AuditOutcome::Success { details: None }, + ) + .await + .expect("append"); + let entries = log.get_session_entries(&session).await.expect("read"); + assert_eq!(entries.len(), 1); + assert!( + render_entry(&entries[0]).is_none(), + "McpToolCall must not render into the admin-history view" + ); +} + +#[tokio::test] +async fn render_admin_request_round_trips() { + let log = AuditLog::in_memory(KeyPair::generate()); + let session = SessionId::from_uuid(uuid::Uuid::nil()); + log.append_with_principal( + session.clone(), + PrincipalId::new("admin").unwrap(), + admin_action("AgentDelete", Some("alice")), + AuthorizationProof::System { + reason: "test".into(), + }, + AuditOutcome::Success { details: None }, + ) + .await + .expect("append"); + let entries = log.get_session_entries(&session).await.expect("read"); + let view = render_entry(&entries[0]).expect("admin entry must render"); + assert_eq!(view.method.as_deref(), Some("AgentDelete")); + assert_eq!(view.principal.as_deref(), Some("admin")); + assert_eq!(view.target_principal.as_deref(), Some("alice")); + assert_eq!(view.outcome, "success"); +} + +#[tokio::test] +async fn pagination_narrows_live_without_hiding_the_callers_records() { + let log = AuditLog::in_memory(KeyPair::generate()); + let session = SessionId::from_uuid(uuid::Uuid::nil()); + for (principal, method) in [ + ("alice", "AliceOwnAfterNarrowing"), + ("bob", "BobHiddenAfterNarrowing"), + ("bob", "BobVisibleBeforeNarrowing"), + ] { + log.append_with_principal( + session.clone(), + PrincipalId::new(principal).unwrap(), + admin_action(method, None), + AuthorizationProof::System { + reason: "test".into(), + }, + AuditOutcome::Success { details: None }, + ) + .await + .expect("append"); + } + let mut entries = log.get_session_entries(&session).await.expect("read"); + entries.reverse(); + + let checks = Arc::new(AtomicUsize::new(0)); + let checks_for_probe = Arc::clone(&checks); + let capability_probe = super::super::events::CapabilityProbe::new(move |_, _, _| { + matches!(checks_for_probe.fetch_add(1, Ordering::SeqCst), 0 | 1) + }); + let caller = PrincipalId::new("alice").unwrap(); + let access = AuditAccess { + capability_probe: &capability_probe, + caller_principal: &caller, + device_key_id: Some("0123456789abcdef"), + requested_principal: None, + }; + + let (page, _) = paginate_page( + entries, + &AuditQuery::default(), + &access, + AuditCursor::default(), + DEFAULT_LIMIT, + ) + .expect("pagination"); + let methods: Vec<_> = page + .iter() + .filter_map(|entry| entry.method.as_deref()) + .collect(); + + assert_eq!( + methods, + vec!["BobVisibleBeforeNarrowing", "AliceOwnAfterNarrowing"] + ); + assert!(!methods.contains(&"BobHiddenAfterNarrowing")); +} + +#[allow(clippy::too_many_lines)] +#[tokio::test] +async fn pagination_cursor_survives_live_narrowing_with_same_second_batch() { + let log = AuditLog::in_memory(KeyPair::generate()); + let session = SessionId::from_uuid(uuid::Uuid::nil()); + for (principal, method) in [ + ("alice", "AliceOlder"), + ("alice", "AliceVisibleAfterNarrowing"), + ("bob", "BobHiddenAfterNarrowing"), + ("bob", "BobVisibleBeforeNarrowing"), + ] { + log.append_with_principal( + session.clone(), + PrincipalId::new(principal).unwrap(), + admin_action(method, None), + AuthorizationProof::System { + reason: "test".into(), + }, + AuditOutcome::Success { details: None }, + ) + .await + .expect("append"); + } + let mut entries = log.get_session_entries(&session).await.expect("read"); + entries.reverse(); + let same_second = Timestamp::from_datetime( + chrono::Utc + .timestamp_opt(1_700_000_000, 0) + .single() + .unwrap(), + ); + let next_second = Timestamp::from_datetime( + chrono::Utc + .timestamp_opt(1_699_999_999, 0) + .single() + .unwrap(), + ); + for entry in &mut entries[..3] { + entry.timestamp = same_second; + } + entries[3].timestamp = next_second; + + let firehose_probe = super::super::events::CapabilityProbe::new(|_, _, _| true); + let self_only_probe = super::super::events::CapabilityProbe::new(|_, _, _| false); + let caller = PrincipalId::new("alice").unwrap(); + let firehose_access = AuditAccess { + capability_probe: &firehose_probe, + caller_principal: &caller, + device_key_id: Some("0123456789abcdef"), + requested_principal: None, + }; + let self_only_access = AuditAccess { + capability_probe: &self_only_probe, + caller_principal: &caller, + device_key_id: Some("0123456789abcdef"), + requested_principal: None, + }; + + let (page_one, next_cursor) = paginate_page( + entries.clone(), + &AuditQuery::default(), + &firehose_access, + AuditCursor::default(), + 1, + ) + .expect("page one"); + assert_eq!( + page_one[0].method.as_deref(), + Some("BobVisibleBeforeNarrowing") + ); + let mut cursor = parse_cursor(next_cursor.as_deref()).expect("page-one cursor parses"); + assert_eq!(cursor.timestamp, Some(1_700_000_000)); + assert_eq!(cursor.same_second_offset, 1); + assert_eq!(cursor.scope, Some(CursorScope::All)); + assert_eq!(cursor.anchor_entry_id, Some(entries[0].id.0)); + cursor.scope = validate_cursor_scope( + cursor.timestamp, + cursor.scope.as_ref(), + &AuditQuery::default(), + &self_only_access, + ) + .expect("all-scope cursor may narrow to self-only"); + let (page_two, next_cursor) = paginate_page( + entries.clone(), + &AuditQuery::default(), + &self_only_access, + cursor, + 1, + ) + .expect("page two"); + assert_eq!( + page_two[0].method.as_deref(), + Some("AliceVisibleAfterNarrowing") + ); + let mut cursor = parse_cursor(next_cursor.as_deref()).expect("page-two cursor parses"); + assert_eq!(cursor.timestamp, Some(1_700_000_000)); + assert_eq!(cursor.same_second_offset, 3); + assert_eq!(cursor.scope, Some(CursorScope::Principal(caller.clone()))); + assert_eq!(cursor.anchor_entry_id, Some(entries[2].id.0)); + cursor.scope = validate_cursor_scope( + cursor.timestamp, + cursor.scope.as_ref(), + &AuditQuery::default(), + &self_only_access, + ) + .expect("self-only cursor continues under same scope"); + let (page_three, next_cursor) = paginate_page( + entries.clone(), + &AuditQuery::default(), + &self_only_access, + cursor, + 1, + ) + .expect("page three"); + assert_eq!(page_three[0].method.as_deref(), Some("AliceOlder")); + assert_eq!(next_cursor.as_deref(), None); +} + +#[tokio::test] +async fn identity_cursor_survives_same_second_append_between_pages() { + let log = AuditLog::in_memory(KeyPair::generate()); + let session = SessionId::from_uuid(uuid::Uuid::nil()); + for method in ["Older", "PageOne"] { + log.append_with_principal( + session.clone(), + PrincipalId::new("alice").unwrap(), + admin_action(method, None), + AuthorizationProof::System { + reason: "test".into(), + }, + AuditOutcome::Success { details: None }, + ) + .await + .expect("append"); + } + let timestamp = Timestamp::from_datetime( + chrono::Utc + .timestamp_opt(1_700_000_000, 0) + .single() + .unwrap(), + ); + let mut first_snapshot = log.get_session_entries(&session).await.expect("read"); + first_snapshot.reverse(); + for entry in &mut first_snapshot { + entry.timestamp = timestamp; + } + + let self_only_probe = super::super::events::CapabilityProbe::new(|_, _, _| false); + let caller = PrincipalId::new("alice").unwrap(); + let access = AuditAccess { + capability_probe: &self_only_probe, + caller_principal: &caller, + device_key_id: Some("0123456789abcdef"), + requested_principal: None, + }; + + let (page_one, next_cursor) = paginate_page( + first_snapshot.clone(), + &AuditQuery::default(), + &access, + AuditCursor::default(), + 1, + ) + .expect("page one"); + assert_eq!(page_one[0].method.as_deref(), Some("PageOne")); + + let mut cursor = parse_cursor(next_cursor.as_deref()).expect("identity cursor"); + assert_eq!(cursor.anchor_entry_id, Some(first_snapshot[0].id.0)); + cursor.scope = validate_cursor_scope( + cursor.timestamp, + cursor.scope.as_ref(), + &AuditQuery::default(), + &access, + ) + .expect("self scope continues"); + + log.append_with_principal( + session.clone(), + caller.clone(), + admin_action("AppendedAfterPageOne", None), + AuthorizationProof::System { + reason: "test".into(), + }, + AuditOutcome::Success { details: None }, + ) + .await + .expect("append between pages"); + let mut second_snapshot = log.get_session_entries(&session).await.expect("read"); + second_snapshot.reverse(); + for entry in &mut second_snapshot { + entry.timestamp = timestamp; + } + + let (page_two, next_cursor) = + paginate_page(second_snapshot, &AuditQuery::default(), &access, cursor, 2) + .expect("page two"); + let methods: Vec<_> = page_two + .iter() + .filter_map(|entry| entry.method.as_deref()) + .collect(); + assert_eq!(methods, vec!["Older"]); + assert_eq!(next_cursor, None); +} + +#[tokio::test] +async fn identity_cursor_missing_anchor_requires_restart() { + let log = AuditLog::in_memory(KeyPair::generate()); + let session = SessionId::from_uuid(uuid::Uuid::nil()); + log.append_with_principal( + session.clone(), + PrincipalId::new("alice").unwrap(), + admin_action("OnlyEntry", None), + AuthorizationProof::System { + reason: "test".into(), + }, + AuditOutcome::Success { details: None }, + ) + .await + .expect("append"); + let mut entries = log.get_session_entries(&session).await.expect("read"); + entries.reverse(); + + let self_only_probe = super::super::events::CapabilityProbe::new(|_, _, _| false); + let caller = PrincipalId::new("alice").unwrap(); + let access = AuditAccess { + capability_probe: &self_only_probe, + caller_principal: &caller, + device_key_id: Some("0123456789abcdef"), + requested_principal: None, + }; + let cursor = AuditCursor { + timestamp: Some(1_700_000_000), + same_second_offset: 1, + scope: Some(CursorScope::Principal(caller.clone())), + anchor_entry_id: Some(uuid::Uuid::new_v4()), + }; + + let err = paginate_page(entries, &AuditQuery::default(), &access, cursor, 1) + .expect_err("missing anchor must restart pagination"); + assert!(err.to_string().contains(CURSOR_ANCHOR_MISSING)); +} + +#[tokio::test] +async fn exact_limit_page_preserves_scope_cursor_after_hidden_row() { + let log = AuditLog::in_memory(KeyPair::generate()); + let session = SessionId::from_uuid(uuid::Uuid::nil()); + for (principal, method) in [ + ("alice", "AliceTerminal"), + ("bob", "BobHiddenBeforeTerminal"), + ] { + log.append_with_principal( + session.clone(), + PrincipalId::new(principal).unwrap(), + admin_action(method, None), + AuthorizationProof::System { + reason: "test".into(), + }, + AuditOutcome::Success { details: None }, + ) + .await + .expect("append"); + } + let mut entries = log.get_session_entries(&session).await.expect("read"); + entries.reverse(); + + let self_only_probe = super::super::events::CapabilityProbe::new(|_, _, _| false); + let firehose_probe = super::super::events::CapabilityProbe::new(|_, _, _| true); + let caller = PrincipalId::new("alice").unwrap(); + let access = AuditAccess { + capability_probe: &self_only_probe, + caller_principal: &caller, + device_key_id: Some("0123456789abcdef"), + requested_principal: None, + }; + let widened_access = AuditAccess { + capability_probe: &firehose_probe, + caller_principal: &caller, + device_key_id: Some("0123456789abcdef"), + requested_principal: None, + }; + + let (page, next_cursor) = paginate_page( + entries, + &AuditQuery::default(), + &access, + AuditCursor::default(), + 1, + ) + .expect("terminal page"); + + assert_eq!(page.len(), 1); + assert_eq!(page[0].method.as_deref(), Some("AliceTerminal")); + let cursor = parse_cursor(next_cursor.as_deref()).expect("scope-bound cursor"); + let err = validate_cursor_scope( + cursor.timestamp, + cursor.scope.as_ref(), + &AuditQuery::default(), + &widened_access, + ) + .expect_err("widening after a hidden row must restart pagination"); + assert!(err.to_string().contains(CURSOR_SCOPE_CHANGED)); +} + +#[tokio::test] +async fn exact_limit_physical_eof_has_no_continuation_cursor() { + let log = AuditLog::in_memory(KeyPair::generate()); + let session = SessionId::from_uuid(uuid::Uuid::nil()); + log.append_with_principal( + session.clone(), + PrincipalId::new("alice").unwrap(), + admin_action("AliceTerminal", None), + AuthorizationProof::System { + reason: "test".into(), + }, + AuditOutcome::Success { details: None }, + ) + .await + .expect("append"); + let mut entries = log.get_session_entries(&session).await.expect("read"); + entries.reverse(); + + let self_only_probe = super::super::events::CapabilityProbe::new(|_, _, _| false); + let caller = PrincipalId::new("alice").unwrap(); + let access = AuditAccess { + capability_probe: &self_only_probe, + caller_principal: &caller, + device_key_id: Some("0123456789abcdef"), + requested_principal: None, + }; + + let (page, next_cursor) = paginate_page( + entries, + &AuditQuery::default(), + &access, + AuditCursor::default(), + 1, + ) + .expect("terminal page"); + + assert_eq!(page.len(), 1); + assert_eq!(page[0].method.as_deref(), Some("AliceTerminal")); + assert_eq!(next_cursor, None); +} + +#[test] +fn parse_cursor_handles_legacy_and_identity_anchored_shapes() { + // v1 (legacy): bare integer, no underscore — offset + // defaults to 0. Validation later accepts this shape only + // for an unambiguous default self-view continuation. + let cursor = parse_cursor(Some("1700000000")).expect("bare ts parses"); + assert_eq!(cursor.timestamp, Some(1_700_000_000)); + assert_eq!(cursor.same_second_offset, 0); + assert_eq!(cursor.scope, None); + assert_eq!(cursor.anchor_entry_id, None); + + // v2: `_` — same-second batches resume cleanly + // without losing or duplicating entries across the page + // boundary. + let cursor = parse_cursor(Some("1700000000_3")).expect("v2 cursor parses"); + assert_eq!(cursor.timestamp, Some(1_700_000_000)); + assert_eq!(cursor.same_second_offset, 3); + assert_eq!(cursor.scope, None); + assert_eq!(cursor.anchor_entry_id, None); + + // New cursors carry both the effective scope and immutable entry anchor. + let entry_id = uuid::Uuid::from_u128(1); + let cursor = parse_cursor(Some( + "1700000000_3_p616c696365_00000000-0000-0000-0000-000000000001", + )) + .expect("identity-anchored cursor parses"); + assert_eq!(cursor.timestamp, Some(1_700_000_000)); + assert_eq!(cursor.same_second_offset, 3); + assert_eq!( + cursor.scope, + Some(CursorScope::Principal(PrincipalId::new("alice").unwrap())) + ); + assert_eq!(cursor.anchor_entry_id, Some(entry_id)); + + // None: no cursor → no positioning, start from newest. + assert_eq!( + parse_cursor(None).expect("None passes"), + AuditCursor::default() + ); + + // Garbage rejected with `BadRequest`. + assert!(parse_cursor(Some("not-a-number")).is_err()); + assert!(parse_cursor(Some("123_not-a-number")).is_err()); + assert!(parse_cursor(Some("not-a-number_4")).is_err()); + assert!(parse_cursor(Some("1700000000_3_p616c696365")).is_err()); + assert!(parse_cursor(Some("1700000000_3_all_not-a-uuid")).is_err()); +} + +#[tokio::test] +async fn pagination_cursor_rejects_scope_widening_after_self_only_page() { + let log = AuditLog::in_memory(KeyPair::generate()); + let session = SessionId::from_uuid(uuid::Uuid::nil()); + for (principal, method) in [ + ("alice", "AliceVisibleWhileScoped"), + ("bob", "BobNewlyVisibleAfterWidening"), + ] { + log.append_with_principal( + session.clone(), + PrincipalId::new(principal).unwrap(), + admin_action(method, None), + AuthorizationProof::System { + reason: "test".into(), + }, + AuditOutcome::Success { details: None }, + ) + .await + .expect("append"); + } + let mut entries = log.get_session_entries(&session).await.expect("read"); + entries.reverse(); + let same_second = Timestamp::from_datetime( + chrono::Utc + .timestamp_opt(1_700_000_000, 0) + .single() + .unwrap(), + ); + for entry in &mut entries { + entry.timestamp = same_second; + } + + let self_only_probe = super::super::events::CapabilityProbe::new(|_, _, _| false); + let firehose_probe = super::super::events::CapabilityProbe::new(|_, _, _| true); + let caller = PrincipalId::new("alice").unwrap(); + let self_only_access = AuditAccess { + capability_probe: &self_only_probe, + caller_principal: &caller, + device_key_id: Some("0123456789abcdef"), + requested_principal: None, + }; + let firehose_access = AuditAccess { + capability_probe: &firehose_probe, + caller_principal: &caller, + device_key_id: Some("0123456789abcdef"), + requested_principal: None, + }; + + let (page_one, next_cursor) = paginate_page( + entries, + &AuditQuery::default(), + &self_only_access, + AuditCursor::default(), + 1, + ) + .expect("page one"); + assert_eq!( + page_one[0].method.as_deref(), + Some("AliceVisibleWhileScoped") + ); + let cursor = parse_cursor(next_cursor.as_deref()).expect("cursor parses"); + let err = validate_cursor_scope( + cursor.timestamp, + cursor.scope.as_ref(), + &AuditQuery::default(), + &firehose_access, + ) + .expect_err("widening must fail closed"); + assert!( + err.to_string() + .contains("restart pagination without a cursor"), + "unexpected error: {err}" + ); +} + +#[tokio::test] +async fn pagination_rejects_mid_page_scope_widening_after_visible_rows() { + let log = AuditLog::in_memory(KeyPair::generate()); + let session = SessionId::from_uuid(uuid::Uuid::nil()); + for (principal, method) in [ + ("bob", "BobVisibleAfterRewidening"), + ("alice", "AliceVisibleWhileScoped"), + ("bob", "BobHiddenWhileScoped"), + ("bob", "BobVisibleBeforeNarrowing"), + ] { + log.append_with_principal( + session.clone(), + PrincipalId::new(principal).unwrap(), + admin_action(method, None), + AuthorizationProof::System { + reason: "test".into(), + }, + AuditOutcome::Success { details: None }, + ) + .await + .expect("append"); + } + let mut entries = log.get_session_entries(&session).await.expect("read"); + entries.reverse(); + + let checks = Arc::new(AtomicUsize::new(0)); + let checks_for_probe = Arc::clone(&checks); + let capability_probe = super::super::events::CapabilityProbe::new(move |_, _, _| { + matches!(checks_for_probe.fetch_add(1, Ordering::SeqCst), 0 | 1 | 4) + }); + let caller = PrincipalId::new("alice").unwrap(); + let access = AuditAccess { + capability_probe: &capability_probe, + caller_principal: &caller, + device_key_id: Some("0123456789abcdef"), + requested_principal: None, + }; + + let err = paginate_page( + entries, + &AuditQuery::default(), + &access, + AuditCursor::default(), + 3, + ) + .expect_err("mid-page widening must restart pagination"); + assert!( + err.to_string().contains(CURSOR_SCOPE_CHANGED), + "unexpected error: {err}" + ); +} + +#[tokio::test] +async fn pagination_rejects_scope_widening_before_first_visible_row() { + let log = AuditLog::in_memory(KeyPair::generate()); + let session = SessionId::from_uuid(uuid::Uuid::nil()); + for (principal, method) in [ + ("alice", "AliceVisibleAfterWidening"), + ("bob", "BobHiddenWhileScoped"), + ] { + log.append_with_principal( + session.clone(), + PrincipalId::new(principal).unwrap(), + admin_action(method, None), + AuthorizationProof::System { + reason: "test".into(), + }, + AuditOutcome::Success { details: None }, + ) + .await + .expect("append"); + } + let mut entries = log.get_session_entries(&session).await.expect("read"); + entries.reverse(); + + let checks = Arc::new(AtomicUsize::new(0)); + let checks_for_probe = Arc::clone(&checks); + let capability_probe = super::super::events::CapabilityProbe::new(move |_, _, _| { + checks_for_probe.fetch_add(1, Ordering::SeqCst) == 2 + }); + let caller = PrincipalId::new("alice").unwrap(); + let access = AuditAccess { + capability_probe: &capability_probe, + caller_principal: &caller, + device_key_id: Some("0123456789abcdef"), + requested_principal: None, + }; + + let err = paginate_page( + entries, + &AuditQuery::default(), + &access, + AuditCursor::default(), + 3, + ) + .expect_err("widening before the first visible row must not signal EOF"); + assert!( + err.to_string().contains(CURSOR_SCOPE_CHANGED), + "unexpected error: {err}" + ); +} + +#[tokio::test] +async fn identity_cursor_rejects_widening_while_seeking_anchor() { + let log = AuditLog::in_memory(KeyPair::generate()); + let session = SessionId::from_uuid(uuid::Uuid::nil()); + for (principal, method) in [ + ("alice", "PriorPageAnchor"), + ("bob", "AppendedBeforeAnchor"), + ] { + log.append_with_principal( + session.clone(), + PrincipalId::new(principal).unwrap(), + admin_action(method, None), + AuthorizationProof::System { + reason: "test".into(), + }, + AuditOutcome::Success { details: None }, + ) + .await + .expect("append"); + } + let mut entries = log.get_session_entries(&session).await.expect("read"); + entries.reverse(); + let cursor_ts = 1_700_000_000_u64; + let timestamp = Timestamp::from_datetime( + chrono::Utc + .timestamp_opt(i64::try_from(cursor_ts).unwrap(), 0) + .single() + .unwrap(), + ); + for entry in &mut entries { + entry.timestamp = timestamp; + } + + let checks = Arc::new(AtomicUsize::new(0)); + let checks_for_probe = Arc::clone(&checks); + let capability_probe = super::super::events::CapabilityProbe::new(move |_, _, _| { + checks_for_probe.fetch_add(1, Ordering::SeqCst) >= 2 + }); + let caller = PrincipalId::new("alice").unwrap(); + let access = AuditAccess { + capability_probe: &capability_probe, + caller_principal: &caller, + device_key_id: Some("0123456789abcdef"), + requested_principal: None, + }; + let mut cursor = AuditCursor { + timestamp: Some(cursor_ts), + same_second_offset: 1, + scope: Some(CursorScope::Principal(caller.clone())), + anchor_entry_id: Some(entries[1].id.0), + }; + + cursor.scope = validate_cursor_scope( + cursor.timestamp, + cursor.scope.as_ref(), + &AuditQuery::default(), + &access, + ) + .expect("principal cursor initially matches principal visibility"); + let err = paginate_page(entries, &AuditQuery::default(), &access, cursor, 1) + .expect_err("widening while seeking the identity anchor must restart pagination"); + assert!( + err.to_string().contains(CURSOR_SCOPE_CHANGED), + "unexpected error: {err}" + ); +} + +#[test] +fn legacy_cursor_self_view_stays_compatible() { + let self_only_probe = super::super::events::CapabilityProbe::new(|_, _, _| false); + let caller = PrincipalId::new("alice").unwrap(); + let access = AuditAccess { + capability_probe: &self_only_probe, + caller_principal: &caller, + device_key_id: Some("0123456789abcdef"), + requested_principal: None, + }; + let cursor = parse_cursor(Some("1700000000_3")).expect("legacy cursor parses"); + + let bound_scope = validate_cursor_scope( + cursor.timestamp, + cursor.scope.as_ref(), + &AuditQuery::default(), + &access, + ) + .expect("legacy self-view cursor remains compatible"); + assert_eq!(bound_scope, Some(CursorScope::Principal(caller))); +} + +#[test] +fn legacy_cursor_rejects_widening_between_validation_and_pagination() { + let checks = Arc::new(AtomicUsize::new(0)); + let checks_for_probe = Arc::clone(&checks); + let capability_probe = super::super::events::CapabilityProbe::new(move |_, _, _| { + checks_for_probe.fetch_add(1, Ordering::SeqCst) >= 1 + }); + let caller = PrincipalId::new("alice").unwrap(); + let access = AuditAccess { + capability_probe: &capability_probe, + caller_principal: &caller, + device_key_id: Some("0123456789abcdef"), + requested_principal: None, + }; + let mut cursor = parse_cursor(Some("1700000000_3")).expect("legacy cursor parses"); + + cursor.scope = validate_cursor_scope( + cursor.timestamp, + cursor.scope.as_ref(), + &AuditQuery::default(), + &access, + ) + .expect("legacy cursor validates while authority is self-only"); + let err = paginate_page(Vec::new(), &AuditQuery::default(), &access, cursor, 1) + .expect_err("widening after legacy validation must restart pagination"); + + assert!( + err.to_string().contains(CURSOR_SCOPE_CHANGED), + "unexpected error: {err}" + ); +} + +#[test] +fn legacy_cursor_rejects_firehose_resume() { + let firehose_probe = super::super::events::CapabilityProbe::new(|_, _, _| true); + let caller = PrincipalId::new("alice").unwrap(); + let access = AuditAccess { + capability_probe: &firehose_probe, + caller_principal: &caller, + device_key_id: Some("0123456789abcdef"), + requested_principal: None, + }; + let cursor = parse_cursor(Some("1700000000_3")).expect("legacy cursor parses"); + let err = validate_cursor_scope( + cursor.timestamp, + cursor.scope.as_ref(), + &AuditQuery::default(), + &access, + ) + .expect_err("legacy firehose resume must restart pagination"); + + assert!( + err.to_string().contains(CURSOR_SCOPE_CHANGED), + "unexpected error: {err}" + ); +} + +#[test] +fn legacy_cursor_rejects_explicit_principal_resume() { + let firehose_probe = super::super::events::CapabilityProbe::new(|_, _, _| true); + let caller = PrincipalId::new("alice").unwrap(); + let requested = PrincipalId::new("bob").unwrap(); + let access = AuditAccess { + capability_probe: &firehose_probe, + caller_principal: &caller, + device_key_id: Some("0123456789abcdef"), + requested_principal: Some(&requested), + }; + let cursor = parse_cursor(Some("1700000000_3")).expect("legacy cursor parses"); + let query = AuditQuery { + principal: Some("bob".into()), + ..AuditQuery::default() + }; + let err = validate_cursor_scope(cursor.timestamp, cursor.scope.as_ref(), &query, &access) + .expect_err("legacy explicit-principal resume must restart pagination"); + + assert!( + err.to_string().contains(CURSOR_SCOPE_CHANGED), + "unexpected error: {err}" + ); +} diff --git a/crates/astrid-gateway/src/routes/events.rs b/crates/astrid-gateway/src/routes/events.rs index e320e86be..c70fc6c4a 100644 --- a/crates/astrid-gateway/src/routes/events.rs +++ b/crates/astrid-gateway/src/routes/events.rs @@ -39,6 +39,32 @@ pub const AUDIT_TOPIC: &str = "astrid.v1.audit.entry"; /// `GET /api/sys/audit` route in [`super::audit`]. pub(super) const AUDIT_FIREHOSE_CAP: &str = "audit:read_all"; +type CapabilityEvaluator = dyn Fn(&PrincipalId, Option<&str>, &str) -> bool + Send + Sync + 'static; + +#[derive(Clone)] +pub(crate) struct CapabilityProbe(Arc); + +impl CapabilityProbe { + pub(crate) fn new( + evaluator: impl Fn(&PrincipalId, Option<&str>, &str) -> bool + Send + Sync + 'static, + ) -> Self { + Self(Arc::new(evaluator)) + } + + pub(crate) fn deny_all() -> Self { + Self::new(|_, _, _| false) + } + + fn allows( + &self, + principal: &PrincipalId, + device_key_id: Option<&str>, + capability: &str, + ) -> bool { + (self.0)(principal, device_key_id, capability) + } +} + /// `GET /api/events` — opens a long-lived Server-Sent Events /// stream. The connection stays open until the client disconnects /// or the daemon shuts down. @@ -57,6 +83,11 @@ pub async fn get_events( req: Request, ) -> GatewayResult>>> { let caller = caller_from(&req)?.clone(); + let capability_probe = req + .extensions() + .get::() + .cloned() + .unwrap_or_else(CapabilityProbe::deny_all); // Without a bus handle (the standalone GatewayState ctor used // by route-level tests), report an honest 502 instead of @@ -68,40 +99,50 @@ pub async fn get_events( ))); }; - // Resolve whether the caller gets the firehose or the - // per-principal filtered view. The caller's capability set is - // expressed in their bearer — we don't have it directly, so - // ask the kernel via AgentList and look for the caller's row. - // AgentList is cap-gated by self:agent:list for agents and by - // `*` for admins, so the call always succeeds for any valid - // bearer. The kernel filters by scope server-side. - // - // Use the bus-direct admin client (not the socket-based one) — - // SSE handshakes happen once per dashboard tab open, and the - // socket-dial latency would otherwise dominate first-byte - // time for the audit stream. - let firehose = caller_holds( - &state, + // The kernel-owned probe applies the caller's live device scope. + let initial_firehose = caller_holds( + &capability_probe, &caller.principal, caller.device_key_id.as_deref(), AUDIT_FIREHOSE_CAP, - ) - .await; + ); // Routed subscription so the audit firehose gets the same // per-(topic, principal) DRR fairness the rest of the gateway // SSE streams now use (#813 Layer 4). The principal-firehose // filter at the post-receive layer is unchanged — it's a // capability gate, not a routing concern. - let mut receiver = bus.subscribe_topic_routed( + let receiver = bus.subscribe_topic_routed( state.gateway_route_uuid, AUDIT_TOPIC, "gateway", "gateway::audit_sse", ); - let caller_principal = caller.principal; + let stream = audit_event_stream( + receiver, + capability_probe, + caller.principal, + caller.device_key_id, + initial_firehose, + ); - let stream = async_stream::stream! { + // Heartbeat every 15s — keeps NAT/proxy state alive and lets + // clients distinguish "idle stream" from "daemon crashed". + Ok(Sse::new(stream.boxed()).keep_alive( + KeepAlive::new() + .interval(Duration::from_secs(15)) + .text("keep-alive"), + )) +} + +fn audit_event_stream( + mut receiver: astrid_events::RoutedEventReceiver, + capability_probe: CapabilityProbe, + caller_principal: PrincipalId, + device_key_id: Option, + initial_firehose: bool, +) -> impl Stream> { + async_stream::stream! { // Initial handshake event so clients can confirm the // stream opened without waiting on the first audit entry. yield Ok::( @@ -109,7 +150,7 @@ pub async fn get_events( .event("ready") .data(serde_json::json!({ "principal": caller_principal.to_string(), - "firehose": firehose, + "firehose": initial_firehose, }).to_string()) ); @@ -120,10 +161,14 @@ pub async fn get_events( let IpcPayload::RawJson(val) = &message.payload else { continue; }; - // Per-principal filter for non-firehose subscribers. - // The kernel-side broadcast embeds `principal` (the - // acting caller); if that's not present or doesn't - // match, skip silently. + // Re-evaluate long-lived streams so policy changes and device + // revocation narrow the feed without waiting for reconnect. + let firehose = caller_holds( + &capability_probe, + &caller_principal, + device_key_id.as_deref(), + AUDIT_FIREHOSE_CAP, + ); if !firehose { let entry_principal = val.get("principal").and_then(serde_json::Value::as_str); if entry_principal != Some(caller_principal.as_str()) { @@ -133,54 +178,103 @@ pub async fn get_events( let Ok(payload) = serde_json::to_string(val) else { continue }; yield Ok(Event::default().event("audit").data(payload)); } - }; - - // Heartbeat every 15s — keeps NAT/proxy state alive and lets - // clients distinguish "idle stream" from "daemon crashed". - Ok(Sse::new(stream.boxed()).keep_alive( - KeepAlive::new() - .interval(Duration::from_secs(15)) - .text("keep-alive"), - )) + } } -/// Best-effort capability check via the kernel's `AgentList`. Returns -/// `false` on any failure (parse error, bus unavailable) so the -/// caller falls back to the safer per-principal filter rather than -/// accidentally widening to the firehose. -pub(super) async fn caller_holds( - state: &GatewayState, +/// The deny-all default keeps callers on the per-principal view. +pub(super) fn caller_holds( + capability_probe: &CapabilityProbe, principal: &PrincipalId, device_key_id: Option<&str>, capability: &str, ) -> bool { - use astrid_core::kernel_api::{AdminRequestKind, AdminResponseBody}; - // Carry the session's device scope: a device-scoped caller whose scope - // denies `self:agent:list` cannot read its own row, so the firehose check - // fails closed to the narrower per-principal view — a scoped device must - // not gain the audit firehose its principal would otherwise hold. - let Ok(client) = state - .admin_client(principal.clone()) - .map(|c| c.with_device_key_id(device_key_id.map(str::to_owned))) - else { - return false; - }; - let Ok(resp) = client.request(AdminRequestKind::AgentList).await else { - return false; - }; - let AdminResponseBody::AgentList(list) = resp else { - return false; - }; - // Approximate: caller holds the cap if their direct grants - // include it or if they're in the admin group. Group-level - // inheritance resolution proper lives kernel-side; the gateway - // doesn't have a public API for it, so we recognise the - // bootstrap shape (admin → universal grant) and explicit - // direct grants. - list.into_iter() - .find(|s| &s.principal == principal) - .is_some_and(|s| { - s.groups.iter().any(|g| g == "admin") - || s.grants.iter().any(|g| g == capability || g == "*") - }) + capability_probe.allows(principal, device_key_id, capability) +} + +#[cfg(test)] +mod tests { + use super::*; + + use std::sync::atomic::{AtomicBool, Ordering}; + + use astrid_events::ipc::{IpcMessage, Topic}; + use astrid_events::{AstridEvent, EventBus, EventMetadata}; + + fn audit_event(principal: &str) -> AstridEvent { + let message = IpcMessage::new( + Topic::from_raw(AUDIT_TOPIC), + IpcPayload::RawJson(serde_json::json!({ "principal": principal })), + uuid::Uuid::nil(), + ) + .with_principal(principal.to_owned()); + AstridEvent::Ipc { + metadata: EventMetadata::new("audit_stream_test"), + message, + } + } + + #[test] + fn audit_firehose_enforcement_id_is_registered() { + assert_eq!(AUDIT_FIREHOSE_CAP, "audit:read_all"); + let registry = astrid_core::capability_registry::capability_registry_revision_1().unwrap(); + assert!( + registry + .entries() + .iter() + .any(|entry| entry.id().as_str() == AUDIT_FIREHOSE_CAP), + "gateway enforcement uses {AUDIT_FIREHOSE_CAP:?} without a registry revision 1 entry" + ); + } + + #[tokio::test] + async fn live_stream_loses_firehose_and_keeps_own_visibility() { + let bus = EventBus::new(); + let receiver = bus.subscribe_topic_routed( + uuid::Uuid::new_v4(), + AUDIT_TOPIC, + "gateway-test", + "audit_stream_test", + ); + let firehose = Arc::new(AtomicBool::new(true)); + let firehose_for_probe = Arc::clone(&firehose); + let probe = CapabilityProbe::new(move |principal, device_key_id, capability| { + principal.as_str() == "alice" + && device_key_id == Some("0123456789abcdef") + && capability == AUDIT_FIREHOSE_CAP + && firehose_for_probe.load(Ordering::SeqCst) + }); + let principal = PrincipalId::new("alice").expect("principal"); + let mut stream = Box::pin(audit_event_stream( + receiver, + probe, + principal, + Some("0123456789abcdef".to_owned()), + true, + )); + + assert!(stream.next().await.is_some()); + let _ = bus.publish(audit_event("bob")); + assert!( + tokio::time::timeout(Duration::from_secs(1), stream.next()) + .await + .expect("firehose event") + .is_some() + ); + + firehose.store(false, Ordering::SeqCst); + let _ = bus.publish(audit_event("bob")); + assert!( + tokio::time::timeout(Duration::from_millis(50), stream.next()) + .await + .is_err() + ); + + let _ = bus.publish(audit_event("alice")); + assert!( + tokio::time::timeout(Duration::from_secs(1), stream.next()) + .await + .expect("own event") + .is_some() + ); + } } diff --git a/crates/astrid-gateway/src/routes/mod.rs b/crates/astrid-gateway/src/routes/mod.rs index 99fa8b2cc..317493a85 100644 --- a/crates/astrid-gateway/src/routes/mod.rs +++ b/crates/astrid-gateway/src/routes/mod.rs @@ -7,8 +7,8 @@ use std::sync::Arc; -use axum::Router; use axum::routing::{delete, get, patch, post, put}; +use axum::{Extension, Router}; use astrid_uplink::KernelClientError; @@ -55,13 +55,42 @@ pub mod sessions; pub mod stream; pub mod system; -/// Build the gateway's HTTP router. +/// Build the gateway's HTTP router with self-only audit visibility. +/// +/// This builder installs a deny-all capability evaluator. Runtimes that need +/// live `audit:read_all` policy must use [`build_with_capability_probe`], which +/// also documents the required connect-info serving path for the unauthenticated +/// redeem routes' per-IP throttling. // A flat list of route registrations: its length tracks the API surface, // not branching complexity, and both the readiness and models surfaces add // rows here. Splitting it into sub-routers would obscure the single // public/authed grouping for no readability gain. -#[allow(clippy::too_many_lines)] pub fn build(state: Arc) -> Router { + build_with_probe(state, events::CapabilityProbe::deny_all()) +} + +/// Build the gateway's HTTP router with an in-process capability evaluator. +/// +/// Co-located runtimes can use this to preserve live audit firehose policy +/// when they embed the router directly instead of calling +/// [`crate::run_with_capability_probe`]. +/// +/// When serving this router over a real socket, use +/// `router.into_make_service_with_connect_info::()` +/// rather than plain `axum::serve(listener, router)`. The unauthenticated +/// redeem routes require the peer address for per-IP throttling. +pub fn build_with_capability_probe(state: Arc, capability_probe: F) -> Router +where + F: Fn(&astrid_core::PrincipalId, Option<&str>, &str) -> bool + Send + Sync + 'static, +{ + build_with_probe(state, events::CapabilityProbe::new(capability_probe)) +} + +#[allow(clippy::too_many_lines)] +pub(crate) fn build_with_probe( + state: Arc, + capability_probe: events::CapabilityProbe, +) -> Router { // Unauthenticated routes — discovery + redeem + ops probes. let public = Router::new() .route("/api/distribution", get(distribution::get_distribution)) @@ -90,6 +119,7 @@ pub fn build(state: Arc) -> Router { let authed = build_authed_router(&state); let combined = public.merge(authed) + .layer(Extension(capability_probe)) // Count every request after it routes — axum's `MatchedPath` // extractor gives the registered template (e.g. // `/api/sys/principals/:id`) so the metric stays bounded diff --git a/crates/astrid-gateway/tests/router.rs b/crates/astrid-gateway/tests/router.rs index 48c34cac3..2e61bedb9 100644 --- a/crates/astrid-gateway/tests/router.rs +++ b/crates/astrid-gateway/tests/router.rs @@ -18,7 +18,10 @@ use std::sync::Arc; +use astrid_audit::{AuditAction, AuditLog, AuditOutcome, AuthorizationProof}; use astrid_core::PrincipalId; +use astrid_core::SessionId; +use astrid_crypto::KeyPair; use astrid_gateway::{ GatewayConfig, GatewayState, auth::{CallerContext, mint_bearer, mint_bearer_scoped, verify_bearer}, @@ -83,6 +86,31 @@ fn fresh_state_with_distro(distro: Option<&str>) -> Arc { }) } +fn fresh_state_with_audit_log() -> (Arc, Arc, SessionId) { + let session_id = SessionId::from_uuid(uuid::Uuid::new_v4()); + let audit_log = Arc::new(AuditLog::in_memory(KeyPair::generate())); + let state = Arc::new(GatewayState { + config: GatewayConfig::default(), + signing: SigningMaterial::fresh(), + distribution: Arc::new(DistributionInfo::single_tenant()), + onboarding: Arc::new(OnboardingFields::default()), + redeem_limiter: tokio::sync::Mutex::default(), + metrics_handle: astrid_gateway::metrics::install_recorder().expect("recorder"), + event_bus: None, + revoked_at: std::sync::Arc::new(std::sync::RwLock::new(std::collections::HashMap::new())), + revoked_key_ids: std::sync::Arc::new(std::sync::RwLock::new( + std::collections::HashMap::new(), + )), + audit_log: Some(Arc::clone(&audit_log)), + session_id: Some(session_id.clone()), + gateway_route_uuid: uuid::Uuid::new_v4(), + readiness_probe: None, + topic_probe: None, + registry_timeout: None, + }); + (state, audit_log, session_id) +} + #[tokio::test] async fn distribution_endpoint_returns_metadata_unauthenticated() { let state = fresh_state_with_distro(Some(SAMPLE_DISTRO)); @@ -225,6 +253,44 @@ async fn me_route_returns_principal_from_valid_bearer() { assert_eq!(body["principal"], "alice"); } +#[tokio::test] +async fn public_router_builder_accepts_live_capability_probe() { + let (state, audit_log, session_id) = fresh_state_with_audit_log(); + audit_log + .append_with_principal( + session_id, + PrincipalId::new("bob").unwrap(), + AuditAction::AdminRequest { + method: "BobVisibleThroughProbe".into(), + required_capability: "audit:read_all".into(), + target_principal: None, + params: None, + device_key_id: None, + }, + AuthorizationProof::System { + reason: "test".into(), + }, + AuditOutcome::Success { details: None }, + ) + .await + .expect("append audit entry"); + let router = routes::build_with_capability_probe(Arc::clone(&state), |_, _, _| true); + + let principal = PrincipalId::new("alice").unwrap(); + let bearer = mint_bearer(&state.signing.signer, &principal, 3600); + let req = Request::builder() + .uri("/api/sys/audit?principal=bob") + .header(header::AUTHORIZATION, String::from("Bearer ") + &bearer) + .body(Body::empty()) + .unwrap(); + let resp = router.oneshot(req).await.unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + let bytes = to_bytes(resp.into_body(), 4096).await.unwrap(); + let body: serde_json::Value = serde_json::from_slice(&bytes).unwrap(); + assert_eq!(body["entries"][0]["principal"], "bob"); + assert_eq!(body["entries"][0]["method"], "BobVisibleThroughProbe"); +} + #[tokio::test] async fn me_route_ignores_principal_claim_in_query_string() { // Adversarial check: a client tries to coerce `/api/auth/me` to diff --git a/crates/astrid-integration-tests/tests/gateway_e2e.rs b/crates/astrid-integration-tests/tests/gateway_e2e.rs index 05787c189..4617d752a 100644 --- a/crates/astrid-integration-tests/tests/gateway_e2e.rs +++ b/crates/astrid-integration-tests/tests/gateway_e2e.rs @@ -20,10 +20,10 @@ //! the live state — `GET /api/distribution`, `GET /api/openapi.json`, //! `GET /healthz`. //! 4. Audit events the kernel publishes on -//! `astrid.v1.audit.entry` reach a subscriber on the shared -//! `event_bus` — i.e. the SSE handler would see them when run -//! against a real listener. We tap the bus directly because -//! `Sse::keep_alive` makes `oneshot` await indefinitely. +//! `astrid.v1.audit.entry` reach both a direct bus subscriber and +//! the gateway's live SSE response. Revoking the authenticating +//! device immediately narrows that open response to the caller's +//! own records through the kernel's live policy evaluator. //! 5. Bearers minted by the gateway and verified by the gateway //! round-trip correctly — i.e. `signed → token → verified //! principal` matches against a real on-disk key. @@ -36,11 +36,6 @@ //! start && curl ...` against a built daemon, and by the //! in-process router tests in `astrid-gateway/tests/router.rs` //! that exercise the auth middleware with mocked state. -//! * **Full SSE streaming.** `axum::Sse` keeps the connection open -//! by design; `oneshot` would block. The bus-tap above proves the -//! audit-event arm of the wiring; the SSE serialisation itself is -//! covered by `astrid-gateway/src/routes/events.rs` unit tests. -//! //! ## Sandbox note //! //! Unix-socket bind permissions are blocked in some sandboxed @@ -59,8 +54,11 @@ #![allow(unsafe_code)] use std::sync::Arc; +use std::time::Duration; -use astrid_core::PrincipalId; +use astrid_audit::{AuditAction, AuditOutcome, AuthorizationProof}; +use astrid_core::kernel_api::{AdminRequestKind, AdminResponseBody}; +use astrid_core::{AuthMethod, DeviceKey, DeviceScope, PrincipalId, PrincipalProfile}; use astrid_events::AstridEvent; use astrid_events::ipc::{IpcMessage, IpcPayload, Topic}; use astrid_gateway::{GatewayConfig, GatewayState, routes}; @@ -87,6 +85,343 @@ fn looks_like_sandbox_block(err: &std::io::Error) -> bool { err.kind() == std::io::ErrorKind::PermissionDenied || err.raw_os_error() == Some(1) // EPERM } +fn publish_audit_marker(event_bus: &astrid_events::EventBus, principal: &str, marker: &str) { + let payload = serde_json::json!({ + "principal": principal, + "method": marker, + }); + let message = IpcMessage::new( + Topic::from_raw("astrid.v1.audit.entry"), + IpcPayload::RawJson(payload), + uuid::Uuid::nil(), + ) + .with_principal(principal.to_owned()); + let _ = event_bus.publish(AstridEvent::Ipc { + metadata: astrid_events::EventMetadata::new("gateway_sse_policy_test"), + message, + }); +} + +async fn append_audit_marker( + kernel: &astrid_kernel::Kernel, + principal: &PrincipalId, + marker: &str, +) { + kernel + .audit_log + .append_with_principal( + kernel.session_id.clone(), + principal.clone(), + AuditAction::AdminRequest { + method: marker.to_owned(), + required_capability: "audit:read_all".to_owned(), + target_principal: None, + params: None, + device_key_id: None, + }, + AuthorizationProof::System { + reason: "gateway audit attenuation regression".to_owned(), + }, + AuditOutcome::Success { details: None }, + ) + .await + .expect("append audit marker"); +} + +async fn wait_for_sse_marker( + response: &mut reqwest::Response, + observed: &mut String, + marker: &str, +) { + tokio::time::timeout(Duration::from_secs(3), async { + while !observed.contains(marker) { + let chunk = response + .chunk() + .await + .expect("read SSE response") + .expect("SSE response closed"); + observed.push_str(&String::from_utf8_lossy(&chunk)); + } + }) + .await + .unwrap_or_else(|_| panic!("SSE response did not contain {marker:?}: {observed}")); +} + +async fn assert_sse_marker_absent( + response: &mut reqwest::Response, + observed: &mut String, + marker: &str, +) { + assert!(!observed.contains(marker)); + let result = tokio::time::timeout(Duration::from_millis(500), async { + loop { + let chunk = response + .chunk() + .await + .expect("read SSE response") + .expect("SSE response closed"); + observed.push_str(&String::from_utf8_lossy(&chunk)); + assert!( + !observed.contains(marker), + "SSE response leaked {marker:?}: {observed}" + ); + } + }) + .await; + assert!(result.is_err()); + assert!(!observed.contains(marker)); +} + +async fn assert_scoped_admin_audit_history( + client: &reqwest::Client, + address: std::net::SocketAddr, + kernel: &astrid_kernel::Kernel, + principal: &PrincipalId, + bob: &PrincipalId, + bearer: &str, +) { + append_audit_marker(kernel, bob, "BobHiddenFromScopedAdminHistory").await; + append_audit_marker(kernel, principal, "CallerOwnScopedAdminHistory").await; + let historical_url = format!("http://{address}/api/sys/audit?limit=1000"); + let historical = tokio::time::timeout(Duration::from_secs(3), async { + loop { + match client.get(&historical_url).bearer_auth(bearer).send().await { + Ok(response) => return response, + Err(_) => tokio::time::sleep(Duration::from_millis(25)).await, + } + } + }) + .await + .unwrap_or_else(|_| panic!("gateway did not become ready at {historical_url}")); + assert_eq!(historical.status(), reqwest::StatusCode::OK); + let historical: serde_json::Value = historical + .json() + .await + .expect("parse historical audit response"); + let historical_methods = historical["entries"] + .as_array() + .expect("historical entries") + .iter() + .filter_map(|entry| entry["method"].as_str()) + .collect::>(); + assert!(historical_methods.contains(&"CallerOwnScopedAdminHistory")); + assert!(!historical_methods.contains(&"BobHiddenFromScopedAdminHistory")); +} + +async fn assert_scoped_admin_audit_stream( + client: &reqwest::Client, + address: std::net::SocketAddr, + kernel: &astrid_kernel::Kernel, + principal: &PrincipalId, + bob: &PrincipalId, + bearer: &str, +) { + let url = format!("http://{address}/api/events"); + let mut scoped_response = client + .get(&url) + .bearer_auth(bearer) + .send() + .await + .expect("open scoped-admin audit stream"); + assert_eq!(scoped_response.status(), reqwest::StatusCode::OK); + let mut scoped_observed = String::new(); + wait_for_sse_marker(&mut scoped_response, &mut scoped_observed, "ready").await; + assert!( + scoped_observed.contains("\"firehose\":false"), + "scoped admin ready frame must report firehose=false: {scoped_observed}" + ); + publish_audit_marker( + &kernel.event_bus, + bob.as_str(), + "BobHiddenFromScopedAdminStream", + ); + assert_sse_marker_absent( + &mut scoped_response, + &mut scoped_observed, + "BobHiddenFromScopedAdminStream", + ) + .await; + publish_audit_marker( + &kernel.event_bus, + principal.as_str(), + "CallerOwnScopedAdminStream", + ); + wait_for_sse_marker( + &mut scoped_response, + &mut scoped_observed, + "CallerOwnScopedAdminStream", + ) + .await; + assert!(!scoped_observed.contains("BobHiddenFromScopedAdminStream")); +} + +async fn assert_live_audit_revocation( + client: &reqwest::Client, + address: std::net::SocketAddr, + kernel: &astrid_kernel::Kernel, + state: &GatewayState, + principal: &PrincipalId, + device_key_id: String, + bearer: &str, +) { + let url = format!("http://{address}/api/events"); + let mut response = tokio::time::timeout(Duration::from_secs(3), async { + loop { + match client.get(&url).bearer_auth(bearer).send().await { + Ok(response) => return response, + Err(_) => tokio::time::sleep(Duration::from_millis(25)).await, + } + } + }) + .await + .unwrap_or_else(|_| panic!("gateway did not become ready at {url}")); + assert_eq!(response.status(), reqwest::StatusCode::OK); + let mut observed = String::new(); + wait_for_sse_marker(&mut response, &mut observed, "ready").await; + + publish_audit_marker(&kernel.event_bus, "bob", "BobVisibleBeforeRevocation"); + wait_for_sse_marker(&mut response, &mut observed, "BobVisibleBeforeRevocation").await; + + let revoke = state + .admin_client(PrincipalId::default()) + .expect("admin client") + .request(AdminRequestKind::PairDeviceRevoke { + principal: principal.clone(), + key_id: device_key_id, + }) + .await + .expect("revoke device"); + assert!(matches!( + revoke, + AdminResponseBody::PairDeviceRevoked { .. } + )); + + publish_audit_marker(&kernel.event_bus, "bob", "BobHiddenAfterRevocation"); + assert_sse_marker_absent(&mut response, &mut observed, "BobHiddenAfterRevocation").await; + publish_audit_marker( + &kernel.event_bus, + principal.as_str(), + "CallerOwnAfterRevocation", + ); + wait_for_sse_marker(&mut response, &mut observed, "CallerOwnAfterRevocation").await; + assert_sse_marker_absent(&mut response, &mut observed, "BobHiddenAfterRevocation").await; +} + +fn seed_audit_test_identity() -> (PrincipalId, String, String) { + let principal = PrincipalId::new("audit-e2e-principal").unwrap(); + let audit_device = DeviceKey::new( + "a".repeat(64), + DeviceScope::Scoped { + allow: vec!["audit:read_all".into()], + deny: vec![], + }, + Some("gateway-e2e".into()), + 0, + ); + let audit_device_key_id = audit_device.key_id.clone(); + let self_list_device = DeviceKey::new( + "b".repeat(64), + DeviceScope::Scoped { + allow: vec!["self:agent:list".into()], + deny: vec!["audit:read_all".into()], + }, + Some("gateway-e2e-self-list".into()), + 0, + ); + let self_list_device_key_id = self_list_device.key_id.clone(); + let profile = PrincipalProfile { + groups: vec!["admin".into()], + auth: astrid_core::AuthConfig { + methods: vec![AuthMethod::Keypair], + public_keys: vec![audit_device, self_list_device], + }, + ..PrincipalProfile::default() + }; + let home = astrid_core::dirs::AstridHome::resolve().expect("ASTRID_HOME"); + profile + .save_to_path(&PrincipalProfile::path_for(&home, &principal)) + .expect("save SSE principal profile"); + (principal, audit_device_key_id, self_list_device_key_id) +} + +async fn check_live_audit_device_attenuation( + kernel: &Arc, + state: &Arc, + listener: tokio::net::TcpListener, +) { + let address = listener.local_addr().expect("gateway listener address"); + let (principal, audit_device_key_id, self_list_device_key_id) = seed_audit_test_identity(); + + let audit_bearer = astrid_gateway::auth::mint_bearer_scoped( + &state.signing.signer, + &principal, + &audit_device_key_id, + 300, + ); + let self_list_bearer = astrid_gateway::auth::mint_bearer_scoped( + &state.signing.signer, + &principal, + &self_list_device_key_id, + 300, + ); + let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel::<()>(); + let server_state = Arc::clone(state); + let policy_kernel = Arc::clone(kernel); + let mut server = tokio::spawn(async move { + astrid_gateway::run_with_capability_probe_on_listener( + server_state, + listener, + async move { + let _ = shutdown_rx.await; + }, + move |principal, device_key_id, capability| { + policy_kernel.runtime_capability_allows(principal, device_key_id, capability) + }, + ) + .await + }); + + let client = reqwest::Client::new(); + let bob = PrincipalId::new("audit-e2e-bob").unwrap(); + assert_scoped_admin_audit_history( + &client, + address, + kernel, + &principal, + &bob, + &self_list_bearer, + ) + .await; + assert_scoped_admin_audit_stream( + &client, + address, + kernel, + &principal, + &bob, + &self_list_bearer, + ) + .await; + assert_live_audit_revocation( + &client, + address, + kernel, + state, + &principal, + audit_device_key_id, + &audit_bearer, + ) + .await; + + let _ = shutdown_tx.send(()); + if let Ok(result) = tokio::time::timeout(Duration::from_secs(3), &mut server).await { + result.expect("gateway task").expect("gateway shutdown"); + } else { + server.abort(); + let _ = server.await; + panic!("gateway shutdown timed out"); + } +} + async fn check_unauthenticated_routes(router: Router) { let req = Request::builder() .uri("/api/distribution") @@ -258,8 +593,18 @@ async fn kernel_and_gateway_boot_against_shared_home() { // every subsequent boot loads. Both paths are covered by the // SigningMaterial unit tests; here we just confirm the on-disk // artefact lands. + let gateway_listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind gateway listener"); + let gateway_address = gateway_listener + .local_addr() + .expect("gateway listener address"); + let gateway_config = GatewayConfig { + listen: gateway_address.to_string(), + ..GatewayConfig::default() + }; let state = GatewayState::new( - GatewayConfig::default(), + gateway_config, Some(Arc::clone(&kernel.event_bus)), Some(Arc::clone(&kernel.audit_log)), Some(kernel.session_id.clone()), @@ -293,6 +638,8 @@ async fn kernel_and_gateway_boot_against_shared_home() { // is loaded. check_bearer_roundtrip(router, &state).await; + check_live_audit_device_attenuation(&kernel, &state, gateway_listener).await; + // ── Cleanup ───────────────────────────────────────────────── kernel.shutdown(Some("e2e-test-complete".into())).await; } diff --git a/crates/astrid-kernel/src/lib.rs b/crates/astrid-kernel/src/lib.rs index c678e5871..9ebed57d5 100644 --- a/crates/astrid-kernel/src/lib.rs +++ b/crates/astrid-kernel/src/lib.rs @@ -41,6 +41,8 @@ pub mod invite; pub mod kernel_router; /// Persistent pair-device token store (issue #756). pub mod pair_token; +#[cfg(all(test, not(all(target_arch = "wasm32", target_os = "unknown"))))] +mod runtime_policy_tests; /// The Unix Domain Socket manager. Unix-only: binds the `UnixListener` and /// acquires the singleton advisory lock. #[cfg(unix)] @@ -1325,6 +1327,47 @@ impl Kernel { }) } + /// Evaluate one capability against the current principal and device policy. + #[doc(hidden)] + #[must_use] + pub fn runtime_capability_allows( + &self, + principal: &PrincipalId, + device_key_id: Option<&str>, + capability: &str, + ) -> bool { + let Ok(profile) = self.profile_cache.resolve(principal) else { + return false; + }; + if !profile.enabled { + return false; + } + + let device_scope = match device_key_id { + Some(key_id) => { + let Ok(key_id) = astrid_core::profile::DeviceKeyId::new(key_id) else { + return false; + }; + let Some(device) = profile.auth.device_by_typed_key_id(&key_id) else { + return false; + }; + Some(&device.scope) + }, + None => None, + }; + + let groups = self.groups.load_full(); + let mut check = astrid_capabilities::CapabilityCheck::new_borrowed( + profile.as_ref(), + groups.as_ref(), + principal, + ); + if let Some(scope) = device_scope { + check = check.with_device_scope(scope); + } + check.has(capability) + } + /// In-process probe for "does a loaded capsule subscribe to this topic", /// computed from the live registry without a capability check. Mirrors /// [`Self::agent_readiness_probe`]; the co-located gateway uses it to diff --git a/crates/astrid-kernel/src/runtime_policy_tests.rs b/crates/astrid-kernel/src/runtime_policy_tests.rs new file mode 100644 index 000000000..269970681 --- /dev/null +++ b/crates/astrid-kernel/src/runtime_policy_tests.rs @@ -0,0 +1,187 @@ +use std::sync::Arc; + +use astrid_core::PrincipalId; +use astrid_core::dirs::AstridHome; +use astrid_core::profile::{AuthMethod, DeviceKey, DeviceScope, PrincipalProfile}; + +use crate::Kernel; + +const AUDIT_FIREHOSE_CAP: &str = "audit:read_all"; + +async fn fixture() -> (tempfile::TempDir, Arc, PrincipalId) { + let dir = tempfile::tempdir().expect("tempdir"); + let home = AstridHome::from_path(dir.path()); + let kernel = crate::test_kernel_with_home(home).await; + let principal = PrincipalId::new("audit_operator").expect("principal"); + (dir, kernel, principal) +} + +fn full_device(seed: char) -> DeviceKey { + DeviceKey::new(seed.to_string().repeat(64), DeviceScope::Full, None, 0) +} + +fn scoped_device(seed: char, allow: &[&str], deny: &[&str]) -> DeviceKey { + DeviceKey::new( + seed.to_string().repeat(64), + DeviceScope::Scoped { + allow: allow.iter().map(|value| (*value).to_owned()).collect(), + deny: deny.iter().map(|value| (*value).to_owned()).collect(), + }, + None, + 0, + ) +} + +fn seed(kernel: &Kernel, principal: &PrincipalId, devices: Vec) { + seed_policy( + kernel, + principal, + true, + &[], + &[AUDIT_FIREHOSE_CAP], + &[], + devices, + ); +} + +fn seed_policy( + kernel: &Kernel, + principal: &PrincipalId, + enabled: bool, + groups: &[&str], + grants: &[&str], + revokes: &[&str], + devices: Vec, +) { + let mut profile = PrincipalProfile { + enabled, + groups: groups.iter().map(|value| (*value).to_owned()).collect(), + grants: grants.iter().map(|value| (*value).to_owned()).collect(), + revokes: revokes.iter().map(|value| (*value).to_owned()).collect(), + ..PrincipalProfile::default() + }; + if !devices.is_empty() { + profile.auth.methods.push(AuthMethod::Keypair); + profile.auth.public_keys = devices; + } + profile + .save_to_path(&PrincipalProfile::path_for(&kernel.astrid_home, principal)) + .expect("save profile"); + kernel.profile_cache.invalidate(principal); +} + +#[tokio::test] +async fn no_device_uses_the_principals_effective_authority() { + let (_dir, kernel, principal) = fixture().await; + seed(&kernel, &principal, vec![]); + + assert!(kernel.runtime_capability_allows(&principal, None, AUDIT_FIREHOSE_CAP)); +} + +#[tokio::test] +async fn full_device_preserves_the_principals_effective_authority() { + let (_dir, kernel, principal) = fixture().await; + let device = full_device('a'); + let key_id = device.key_id.clone(); + seed(&kernel, &principal, vec![device]); + + assert!(kernel.runtime_capability_allows(&principal, Some(&key_id), AUDIT_FIREHOSE_CAP)); +} + +#[tokio::test] +async fn scoped_device_must_admit_the_exact_firehose_capability() { + let (_dir, kernel, principal) = fixture().await; + let allowed = scoped_device('a', &[AUDIT_FIREHOSE_CAP], &[]); + let denied = scoped_device('b', &["*"], &[AUDIT_FIREHOSE_CAP]); + let allowed_id = allowed.key_id.clone(); + let denied_id = denied.key_id.clone(); + seed(&kernel, &principal, vec![allowed, denied]); + assert!(kernel.runtime_capability_allows(&principal, Some(&allowed_id), AUDIT_FIREHOSE_CAP)); + assert!(!kernel.runtime_capability_allows(&principal, Some(&denied_id), AUDIT_FIREHOSE_CAP)); +} + +#[tokio::test] +async fn scoped_admin_self_list_does_not_imply_audit_firehose() { + let (_dir, kernel, principal) = fixture().await; + let device = scoped_device('a', &["self:agent:list"], &[AUDIT_FIREHOSE_CAP]); + let key_id = device.key_id.clone(); + seed_policy( + &kernel, + &principal, + true, + &["admin"], + &[], + &[], + vec![device], + ); + + assert!(kernel.runtime_capability_allows(&principal, Some(&key_id), "self:agent:list")); + assert!(!kernel.runtime_capability_allows(&principal, Some(&key_id), AUDIT_FIREHOSE_CAP)); +} + +#[tokio::test] +async fn malformed_and_unknown_device_ids_fail_closed() { + let (_dir, kernel, principal) = fixture().await; + seed(&kernel, &principal, vec![full_device('a')]); + assert!(!kernel.runtime_capability_allows( + &principal, + Some("not-a-device-id"), + AUDIT_FIREHOSE_CAP + )); + assert!(!kernel.runtime_capability_allows( + &principal, + Some("ffffffffffffffff"), + AUDIT_FIREHOSE_CAP + )); +} + +#[tokio::test] +async fn revoked_device_id_fails_closed() { + let (_dir, kernel, principal) = fixture().await; + let device = full_device('a'); + let key_id = device.key_id.clone(); + seed(&kernel, &principal, vec![device]); + assert!(kernel.runtime_capability_allows(&principal, Some(&key_id), AUDIT_FIREHOSE_CAP)); + + seed(&kernel, &principal, vec![]); + + assert!(!kernel.runtime_capability_allows(&principal, Some(&key_id), AUDIT_FIREHOSE_CAP)); +} + +#[tokio::test] +async fn disabled_principal_loses_live_authority() { + let (_dir, kernel, principal) = fixture().await; + seed(&kernel, &principal, vec![]); + assert!(kernel.runtime_capability_allows(&principal, None, AUDIT_FIREHOSE_CAP)); + + seed_policy( + &kernel, + &principal, + false, + &[], + &[AUDIT_FIREHOSE_CAP], + &[], + vec![], + ); + + assert!(!kernel.runtime_capability_allows(&principal, None, AUDIT_FIREHOSE_CAP)); +} + +#[tokio::test] +async fn group_authority_is_live_and_principal_revoke_wins() { + let (_dir, kernel, principal) = fixture().await; + seed_policy(&kernel, &principal, true, &["admin"], &[], &[], vec![]); + assert!(kernel.runtime_capability_allows(&principal, None, AUDIT_FIREHOSE_CAP)); + + seed_policy( + &kernel, + &principal, + true, + &["admin"], + &[], + &[AUDIT_FIREHOSE_CAP], + vec![], + ); + + assert!(!kernel.runtime_capability_allows(&principal, None, AUDIT_FIREHOSE_CAP)); +}