diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index d2d34573..dca57421 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -384,10 +384,11 @@ jobs: set -euo pipefail fga model transform --file model.fga | diff - model.json - # RFC 0047 §5 (RFC0047.1–.3): the OpenFGA resolver on the served binary - # against a real `openfga/openfga` container (testcontainers, image - # pinned by digest in the test) loaded with the in-tree model — the same - # posture as `dex-oidc`. A required check (in `ci-success`'s `needs`). + # RFC 0047 §5 (RFC0047.1–.8): the OpenFGA resolver and the layer-2 + # visibility two-step on the served binary against a real + # `openfga/openfga` container (testcontainers, image pinned by digest in + # the tests) loaded with the in-tree model — the same posture as + # `dex-oidc`. A required check (in `ci-success`'s `needs`). openfga-resolver: name: openfga resolver (testcontainers) # Not part of the scheduled hygiene canary (see `on:`). @@ -405,6 +406,7 @@ jobs: cargo test -p ourios-server --test it -- --ignored --exact rfc0047_openfga::rfc0047_1_to_3_resolver_end_to_end + rfc0047_visibility::rfc0047_4_to_8_visibility_end_to_end # Emission-time semconv conformance: boot the real `ourios-server`, # point its OTLP export (metrics + the dogfooded logs signal) at diff --git a/crates/ourios-core/src/auth/openfga/client.rs b/crates/ourios-core/src/auth/openfga/client.rs index a91d5426..c82b5b60 100644 --- a/crates/ourios-core/src/auth/openfga/client.rs +++ b/crates/ourios-core/src/auth/openfga/client.rs @@ -13,7 +13,10 @@ use std::time::{Duration, Instant}; use serde::{Deserialize, Serialize}; -use super::{Consistency, OpenFgaConfig, Principal, PrincipalKind, TENANT_TYPE}; +use super::{ + CONVERSATION_TYPE, Consistency, OpenFgaConfig, Principal, PrincipalKind, TENANT_TYPE, + TenantObjects, VisibilityConfig, is_object_id, +}; /// `OpenFGA`'s cap on contextual tuples per request (RFC 0047 §3.1): a /// token carrying more groups than this fails resolution closed. @@ -34,8 +37,6 @@ const CACHE_SWEEP_THRESHOLD: usize = 1024; const MAX_CACHE_ENTRIES: usize = 4096; /// How much of a non-2xx body is kept for the error message. const MAX_ERROR_BODY_BYTES: usize = 512; -/// `OpenFGA`'s object-id limit. -const MAX_OBJECT_ID_BYTES: usize = 256; /// One relationship tuple / tuple key on the wire. #[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)] @@ -77,6 +78,12 @@ pub enum OpenFgaError { /// How many the caller wanted to send. count: usize, }, + /// A principal id (OIDC `sub` / static token name) that cannot form an + /// `OpenFGA` user id — a credential/config defect, 401-class. + InvalidPrincipal, + /// A tenant id that cannot form an `OpenFGA` object id — nothing in the + /// graph can refer to it, so every question about it fails closed. + InvalidTenant, /// A token group name that cannot form an `OpenFGA` object id /// (empty, over 256 bytes, or containing `:`, `#` or whitespace) — a /// credential defect, answered like the cap: named, 401-class. @@ -103,6 +110,14 @@ impl fmt::Display for OpenFgaError { "openfga: {count} contextual tuples exceed the per-request cap of \ {MAX_CONTEXTUAL_TUPLES}" ), + Self::InvalidPrincipal => f.write_str( + "openfga: principal id cannot form a user id (empty, too long, or contains \ + ':', '#' or whitespace)", + ), + Self::InvalidTenant => f.write_str( + "openfga: tenant id cannot form an object id (empty, too long, or contains \ + ':', '#' or whitespace)", + ), Self::InvalidGroup { index } => write!( f, "openfga: token group #{index} cannot form an object id (empty, too long, or \ @@ -414,15 +429,6 @@ impl OpenFgaClient { } } -/// Whether `id` can be the id half of an `OpenFGA` object (`type:id`). -fn is_object_id(id: &str) -> bool { - !id.is_empty() - && id.len() <= MAX_OBJECT_ID_BYTES - && !id - .chars() - .any(|c| c == ':' || c == '#' || c.is_whitespace()) -} - fn transport(e: &reqwest::Error) -> OpenFgaError { OpenFgaError::Unavailable(if e.is_timeout() { "request timed out".to_string() @@ -522,7 +528,43 @@ struct CacheKey { pub struct OpenFgaResolver { client: OpenFgaClient, session_ttl: Duration, + visibility: VisibilityConfig, cache: Mutex>, + /// The RFC 0047 §3.4 two-step outcome per (session, tenant), cached + /// like the binding — the enumeration behind `Scoped` is never cached. + branches: Mutex>, +} + +struct CachedBranch { + expires: Instant, + branch: Branch, +} + +/// The two-step's outcome for a principal in a tenant, before any +/// enumeration. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum Branch { + TenantWide, + MetadataOnly, + Scoped, +} + +/// What a principal may see inside a tenant (RFC 0047 §3.4): the whole +/// tenant, the whole tenant with content masked, or exactly the listed +/// conversations (ids with the `conversation:/` prefix stripped). +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum Visibility { + /// `Check(can_read_content, tenant)` allowed — the tenant predicate only. + TenantWide, + /// `Check(can_read_metadata, tenant)` allowed — every row, content + /// columns masked. + MetadataOnly, + /// A scoped principal: the tenant's conversation ids it may read + /// (empty when none, or when no object type is bound). + Scoped { + /// Conversation ids, prefix stripped, tenant-filtered. + conversations: BTreeSet, + }, } impl fmt::Debug for OpenFgaResolver { @@ -544,10 +586,18 @@ impl OpenFgaResolver { Ok(Self { client: OpenFgaClient::new(config)?, session_ttl: config.session_ttl(), + visibility: config.visibility().clone(), cache: Mutex::new(HashMap::new()), + branches: Mutex::new(HashMap::new()), }) } + /// The layer-2 visibility configuration. + #[must_use] + pub fn visibility_config(&self) -> &VisibilityConfig { + &self.visibility + } + /// The underlying client, for the planner and tool gate. #[must_use] pub fn client(&self) -> &OpenFgaClient { @@ -599,13 +649,7 @@ impl OpenFgaResolver { principal: &Principal, groups: &[String], ) -> Result { - let mut groups: Vec = groups.to_vec(); - groups.sort(); - groups.dedup(); - let cache_key = CacheKey { - principal: principal.clone(), - groups, - }; + let cache_key = Self::cache_key(principal, groups)?; let now = Instant::now(); { let cache = self.cache.lock().unwrap_or_else(PoisonError::into_inner); @@ -644,6 +688,140 @@ impl OpenFgaResolver { Ok(grants) } + fn cache_key(principal: &Principal, groups: &[String]) -> Result { + if !is_object_id(principal.id()) { + return Err(OpenFgaError::InvalidPrincipal); + } + let mut groups: Vec = groups.to_vec(); + groups.sort(); + groups.dedup(); + Ok(CacheKey { + principal: principal.clone(), + groups, + }) + } + + /// The RFC 0047 §3.4 two-step for `principal` in `tenant`: + /// `Check(can_read_content)` → [`Visibility::TenantWide`]; else + /// `Check(can_read_metadata)` → [`Visibility::MetadataOnly`]; else the + /// scoped enumeration — the **streamed** `ListObjects(can_read_content, + /// conversation)`, filtered to this tenant's `conversation:/` + /// prefix, counting only those toward `visibility.max_objects`, within + /// `visibility.list_timeout` — a truncated or cut-off set is never an + /// answer. The two checks are cached with the session TTL; the + /// enumeration never is. + /// + /// # Errors + /// + /// [`OpenFgaError::BoundExceeded`] past `max_objects` tenant ids, + /// [`OpenFgaError::Incomplete`] when the stream is cut off, + /// [`OpenFgaError::Unavailable`] on transport/status failure, and the + /// credential-defect variants ([`OpenFgaError::InvalidPrincipal`], + /// [`OpenFgaError::InvalidGroup`], [`OpenFgaError::TooManyContextualTuples`]). + pub async fn visibility( + &self, + principal: &Principal, + groups: &[String], + tenant: &str, + ) -> Result { + let cache_key = Self::cache_key(principal, groups)?; + let contextual = Self::group_tuples(principal, &cache_key.groups)?; + let user = principal.to_string(); + let objects = TenantObjects::new(tenant).ok_or(OpenFgaError::InvalidTenant)?; + let tenant_object = objects.tenant().to_string(); + let now = Instant::now(); + let branch_key = (cache_key, tenant.to_string()); + let cached = { + let branches = self.branches.lock().unwrap_or_else(PoisonError::into_inner); + branches + .get(&branch_key) + .filter(|entry| entry.expires > now) + .map(|entry| entry.branch) + }; + let branch = if let Some(branch) = cached { + branch + } else { + let branch = self.two_step(&user, &tenant_object, &contextual).await?; + if !self.session_ttl.is_zero() { + let mut branches = self.branches.lock().unwrap_or_else(PoisonError::into_inner); + if branches.len() >= CACHE_SWEEP_THRESHOLD { + branches.retain(|_, entry| entry.expires > now); + } + if branches.len() >= MAX_CACHE_ENTRIES + && let Some(soonest) = branches + .iter() + .min_by_key(|(_, entry)| entry.expires) + .map(|(key, _)| key.clone()) + { + branches.remove(&soonest); + } + branches.insert( + branch_key, + CachedBranch { + expires: now + self.session_ttl, + branch, + }, + ); + } + branch + }; + match branch { + Branch::TenantWide => Ok(Visibility::TenantWide), + Branch::MetadataOnly => Ok(Visibility::MetadataOnly), + Branch::Scoped => { + let bound = self + .visibility + .objects() + .iter() + .any(|object| object.object_type() == CONVERSATION_TYPE); + if !bound { + return Ok(Visibility::Scoped { + conversations: BTreeSet::new(), + }); + } + let prefix = objects.conversation_prefix().to_string(); + let objects = self + .client + .streamed_list_objects( + ListObjectsRequest { + user: &user, + relation: "can_read_content", + object_type: CONVERSATION_TYPE, + contextual_tuples: &contextual, + }, + self.visibility.list_timeout(), + self.visibility.max_objects(), + |object| object.starts_with(&prefix), + ) + .await?; + Ok(Visibility::Scoped { + conversations: objects + .into_iter() + .map(|object| object[prefix.len()..].to_string()) + .collect(), + }) + } + } + } + + /// The two `Check`s of RFC 0047 §3.4 steps 1–2. + async fn two_step( + &self, + user: &str, + tenant_object: &str, + contextual: &[TupleKey], + ) -> Result { + let content = TupleKey::new(user, "can_read_content", tenant_object); + if self.client.check(&content, contextual).await? { + return Ok(Branch::TenantWide); + } + let metadata = TupleKey::new(user, "can_read_metadata", tenant_object); + if self.client.check(&metadata, contextual).await? { + return Ok(Branch::MetadataOnly); + } + Ok(Branch::Scoped) + } + async fn list_tenants( &self, user: &str, @@ -690,7 +868,7 @@ mod tests { use super::super::{OpenFgaSpec, Principal, PrincipalKind, build_openfga_config}; use super::{ Grants, ListObjectsRequest, MAX_CONTEXTUAL_TUPLES, OpenFgaClient, OpenFgaError, - OpenFgaResolver, TupleKey, + OpenFgaResolver, TupleKey, Visibility, }; /// A loopback stand-in for the `OpenFGA` HTTP API: `check` answers from @@ -927,6 +1105,276 @@ mod tests { ); } + /// A grant-table fake for the two-step: `check` answers membership of + /// (user, relation, object); `streamed-list-objects` lists the grants of + /// the requested type/relation for the user, optionally stalling after + /// the first frame. + #[derive(Clone)] + struct GrantFake { + calls: Arc, + streams: Arc, + grants: Arc>, + stall: bool, + } + + async fn grant_check( + State(fake): State, + body: axum::body::Bytes, + ) -> impl IntoResponse { + fake.calls.fetch_add(1, Ordering::SeqCst); + let request: Value = serde_json::from_slice(&body).expect("json"); + let key = &request["tuple_key"]; + let allowed = fake + .grants + .iter() + .any(|(u, r, o)| key["user"] == *u && key["relation"] == *r && key["object"] == *o); + axum::Json(json!({ "allowed": allowed })).into_response() + } + + async fn grant_streamed( + State(fake): State, + body: axum::body::Bytes, + ) -> impl IntoResponse { + fake.calls.fetch_add(1, Ordering::SeqCst); + fake.streams.fetch_add(1, Ordering::SeqCst); + let request: Value = serde_json::from_slice(&body).expect("json"); + let user = request["user"].as_str().expect("user").to_string(); + let relation = request["relation"].as_str().expect("relation").to_string(); + let prefix = format!("{}:", request["type"].as_str().expect("type")); + let objects: Vec = fake + .grants + .iter() + .filter(|(u, r, o)| *u == user && *r == relation && o.starts_with(&prefix)) + .map(|(_, _, o)| (*o).to_string()) + .collect(); + let stall = fake.stall; + let stream = async_stream(move |tx| async move { + for object in objects { + let line = format!("{{\"result\":{{\"object\":\"{object}\"}}}}\n"); + if tx.send(Ok::<_, std::io::Error>(line)).await.is_err() { + return; + } + if stall { + tokio::time::sleep(Duration::from_secs(30)).await; + } + } + }); + axum::response::Response::builder() + .header("content-type", "application/x-ndjson") + .body(Body::from_stream(stream)) + .expect("response") + } + + async fn serve_grants(fake: GrantFake) -> String { + let app = Router::new() + .route("/stores/{store}/check", post(grant_check)) + .route( + "/stores/{store}/streamed-list-objects", + post(grant_streamed), + ) + .with_state(fake); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind"); + let url = format!("http://{}", listener.local_addr().expect("addr")); + tokio::spawn(async move { + axum::serve(listener, app).await.expect("serve"); + }); + url + } + + fn visibility_resolver(url: &str, max_objects: &str, list_timeout_ms: &str) -> OpenFgaResolver { + use super::super::{VisibilityObjectSpec, VisibilitySpec}; + let config = build_openfga_config(&OpenFgaSpec { + api_url: Some(url.to_string()), + store_id: Some("s".to_string()), + request_timeout_secs: Some("1".to_string()), + visibility: VisibilitySpec { + objects: vec![VisibilityObjectSpec { + object_type: Some("conversation".to_string()), + column: Some("attr.gen_ai.conversation.id".to_string()), + }], + max_objects: Some(max_objects.to_string()), + list_timeout_ms: Some(list_timeout_ms.to_string()), + ..VisibilitySpec::default() + }, + ..OpenFgaSpec::default() + }) + .expect("config"); + OpenFgaResolver::new(&config).expect("resolver") + } + + /// RFC 0047 §3.4 two-step (RFC0047.4/.5/.7 at the resolver): a + /// tenant-wide reader never enumerates; a metadata reader is masked, + /// never enumerated; a scoped principal's conversations are enumerated + /// through the **streamed** call, filtered to the tenant prefix, and + /// only tenant ids count toward the bound; the branch is cached per + /// session while the enumeration is not. + #[tokio::test] + #[allow(clippy::too_many_lines)] // one grant table, every branch of the two-step in sequence + async fn two_step_visibility() { + let fake = GrantFake { + calls: Arc::new(AtomicUsize::new(0)), + streams: Arc::new(AtomicUsize::new(0)), + grants: Arc::new(vec![ + ("user:alice", "can_read_content", "tenant:acme"), + ("user:alice", "can_read_metadata", "tenant:acme"), + ("user:fin", "can_read_metadata", "tenant:acme"), + ("user:bob", "can_read_content", "conversation:acme/c-1"), + ("user:bob", "can_read_content", "conversation:acme/c-2"), + ("user:bob", "can_read_content", "conversation:globex/c-1"), + ("agent:bot", "can_read_content", "conversation:acme/c-3"), + ("agent:bot", "can_read_content", "conversation:acme/c-4"), + ("agent:bot", "can_read_content", "conversation:globex/c-5"), + ("agent:bot", "can_read_content", "conversation:globex/c-6"), + ("agent:bot", "can_read_content", "conversation:globex/c-7"), + ]), + stall: false, + }; + let streams = Arc::clone(&fake.streams); + let calls = Arc::clone(&fake.calls); + let url = serve_grants(fake).await; + let resolver = visibility_resolver(&url, "2", "500"); + let alice = Principal::new(PrincipalKind::User, "alice"); + let fin = Principal::new(PrincipalKind::User, "fin"); + let bob = Principal::new(PrincipalKind::User, "bob"); + let bot = Principal::new(PrincipalKind::Agent, "bot"); + + assert_eq!( + resolver + .visibility(&alice, &[], "acme") + .await + .expect("alice"), + Visibility::TenantWide + ); + assert_eq!( + streams.load(Ordering::SeqCst), + 0, + "RFC0047.4: no enumeration" + ); + assert_eq!( + resolver.visibility(&fin, &[], "acme").await.expect("fin"), + Visibility::MetadataOnly + ); + assert_eq!( + streams.load(Ordering::SeqCst), + 0, + "RFC0047.8: no enumeration" + ); + assert_eq!( + resolver.visibility(&bob, &[], "acme").await.expect("bob"), + Visibility::Scoped { + conversations: BTreeSet::from(["c-1".to_string(), "c-2".to_string()]) + }, + "RFC0047.5: exactly bob's acme conversations, globex filtered out" + ); + assert_eq!(streams.load(Ordering::SeqCst), 1); + // The branch is cached; the enumeration is not. + let before = calls.load(Ordering::SeqCst); + resolver.visibility(&bob, &[], "acme").await.expect("bob"); + assert_eq!( + calls.load(Ordering::SeqCst), + before + 1, + "one stream, no checks" + ); + // RFC0047.7: bot has 2 acme + 3 globex conversations under a bound + // of 2 — succeeds with exactly the 2 (only acme ids count) … + assert_eq!( + resolver.visibility(&bot, &[], "acme").await.expect("bot"), + Visibility::Scoped { + conversations: BTreeSet::from(["c-3".to_string(), "c-4".to_string()]) + } + ); + // … and fails closed past the bound in globex (3 > 2). + assert_eq!( + resolver + .visibility(&bot, &[], "globex") + .await + .expect_err("bound"), + OpenFgaError::BoundExceeded { bound: 2 } + ); + // A principal with no grant on the tenant scopes to nothing. + assert_eq!( + resolver + .visibility(&fin, &[], "globex") + .await + .expect("fin/globex"), + Visibility::Scoped { + conversations: BTreeSet::new() + } + ); + // Enumeration errors are never cached: a stalled stream fails closed + // with `Incomplete`, every time. + let stalled = serve_grants(GrantFake { + calls: Arc::new(AtomicUsize::new(0)), + streams: Arc::new(AtomicUsize::new(0)), + grants: Arc::new(vec![ + ("user:bob", "can_read_content", "conversation:acme/c-1"), + ("user:bob", "can_read_content", "conversation:acme/c-2"), + ]), + stall: true, + }) + .await; + let resolver = visibility_resolver(&stalled, "10", "200"); + for _ in 0..2 { + assert_eq!( + resolver + .visibility(&bob, &[], "acme") + .await + .expect_err("stalled"), + OpenFgaError::Incomplete + ); + } + // A principal whose id is no object id is a credential defect; a + // tenant that is none fails closed before any call. + assert_eq!( + resolver + .visibility(&Principal::new(PrincipalKind::User, "a b"), &[], "acme") + .await + .expect_err("invalid principal"), + OpenFgaError::InvalidPrincipal + ); + assert_eq!( + resolver + .visibility(&bob, &[], "acme:prod") + .await + .expect_err("invalid tenant"), + OpenFgaError::InvalidTenant + ); + } + + /// The conversation prefix is the injective tenant encoding: a tenant + /// containing `/` filters on `conversation:/` — `a` never sees + /// `a/b`'s conversations even though both spell `conversation:a/b/...` + /// without the encoding. + #[tokio::test] + async fn scoped_enumeration_uses_the_encoded_tenant_prefix() { + let fake = GrantFake { + calls: Arc::new(AtomicUsize::new(0)), + streams: Arc::new(AtomicUsize::new(0)), + grants: Arc::new(vec![ + ("user:bob", "can_read_content", "conversation:a/b/c-1"), + ("user:bob", "can_read_content", "conversation:a%2Fb/c-2"), + ]), + stall: false, + }; + let url = serve_grants(fake).await; + let resolver = visibility_resolver(&url, "10", "500"); + let bob = Principal::new(PrincipalKind::User, "bob"); + assert_eq!( + resolver.visibility(&bob, &[], "a").await.expect("a"), + Visibility::Scoped { + conversations: BTreeSet::from(["b/c-1".to_string()]) + } + ); + assert_eq!( + resolver.visibility(&bob, &[], "a/b").await.expect("a/b"), + Visibility::Scoped { + conversations: BTreeSet::from(["c-2".to_string()]) + } + ); + } + /// The resolver: `can_query` / `can_write` sets with the `tenant:` /// prefix stripped, cached per (principal, groups) for the TTL — and /// re-resolved past it; errors are not cached; a group list past the @@ -999,11 +1447,19 @@ mod tests { 1 ); - // The cache key is structured: a subject or group carrying a - // separator can never alias another session (`"a\nb"` with no - // groups vs `"a"` in group `"b"`). + // A subject that is no object id is a credential defect, before any + // call; and the cache key is structured, so a subject or group + // carrying a separator can never alias another session (`"a|b"` + // with no groups vs `"a"` in group `"b"`). + assert_eq!( + resolver + .resolve(&Principal::new(PrincipalKind::User, "a\nb"), &[]) + .await + .expect_err("whitespace in sub"), + OpenFgaError::InvalidPrincipal + ); let before = calls.load(Ordering::SeqCst); - let odd = Principal::new(PrincipalKind::User, "a\nb"); + let odd = Principal::new(PrincipalKind::User, "a|b"); resolver.resolve(&odd, &[]).await.expect("resolve"); resolver .resolve( diff --git a/crates/ourios-core/src/auth/openfga/mod.rs b/crates/ourios-core/src/auth/openfga/mod.rs index 3670af03..e04f9f17 100644 --- a/crates/ourios-core/src/auth/openfga/mod.rs +++ b/crates/ourios-core/src/auth/openfga/mod.rs @@ -17,7 +17,7 @@ mod client; #[cfg(feature = "openfga")] pub use client::{ Grants, ListObjectsRequest, MAX_CONTEXTUAL_TUPLES, OpenFgaClient, OpenFgaError, - OpenFgaResolver, TupleKey, + OpenFgaResolver, TupleKey, Visibility, }; /// The default `session_ttl_secs` (RFC 0047 §3.1): revocation latency. @@ -46,6 +46,38 @@ pub struct OpenFgaSpec { pub consistency: Option, /// The per-call request timeout in seconds. pub request_timeout_secs: Option, + /// The layer-2 visibility section (RFC 0047 §3.4). + pub visibility: VisibilitySpec, + /// The `OpenFGA` server's own `OPENFGA_LIST_OBJECTS_DEADLINE`, in + /// milliseconds, as the operator declares it (default 3000). + pub server_list_objects_deadline_ms: Option, +} + +/// The raw `auth.openfga.visibility` section (RFC 0047 §3.4) — nothing +/// here is secret. +#[derive(Debug, Default, Clone)] +pub struct VisibilitySpec { + /// `objects[]`: graph object type → the promoted column carrying its id. + pub objects: Vec, + /// The promoted column compared to a `user:` principal's subject (the + /// §3.3 self fast path); unset disables the path. + pub self_principal_column: Option, + /// The columns a metadata-only reader sees as NULL and may not filter or + /// aggregate on. `None` = the `GenAI` content default set. + pub content_columns: Option>, + /// The bound on tenant-scoped ids per enumeration (default 10 000). + pub max_objects: Option, + /// The client-side enumeration timeout in milliseconds (default 2000). + pub list_timeout_ms: Option, +} + +/// One `visibility.objects[]` entry, raw. +#[derive(Debug, Default, Clone)] +pub struct VisibilityObjectSpec { + /// The `OpenFGA` object type (`conversation`). + pub object_type: Option, + /// The promoted column (`attr.gen_ai.conversation.id`). + pub column: Option, } impl fmt::Debug for OpenFgaSpec { @@ -58,6 +90,11 @@ impl fmt::Debug for OpenFgaSpec { .field("session_ttl_secs", &self.session_ttl_secs) .field("consistency", &self.consistency) .field("request_timeout_secs", &self.request_timeout_secs) + .field("visibility", &self.visibility) + .field( + "server_list_objects_deadline_ms", + &self.server_list_objects_deadline_ms, + ) .finish() } } @@ -94,8 +131,91 @@ pub struct OpenFgaConfig { session_ttl: Duration, consistency: Consistency, request_timeout: Duration, + visibility: VisibilityConfig, } +/// The validated `auth.openfga.visibility` configuration (RFC 0047 §3.4). +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct VisibilityConfig { + objects: Vec, + self_principal_column: Option, + content_columns: Vec, + max_objects: usize, + list_timeout: Duration, +} + +/// One bound object type: which promoted column carries its ids. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct VisibilityObject { + object_type: String, + column: String, +} + +impl VisibilityObject { + /// The `OpenFGA` object type. + #[must_use] + pub fn object_type(&self) -> &str { + &self.object_type + } + + /// The promoted column carrying the object ids. + #[must_use] + pub fn column(&self) -> &str { + &self.column + } +} + +impl VisibilityConfig { + /// The bound object types, in configuration order. + #[must_use] + pub fn objects(&self) -> &[VisibilityObject] { + &self.objects + } + + /// The self-fast-path column, when enabled. + #[must_use] + pub fn self_principal_column(&self) -> Option<&str> { + self.self_principal_column.as_deref() + } + + /// The content columns (DSL names: `body`, `attr.`). + #[must_use] + pub fn content_columns(&self) -> &[String] { + &self.content_columns + } + + /// The per-tenant enumeration bound. + #[must_use] + pub fn max_objects(&self) -> usize { + self.max_objects + } + + /// The client-side enumeration timeout. + #[must_use] + pub fn list_timeout(&self) -> Duration { + self.list_timeout + } +} + +/// The RFC 0047 §3.4 default content columns: the `GenAI` semconv content +/// attributes plus the log body. +pub const DEFAULT_CONTENT_COLUMNS: [&str; 6] = [ + "body", + "attr.gen_ai.input.messages", + "attr.gen_ai.output.messages", + "attr.gen_ai.system_instructions", + "attr.gen_ai.tool.call.arguments", + "attr.gen_ai.tool.call.result", +]; +/// The default `visibility.max_objects`. +pub const DEFAULT_MAX_OBJECTS: usize = 10_000; +/// The default `visibility.list_timeout_ms`. +pub const DEFAULT_LIST_TIMEOUT_MS: u64 = 2_000; +/// The default `server_list_objects_deadline_ms` (`OpenFGA`'s own default). +pub const DEFAULT_SERVER_LIST_OBJECTS_DEADLINE_MS: u64 = 3_000; +/// The `OpenFGA` object type of a conversation — the one bindable type in v1. +pub const CONVERSATION_TYPE: &str = "conversation"; + impl fmt::Debug for OpenFgaConfig { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { f.debug_struct("OpenFgaConfig") @@ -106,6 +226,7 @@ impl fmt::Debug for OpenFgaConfig { .field("session_ttl", &self.session_ttl) .field("consistency", &self.consistency) .field("request_timeout", &self.request_timeout) + .field("visibility", &self.visibility) .finish() } } @@ -152,6 +273,12 @@ impl OpenFgaConfig { pub fn request_timeout(&self) -> Duration { self.request_timeout } + + /// The layer-2 visibility configuration. + #[must_use] + pub fn visibility(&self) -> &VisibilityConfig { + &self.visibility + } } /// Validate a raw [`OpenFgaSpec`] into the resolved [`OpenFgaConfig`] @@ -228,6 +355,10 @@ pub fn build_openfga_config(spec: &OpenFgaSpec) -> Result ); } }; + let visibility = build_visibility_config( + &spec.visibility, + spec.server_list_objects_deadline_ms.as_deref(), + )?; Ok(OpenFgaConfig { api_url, store_id, @@ -236,6 +367,140 @@ pub fn build_openfga_config(spec: &OpenFgaSpec) -> Result session_ttl, consistency, request_timeout: Duration::from_secs(request_timeout_secs), + visibility, + }) +} + +/// Validate `visibility.objects[]`: v1 binds at most the `conversation` +/// type, once, to a promoted column. +fn build_visibility_objects( + specs: &[VisibilityObjectSpec], + promoted_column: impl Fn(&str, &str) -> Result, +) -> Result, String> { + let mut objects: Vec = Vec::with_capacity(specs.len()); + for (index, object) in specs.iter().enumerate() { + let object_type = match object.object_type.as_deref() { + Some(CONVERSATION_TYPE) => CONVERSATION_TYPE.to_string(), + _ => { + return Err(format!( + "auth.openfga.visibility.objects[{index}].type must be \ + `conversation` — the one object type v1 binds (RFC 0047 §3.4)" + )); + } + }; + if objects.iter().any(|o| o.object_type == object_type) { + return Err(format!( + "auth.openfga.visibility.objects[{index}]: type `{object_type}` \ + bound twice (RFC 0047 §3.4)" + )); + } + let column = promoted_column( + &format!("objects[{index}].column"), + object.column.as_deref().unwrap_or_default(), + )?; + objects.push(VisibilityObject { + object_type, + column, + }); + } + Ok(objects) +} + +/// Validate the raw visibility section (RFC 0047 §3.4). +/// +/// # Errors +/// +/// v1 binds at most the `conversation` type, to an `attr.`/`resource.` +/// promoted column; `self_principal_column` must be such a column; +/// `content_columns` entries must be `body` or `attr.`/`resource.` names; +/// `max_objects` ≥ 1; `list_timeout_ms` ≥ 1 and **strictly below** +/// `server_list_objects_deadline_ms` — the client timeout must be the one +/// that fires, so an incomplete enumeration is always detected here. +fn build_visibility_config( + spec: &VisibilitySpec, + server_deadline_ms: Option<&str>, +) -> Result { + let promoted_column = |key: &str, value: &str| -> Result { + if value.is_empty() + || value.trim() != value + || !(value.starts_with("attr.") || value.starts_with("resource.")) + { + return Err(format!( + "auth.openfga.visibility.{key} must name a promoted column as \ + `attr.` or `resource.` (RFC 0047 §3.4)" + )); + } + Ok(value.to_string()) + }; + let objects = build_visibility_objects(&spec.objects, promoted_column)?; + let self_principal_column = match spec.self_principal_column.as_deref() { + None | Some("") => None, + Some(column) => Some(promoted_column("self_principal_column", column)?), + }; + let content_columns = match &spec.content_columns { + None => DEFAULT_CONTENT_COLUMNS + .iter() + .map(|c| (*c).to_string()) + .collect(), + // Masking is never silently disabled: an empty list would let a + // metadata-only reader read every content column. + Some(columns) if columns.is_empty() => { + return Err( + "auth.openfga.visibility.content_columns must not be empty — omit it for the \ + default set; metadata-only readers always have content masked (RFC 0047 §3.4)" + .to_string(), + ); + } + Some(columns) => columns + .iter() + .enumerate() + .map(|(index, column)| match column.as_str() { + "body" => Ok(column.clone()), + other => promoted_column(&format!("content_columns[{index}]"), other), + }) + .collect::, _>>()?, + }; + let count = |key: &str, raw: Option<&str>, default: u64| -> Result { + match raw { + None => Ok(default), + Some(raw) => match raw.trim().parse::() { + Ok(n) if n >= 1 => Ok(n), + _ => Err(format!( + "auth.openfga.{key} must be a positive integer (RFC 0047 §3.4)" + )), + }, + } + }; + let max_objects = usize::try_from(count( + "visibility.max_objects", + spec.max_objects.as_deref(), + DEFAULT_MAX_OBJECTS as u64, + )?) + .map_err(|_| "auth.openfga.visibility.max_objects is out of range".to_string())?; + let list_timeout_ms = count( + "visibility.list_timeout_ms", + spec.list_timeout_ms.as_deref(), + DEFAULT_LIST_TIMEOUT_MS, + )?; + let server_deadline_ms = count( + "server_list_objects_deadline_ms", + server_deadline_ms, + DEFAULT_SERVER_LIST_OBJECTS_DEADLINE_MS, + )?; + if list_timeout_ms >= server_deadline_ms { + return Err(format!( + "auth.openfga.visibility.list_timeout_ms ({list_timeout_ms}) must be strictly \ + below auth.openfga.server_list_objects_deadline_ms ({server_deadline_ms}): the \ + client timeout must be the one that fires, so an incomplete enumeration is \ + detected and failed closed here (RFC 0047 §3.4)" + )); + } + Ok(VisibilityConfig { + objects, + self_principal_column, + content_columns, + max_objects, + list_timeout: Duration::from_millis(list_timeout_ms), }) } @@ -304,6 +569,93 @@ impl fmt::Display for Principal { /// `tenant` type, the object every resource hangs off. pub const TENANT_TYPE: &str = "tenant"; +/// `OpenFGA`'s object-id limit. +pub const MAX_OBJECT_ID_BYTES: usize = 256; + +/// Whether `id` can be the id half of an `OpenFGA` object or user +/// (`type:id`): non-empty, at most 256 bytes, no `:`, `#` or whitespace. +#[must_use] +pub fn is_object_id(id: &str) -> bool { + !id.is_empty() + && id.len() <= MAX_OBJECT_ID_BYTES + && !id + .chars() + .any(|c| c == ':' || c == '#' || c.is_whitespace()) +} + +/// The naming rule for tenant-scoped objects (RFC 0047 §3.3) — the **one** +/// place it lives, used by the planner and the emitter alike. The tenant is +/// its own object, `tenant:`; a conversation inside it is +/// `conversation:/` where `enc` percent-encodes `%` and `/` in +/// the tenant so the `/` separator is unambiguous — `a` + `b/c-1` and +/// `a/b` + `c-1` are two different objects — and the raw conversation id +/// follows verbatim (it may itself contain `/`). +/// +/// A tenant that cannot be an object id at all (`:`, `#`, whitespace, too +/// long, empty) has no graph objects; callers fail closed. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct TenantObjects { + tenant_object: String, + conversation_prefix: String, + tool_prefix: String, +} + +impl TenantObjects { + /// The graph objects of `tenant`, or `None` when the tenant id cannot + /// form an object id. + #[must_use] + pub fn new(tenant: &str) -> Option { + if !is_object_id(tenant) { + return None; + } + let encoded = encode_tenant_segment(tenant); + Some(Self { + tenant_object: format!("{TENANT_TYPE}:{tenant}"), + conversation_prefix: format!("{CONVERSATION_TYPE}:{encoded}/"), + tool_prefix: format!("tool:{encoded}/"), + }) + } + + /// `tenant:`. + #[must_use] + pub fn tenant(&self) -> &str { + &self.tenant_object + } + + /// `conversation:/` — every conversation of the tenant starts + /// with this; the remainder is the raw conversation id. + #[must_use] + pub fn conversation_prefix(&self) -> &str { + &self.conversation_prefix + } + + /// `conversation:/`. + #[must_use] + pub fn conversation(&self, id: &str) -> String { + format!("{}{id}", self.conversation_prefix) + } + + /// `tool:/`. + #[must_use] + pub fn tool(&self, name: &str) -> String { + format!("{}{name}", self.tool_prefix) + } +} + +/// Percent-encode the two bytes that would make the tenant segment of a +/// `conversation:/` object ambiguous. +fn encode_tenant_segment(tenant: &str) -> String { + let mut out = String::with_capacity(tenant.len()); + for c in tenant.chars() { + match c { + '%' => out.push_str("%25"), + '/' => out.push_str("%2F"), + other => out.push(other), + } + } + out +} + #[cfg(test)] mod tests { use std::time::Duration; @@ -403,6 +755,161 @@ mod tests { } } + /// RFC 0047 §3.4 visibility validation: defaults, the conversation-only + /// binding, promoted-column names, and the client-below-server timeout + /// rule — each rejection naming its key. + #[test] + #[allow(clippy::too_many_lines)] // one validation matrix, one test + fn visibility_config_defaults_and_rules() { + use super::{VisibilityObjectSpec, VisibilitySpec}; + let defaults = build_openfga_config(&spec()) + .expect("valid") + .visibility() + .clone(); + assert!(defaults.objects().is_empty()); + assert_eq!(defaults.self_principal_column(), None); + assert_eq!(defaults.content_columns()[0], "body"); + assert_eq!(defaults.content_columns().len(), 6); + assert_eq!(defaults.max_objects(), 10_000); + assert_eq!(defaults.list_timeout(), Duration::from_secs(2)); + + let bound = build_openfga_config(&OpenFgaSpec { + visibility: VisibilitySpec { + objects: vec![VisibilityObjectSpec { + object_type: Some("conversation".to_string()), + column: Some("attr.gen_ai.conversation.id".to_string()), + }], + self_principal_column: Some("attr.user.hash".to_string()), + content_columns: Some(vec!["body".to_string(), "attr.prompt".to_string()]), + max_objects: Some("100".to_string()), + list_timeout_ms: Some("500".to_string()), + }, + server_list_objects_deadline_ms: Some("1000".to_string()), + ..spec() + }) + .expect("valid"); + let visibility = bound.visibility(); + assert_eq!(visibility.objects()[0].object_type(), "conversation"); + assert_eq!( + visibility.objects()[0].column(), + "attr.gen_ai.conversation.id" + ); + assert_eq!(visibility.self_principal_column(), Some("attr.user.hash")); + assert_eq!(visibility.content_columns(), ["body", "attr.prompt"]); + assert_eq!(visibility.max_objects(), 100); + assert_eq!(visibility.list_timeout(), Duration::from_millis(500)); + + let object = |object_type: &str, column: &str| VisibilityObjectSpec { + object_type: Some(object_type.to_string()), + column: Some(column.to_string()), + }; + for (key, visibility, deadline) in [ + ( + "objects[0].type", + VisibilitySpec { + objects: vec![object("tool", "attr.tool")], + ..VisibilitySpec::default() + }, + None, + ), + ( + "objects[1]", + VisibilitySpec { + objects: vec![ + object("conversation", "attr.a"), + object("conversation", "attr.b"), + ], + ..VisibilitySpec::default() + }, + None, + ), + ( + "objects[0].column", + VisibilitySpec { + objects: vec![object("conversation", "gen_ai.conversation.id")], + ..VisibilitySpec::default() + }, + None, + ), + ( + "self_principal_column", + VisibilitySpec { + self_principal_column: Some("user.hash".to_string()), + ..VisibilitySpec::default() + }, + None, + ), + ( + "content_columns[1]", + VisibilitySpec { + content_columns: Some(vec!["body".to_string(), "severity".to_string()]), + ..VisibilitySpec::default() + }, + None, + ), + ( + "visibility.max_objects", + VisibilitySpec { + max_objects: Some("0".to_string()), + ..VisibilitySpec::default() + }, + None, + ), + ( + "visibility.list_timeout_ms", + VisibilitySpec { + list_timeout_ms: Some("3000".to_string()), + ..VisibilitySpec::default() + }, + None, + ), + ( + "visibility.list_timeout_ms", + VisibilitySpec::default(), + Some("2000"), + ), + ( + "content_columns must not be empty", + VisibilitySpec { + content_columns: Some(Vec::new()), + ..VisibilitySpec::default() + }, + None, + ), + ] { + let err = build_openfga_config(&OpenFgaSpec { + visibility, + server_list_objects_deadline_ms: deadline.map(str::to_string), + ..spec() + }) + .expect_err("invalid"); + assert!(err.contains(key), "{key} named: {err}"); + } + } + + /// The tenant-scoped object naming rule is injective in the tenant + /// (`/` and `%` percent-encoded in the tenant segment) and refuses a + /// tenant that cannot be an object id. + #[test] + fn tenant_objects_are_unambiguous() { + use super::TenantObjects; + let a = TenantObjects::new("a").expect("valid"); + let ab = TenantObjects::new("a/b").expect("valid"); + assert_eq!(a.tenant(), "tenant:a"); + assert_eq!(a.conversation("b/c-1"), "conversation:a/b/c-1"); + assert_eq!(ab.conversation("c-1"), "conversation:a%2Fb/c-1"); + assert_ne!(a.conversation("b/c-1"), ab.conversation("c-1")); + assert_eq!(ab.conversation_prefix(), "conversation:a%2Fb/"); + assert_eq!( + TenantObjects::new("100%").expect("valid").conversation("x"), + "conversation:100%25/x" + ); + assert_eq!(a.tool("query_logs"), "tool:a/query_logs"); + for bad in ["", "a b", "a:b", "a#b"] { + assert!(TenantObjects::new(bad).is_none(), "{bad:?}"); + } + } + /// The principal vocabulary renders exactly the model's type names. #[test] fn principals_render_model_types() { diff --git a/crates/ourios-ingester/src/receiver.rs b/crates/ourios-ingester/src/receiver.rs index bb4958f5..17350fc0 100644 --- a/crates/ourios-ingester/src/receiver.rs +++ b/crates/ourios-ingester/src/receiver.rs @@ -45,7 +45,7 @@ pub mod tenant; pub mod tls; pub mod tls_serve; -pub use auth::{AuthBinding, AuthError, AuthResolver, authenticate_bearer}; +pub use auth::{AuthBinding, AuthError, AuthResolver, GraphIdentity, authenticate_bearer}; pub use commit::CommitCoordinator; pub use decode::{DecodeError, decode_json, decode_protobuf}; pub use materialize::{materialize_record, materialize_resource_logs}; diff --git a/crates/ourios-ingester/src/receiver/auth.rs b/crates/ourios-ingester/src/receiver/auth.rs index 618cabea..ef6e5a17 100644 --- a/crates/ourios-ingester/src/receiver/auth.rs +++ b/crates/ourios-ingester/src/receiver/auth.rs @@ -42,6 +42,30 @@ pub struct AuthBinding { token_name: String, read: TenantSet, write: TenantSet, + /// The RFC 0047 principal and its token groups — `Some` only when the + /// graph resolver bound this session, for the planner's two-step. + graph: Option, +} + +/// The graph-side identity of a binding (RFC 0047 §3.1/§3.4). +#[derive(Debug, Clone)] +pub struct GraphIdentity { + principal: Principal, + groups: Vec, +} + +impl GraphIdentity { + /// The `OpenFGA` principal. + #[must_use] + pub fn principal(&self) -> &Principal { + &self.principal + } + + /// The token's group claim (contextual `team#member` tuples). + #[must_use] + pub fn groups(&self) -> &[String] { + &self.groups + } } impl AuthBinding { @@ -75,11 +99,18 @@ impl AuthBinding { self.write.allows(tenant) } + /// The graph identity, when the RFC 0047 resolver bound this session. + #[must_use] + pub fn graph(&self) -> Option<&GraphIdentity> { + self.graph.as_ref() + } + fn same(token_name: String, tenants: TenantSet) -> Self { Self { token_name, read: tenants.clone(), write: tenants, + graph: None, } } } @@ -235,6 +266,14 @@ impl AuthResolver { self } + /// The RFC 0047 graph resolver, when configured — the planner's + /// two-step and the tool gate consult it. + #[cfg(feature = "openfga")] + #[must_use] + pub fn openfga(&self) -> Option<&Arc> { + self.openfga.as_ref() + } + /// Whether every request passes unbound (§3.1 open mode). #[must_use] pub fn is_open(&self) -> bool { @@ -331,7 +370,8 @@ impl AuthResolver { // name that is no object id): named, 401-class. Err( e @ (OpenFgaError::TooManyContextualTuples { .. } - | OpenFgaError::InvalidGroup { .. }), + | OpenFgaError::InvalidGroup { .. } + | OpenFgaError::InvalidPrincipal), ) => { tracing::warn!( token_name = %identity.name, @@ -359,6 +399,10 @@ impl AuthResolver { token_name: identity.name, read, write, + graph: Some(GraphIdentity { + principal: identity.principal, + groups: identity.groups, + }), })); } // Without the graph a credential must carry its own tenant list — diff --git a/crates/ourios-querier/src/compile.rs b/crates/ourios-querier/src/compile.rs index c92e8dcd..a9ade633 100644 --- a/crates/ourios-querier/src/compile.rs +++ b/crates/ourios-querier/src/compile.rs @@ -99,6 +99,9 @@ pub(crate) struct Plan { body_equalities: BTreeMap, pub(crate) limit: Option, pub(crate) aggregate: Option, + /// RFC 0047 §3.4 layer-2 visibility, applied in [`apply`] as one more + /// filter over the promoted columns (so it prunes like any predicate). + visibility: Option, } /// A validated aggregation stage (RFC 0002 §6.3/§6.5 amendment 2026-07-15 for @@ -320,6 +323,7 @@ pub(crate) fn compile( default_window_nanos: u64, alias_map: &AliasMap, registry: &TemplateRegistry, + visibility: Option, ) -> Result { let Validated { window, @@ -343,6 +347,7 @@ pub(crate) fn compile( body_equalities, limit, aggregate, + visibility, }) } @@ -487,6 +492,7 @@ pub(crate) fn apply(df: DataFrame, plan: Plan) -> Result, Quer // the `Aggregate` node, over the same filtered scan); the caller // reads it off the plan before handing the plan here. aggregate: _, + visibility, } = plan; // The window filters the *effective* timestamp (RFC 0002 §6.2 amendment // 2026-06-11), with the RFC 0005 §3.9 fallback for pre-amendment files; @@ -504,6 +510,19 @@ pub(crate) fn apply(df: DataFrame, plan: Plan) -> Result, Quer } } + // RFC 0047 §3.4: the scoped principal's `IN (…)` / self fast path — an + // ordinary predicate over promoted columns, so it prunes; nothing to + // see ⇒ an empty result, not an error. + if let Some(visibility) = &visibility { + match visibility.filter(&df)? { + crate::visibility::VisibilityFilter::Nothing => return Ok(None), + crate::visibility::VisibilityFilter::Everything => {} + crate::visibility::VisibilityFilter::Only(expr) => { + df = df.filter(expr).map_err(crate::storage_err)?; + } + } + } + Ok(Some(df)) } diff --git a/crates/ourios-querier/src/lib.rs b/crates/ourios-querier/src/lib.rs index 8c4c0a2e..48b2fcb6 100644 --- a/crates/ourios-querier/src/lib.rs +++ b/crates/ourios-querier/src/lib.rs @@ -47,6 +47,7 @@ mod log_row; mod schema_adapt; mod template_map; mod template_registry; +pub mod visibility; pub use alias_store::derive_alias_map; pub use audit_scan::StoreRef; @@ -59,6 +60,7 @@ pub use template_map::{ load_or_derive, }; pub use template_registry::{TemplateRegistry, derive_template_registry}; +pub use visibility::{ScopedIds, SelfMatch, Visibility}; use std::path::PathBuf; use std::sync::Arc; @@ -151,7 +153,7 @@ pub struct QueryStats { /// Additive execution options for [`Querier::run_query_with`]. The /// `Default` is byte-for-byte the [`Querier::run_query`] behavior. -#[derive(Debug, Clone, Copy, Default)] +#[derive(Debug, Clone, Default)] #[non_exhaustive] pub struct QueryOptions { /// Single-pass execution for limited queries (RFC 0031 §3.6): run the @@ -175,6 +177,10 @@ pub struct QueryOptions { /// total IO. Callers needing the pinned "a limited query's `stats` /// equal a count-only query's" shape (RFC 0017 §3.4) keep the default. pub elide_count_scan: bool, + /// RFC 0047 §3.4 layer-2 visibility: `None` = no rewrite (open mode, + /// or a resolver without the graph); `Some` = the caller's two-step + /// decision, applied at plan time (see [`Visibility`]). + pub visibility: Option, } impl QueryOptions { @@ -184,8 +190,16 @@ impl QueryOptions { pub const fn single_pass() -> Self { Self { elide_count_scan: true, + visibility: None, } } + + /// These options with the RFC 0047 §3.4 visibility decision attached. + #[must_use] + pub fn with_visibility(mut self, visibility: Visibility) -> Self { + self.visibility = Some(visibility); + self + } } /// Result of a query: the matching-row count (`rows`) and the scan's pruning @@ -291,6 +305,10 @@ pub enum QueryError { /// implementation specifics the public surface must not expose /// (hazard §4.6 / RFC0007.3). Storage { detail: String }, + /// The query reads a column the principal may not read (RFC 0047 + /// §3.4 masking — a filter or aggregation on a content column). Names + /// the column: it is configuration, not data. + Forbidden { column: String }, } impl std::fmt::Display for QueryError { @@ -302,6 +320,9 @@ impl std::fmt::Display for QueryError { // message would leak `DataFusion`/SQL specifics (§4.6). // The detail is preserved on the variant for `Debug`. Self::Storage { .. } => write!(f, "failed to read storage"), + Self::Forbidden { column } => { + write!(f, "column `{column}` is not readable by this principal") + } } } } @@ -1154,6 +1175,11 @@ impl Querier { // pure validation internally — one source of truth, negligible // cost. compile::validate(query, now_unix_nano, default_window_nanos)?; + // RFC 0047 §3.4: a metadata-only reader's query must not touch a + // content column — rejected before any IO, naming the column. + if let Some(visibility) = &options.visibility { + visibility.validate(query)?; + } // A `body ==`/`!=` needs the RFC 0017 registry for the RFC 0044 // template arm; the `resolves_to` alias fold needs the alias map. // Both ride the one RFC 0033 cached-map acquisition (artifact hit or @@ -1205,6 +1231,7 @@ impl Querier { default_window_nanos, map, registry, + options.visibility.clone(), )?; // The DSL `limit` (RFC 0002) doubles as the RFC 0017 row cap; read it // — and the aggregation stage — before `plan` moves into the filter @@ -1401,13 +1428,17 @@ impl Querier { .collect_records(df.clone(), n, tenant, ctx.task_ctx(), acquired.take()) .await?; if collected.records.len() < n { + let mut records = collected.records; + if let Some(visibility) = &query_options.visibility { + visibility.mask(&mut records); + } return Ok(QueryResult { - rows: collected.records.len() as u64, + rows: records.len() as u64, stats: QueryStats { bytes_read: 0, ..collected.scan }, - records: collected.records, + records, aggregate: None, materialize_bytes_read: collected.scan.bytes_read, registry_bytes_read: collected.registry_bytes_read, @@ -1449,10 +1480,14 @@ impl Querier { } (None, None) => CollectedRecords::default(), }; + let mut records = collected.records; + if let Some(visibility) = &query_options.visibility { + visibility.mask(&mut records); + } Ok(QueryResult { rows, stats, - records: collected.records, + records, aggregate: None, materialize_bytes_read: collected.scan.bytes_read, registry_bytes_read: collected.registry_bytes_read, diff --git a/crates/ourios-querier/src/log_row.rs b/crates/ourios-querier/src/log_row.rs index dca49a52..ac080865 100644 --- a/crates/ourios-querier/src/log_row.rs +++ b/crates/ourios-querier/src/log_row.rs @@ -129,6 +129,10 @@ pub enum LogBody { /// JSON `body`, returned as the typed value (any non-`String` variant: /// kvlist / array or a scalar int / bool / bytes), never flattened. Structured(AnyValue), + /// The body was withheld from a metadata-only reader (RFC 0047 §3.4 + /// masking). Distinct from [`LogBody::Absent`]: the record has a body; + /// this principal may not read it. + Masked, /// `body_kind = Absent` — the wire delivered no body (RFC 0025 /// §3.2). Deliberately distinct from /// [`LogBody::Rendered`] with an empty line: an empty-string @@ -184,6 +188,15 @@ pub fn render_log_body(record: &MinedRecord, registry: &TemplateRegistry) -> Log } } +#[cfg(test)] +impl LogRow { + /// A row with every field at its zero value — the fixture other + /// in-crate unit tests mutate. + pub(crate) fn test_row() -> Self { + Self::from_record(&tests::record(BodyKind::Absent), &TemplateRegistry::new()) + } +} + #[cfg(test)] mod tests { use ourios_core::otlp::any_value::Value; @@ -196,7 +209,7 @@ mod tests { AnyValue, BodyKind, LogBody, MinedRecord, Reconstruction, TemplateRegistry, render_log_body, }; - fn record(body_kind: BodyKind) -> MinedRecord { + pub(super) fn record(body_kind: BodyKind) -> MinedRecord { MinedRecord { tenant_id: TenantId::new("t"), template_id: 0, diff --git a/crates/ourios-querier/src/visibility.rs b/crates/ourios-querier/src/visibility.rs new file mode 100644 index 00000000..1b8ab585 --- /dev/null +++ b/crates/ourios-querier/src/visibility.rs @@ -0,0 +1,351 @@ +//! RFC 0047 §3.4 — layer-2 visibility as the engine sees it: a decision the +//! caller already made against the authorization graph, applied here as +//! **query rewrite at plan time** — an extra predicate over a promoted +//! column, or column masking on the returned rows — never as per-record +//! checks. The engine knows nothing about `OpenFGA`; it receives one of +//! three shapes and applies it. + +use datafusion::common::Column; +use datafusion::dataframe::DataFrame; +use datafusion::logical_expr::Expr; +use datafusion::prelude::lit; + +use crate::dsl::ir::{Call, Field, GroupTerm, Predicate, Query, Stage}; +use crate::log_row::{LogBody, LogRow}; +use crate::{QueryError, has_column}; +use ourios_parquet::promoted; + +/// What the principal may see inside the tenant (RFC 0047 §3.4). Column +/// names are DSL names: `body`, `attr.`, `resource.`. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum Visibility { + /// Step 1 allowed: the tenant predicate only — today's plan, unchanged. + TenantWide, + /// Step 2 allowed: every row of the tenant, with `content_columns` + /// masked on the way out; a query that filters or aggregates on one of + /// them is rejected (`403`, named column) rather than answered. + Masked { + /// The columns a metadata-only reader may not read. + content_columns: Vec, + }, + /// Step 3: only rows whose conversation is one the principal may read + /// — OR'd with the §3.3 self fast path when configured. No bound + /// object type, or an empty id set, and no self match ⇒ an empty + /// result, not an error. + Scoped { + /// The enumerated conversations over their promoted column; `None` + /// when no object type is bound (nothing to enumerate). + conversations: Option, + /// The self fast path: rows whose `column` equals `value`. + self_match: Option, + }, +} + +/// The enumerated object ids of a scoped principal and the promoted +/// column they are matched against. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ScopedIds { + /// The promoted column carrying the object ids (`attr.`). + pub column: String, + /// The ids the principal may read (prefix stripped). + pub ids: Vec, +} + +/// The §3.3 self fast path: ` == `, where `value` is a +/// `user:` principal's subject with the prefix stripped. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct SelfMatch { + /// The promoted column carrying the principal identity. + pub column: String, + /// The principal's subject. + pub value: String, +} + +impl Visibility { + /// Reject a query that *reads* a masked column (RFC0047.8): any content + /// column in the predicate, an aggregation path, or a `by`-list. A + /// projection is not a read — a projected content column comes back + /// masked, like every returned row's. Never an oracle about the data — + /// the columns are configuration. + /// + /// # Errors + /// + /// [`QueryError::Forbidden`] naming the first masked column referenced. + pub(crate) fn validate(&self, query: &Query) -> Result<(), QueryError> { + let Self::Masked { content_columns } = self else { + return Ok(()); + }; + let mut fields: Vec<&Field> = Vec::new(); + collect_predicate_fields(&query.predicate, &mut fields); + for stage in &query.stages { + match stage { + Stage::Count { by } => collect_group_fields(by, &mut fields), + Stage::Agg { path, by, .. } => { + fields.push(path); + collect_group_fields(by, &mut fields); + } + Stage::Range(..) + | Stage::Sort { .. } + | Stage::Limit(_) + | Stage::Project(_) + | Stage::Render => {} + } + } + for field in fields { + if let Some(name) = dsl_name(field) + && content_columns.contains(&name) + { + return Err(QueryError::Forbidden { column: name }); + } + } + Ok(()) + } + + /// The plan-time filter for a scoped principal: `column IN (ids)` OR + /// the self fast path, over the promoted columns. [`VisibilityFilter::Nothing`] + /// when the principal can see no row at all (nothing to enumerate and no + /// fast path, or the promoted columns are absent from the scanned schema — + /// an absent column carries no id, so it matches nothing); + /// [`VisibilityFilter::Everything`] for the two branches that add no + /// predicate. + /// + /// # Errors + /// + /// [`QueryError::InvalidQuery`] when a configured column is not a + /// promoted-column name. + pub(crate) fn filter(&self, df: &DataFrame) -> Result { + let Self::Scoped { + conversations, + self_match, + } = self + else { + return Ok(VisibilityFilter::Everything); + }; + let mut arms: Vec = Vec::new(); + if let Some(ScopedIds { column, ids }) = conversations + && !ids.is_empty() + && let Some(promoted) = promoted_expr(df, column)? + { + arms.push(promoted.in_list(ids.iter().map(|id| lit(id.clone())).collect(), false)); + } + if let Some(SelfMatch { column, value }) = self_match + && let Some(promoted) = promoted_expr(df, column)? + { + arms.push(promoted.eq(lit(value.clone()))); + } + let mut arms = arms.into_iter(); + let Some(first) = arms.next() else { + return Ok(VisibilityFilter::Nothing); + }; + Ok(VisibilityFilter::Only(arms.fold(first, Expr::or))) + } + + /// Mask the content columns of returned rows (RFC0047.8): the body + /// becomes [`LogBody::Masked`], masked attributes keep their key with + /// the value unset — the OTLP null (`"value": null` on the JSON API). + /// The column vocabulary (`body`, `attr.`, `resource.`) is + /// validated where it is configured (`auth.openfga.visibility`), so a + /// name of any other shape cannot reach here. + pub(crate) fn mask(&self, rows: &mut [LogRow]) { + let Self::Masked { content_columns } = self else { + return; + }; + let body = content_columns.iter().any(|column| column == "body"); + let attrs: Vec<&str> = content_columns + .iter() + .filter_map(|column| column.strip_prefix(promoted::ATTR_PREFIX)) + .collect(); + let resources: Vec<&str> = content_columns + .iter() + .filter_map(|column| column.strip_prefix(promoted::RESOURCE_PREFIX)) + .collect(); + for row in rows { + if body { + row.body = LogBody::Masked; + } + for kv in &mut row.attributes { + if attrs.contains(&kv.key.as_str()) { + kv.value = None; + } + } + for kv in &mut row.resource_attributes { + if resources.contains(&kv.key.as_str()) { + kv.value = None; + } + } + } + } +} + +/// What [`Visibility::filter`] adds to the plan. +pub(crate) enum VisibilityFilter { + /// No visibility predicate — every row the query matches. + Everything, + /// The principal can see no row: the plan short-circuits to empty. + Nothing, + /// Rows must additionally satisfy this predicate. + Only(Expr), +} + +/// The DSL name of a field the masking vocabulary can name; `None` for +/// fields that are never content. +fn dsl_name(field: &Field) -> Option { + match field { + Field::Body => Some("body".to_string()), + Field::Attr(key) => Some(format!("{}{key}", promoted::ATTR_PREFIX)), + Field::Resource(key) => Some(format!("{}{key}", promoted::RESOURCE_PREFIX)), + _ => None, + } +} + +fn collect_predicate_fields<'a>(predicate: &'a Predicate, out: &mut Vec<&'a Field>) { + match predicate { + Predicate::Bool(_) | Predicate::Severity { .. } => {} + Predicate::Comparison { field, .. } => out.push(field), + Predicate::Call(call) => match call { + Call::Matches { field, .. } + | Call::Contains { field, .. } + | Call::StartsWith { field, .. } + | Call::EndsWith { field, .. } => out.push(field), + Call::ResolvesTo(_) => {} + }, + Predicate::Not(inner) => collect_predicate_fields(inner, out), + Predicate::And(terms) | Predicate::Or(terms) => { + for term in terms { + collect_predicate_fields(term, out); + } + } + } +} + +fn collect_group_fields<'a>(by: &'a [GroupTerm], out: &mut Vec<&'a Field>) { + for term in by { + if let GroupTerm::Field(field) = term { + out.push(field); + } + } +} + +/// The unqualified column expression for a DSL `attr.` / +/// `resource.` name when the promoted column is in the scanned +/// schema; `Ok(None)` when absent (it matches nothing). +fn promoted_expr(df: &DataFrame, name: &str) -> Result, QueryError> { + if !(name.starts_with(promoted::ATTR_PREFIX) || name.starts_with(promoted::RESOURCE_PREFIX)) { + return Err(QueryError::InvalidQuery { + detail: format!("visibility column `{name}` is not a promoted attribute column"), + }); + } + // The promoted column is literally named `attr.` — the same string + // as the DSL name — and must be addressed as an unqualified `Column` + // (`col()` would parse the dots as a qualifier). + Ok(has_column(df, name).then(|| Expr::Column(Column::new_unqualified(name)))) +} + +#[cfg(test)] +mod tests { + use super::{ScopedIds, SelfMatch, Visibility}; + use crate::QueryError; + use crate::dsl::ir::{Query, Statement}; + use crate::dsl::parse_statement; + use crate::log_row::{LogBody, LogRow}; + + fn logs(statement: &str) -> Query { + match parse_statement(statement).expect("parses") { + Statement::Logs(query) => query, + Statement::Drift(_) => panic!("not a logs query"), + } + } + + fn masked() -> Visibility { + Visibility::Masked { + content_columns: vec!["body".to_string(), "attr.gen_ai.input.messages".to_string()], + } + } + + /// RFC0047.8: a masked column in the predicate, an aggregation path, or + /// a by-list is rejected naming the column; anything else passes, and + /// the other branches never reject. + #[test] + fn masked_columns_are_forbidden_in_filters_and_aggregations() { + for statement in [ + "attr.gen_ai.input.messages == \"hi\"", + "contains(body, \"x\")", + "not (severity >= 9 or attr.gen_ai.input.messages == \"a\")", + "true | sum(attr.gen_ai.input.messages) by attr.model", + "true | count by attr.gen_ai.input.messages", + ] { + let query = logs(statement); + match masked().validate(&query) { + Err(QueryError::Forbidden { column }) => assert!( + column == "body" || column == "attr.gen_ai.input.messages", + "{statement}: names the column, got {column}" + ), + other => panic!("{statement}: expected Forbidden, got {other:?}"), + } + } + for statement in [ + "true", + "attr.model == \"gpt\" | sum(attr.cost_usd) by attr.model", + "severity >= 9 | count by service", + ] { + let query = logs(statement); + masked().validate(&query).expect("not a content column"); + Visibility::TenantWide + .validate(&query) + .expect("never rejects"); + } + let query = logs("attr.gen_ai.input.messages == \"hi\""); + Visibility::Scoped { + conversations: None, + self_match: None, + } + .validate(&query) + .expect("scoped principals read their content"); + } + + /// RFC0047.8: masking sets the body to `Masked` and unsets the value of + /// masked attributes, leaving keys (and every other column) intact. + #[test] + fn masking_nulls_content_columns_only() { + use ourios_core::otlp::{AnyValue, KeyValue, any_value}; + let kv = |key: &str, value: &str| KeyValue { + key: key.to_string(), + value: Some(AnyValue { + value: Some(any_value::Value::StringValue(value.to_string())), + }), + ..Default::default() + }; + let mut row = LogRow::test_row(); + row.body = LogBody::Rendered { + line: b"secret prompt".to_vec(), + reconstruction: ourios_miner::reconstruct::Reconstruction::Faithful, + }; + row.attributes = vec![kv("gen_ai.input.messages", "hi"), kv("model", "gpt")]; + let mut rows = vec![row]; + masked().mask(&mut rows); + assert_eq!(rows[0].body, LogBody::Masked); + assert_eq!(rows[0].attributes[0].key, "gen_ai.input.messages"); + assert!( + rows[0].attributes[0].value.is_none(), + "value unset (OTLP null)" + ); + assert!( + rows[0].attributes[1].value.is_some(), + "other columns intact" + ); + + let mut rows = vec![LogRow::test_row()]; + Visibility::Scoped { + conversations: Some(ScopedIds { + column: "attr.gen_ai.conversation.id".to_string(), + ids: vec!["c-1".to_string()], + }), + self_match: Some(SelfMatch { + column: "attr.user.hash".to_string(), + value: "bob".to_string(), + }), + } + .mask(&mut rows); + assert_ne!(rows[0].body, LogBody::Masked, "scoped rows are not masked"); + } +} diff --git a/crates/ourios-querier/tests/it/main.rs b/crates/ourios-querier/tests/it/main.rs index ede097fe..9e6d46e4 100644 --- a/crates/ourios-querier/tests/it/main.rs +++ b/crates/ourios-querier/tests/it/main.rs @@ -36,3 +36,4 @@ mod rfc0033_cached_template_map; mod rfc0036_window_materialization; mod rfc0042_mixed_schema; mod rfc0044_body_equality; +mod rfc0047_visibility; diff --git a/crates/ourios-querier/tests/it/rfc0047_visibility.rs b/crates/ourios-querier/tests/it/rfc0047_visibility.rs new file mode 100644 index 00000000..d3a6f48e --- /dev/null +++ b/crates/ourios-querier/tests/it/rfc0047_visibility.rs @@ -0,0 +1,238 @@ +//! RFC 0047 §3.4 at the engine (RFC0047.5/.6/.8, the predicate-composition +//! half of §6): a scoped [`Visibility`] becomes an `IN (…)` over the +//! promoted conversation column OR'd with the self fast path; a masked one +//! returns every row with the content columns nulled and rejects a filter or +//! aggregation on them; tenant-wide is today's plan. The graph decision is +//! the caller's — here it is handed in directly. +//! See `docs/rfcs/0047-rebac-resolver-and-graph-visibility.md` §5. + +use ourios_core::tenant::TenantId; +use ourios_parquet::{PromotedAttributes, PromotedClass, PromotedKey}; +use ourios_querier::{ + LogBody, Querier, QueryError, QueryOptions, QueryResult, ScopedIds, SelfMatch, Visibility, +}; + +use crate::common::{ + DEFAULT_WINDOW_NS, NOW, TS0, kv, no_aliases, rec_with_attrs, write_all_with_promoted, +}; + +const CONVERSATION: &str = "attr.gen_ai.conversation.id"; +const USER_HASH: &str = "attr.user.hash"; + +fn kv_double(key: &str, value: f64) -> ourios_core::otlp::KeyValue { + ourios_core::otlp::KeyValue { + key: key.to_string(), + value: Some(ourios_core::otlp::AnyValue { + value: Some(ourios_core::otlp::any_value::Value::DoubleValue(value)), + }), + ..Default::default() + } +} + +/// Six rows in `acme`: conversations c-1..c-3 (two rows each), each with a +/// `user.hash`, a `gen_ai.input.messages` content attribute and a +/// `cost_usd`; plus one `globex` row that shares an id with `acme/c-1`. +fn seed() -> tempfile::TempDir { + let bucket = tempfile::TempDir::new().expect("temp"); + let promoted = PromotedAttributes::new_typed( + [], + [ + PromotedKey::string("gen_ai.conversation.id".to_string()), + PromotedKey::string("user.hash".to_string()), + PromotedKey::string("model".to_string()), + PromotedKey { + key: "cost_usd".into(), + class: PromotedClass::F64, + }, + ], + ); + let row = |tenant: &str, i: u64, conversation: &str, user: &str| { + rec_with_attrs( + tenant, + TS0 + i * 1_000, + vec![kv("service.name", "agent")], + vec![ + kv("gen_ai.conversation.id", conversation), + kv("user.hash", user), + kv("gen_ai.input.messages", "the secret prompt"), + kv("model", "gpt"), + kv_double("cost_usd", 1.5), + ], + ) + }; + let recs = vec![ + row("acme", 1, "c-1", "alice"), + row("acme", 2, "c-1", "alice"), + row("acme", 3, "c-2", "bob"), + row("acme", 4, "c-2", "carol"), + row("acme", 5, "c-3", "carol"), + row("acme", 6, "c-3", "carol"), + row("globex", 7, "c-1", "mallory"), + ]; + write_all_with_promoted(bucket.path(), &recs, &promoted); + bucket +} + +async fn run( + bucket: &std::path::Path, + dsl: &str, + visibility: Visibility, +) -> Result { + let query = ourios_querier::dsl::parse(dsl).expect("parse DSL"); + Querier::new(bucket) + .run_query_with( + &query, + &TenantId::new("acme"), + NOW, + DEFAULT_WINDOW_NS, + Some(&no_aliases()), + QueryOptions::default().with_visibility(visibility), + ) + .await +} + +fn conversations(result: &QueryResult) -> Vec { + let mut ids: Vec = result + .records + .iter() + .map(|row| { + row.attributes + .iter() + .find(|kv| kv.key == "gen_ai.conversation.id") + .and_then(|kv| kv.value.as_ref()) + .and_then(|v| match &v.value { + Some(ourios_core::otlp::any_value::Value::StringValue(s)) => Some(s.clone()), + _ => None, + }) + .expect("conversation id") + }) + .collect(); + ids.sort(); + ids +} + +fn scoped(ids: &[&str], self_match: Option<&str>) -> Visibility { + Visibility::Scoped { + conversations: Some(ScopedIds { + column: CONVERSATION.to_string(), + ids: ids.iter().map(|s| (*s).to_string()).collect(), + }), + self_match: self_match.map(|value| SelfMatch { + column: USER_HASH.to_string(), + value: value.to_string(), + }), + } +} + +/// RFC0047.5/.6 (engine half): exactly the scoped ids' rows return, the +/// self fast path adds the principal's own rows, an empty scope with no +/// fast path is an empty result (not an error), and the tenant-wide branch +/// is untouched. `true | limit 100` is a match-all query. +#[tokio::test] +async fn scoped_visibility_filters_to_the_ids_and_self() { + let bucket = seed(); + let all = run(bucket.path(), "true | limit 100", Visibility::TenantWide) + .await + .expect("query"); + assert_eq!(all.rows, 6, "tenant-wide: every acme row (never globex)"); + + let bob = run(bucket.path(), "true | limit 100", scoped(&["c-2"], None)) + .await + .expect("query"); + assert_eq!(conversations(&bob), ["c-2", "c-2"]); + assert_eq!(bob.rows, 2, "the count follows the same predicate"); + + // Scoped to c-2 plus the self fast path on `carol` picks up c-3 too. + let carol = run( + bucket.path(), + "true | limit 100", + scoped(&["c-2"], Some("carol")), + ) + .await + .expect("query"); + assert_eq!(conversations(&carol), ["c-2", "c-2", "c-3", "c-3"]); + + // The user's own predicate composes (AND) with the visibility filter. + let narrowed = run( + bucket.path(), + "attr.user.hash == \"bob\" | limit 100", + scoped(&["c-2", "c-3"], None), + ) + .await + .expect("query"); + assert_eq!(narrowed.rows, 1); + + let nothing = run(bucket.path(), "true | limit 100", scoped(&[], None)) + .await + .expect("empty scope is not an error"); + assert_eq!(nothing.rows, 0); + assert!(nothing.records.is_empty()); + + // Aggregations run over the scoped rows only. + let spend = run( + bucket.path(), + "true | sum(attr.cost_usd) by attr.model", + scoped(&["c-1"], None), + ) + .await + .expect("query"); + let groups = spend.aggregate.expect("aggregate"); + assert_eq!(groups.len(), 1); + let value = groups[0].value.flatten().expect("sum"); + assert!((value - 3.0).abs() < 1e-9, "two c-1 rows × 1.5"); +} + +/// RFC0047.8 (engine half): a masked reader gets every row with the +/// content columns nulled — body `Masked`, the attribute's value unset — +/// while other attributes and aggregations over metadata are intact; a +/// filter or aggregation on a content column is `Forbidden` naming it. +#[tokio::test] +async fn masked_visibility_nulls_content_and_forbids_reading_it() { + let bucket = seed(); + let masked = Visibility::Masked { + content_columns: vec!["body".to_string(), "attr.gen_ai.input.messages".to_string()], + }; + let rows = run(bucket.path(), "true | limit 100", masked.clone()) + .await + .expect("query"); + assert_eq!(rows.rows, 6, "every row of the tenant"); + for row in &rows.records { + assert_eq!(row.body, LogBody::Masked); + let content = row + .attributes + .iter() + .find(|kv| kv.key == "gen_ai.input.messages") + .expect("key kept"); + assert!(content.value.is_none(), "value unset"); + assert!( + row.attributes + .iter() + .any(|kv| kv.key == "model" && kv.value.is_some()), + "metadata intact" + ); + } + let spend = run( + bucket.path(), + "true | sum(attr.cost_usd) by attr.model", + masked.clone(), + ) + .await + .expect("metadata aggregation"); + let groups = spend.aggregate.expect("aggregate"); + let value = groups[0].value.flatten().expect("sum"); + assert!((value - 9.0).abs() < 1e-9, "six rows × 1.5"); + + for dsl in [ + "attr.gen_ai.input.messages == \"x\"", + "contains(body, \"secret\")", + "true | count by attr.gen_ai.input.messages", + ] { + match run(bucket.path(), dsl, masked.clone()).await { + Err(QueryError::Forbidden { column }) => assert!( + column == "body" || column == "attr.gen_ai.input.messages", + "{dsl}: {column}" + ), + other => panic!("{dsl}: expected Forbidden, got {other:?}"), + } + } +} diff --git a/crates/ourios-semconv/src/lib.rs b/crates/ourios-semconv/src/lib.rs index 5e5a0db1..9c4acf3e 100644 --- a/crates/ourios-semconv/src/lib.rs +++ b/crates/ourios-semconv/src/lib.rs @@ -127,6 +127,9 @@ pub const OURIOS_QUERY_DURATION: &str = "ourios.query.duration"; /// `ourios.query.row_groups` (counter, unit `{row_group}`). pub const OURIOS_QUERY_ROW_GROUPS: &str = "ourios.query.row_groups"; +/// `ourios.query.visibility` (counter, unit `{query}`). +pub const OURIOS_QUERY_VISIBILITY: &str = "ourios.query.visibility"; + /// `ourios.receiver.tenant.divergences` (counter, unit `{divergence}`). pub const OURIOS_RECEIVER_TENANT_DIVERGENCES: &str = "ourios.receiver.tenant.divergences"; @@ -207,6 +210,9 @@ pub const OURIOS_QUERY_KIND: &str = "ourios.query.kind"; /// `ourios.query.row_group.state` attribute key. pub const OURIOS_QUERY_ROW_GROUP_STATE: &str = "ourios.query.row_group.state"; +/// `ourios.query.visibility.branch` attribute key. +pub const OURIOS_QUERY_VISIBILITY_BRANCH: &str = "ourios.query.visibility.branch"; + /// `ourios.service` attribute key. pub const OURIOS_SERVICE: &str = "ourios.service"; diff --git a/crates/ourios-server/src/auth.rs b/crates/ourios-server/src/auth.rs index 76486b21..713a7f62 100644 --- a/crates/ourios-server/src/auth.rs +++ b/crates/ourios-server/src/auth.rs @@ -9,7 +9,7 @@ //! ([`config::file::AuthSection`](crate::config::file::AuthSection)) onto //! the core spec shapes — the single validation path stays in core. -use ourios_core::auth::openfga::OpenFgaSpec; +use ourios_core::auth::openfga::{OpenFgaSpec, VisibilityObjectSpec, VisibilitySpec}; pub use ourios_core::auth::{AuthConfig, OidcConfig, ResolvedToken, TenantSet, TokenStore}; use ourios_core::auth::{OidcSpec, TokenSpec}; @@ -58,6 +58,22 @@ pub fn build_auth_config(section: Option<&AuthSection>) -> Result, + /// The layer-2 visibility section (RFC 0047 §3.4). + pub visibility: VisibilitySection, + /// The `OpenFGA` server's `OPENFGA_LIST_OBJECTS_DEADLINE` in + /// milliseconds (default 3000); `visibility.list_timeout_ms` must stay + /// strictly below it. + #[serde(deserialize_with = "scalar_opt")] + pub server_list_objects_deadline_ms: Option, +} + +/// `auth.openfga.visibility.*` — RFC 0047 §3.4. Nothing here is secret. +#[derive(Debug, Default, Deserialize)] +#[serde(default, deny_unknown_fields)] +pub struct VisibilitySection { + /// Object type → promoted column bindings (v1: `conversation` only). + pub objects: Vec, + /// The promoted column compared to a `user:` principal's subject. + #[serde(deserialize_with = "scalar_opt")] + pub self_principal_column: Option, + /// The content columns a metadata-only reader may not read. `None` = + /// the `GenAI` default set; an explicit list **replaces** it and must + /// not be empty (masking is never disabled — validated at startup). + #[serde(default, deserialize_with = "scalar_vec_opt")] + pub content_columns: Option>, + /// The per-tenant enumeration bound (default 10 000). + #[serde(deserialize_with = "scalar_opt")] + pub max_objects: Option, + /// The client-side enumeration timeout in milliseconds (default 2000). + #[serde(deserialize_with = "scalar_opt")] + pub list_timeout_ms: Option, +} + +/// One `auth.openfga.visibility.objects[]` entry. +#[derive(Debug, Default, Deserialize)] +#[serde(default, deny_unknown_fields)] +pub struct VisibilityObjectSection { + /// The `OpenFGA` object type (`conversation`). + #[serde(rename = "type", deserialize_with = "scalar_opt")] + pub object_type: Option, + /// The promoted column carrying the object ids. + #[serde(deserialize_with = "scalar_opt")] + pub column: Option, } impl fmt::Debug for OpenFgaSection { @@ -487,6 +528,11 @@ impl fmt::Debug for OpenFgaSection { .field("session_ttl_secs", &self.session_ttl_secs) .field("consistency", &self.consistency) .field("request_timeout_secs", &self.request_timeout_secs) + .field("visibility", &self.visibility) + .field( + "server_list_objects_deadline_ms", + &self.server_list_objects_deadline_ms, + ) .finish() } } @@ -688,6 +734,18 @@ impl AuthSection { substitute(&mut openfga.session_ttl_secs, lookup)?; substitute(&mut openfga.consistency, lookup)?; substitute(&mut openfga.request_timeout_secs, lookup)?; + substitute(&mut openfga.server_list_objects_deadline_ms, lookup)?; + let visibility = &mut openfga.visibility; + for object in &mut visibility.objects { + substitute(&mut object.object_type, lookup)?; + substitute(&mut object.column, lookup)?; + } + substitute(&mut visibility.self_principal_column, lookup)?; + for column in visibility.content_columns.iter_mut().flatten() { + *column = env_subst::resolve(column, lookup)?; + } + substitute(&mut visibility.max_objects, lookup)?; + substitute(&mut visibility.list_timeout_ms, lookup)?; } Ok(()) } @@ -810,6 +868,17 @@ where .collect()) } +/// [`scalar_vec`] for an optional list — absent and present differ (an +/// absent `content_columns` takes the default set; a present list replaces +/// it — and, validated at startup, may not be empty). +fn scalar_vec_opt<'de, D>(deserializer: D) -> Result>, D::Error> +where + D: serde::Deserializer<'de>, +{ + Ok(Option::>::deserialize(deserializer)? + .map(|scalars| scalars.into_iter().map(|s| s.0).collect())) +} + /// A YAML scalar captured as its string form (see [`scalar_opt`]). struct Scalar(String); @@ -1453,6 +1522,67 @@ auth: assert!(matches!(err, FileConfigError::Schema(_)), "got {err:?}"); } + /// RFC 0047 §3.4: the visibility section — `type` (a keyword, renamed + /// onto `object_type`), substituted leaves, an explicit + /// `content_columns` list distinct from an absent one, unknown keys + /// rejected. + #[test] + fn openfga_visibility_section_parses() { + let lookup = env(&[("CONV", "attr.gen_ai.conversation.id")]); + let cfg = parse( + "auth:\n tokens:\n - name: a\n token: ${env:CONV}\n tenants: [x]\n openfga:\n api_url: http://fga:8080\n store_id: s\n server_list_objects_deadline_ms: 3000\n visibility:\n objects:\n - type: conversation\n column: ${env:CONV}\n self_principal_column: attr.user.hash\n content_columns: [body, attr.prompt]\n max_objects: 100\n list_timeout_ms: 500\n", + &lookup, + ) + .expect("valid"); + let openfga = cfg.auth.expect("auth").openfga.expect("openfga"); + assert_eq!( + openfga.server_list_objects_deadline_ms.as_deref(), + Some("3000") + ); + let visibility = &openfga.visibility; + assert_eq!( + visibility.objects[0].object_type.as_deref(), + Some("conversation") + ); + assert_eq!( + visibility.objects[0].column.as_deref(), + Some("attr.gen_ai.conversation.id"), + "substituted" + ); + assert_eq!( + visibility.self_principal_column.as_deref(), + Some("attr.user.hash") + ); + assert_eq!( + visibility.content_columns.as_deref(), + Some(&["body".to_string(), "attr.prompt".to_string()][..]), + "an explicit list replaces the default set (non-empty by startup validation)" + ); + assert_eq!(visibility.max_objects.as_deref(), Some("100")); + assert_eq!(visibility.list_timeout_ms.as_deref(), Some("500")); + let cfg = parse( + "auth:\n tokens:\n - name: a\n token: ${env:CONV}\n tenants: [x]\n openfga:\n api_url: http://fga:8080\n store_id: s\n", + &lookup, + ) + .expect("valid"); + assert!( + cfg.auth + .expect("auth") + .openfga + .expect("openfga") + .visibility + .content_columns + .is_none(), + "absent = the default set" + ); + let err = parse( + "auth:\n openfga:\n visibility:\n objects:\n - kind: conversation\n", + &lookup, + ) + .expect_err("typo"); + assert!(matches!(err, FileConfigError::Schema(_)), "got {err:?}"); + } + /// An omitted section leaves its fields unset (`None`), matching an unset /// environment variable — the schema does not require every section. #[test] diff --git a/crates/ourios-server/src/lib.rs b/crates/ourios-server/src/lib.rs index 1b6c3ccd..d179c87a 100644 --- a/crates/ourios-server/src/lib.rs +++ b/crates/ourios-server/src/lib.rs @@ -15,3 +15,4 @@ pub mod auth; pub mod config; mod mcp; pub mod querier; +mod visibility; diff --git a/crates/ourios-server/src/mcp.rs b/crates/ourios-server/src/mcp.rs index 8490a1ca..f9c04f4c 100644 --- a/crates/ourios-server/src/mcp.rs +++ b/crates/ourios-server/src/mcp.rs @@ -256,9 +256,9 @@ impl OuriosMcp { &self, ctx: &rmcp::service::RequestContext, tenant: &str, - ) -> Result<(), ErrorData> { + ) -> Result, ErrorData> { if self.auth.is_open() { - return Ok(()); + return Ok(None); } // The transport layer (`require_bearer`) already authenticated and // cached the resolved binding on the request — read it from the @@ -270,7 +270,7 @@ impl OuriosMcp { .get::() .and_then(|parts| parts.extensions.get::()); match binding { - Some(binding) if binding.may_read(tenant) => Ok(()), + Some(binding) if binding.may_read(tenant) => Ok(Some(binding.clone())), Some(_) => Err(ErrorData::invalid_request( "the tenant is outside the authenticated token's allowed set", None, @@ -281,6 +281,29 @@ impl OuriosMcp { )), } } + + /// RFC 0047 §3.4 for a tool call: the two-step decision for the bound + /// principal in `tenant`, as an MCP error when refused. + async fn visibility( + &self, + binding: Option<&AuthBinding>, + tenant: &str, + ) -> Result, ErrorData> { + crate::visibility::resolve(&self.auth, binding, tenant, &self.metrics) + .await + .map_err(|rejection| visibility_error(&rejection)) + } +} + +/// Map a visibility refusal onto the MCP error vocabulary: `403`-class → +/// `invalid_request` (the caller may not), `503`-class → `internal_error` +/// (retry), `401`-class → `invalid_request` with the bearer message. +fn visibility_error(rejection: &crate::visibility::VisibilityRejection) -> ErrorData { + if rejection.status.is_server_error() { + ErrorData::internal_error(rejection.message.clone(), None) + } else { + ErrorData::invalid_request(rejection.message.clone(), None) + } } #[tool_router] @@ -335,6 +358,7 @@ impl OuriosMcp { gen_ai.tool.name = "query_logs", mcp.method.name = "tools/call", mcp.session.id = tracing::field::Empty, + ourios.query.visibility.branch = tracing::field::Empty, ) )] async fn query_logs_traced( @@ -346,7 +370,7 @@ impl OuriosMcp { tracing::Span::current().record("mcp.session.id", session); } let tenant_arg = normalize_tenant(&args.tenant)?; - self.check_tenant(&ctx, tenant_arg)?; + let binding = self.check_tenant(&ctx, tenant_arg)?; let statement = dsl::parse_statement(&args.query) .map_err(|e| ErrorData::invalid_params(format!("invalid query: {e}"), None))?; let Statement::Logs(mut query) = statement else { @@ -360,16 +384,21 @@ impl OuriosMcp { // documented "maximum rendered rows" contract holds. let cap = args.limit.unwrap_or(DEFAULT_LIMIT).min(MAX_LIMIT); cap_rows_unless_aggregation(&mut query.stages, cap); + let visibility = self.visibility(binding.as_ref(), tenant_arg).await?; + let options = visibility.map_or_else(ourios_querier::QueryOptions::default, |v| { + ourios_querier::QueryOptions::default().with_visibility(v) + }); let tenant = TenantId::new(tenant_arg); let started = std::time::Instant::now(); let result = self .querier - .run_query( + .run_query_with( &query, &tenant, now_unix_nano(), self.default_window_nanos, None, + options, ) .await; // The same instruments as the JSON API — an MCP query IS a query @@ -423,6 +452,7 @@ impl OuriosMcp { gen_ai.tool.name = "list_templates", mcp.method.name = "tools/call", mcp.session.id = tracing::field::Empty, + ourios.query.visibility.branch = tracing::field::Empty, ) )] async fn list_templates_traced( @@ -434,7 +464,11 @@ impl OuriosMcp { tracing::Span::current().record("mcp.session.id", session); } let tenant_arg = normalize_tenant(&args.tenant)?; - self.check_tenant(&ctx, tenant_arg)?; + let binding = self.check_tenant(&ctx, tenant_arg)?; + let visibility = self.visibility(binding.as_ref(), tenant_arg).await?; + if let Some(rejection) = crate::visibility::require_tenant_wide(visibility.as_ref()) { + return Err(visibility_error(&rejection)); + } let tenant = TenantId::new(tenant_arg); let started = std::time::Instant::now(); let registry = match self.querier.template_registry(&tenant).await { @@ -500,6 +534,7 @@ impl OuriosMcp { gen_ai.tool.name = "template_drift", mcp.method.name = "tools/call", mcp.session.id = tracing::field::Empty, + ourios.query.visibility.branch = tracing::field::Empty, ) )] async fn template_drift_traced( @@ -511,7 +546,11 @@ impl OuriosMcp { tracing::Span::current().record("mcp.session.id", session); } let tenant_arg = normalize_tenant(&args.tenant)?; - self.check_tenant(&ctx, tenant_arg)?; + let binding = self.check_tenant(&ctx, tenant_arg)?; + let visibility = self.visibility(binding.as_ref(), tenant_arg).await?; + if let Some(rejection) = crate::visibility::require_tenant_wide(visibility.as_ref()) { + return Err(visibility_error(&rejection)); + } // One grammar, one boundary rule: the drift window parses through // the DSL front-end exactly as the JSON API's statement does // (RFC0010.2's half-open rule inherited verbatim). @@ -636,7 +675,14 @@ fn now_unix_nano() -> u64 { /// H6-scrubbed surface the JSON API already exposes — no DataFusion/SQL /// leaks through either boundary. fn query_tool_error(e: &ourios_querier::QueryError) -> ErrorData { - ErrorData::internal_error(e.to_string(), None) + match e { + // RFC 0047 §3.4: a content column the principal may not read — the + // caller's request is wrong, not the server. + ourios_querier::QueryError::Forbidden { .. } => { + ErrorData::invalid_request(e.to_string(), None) + } + _ => ErrorData::internal_error(e.to_string(), None), + } } /// Serialize an RFC 0016 response shape as the tool's JSON content — the diff --git a/crates/ourios-server/src/querier.rs b/crates/ourios-server/src/querier.rs index 888cf040..e932a1ac 100644 --- a/crates/ourios-server/src/querier.rs +++ b/crates/ourios-server/src/querier.rs @@ -49,7 +49,7 @@ use ourios_parquet::{PromotedAttributes, StoreConfig}; use ourios_querier::dsl::ir::Stage; use ourios_querier::dsl::{self, Statement}; use ourios_querier::{ - AggregateGroup, DriftResult, LogBody, LogRow, Querier, QueryResult, QueryStats, + AggregateGroup, DriftResult, LogBody, LogRow, Querier, QueryOptions, QueryResult, QueryStats, }; use ourios_semconv as semconv; @@ -149,6 +149,8 @@ const ERROR_TYPE: &str = "error.type"; pub(crate) struct QuerierMetrics { duration: Histogram, row_groups: Counter, + /// `ourios.query.visibility` (RFC 0047 §3.4): queries by two-step branch. + visibility: Counter, } impl QuerierMetrics { @@ -168,12 +170,36 @@ impl QuerierMetrics { // ingester's single attribute-free `add(0, &[])`. row_groups.add(0, &Self::state_attrs(ROW_GROUP_SCANNED)); row_groups.add(0, &Self::state_attrs(ROW_GROUP_PRUNED)); + let visibility = meter + .u64_counter(semconv::OURIOS_QUERY_VISIBILITY) + .with_unit("{query}") + .build(); + for branch in [ + crate::visibility::BRANCH_TENANT_WIDE, + crate::visibility::BRANCH_METADATA_MASKED, + crate::visibility::BRANCH_SCOPED, + ] { + visibility.add(0, &Self::branch_attrs(branch)); + } Self { duration, row_groups, + visibility, } } + fn branch_attrs(branch: &'static str) -> [KeyValue; 1] { + [KeyValue::new( + semconv::OURIOS_QUERY_VISIBILITY_BRANCH, + branch, + )] + } + + /// Record which RFC 0047 §3.4 branch a query took. + pub(crate) fn record_visibility(&self, branch: &'static str) { + self.visibility.add(1, &Self::branch_attrs(branch)); + } + fn state_attrs(state: &'static str) -> [KeyValue; 1] { [KeyValue::new(semconv::OURIOS_QUERY_ROW_GROUP_STATE, state)] } @@ -436,6 +462,7 @@ async fn handle_query( http.route = "/v1/query", ourios.tenant = tracing::field::Empty, http.response.status_code = tracing::field::Empty, + ourios.query.visibility.branch = tracing::field::Empty, ) )] async fn handle_query_traced(state: QuerierState, headers: HeaderMap, body: Bytes) -> Response { @@ -497,44 +524,31 @@ async fn handle_query_inner(state: QuerierState, headers: HeaderMap, body: Bytes Err(message) => return error_response(StatusCode::BAD_REQUEST, "invalid_query", &message), }; + // RFC 0047 §3.4: the two-step decides which rows this principal may see + // inside the tenant — after the syntactic gates (a malformed query never + // costs an enumeration), before the engine. + let visibility = match crate::visibility::resolve( + &state.auth, + binding.as_ref(), + tenant.as_str(), + &state.metrics, + ) + .await + { + Ok(visibility) => visibility, + Err(rejection) => return reject_visibility(&state, gate_started, &rejection), + }; + let now = now_unix_nano(); let started = Instant::now(); match statement { - Statement::Logs(mut query) => { - // An aggregation — `count [by …]` or a scalar `sum`/`min`/`max`/ - // `avg` — and `limit` are mutually exclusive (`compile::validate` - // rejects the combination, RFC 0002 amendments 2026-07-15 and - // 2026-07-23): the query answers with its grouped map, not a - // capped row set, so the §7 default/cap limit is meaningless for - // it and must not be injected. - let is_aggregation = query - .stages - .iter() - .any(|s| matches!(s, Stage::Count { .. } | Stage::Agg { .. })); - if !is_aggregation { - apply_limit(&mut query.stages, DEFAULT_LIMIT, MAX_LIMIT); - } - let result = state - .querier - .run_query(&query, &tenant, now, state.default_window_nanos, None) - .await; - let elapsed = started.elapsed(); - match result { - Ok(result) => { - state - .metrics - .record_ok(QUERY_KIND_LOGS, elapsed, &result.stats); - json_ok(&LogQueryResponse::from(&result)) - } - Err(e) => { - state - .metrics - .record_err(QUERY_KIND_LOGS, elapsed, query_error_type(&e)); - query_error_response(&e) - } - } + Statement::Logs(query) => { + run_logs_query(&state, query, &tenant, now, visibility, started).await } Statement::Drift(query) => { + if let Some(rejection) = crate::visibility::require_tenant_wide(visibility.as_ref()) { + return reject_visibility(&state, gate_started, &rejection); + } let result = state.querier.run_drift(&query, &tenant, now).await; let elapsed = started.elapsed(); match result { @@ -563,6 +577,7 @@ pub(crate) fn query_error_type(error: &ourios_querier::QueryError) -> &'static s QueryError::TenantRequired => "tenant_required", QueryError::InvalidQuery { .. } => "invalid_query", QueryError::Storage { .. } => "storage", + QueryError::Forbidden { .. } => "permission_denied", // OpenTelemetry's fallback for an unclassified error class. _ => "_OTHER", } @@ -804,6 +819,9 @@ enum LogBodyDto { Structured { value: serde_json::Value, }, + /// The body exists but this principal may not read it (RFC 0047 §3.4 + /// masking) — distinct from a missing `body` key (no body on the wire). + Masked, } impl From<&LogBody> for LogBodyDto { @@ -825,6 +843,7 @@ impl From<&LogBody> for LogBodyDto { LogBody::Structured(value) => Self::Structured { value: any_value_json(value), }, + LogBody::Masked => Self::Masked, // `LogBody` is `#[non_exhaustive]`; a future body shape degrades to // an empty retained line rather than failing the whole response. // (`Absent` never reaches here — the row DTO maps it to a missing @@ -939,6 +958,74 @@ fn json_ok(value: &T) -> Response { /// A `status` JSON error body `{ "error": { "kind", "message" } }` (RFC 0016 /// §3.5). `message` is Ourios-owned text — never engine internals. +/// The logs arm of the query endpoint: cap the rows, run under the +/// RFC 0047 visibility decision, record, encode. +async fn run_logs_query( + state: &QuerierState, + mut query: ourios_querier::dsl::ir::Query, + tenant: &TenantId, + now: u64, + visibility: Option, + started: Instant, +) -> Response { + // An aggregation — `count [by …]` or a scalar `sum`/`min`/`max`/ + // `avg` — and `limit` are mutually exclusive (`compile::validate` + // rejects the combination, RFC 0002 amendments 2026-07-15 and + // 2026-07-23): the query answers with its grouped map, not a + // capped row set, so the §7 default/cap limit is meaningless for + // it and must not be injected. + let is_aggregation = query + .stages + .iter() + .any(|s| matches!(s, Stage::Count { .. } | Stage::Agg { .. })); + if !is_aggregation { + apply_limit(&mut query.stages, DEFAULT_LIMIT, MAX_LIMIT); + } + let options = visibility.map_or_else(QueryOptions::default, |v| { + QueryOptions::default().with_visibility(v) + }); + let result = state + .querier + .run_query_with( + &query, + tenant, + now, + state.default_window_nanos, + None, + options, + ) + .await; + let elapsed = started.elapsed(); + match result { + Ok(result) => { + state + .metrics + .record_ok(QUERY_KIND_LOGS, elapsed, &result.stats); + json_ok(&LogQueryResponse::from(&result)) + } + Err(e) => { + state + .metrics + .record_err(QUERY_KIND_LOGS, elapsed, query_error_type(&e)); + query_error_response(&e) + } + } +} + +/// An RFC 0047 §3.4 refusal, recorded like every pre-dispatch rejection. +fn reject_visibility( + state: &QuerierState, + gate_started: Instant, + rejection: &crate::visibility::VisibilityRejection, +) -> Response { + state.metrics.record_err( + QUERY_KIND_REJECTED, + gate_started.elapsed(), + rejection.error_type, + ); + error_response(rejection.status, rejection.kind, &rejection.message) +} + /// The query endpoint's bearer gate: `Ok(binding)` (`None` in open mode) /// or the finished rejection response, recorded on `ourios.query.duration` /// (kind `rejected`, RFC 0026 §3.4) with the failure class as `error.type`. @@ -1007,6 +1094,14 @@ fn query_error_response(error: &ourios_querier::QueryError) -> Response { "missing_tenant", "the X-Ourios-Tenant header is required and must be non-empty", ), + // RFC 0047 §3.4: a metadata-only reader filtered or aggregated on a + // content column — refused, naming the column (configuration, not + // data). + QueryError::Forbidden { .. } => error_response( + StatusCode::FORBIDDEN, + "column_forbidden", + &error.to_string(), + ), // `Storage` and any future `#[non_exhaustive]` variant → a scrubbed // 500. `QueryError::Display` is the H6-safe surface (it withholds the // engine detail), so this never leaks DataFusion/SQL text. diff --git a/crates/ourios-server/src/visibility.rs b/crates/ourios-server/src/visibility.rs new file mode 100644 index 00000000..bc90dee8 --- /dev/null +++ b/crates/ourios-server/src/visibility.rs @@ -0,0 +1,267 @@ +//! RFC 0047 §3.4 — the two-step, at the query surfaces. Runs once per +//! request after the tenant gate: asks the graph resolver which branch the +//! principal takes inside the tenant and hands the engine the matching +//! [`Visibility`], recording the branch on `ourios.query.visibility` and +//! the request span. Every failure is fail-closed and named — never a +//! partial predicate, never an open door. + +use axum::http::StatusCode; +use ourios_core::auth::openfga::{ + CONVERSATION_TYPE, OpenFgaError, PrincipalKind, Visibility as GraphVisibility, +}; +use ourios_ingester::receiver::{AuthBinding, AuthResolver}; +use ourios_querier::{ScopedIds, SelfMatch, Visibility}; + +use crate::querier::QuerierMetrics; + +/// The `ourios.query.visibility.branch` values. +pub(crate) const BRANCH_TENANT_WIDE: &str = "tenant_wide"; +pub(crate) const BRANCH_METADATA_MASKED: &str = "metadata_masked"; +pub(crate) const BRANCH_SCOPED: &str = "scoped"; + +/// Why a query was refused before the engine ran: the HTTP status, a stable +/// `kind`, the message, and the `error.type` for the duration histogram. +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct VisibilityRejection { + pub(crate) status: StatusCode, + pub(crate) kind: &'static str, + pub(crate) message: String, + pub(crate) error_type: &'static str, +} + +/// The layer-2 decision for `binding` in `tenant`: `Ok(None)` when no graph +/// resolver bound this session (open mode, static/OIDC-only deployments — +/// today's plan), else the branch the engine applies. +/// +/// # Errors +/// +/// [`VisibilityRejection`]: the bound (`403`, "ask for tenant-wide read"), +/// an incomplete enumeration or unreachable `OpenFGA` (`503`), a +/// credential defect (`401`). +pub(crate) async fn resolve( + auth: &AuthResolver, + binding: Option<&AuthBinding>, + tenant: &str, + metrics: &QuerierMetrics, +) -> Result, VisibilityRejection> { + let (Some(graph), Some(resolver)) = (binding.and_then(AuthBinding::graph), auth.openfga()) + else { + return Ok(None); + }; + let decision = match resolver + .visibility(graph.principal(), graph.groups(), tenant) + .await + { + Ok(decision) => decision, + Err(e) => { + // A scoped enumeration that failed closed still happened — + // count it, so `scoped` on `ourios.query.visibility` is exactly + // "a stream was issued". + if matches!( + e, + OpenFgaError::BoundExceeded { .. } | OpenFgaError::Incomplete + ) { + metrics.record_visibility(BRANCH_SCOPED); + tracing::Span::current().record("ourios.query.visibility.branch", BRANCH_SCOPED); + } + return Err(reject( + &e, + tenant, + resolver.visibility_config().max_objects(), + )); + } + }; + let config = resolver.visibility_config(); + let (branch, visibility) = match decision { + GraphVisibility::TenantWide => (BRANCH_TENANT_WIDE, Visibility::TenantWide), + GraphVisibility::MetadataOnly => ( + BRANCH_METADATA_MASKED, + Visibility::Masked { + content_columns: config.content_columns().to_vec(), + }, + ), + GraphVisibility::Scoped { conversations } => { + // No bound conversation object ⇒ nothing to enumerate (the + // resolver returned no ids) — represented as such, never as an + // empty column name. + let conversations = config + .objects() + .iter() + .find(|object| object.object_type() == CONVERSATION_TYPE) + .map(|object| ScopedIds { + column: object.column().to_string(), + ids: conversations.into_iter().collect(), + }); + // The self fast path is for `user:` principals only (§3.3): + // agents and service accounts never get it. + let self_match = match (graph.principal().kind(), config.self_principal_column()) { + (PrincipalKind::User, Some(self_column)) => Some(SelfMatch { + column: self_column.to_string(), + value: graph.principal().id().to_string(), + }), + _ => None, + }; + ( + BRANCH_SCOPED, + Visibility::Scoped { + conversations, + self_match, + }, + ) + } + }; + metrics.record_visibility(branch); + tracing::Span::current().record("ourios.query.visibility.branch", branch); + Ok(Some(visibility)) +} + +fn reject(error: &OpenFgaError, tenant: &str, bound: usize) -> VisibilityRejection { + match error { + OpenFgaError::BoundExceeded { .. } => VisibilityRejection { + status: StatusCode::FORBIDDEN, + kind: "visibility_bound", + message: format!( + "visibility set exceeds {bound} objects in tenant {tenant}; ask for tenant-wide read" + ), + error_type: "visibility_bound", + }, + OpenFgaError::Incomplete => VisibilityRejection { + status: StatusCode::SERVICE_UNAVAILABLE, + kind: "visibility_incomplete", + message: "visibility enumeration incomplete; retry later".to_string(), + error_type: "visibility_incomplete", + }, + OpenFgaError::Unavailable(_) => VisibilityRejection { + status: StatusCode::SERVICE_UNAVAILABLE, + kind: "auth_unavailable", + message: "the authorization resolver is unavailable; retry later".to_string(), + error_type: "upstream_unavailable", + }, + // A tenant no graph object can name: nothing can have been granted + // on it, so nothing in it is readable — and the operator should hear + // why. + OpenFgaError::InvalidTenant => VisibilityRejection { + status: StatusCode::FORBIDDEN, + kind: "tenant_unaddressable", + message: format!( + "tenant `{tenant}` cannot be named in the authorization graph (an object id \ + may not be empty, exceed 256 bytes, or contain ':', '#' or whitespace)" + ), + error_type: "permission_denied", + }, + OpenFgaError::TooManyContextualTuples { .. } + | OpenFgaError::InvalidGroup { .. } + | OpenFgaError::InvalidPrincipal => VisibilityRejection { + status: StatusCode::UNAUTHORIZED, + kind: "unauthenticated", + message: "a valid bearer token is required".to_string(), + error_type: "unauthenticated", + }, + } +} + +/// A principal that is not a tenant-wide content reader may not run +/// template-level queries (`drift`, the registry) — templates are mined +/// from bodies, so listing them would leak content past the row-level +/// enforcement (RFC 0047 §3.4). `None` = allowed. +pub(crate) fn require_tenant_wide(visibility: Option<&Visibility>) -> Option { + match visibility { + None | Some(Visibility::TenantWide) => None, + Some(Visibility::Masked { .. } | Visibility::Scoped { .. }) => Some(VisibilityRejection { + status: StatusCode::FORBIDDEN, + kind: "visibility_scoped", + message: "template-level queries require tenant-wide content read".to_string(), + error_type: "permission_denied", + }), + } +} + +#[cfg(test)] +mod tests { + use axum::http::StatusCode; + use ourios_core::auth::openfga::OpenFgaError; + use ourios_querier::Visibility; + + use super::{reject, require_tenant_wide}; + + /// The refusal contract per resolver error: status, stable kind, + /// `error.type` — pinned so a reword cannot change a class unnoticed. + #[test] + fn rejections_map_each_error_class() { + let cases = [ + ( + OpenFgaError::BoundExceeded { bound: 3 }, + StatusCode::FORBIDDEN, + "visibility_bound", + "visibility_bound", + ), + ( + OpenFgaError::Incomplete, + StatusCode::SERVICE_UNAVAILABLE, + "visibility_incomplete", + "visibility_incomplete", + ), + ( + OpenFgaError::Unavailable("down".to_string()), + StatusCode::SERVICE_UNAVAILABLE, + "auth_unavailable", + "upstream_unavailable", + ), + ( + OpenFgaError::InvalidTenant, + StatusCode::FORBIDDEN, + "tenant_unaddressable", + "permission_denied", + ), + ( + OpenFgaError::TooManyContextualTuples { count: 101 }, + StatusCode::UNAUTHORIZED, + "unauthenticated", + "unauthenticated", + ), + ( + OpenFgaError::InvalidGroup { index: 0 }, + StatusCode::UNAUTHORIZED, + "unauthenticated", + "unauthenticated", + ), + ( + OpenFgaError::InvalidPrincipal, + StatusCode::UNAUTHORIZED, + "unauthenticated", + "unauthenticated", + ), + ]; + for (error, status, kind, error_type) in cases { + let rejection = reject(&error, "acme", 100); + assert_eq!(rejection.status, status, "{error:?}"); + assert_eq!(rejection.kind, kind, "{error:?}"); + assert_eq!(rejection.error_type, error_type, "{error:?}"); + } + let bound = reject(&OpenFgaError::BoundExceeded { bound: 3 }, "acme", 100); + assert!(bound.message.contains("exceeds 100 objects in tenant acme")); + assert!(bound.message.contains("ask for tenant-wide read")); + } + + /// Template-level surfaces: open mode and tenant-wide pass; masked and + /// scoped principals are refused with the stable kind. + #[test] + fn template_level_queries_need_tenant_wide_read() { + assert!(require_tenant_wide(None).is_none()); + assert!(require_tenant_wide(Some(&Visibility::TenantWide)).is_none()); + for visibility in [ + Visibility::Masked { + content_columns: vec!["body".to_string()], + }, + Visibility::Scoped { + conversations: None, + self_match: None, + }, + ] { + let rejection = require_tenant_wide(Some(&visibility)).expect("refused"); + assert_eq!(rejection.status, StatusCode::FORBIDDEN); + assert_eq!(rejection.kind, "visibility_scoped"); + assert_eq!(rejection.error_type, "permission_denied"); + } + } +} diff --git a/crates/ourios-server/tests/it/main.rs b/crates/ourios-server/tests/it/main.rs index 293b087f..9607f933 100644 --- a/crates/ourios-server/tests/it/main.rs +++ b/crates/ourios-server/tests/it/main.rs @@ -34,3 +34,4 @@ mod rfc0039_4_sampling; mod rfc0043_5_event_name_query; mod rfc0046_out_of_band_tenancy; mod rfc0047_openfga; +mod rfc0047_visibility; diff --git a/crates/ourios-server/tests/it/rfc0029_oidc.rs b/crates/ourios-server/tests/it/rfc0029_oidc.rs index 78321f60..290a9031 100644 --- a/crates/ourios-server/tests/it/rfc0029_oidc.rs +++ b/crates/ourios-server/tests/it/rfc0029_oidc.rs @@ -511,6 +511,17 @@ pub(crate) mod claim_binding { tmp: &tempfile::TempDir, auth_yaml: &str, envs: &[(&str, &str)], + ) -> (tokio::process::Child, String, String, std::net::SocketAddr) { + spawn_with_auth_and_storage(tmp, "", auth_yaml, envs).await + } + + /// [`spawn_with_auth`] with extra `storage:` keys (e.g. a + /// `promoted_attributes` block, indented two spaces) after `local`. + pub(crate) async fn spawn_with_auth_and_storage( + tmp: &tempfile::TempDir, + storage_yaml: &str, + auth_yaml: &str, + envs: &[(&str, &str)], ) -> (tokio::process::Child, String, String, std::net::SocketAddr) { let wal = tmp.path().join("wal"); std::fs::create_dir_all(&wal).expect("wal dir"); @@ -518,7 +529,7 @@ pub(crate) mod claim_binding { let mut file = std::fs::File::create(&config_path).expect("create config"); write!( file, - "storage:\n local:\n bucket_root: {}\n\ + "storage:\n local:\n bucket_root: {}\n{storage_yaml}\ receiver:\n enabled: true\n grpc_addr: 127.0.0.1:0\n http_addr: 127.0.0.1:0\n wal_root: {}\n\ querier:\n enabled: true\n http_addr: 127.0.0.1:0\n\ {auth_yaml}", diff --git a/crates/ourios-server/tests/it/rfc0047_openfga.rs b/crates/ourios-server/tests/it/rfc0047_openfga.rs index 8dafe9f6..f02d44e6 100644 --- a/crates/ourios-server/tests/it/rfc0047_openfga.rs +++ b/crates/ourios-server/tests/it/rfc0047_openfga.rs @@ -20,16 +20,22 @@ use tokio::time::timeout; use crate::rfc0029_oidc::claim_binding::{query_status, spawn_with_auth}; use crate::rfc0029_oidc::ingest_binding::{make_key, serve_issuer, tenant_request}; -const OPENFGA_IMAGE: &str = "openfga/openfga"; +pub(crate) const OPENFGA_IMAGE: &str = "openfga/openfga"; /// Pinned by digest for reproducibility (v1.11.1). -const OPENFGA_TAG: &str = +pub(crate) const OPENFGA_TAG: &str = "v1.11.1@sha256:1f9187961aded3ce60e3c4b7ccc39074ce88291aa51d9e3be09db4ff51e7b692"; -const MODEL_JSON: &str = include_str!("../../../../deploy/openfga/model.json"); +pub(crate) const MODEL_JSON: &str = include_str!("../../../../deploy/openfga/model.json"); const COLLECTOR_TOKEN: &str = "tok-collector-cluster1"; /// A JWT for `sub` from the fixture issuer — no tenant claim (the graph -/// binds), optional groups. -fn mint(encoding: &jsonwebtoken::EncodingKey, issuer: &str, sub: &str, groups: &[&str]) -> String { +/// binds), optional groups, optionally carrying the agent claim. +pub(crate) fn mint( + encoding: &jsonwebtoken::EncodingKey, + issuer: &str, + sub: &str, + groups: &[&str], + agent: bool, +) -> String { let now = i64::try_from( std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) @@ -43,13 +49,16 @@ fn mint(encoding: &jsonwebtoken::EncodingKey, issuer: &str, sub: &str, groups: & if !groups.is_empty() { claims["groups"] = serde_json::json!(groups); } + if agent { + claims["ourios_principal_type"] = serde_json::json!("agent"); + } let mut header = jsonwebtoken::Header::new(jsonwebtoken::Algorithm::ES256); header.kid = Some("key-1".to_string()); jsonwebtoken::encode(&header, &claims, encoding).expect("mint") } /// Create a store and load the in-tree model; returns `(store_id, model_id)`. -async fn provision(api_url: &str) -> (String, String) { +pub(crate) async fn provision(api_url: &str) -> (String, String) { let http = reqwest::Client::new(); let store: serde_json::Value = http .post(format!("{api_url}/stores")) @@ -82,7 +91,7 @@ async fn provision(api_url: &str) -> (String, String) { (store_id, model_id) } -fn tuple(user: &str, relation: &str, object: &str) -> TupleKey { +pub(crate) fn tuple(user: &str, relation: &str, object: &str) -> TupleKey { TupleKey::new(user, relation, object) } @@ -199,7 +208,7 @@ async fn rfc0047_1_to_3_resolver_end_to_end() { }; // --- RFC0047.1: resolver binding --------------------------------------- - let alice = mint(&encoding, &issuer, "alice", &[]); + let alice = mint(&encoding, &issuer, "alice", &[], false); assert!( query_status(querier, Some(&alice), Some("acme")) .await @@ -225,8 +234,8 @@ async fn rfc0047_1_to_3_resolver_end_to_end() { ); // bob (participant + binding tuple) and fin (metadata only) reach the // planner — bound to acme — while holding no tenant-wide content read. - let bob = mint(&encoding, &issuer, "bob", &[]); - let fin = mint(&encoding, &issuer, "fin", &[]); + let bob = mint(&encoding, &issuer, "bob", &[], false); + let fin = mint(&encoding, &issuer, "fin", &[], false); for (who, token) in [("bob", &bob), ("fin", &fin)] { assert!( query_status(querier, Some(token), Some("acme")) @@ -251,7 +260,7 @@ async fn rfc0047_1_to_3_resolver_end_to_end() { "alice does" ); // A principal with no tuples is unbound — 401, never empty-but-open. - let nobody = mint(&encoding, &issuer, "nobody", &[]); + let nobody = mint(&encoding, &issuer, "nobody", &[], false); assert!( query_status(querier, Some(&nobody), Some("acme")) .await @@ -270,7 +279,7 @@ async fn rfc0047_1_to_3_resolver_end_to_end() { ) .await .expect("team edge"); - let carol = mint(&encoding, &issuer, "carol", &["platform"]); + let carol = mint(&encoding, &issuer, "carol", &["platform"], false); assert!( query_status(querier, Some(&carol), Some("globex")) .await diff --git a/crates/ourios-server/tests/it/rfc0047_visibility.rs b/crates/ourios-server/tests/it/rfc0047_visibility.rs new file mode 100644 index 00000000..a994a67b --- /dev/null +++ b/crates/ourios-server/tests/it/rfc0047_visibility.rs @@ -0,0 +1,453 @@ +//! RFC 0047 §3.4 — layer-2 visibility on the served binary against a +//! **real `OpenFGA` container** (testcontainers; CI-gated like the layer-1 +//! test — `#[ignore]`d in the default run), over Parquet pre-written with +//! the promoted `gen_ai.conversation.id` / `user.hash` / `cost_usd` +//! columns. +//! +//! Scenarios RFC0047.4 (tenant-wide reader), .5 (participant + self fast +//! path), .6 (agent principal + revocable delegation), .7 (bounded +//! enumeration, per tenant), .8 (metadata without content). +//! See `docs/rfcs/0047-rebac-resolver-and-graph-visibility.md` §5. + +use std::collections::HashMap; +use std::path::Path; +use std::time::Duration; + +use ourios_core::auth::openfga::{OpenFgaClient, OpenFgaSpec, build_openfga_config}; +use ourios_core::otlp::any_value::Value as AvValue; +use ourios_core::otlp::{AnyValue, KeyValue}; +use ourios_core::record::{BodyKind, MinedRecord, Param}; +use ourios_core::tenant::TenantId; +use ourios_parquet::{ + DEFAULT_ZSTD_LEVEL, PartitionKey, PromotedAttributes, PromotedClass, PromotedKey, Store, Writer, +}; +use testcontainers_modules::testcontainers::core::ContainerPort; +use testcontainers_modules::testcontainers::runners::AsyncRunner; +use testcontainers_modules::testcontainers::{GenericImage, ImageExt}; +use tokio::time::timeout; + +use crate::rfc0029_oidc::claim_binding::spawn_with_auth_and_storage; +use crate::rfc0029_oidc::ingest_binding::{make_key, serve_issuer}; +use crate::rfc0047_openfga::{OPENFGA_IMAGE, OPENFGA_TAG, mint, provision, tuple}; + +fn kv(key: &str, value: &str) -> KeyValue { + KeyValue { + key: key.to_string(), + value: Some(AnyValue { + value: Some(AvValue::StringValue(value.to_string())), + }), + ..Default::default() + } +} + +fn kv_double(key: &str, value: f64) -> KeyValue { + KeyValue { + key: key.to_string(), + value: Some(AnyValue { + value: Some(AvValue::DoubleValue(value)), + }), + ..Default::default() + } +} + +fn recent_ns(offset: u64) -> u64 { + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .expect("epoch") + .as_nanos(); + u64::try_from(now).expect("fits") - 60_000_000_000 + offset * 1_000_000 +} + +/// One `acme` row in `conversation` with the given `user.hash`, a content +/// attribute, a model and a $1.50 cost. +fn row(i: u64, conversation: &str, user: &str) -> MinedRecord { + MinedRecord { + tenant_id: TenantId::new("acme"), + template_id: 1, + template_version: 1, + severity_number: 9, + severity_text: Some("INFO".to_string()), + scope_name: Some("agent".to_string()), + scope_version: None, + scope_attributes: Vec::new(), + resource_schema_url: None, + scope_schema_url: None, + time_unix_nano: recent_ns(i), + observed_time_unix_nano: None, + attributes: vec![ + kv("gen_ai.conversation.id", conversation), + kv("user.hash", user), + kv("gen_ai.input.messages", "the secret prompt"), + kv("model", "gpt"), + kv_double("cost_usd", 1.5), + ], + dropped_attributes_count: 0, + resource_attributes: vec![kv("service.name", "agent")], + trace_id: None, + span_id: None, + flags: 0, + event_name: None, + body_kind: BodyKind::String, + params: vec![Param { + type_tag: ourios_core::audit::ParamType::Num, + value: "42".to_string(), + }], + separators: vec![String::new(), " ".to_string()], + body: None, + confidence: 1.0, + lossy_flag: false, + } +} + +fn promoted() -> PromotedAttributes { + PromotedAttributes::new_typed( + [], + [ + PromotedKey::string("gen_ai.conversation.id".to_string()), + PromotedKey::string("user.hash".to_string()), + PromotedKey::string("model".to_string()), + PromotedKey { + key: "cost_usd".into(), + class: PromotedClass::F64, + }, + ], + ) +} + +fn write_records(bucket: &Path, recs: &[MinedRecord]) { + let store = Store::local(bucket).expect("local store"); + let mut by_part: HashMap> = HashMap::new(); + for r in recs { + by_part + .entry(PartitionKey::derive(r).expect("derive partition")) + .or_default() + .push(r.clone()); + } + for (part, rs) in by_part { + let mut w = Writer::open_in_with_promoted(&store, part, DEFAULT_ZSTD_LEVEL, promoted()) + .expect("open writer"); + w.append_records(&rs).expect("append"); + w.close().expect("close"); + } +} + +/// `POST /v1/query` with a bearer and tenant; returns (status, JSON body). +async fn query( + http: &reqwest::Client, + addr: std::net::SocketAddr, + bearer: &str, + tenant: &str, + dsl: &str, +) -> (u16, serde_json::Value) { + let response = http + .post(format!("http://{addr}/v1/query")) + .bearer_auth(bearer) + .header("x-ourios-tenant", tenant) + .header("content-type", "text/plain") + .body(dsl.to_string()) + .send() + .await + .expect("query"); + let status = response.status().as_u16(); + let body = response.text().await.expect("body"); + ( + status, + serde_json::from_str(&body).unwrap_or(serde_json::Value::String(body)), + ) +} + +/// The sorted conversation ids of a row response. +fn conversations(body: &serde_json::Value) -> Vec { + let mut ids: Vec = body["records"] + .as_array() + .expect("records") + .iter() + .map(|record| { + record["attributes"] + .as_array() + .expect("attributes") + .iter() + .find(|kv| kv["key"] == "gen_ai.conversation.id") + .and_then(|kv| kv["value"]["stringValue"].as_str()) + .expect("conversation id") + .to_string() + }) + .collect(); + ids.sort(); + ids +} + +/// Scenarios RFC0047.4–.8 on the served binary (one container, one +/// server, one seeded store). +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +#[allow(clippy::too_many_lines)] // one container + one server, every arm in sequence +#[ignore = "RFC0047.4–.8 — needs Docker (real OpenFGA container); run by the openfga-resolver CI job via --ignored"] +async fn rfc0047_4_to_8_visibility_end_to_end() { + // --- OpenFGA ----------------------------------------------------------- + let container = GenericImage::new(OPENFGA_IMAGE, OPENFGA_TAG) + .with_exposed_port(ContainerPort::Tcp(8080)) + .with_cmd(["run"]) + .start() + .await + .expect("openfga started"); + let port = container + .get_host_port_ipv4(8080) + .await + .expect("mapped port"); + let api_url = format!("http://127.0.0.1:{port}"); + let http = reqwest::Client::new(); + timeout(Duration::from_secs(60), async { + loop { + if let Ok(response) = http.get(format!("{api_url}/healthz")).send().await + && response.status().is_success() + { + return; + } + tokio::time::sleep(Duration::from_millis(250)).await; + } + }) + .await + .expect("openfga healthy before timeout"); + let (store_id, model_id) = provision(&api_url).await; + let fga = OpenFgaClient::new( + &build_openfga_config(&OpenFgaSpec { + api_url: Some(api_url.clone()), + store_id: Some(store_id.clone()), + authorization_model_id: Some(model_id.clone()), + ..OpenFgaSpec::default() + }) + .expect("config"), + ) + .expect("client"); + let conv = |id: &str| format!("conversation:acme/{id}"); + let mut tuples = vec![ + tuple("user:alice", "reader", "tenant:acme"), + tuple("user:fin", "metadata_reader", "tenant:acme"), + tuple("user:mallory", "reader", "tenant:globex"), + // bob: participant of c-1? No — bob is participant of c-2 in acme + // and of globex/c-1 (another tenant's id that collides with acme's). + tuple("user:bob", "participant", &conv("c-2")), + tuple("user:bob", "scoped_reader", "tenant:acme"), + tuple("user:bob", "participant", "conversation:globex/c-1"), + tuple("user:bob", "scoped_reader", "tenant:globex"), + // bot: actor on c-3, c-4; delegate on alice's c-7. + tuple("agent:bot", "actor", &conv("c-3")), + tuple("agent:bot", "actor", &conv("c-4")), + tuple("agent:bot", "delegate", &conv("c-7")), + tuple("agent:bot", "scoped_reader", "tenant:acme"), + // other: actor on c-5. + tuple("agent:other", "actor", &conv("c-5")), + tuple("agent:other", "scoped_reader", "tenant:acme"), + // big: 4 acme conversations under a bound of 3 → refused. + tuple("agent:big", "scoped_reader", "tenant:acme"), + // mixed: 2 acme + 4 globex conversations → acme succeeds with 2. + tuple("agent:mixed", "scoped_reader", "tenant:acme"), + ]; + for id in ["c-11", "c-12", "c-13", "c-14"] { + tuples.push(tuple("agent:big", "actor", &conv(id))); + } + for id in ["c-15", "c-16"] { + tuples.push(tuple("agent:mixed", "actor", &conv(id))); + } + for id in ["g-1", "g-2", "g-3", "g-4"] { + tuples.push(tuple( + "agent:mixed", + "actor", + &format!("conversation:globex/{id}"), + )); + } + for id in [ + "c-1", "c-2", "c-3", "c-4", "c-5", "c-7", "c-9", "c-11", "c-12", "c-13", "c-14", "c-15", + "c-16", + ] { + tuples.push(tuple("tenant:acme", "parent", &conv(id))); + } + for chunk in tuples.chunks(100) { + fga.write(chunk, &[]).await.expect("seed tuples"); + } + + // --- Parquet + server -------------------------------------------------- + let tmp = tempfile::TempDir::new().expect("temp"); + let recs = vec![ + row(1, "c-1", "alice"), + row(2, "c-1", "alice"), + row(3, "c-2", "bob"), + row(4, "c-3", "bot"), + row(5, "c-4", "bot"), + row(6, "c-5", "other"), + row(7, "c-7", "alice"), + // c-9: rows carry user.hash = bob but no tuple yet (self fast path). + row(8, "c-9", "bob"), + row(9, "c-11", "big"), + row(10, "c-12", "big"), + row(11, "c-13", "big"), + row(12, "c-14", "big"), + row(13, "c-15", "mixed"), + row(14, "c-16", "mixed"), + ]; + write_records(tmp.path(), &recs); + let total = recs.len(); + + let (encoding, jwk) = make_key("key-1"); + let issuer = serve_issuer(jwk).await; + let storage_yaml = " promoted_attributes:\n log: [gen_ai.conversation.id, user.hash, model, {key: cost_usd, type: f64}]\n"; + let auth_yaml = format!( + "auth:\n\ + \x20\x20oidc:\n\ + \x20\x20\x20\x20issuer: {issuer}\n\ + \x20\x20\x20\x20audience: ourios\n\ + \x20\x20\x20\x20agent_claim: ourios_principal_type=agent\n\ + \x20\x20openfga:\n\ + \x20\x20\x20\x20api_url: {api_url}\n\ + \x20\x20\x20\x20store_id: {store_id}\n\ + \x20\x20\x20\x20authorization_model_id: {model_id}\n\ + \x20\x20\x20\x20session_ttl_secs: 1\n\ + \x20\x20\x20\x20request_timeout_secs: 2\n\ + \x20\x20\x20\x20visibility:\n\ + \x20\x20\x20\x20\x20\x20objects:\n\ + \x20\x20\x20\x20\x20\x20\x20\x20- type: conversation\n\ + \x20\x20\x20\x20\x20\x20\x20\x20\x20\x20column: attr.gen_ai.conversation.id\n\ + \x20\x20\x20\x20\x20\x20self_principal_column: attr.user.hash\n\ + \x20\x20\x20\x20\x20\x20max_objects: 3\n\ + \x20\x20\x20\x20\x20\x20list_timeout_ms: 1500\n" + ); + let (mut child, _grpc, _http, querier) = + spawn_with_auth_and_storage(&tmp, storage_yaml, &auth_yaml, &[]).await; + let all = "true | limit 100"; + + // --- RFC0047.4: tenant-wide reader ------------------------------------ + let alice = mint(&encoding, &issuer, "alice", &[], false); + let (status, body) = query(&http, querier, &alice, "acme", all).await; + assert_eq!(status, 200, "{body}"); + assert_eq!(body["rows"], total, "every row, tenant predicate only"); + // No enumeration is issued for a tenant-wide reader — pinned by the + // resolver unit test (`two_step_visibility`); here: the plan is complete. + + // --- RFC0047.5: participant + self fast path --------------------------- + let bob = mint(&encoding, &issuer, "bob", &[], false); + let (status, body) = query(&http, querier, &bob, "acme", all).await; + assert_eq!(status, 200, "{body}"); + assert_eq!( + conversations(&body), + ["c-2", "c-9"], + "bob: his participant conversation + rows carrying his subject; \ + never acme/c-1 although he is participant of globex/c-1" + ); + let (status, body) = query( + &http, + querier, + &bob, + "acme", + "attr.user.hash == \"alice\" | limit 100", + ) + .await; + assert_eq!(status, 200, "{body}"); + assert_eq!( + body["rows"], 0, + "the user's own predicate ANDs with visibility" + ); + + // --- RFC0047.6: agent as principal + revocable delegation -------------- + let bot = mint(&encoding, &issuer, "bot", &[], true); + let (status, body) = query(&http, querier, &bot, "acme", all).await; + assert_eq!(status, 200, "{body}"); + assert_eq!( + conversations(&body), + ["c-3", "c-4", "c-7"], + "bot: its actor conversations + the delegated c-7, none of other's" + ); + let other = mint(&encoding, &issuer, "other", &[], true); + let (status, body) = query(&http, querier, &other, "acme", all).await; + assert_eq!(status, 200, "{body}"); + assert_eq!(conversations(&body), ["c-5"]); + fga.write(&[], &[tuple("agent:bot", "delegate", &conv("c-7"))]) + .await + .expect("revoke delegation"); + tokio::time::sleep(Duration::from_millis(1200)).await; + let (status, body) = query(&http, querier, &bot, "acme", all).await; + assert_eq!(status, 200, "{body}"); + assert_eq!( + conversations(&body), + ["c-3", "c-4"], + "delegation revoked past the TTL" + ); + + // --- RFC0047.7: bounded enumeration, per tenant ------------------------ + let big = mint(&encoding, &issuer, "big", &[], true); + let (status, body) = query(&http, querier, &big, "acme", all).await; + assert_eq!(status, 403, "{body}"); + assert_eq!(body["error"]["kind"], "visibility_bound", "{body}"); + assert!( + body["error"]["message"] + .as_str() + .unwrap_or_default() + .contains("exceeds 3 objects in tenant acme"), + "{body}" + ); + let mixed = mint(&encoding, &issuer, "mixed", &[], true); + let (status, body) = query(&http, querier, &mixed, "acme", all).await; + assert_eq!(status, 200, "{body}"); + assert_eq!( + conversations(&body), + ["c-15", "c-16"], + "only tenant-acme ids count toward the bound (4 globex ids ignored)" + ); + + // --- RFC0047.8: metadata without content ------------------------------- + let fin = mint(&encoding, &issuer, "fin", &[], false); + let (status, body) = query( + &http, + querier, + &fin, + "acme", + "true | sum(attr.cost_usd) by attr.model", + ) + .await; + assert_eq!(status, 200, "{body}"); + let groups = body["aggregate"].as_array().expect("aggregate"); + assert_eq!(groups.len(), 1, "{body}"); + let sum = groups[0]["value"].as_f64().expect("sum"); + assert!( + (sum - 1.5 * f64::from(u32::try_from(total).expect("small"))).abs() < 1e-6, + "every row of the tenant: {body}" + ); + let (status, body) = query(&http, querier, &fin, "acme", all).await; + assert_eq!(status, 200, "{body}"); + assert_eq!(body["rows"], total); + for record in body["records"].as_array().expect("records") { + assert_eq!(record["body"]["kind"], "masked", "{record}"); + let content = record["attributes"] + .as_array() + .expect("attributes") + .iter() + .find(|kv| kv["key"] == "gen_ai.input.messages") + .expect("key kept"); + assert!(content["value"].is_null(), "value is null: {content}"); + } + let (status, body) = query( + &http, + querier, + &fin, + "acme", + "contains(body, \"secret\") | limit 10", + ) + .await; + assert_eq!(status, 403, "{body}"); + assert_eq!(body["error"]["kind"], "column_forbidden", "{body}"); + assert!( + body["error"]["message"] + .as_str() + .unwrap_or_default() + .contains("`body`"), + "{body}" + ); + // Template-level queries need tenant-wide content read. + let (status, body) = query(&http, querier, &fin, "acme", "drift from -1h to now").await; + assert_eq!(status, 403, "{body}"); + assert_eq!(body["error"]["kind"], "visibility_scoped", "{body}"); + let (status, _) = query(&http, querier, &alice, "acme", "drift from -1h to now").await; + assert_eq!(status, 200, "tenant-wide readers may"); + + child.kill().await.expect("kill the server"); + drop(container); +} diff --git a/docs/guides/authentication.md b/docs/guides/authentication.md index e9a6909e..57e23a95 100644 --- a/docs/guides/authentication.md +++ b/docs/guides/authentication.md @@ -134,6 +134,57 @@ pipeline. A credential's own tenant list — a static token's `tenants`, an OIDC `tenant_claim` — can only narrow what the graph grants, never widen it; a principal the graph grants nothing is unbound (401). +### Visibility inside a tenant (layer 2) + +With the graph configured, every query also runs the RFC 0047 §3.4 +**two-step** for the principal in the tenant it queries — query rewrite +at plan time, never per-record checks: + +1. `Check(principal, can_read_content, tenant)` allowed → the tenant + predicate only (today's plan). +2. else `Check(can_read_metadata, tenant)` allowed → every row, with the + configured `content_columns` returned as null (`body` as + `{"kind":"masked"}`); a query that filters or aggregates on one of them + is `403 column_forbidden`, naming the column. +3. else the principal is **scoped** (bound through `scoped_reader`): its + readable conversations are enumerated through the *streamed* + `ListObjects` — filtered to this tenant, at most `max_objects` tenant + ids, within `list_timeout_ms` — and become + `attr.gen_ai.conversation.id IN (…)`, OR'd with the self fast path + (`self_principal_column == `, `user:` principals only). Past + the bound: `403 visibility_bound` ("ask for tenant-wide read"); a + cut-off stream: `503 visibility_incomplete` — never a partial predicate. + Template-level queries (`drift`, `list_templates`, `template_drift`) + need tenant-wide content read (`403 visibility_scoped`). + +```yaml +auth: + openfga: + # … + server_list_objects_deadline_ms: 3000 # OPENFGA_LIST_OBJECTS_DEADLINE + visibility: + objects: + - type: conversation + column: attr.gen_ai.conversation.id # a promoted column + self_principal_column: attr.user.hash # optional; user: principals only + # content_columns: [...] # optional; REPLACES the default set + max_objects: 10000 # tenant ids only + list_timeout_ms: 2000 # MUST be < the server deadline +``` + +`content_columns` defaults to the GenAI content attributes plus `body`; +an explicit list **replaces** that set (list every column to mask) and +may not be empty — masking is never silently disabled. `objects` unset +means scoped principals see nothing (their bound is not enumerable). +Tenant-scoped graph objects are `tenant:` and +`conversation:/` where `enc` percent-encodes `/` and `%` in +the tenant, so tenants containing `/` never alias; a tenant that cannot +be an object id (`:`, `#`, whitespace) has no graph objects and every +question about it fails closed. The two `Check`s cache with the session TTL; the +enumeration runs per query. The branch a query took is recorded on +`ourios.query.visibility` (`ourios.query.visibility.branch`) and the +request span. + The binding is cached per credential for `session_ttl_secs` and is **fail-closed**: an unreachable or slow OpenFGA answers `503` on the query and MCP surfaces and `UNAVAILABLE`/`503` on ingest, and diff --git a/docs/guides/configuration.md b/docs/guides/configuration.md index 2464b1cb..3467f167 100644 --- a/docs/guides/configuration.md +++ b/docs/guides/configuration.md @@ -78,6 +78,17 @@ auth: session_ttl_secs: 60 consistency: minimize_latency # or higher_consistency request_timeout_secs: 5 + server_list_objects_deadline_ms: 3000 + visibility: # RFC 0047 §3.4 — layer 2 inside a tenant + objects: + - type: conversation + column: attr.gen_ai.conversation.id + self_principal_column: attr.user.hash + # Optional. REPLACES the default set (body + the GenAI content + # attributes) — list every column to mask; must not be empty. + # content_columns: [body, attr.gen_ai.input.messages, attr.gen_ai.output.messages] + max_objects: 10000 + list_timeout_ms: 2000 # must be < server_list_objects_deadline_ms ``` ## Environment variables (no `--config`) diff --git a/docs/rfcs/0047-rebac-resolver-and-graph-visibility.md b/docs/rfcs/0047-rebac-resolver-and-graph-visibility.md index b9aa45f2..a3674e57 100644 --- a/docs/rfcs/0047-rebac-resolver-and-graph-visibility.md +++ b/docs/rfcs/0047-rebac-resolver-and-graph-visibility.md @@ -11,11 +11,13 @@ superseded-by: — # RFC 0047 — ReBAC resolver and graph-fed visibility -> **Status: `red` (2026-08-17).** Slice 1 — the layer-1 resolver — is -> green: RFC0047.1–.3 pass on the served binary against a real OpenFGA +> **Status: `red` (2026-08-18).** Slices 1–2 are green: RFC0047.1–.3 +> (the layer-1 resolver) and RFC0047.4–.8 (the planner two-step, masking, +> bounded enumeration) pass on the served binary against a real OpenFGA > container (`openfga-resolver` CI job) with the in-tree model; RFC0047.12 -> gates CI. RFC0047.4–.11 (planner two-step, tool gate, emitter, erasure) -> are the remaining slices. Prerequisite: RFC 0046 (out-of-band tenancy, `green`) — the tenant is an +> gates CI. RFC0047.9–.11 (tool gate, emitter, erasure) are the remaining +> slices; the RFC0047.5 request-carried contextual-tuple arm is deferred +> (§3.3, §7). Prerequisite: RFC 0046 (out-of-band tenancy, `green`) — the tenant is an > opaque, coarse, credential-selected object, which is exactly the object > type this RFC binds the authorization graph to. Grounded in the #688 > OpenFGA spike (resolver seam holds, p50 1.4 ms), two OpenFGA-assistant @@ -235,7 +237,14 @@ the emitter writes and every id the planner reads: OpenFGA object ids are opaque strings in one store, so the same raw conversation id in two tenants must be two objects; the `parent` tuple carries the tenant edge and the planner strips the prefix when it builds predicates (§3.4). A pure naming -rule, mirrored in exactly two places (emitter, planner). +rule, held in **one** place (`TenantObjects` in the core `openfga` module) +that both the emitter and the planner call. *Slice-2 decision:* the tenant +segment is percent-encoded for `/` and `%` (`conversation:/`), +so a tenant containing `/` can never alias another tenant's conversation +(`a` + `b/c-1` vs `a/b` + `c-1`); the raw conversation id follows verbatim. +A tenant that cannot itself be an object id (`:`, `#`, whitespace, > 256 +bytes) has no graph objects at all — every graph question about it fails +closed (`403 tenant_unaddressable`, naming the rule). The **binding tuple** (`tenant:T#scoped_reader@`) rides along with every conversation grant so the principal can bind the tenant at @@ -268,6 +277,17 @@ path is disabled — never a mismatched comparison; (b) **contextual tuples** passed on `Check`/`ListObjects` and never persisted. Tenant-wide readers (the FinOps/operator case) never wait: they resolve at layer 1. +**Slice-2 decision — bridge (b) is deferred, not built.** A contextual +tuple carried *by the request* is asserted by the very principal it +grants: any scoped caller could name any conversation id and read it — +a self-granted escalation the graph never checked. Contextual tuples are +an application-trusted input (the group claim, minted by the IdP, is one); +a caller-supplied one is not. Until a trusted carrier exists (a signed +claim from the producer, or the emitter's flush-cadence hook closing the +gap), freshness bridges are the self fast path (a) — verified against the +stored `user.hash` — and the emitter cadence. The RFC0047.5 contextual arm +is therefore deferred with this question in §7. + ### 3.4 Layer 2 — query rewrite at plan time For a query over tenant T by principal P, the planner runs the **two-step**: @@ -326,22 +346,42 @@ auth: - type: conversation column: attr.gen_ai.conversation.id self_principal_column: attr.user.hash # the §3.3 fast path - content_columns: [body, attr.gen_ai.input.messages, attr.gen_ai.output.messages] + content_columns: [body, attr.gen_ai.input.messages, attr.gen_ai.output.messages] # replaces the default set max_objects: 10000 # tenant-T ids only (§3.4 step 3) - list_timeout: 2s # MUST stay below OPENFGA_LIST_OBJECTS_DEADLINE (3s) + list_timeout_ms: 2000 # MUST stay below server_list_objects_deadline_ms (3000) + server_list_objects_deadline_ms: 3000 # the server's OPENFGA_LIST_OBJECTS_DEADLINE ``` -`list_timeout` is deliberately **below** OpenFGA's own +`list_timeout_ms` is deliberately **below** OpenFGA's own `OPENFGA_LIST_OBJECTS_DEADLINE` (server default 3 s, which bounds the streamed call too): the client-side timeout must be the one that fires, so an incomplete enumeration is always detected here and failed closed, never -ended quietly by the server. Startup validation rejects a `list_timeout` -that is not below the configured server deadline when the operator declares -one (`auth.openfga.server_list_objects_deadline`, default 3 s). +ended quietly by the server. Startup validation rejects a `list_timeout_ms` +that is not below the configured server deadline +(`auth.openfga.server_list_objects_deadline_ms`, default 3000). Per-record `Check` calls in the scan path are **never** performed (the architectural line from the first spike, confirmed by both reviews). +**Slice-2 decisions (implemented).** Durations are milliseconds +(`list_timeout_ms`, `server_list_objects_deadline_ms`) like every other +knob; `objects[].type` accepts only `conversation` in v1 (the one bindable +type) and columns must be `attr.`/`resource.` promoted names; the two +`Check`s cache with the session TTL, the enumeration never does; masking +renders `body` as `{"kind":"masked"}` and a masked attribute as `"value": +null` (the OTLP unset value) — a reader can tell withheld from absent; +template-level surfaces (`drift`, `list_templates`, `template_drift`) need +tenant-wide content read (`403 visibility_scoped`) because templates are +mined from bodies; the branch taken is recorded on +`ourios.query.visibility{ourios.query.visibility.branch}` and the request +span (the MCP tool spans carry the same field), so RFC0047.4's "no +enumeration" is a counter assertion; an explicit `content_columns` list +replaces the default set and may not be empty (masking is never silently +disabled). The self +fast path is `user:` principals only, and principal ids are validated as +object ids (a `sub` with `:`/`#`/whitespace is a 401-class credential +defect, not a 503). + ### 3.5 MCP tools as objects Every RFC 0027 tool (`query_logs`, `list_templates`, `template_drift`, the @@ -446,10 +486,11 @@ container (testcontainers, like Dex for RFC 0029) with the in-tree model. > tuple), When bob queries `true` on tenant `acme`, Then exactly the rows of > `c-1`/`c-2` return; And Given a further conversation `c-9` whose rows > carry `attr.user.hash = bob` (the principal's subject, prefix stripped) -> but no tuple yet, Then those rows also return (self fast path); And Given -> a contextual tuple for `c-10` on the request, Then `c-10`'s rows return -> too; And Given bob is also participant on `globex/c-1` (another tenant), -> Then that id never appears in the `acme` predicate. +> but no tuple yet, Then those rows also return (self fast path); And +> *(deferred — §3.3 slice-2 decision, §7)* Given a contextual tuple for +> `c-10` on the request, Then `c-10`'s rows return too; And Given bob is +> also participant on `globex/c-1` (another tenant), Then that id never +> appears in the `acme` predicate. > **RFC0047.6 — agent as principal.** Given `agent:bot` actor on 500 > conversations and `agent:other` actor on 500 different ones, When bot @@ -526,6 +567,11 @@ planner's returned row set equals the naive "rows whose conversation ∈ - [ ] **Where the emitter runs** — compaction sweep only, or also the receiver flush cadence (§3.3 proposes both); the freshness bridges make the answer a tuning question. +- [ ] **Request-carried contextual tuples (§3.3 bridge b)** — deferred in + slice 2: a caller-asserted `participant` tuple is a self-grant. Who + may assert one (a producer-signed claim? the emitter's flush hook + instead?), or drop the bridge and rely on the self fast path + + emitter cadence. - [ ] **OpenFGA MCP for design time** — community servers exist (`evansims/openfga-mcp`, read-only by default); adopt for authoring the `.fga.yaml` tests, never as a runtime dependency (assistant review 2 Q3). diff --git a/semconv/registry/attributes.yaml b/semconv/registry/attributes.yaml index ffbf1198..5b00556c 100644 --- a/semconv/registry/attributes.yaml +++ b/semconv/registry/attributes.yaml @@ -181,6 +181,31 @@ groups: the total), so the B1 pruned fraction is derived in the backend as `pruned / (scanned + pruned)` (RFC 0016; OTel usage/state convention — record raw counts, derive the ratio). + - id: ourios.query.visibility.branch + type: + members: + - id: tenant_wide + value: "tenant_wide" + stability: development + brief: >- + `Check(can_read_content, tenant)` allowed — the tenant + predicate only, no enumeration (RFC 0047 §3.4 step 1). + - id: metadata_masked + value: "metadata_masked" + stability: development + brief: >- + `Check(can_read_metadata, tenant)` allowed — every row with + the content columns masked, no enumeration (step 2). + - id: scoped + value: "scoped" + stability: development + brief: >- + A scoped principal — the streamed, bounded conversation + enumeration became an `IN (…)` predicate (step 3). + stability: development + brief: >- + Which branch of the RFC 0047 §3.4 two-step a query took inside its + tenant. Absent when no graph resolver is configured. - id: ourios.template_map.lookup.outcome type: members: diff --git a/semconv/registry/metrics.yaml b/semconv/registry/metrics.yaml index 03374658..e85d1ac3 100644 --- a/semconv/registry/metrics.yaml +++ b/semconv/registry/metrics.yaml @@ -412,6 +412,21 @@ groups: - ref: ourios.query.row_group.state requirement_level: required + - id: metric.ourios.query.visibility + type: metric + metric_name: ourios.query.visibility + stability: development + brief: >- + Queries answered under RFC 0047 §3.4 layer-2 visibility, by the + two-step branch taken. `scoped` counts every streamed conversation + enumeration issued; a tenant-wide or metadata reader never + enumerates (RFC0047.4 / RFC0047.8). + instrument: counter + unit: "{query}" + attributes: + - ref: ourios.query.visibility.branch + requirement_level: required + # Audit sink (issue #302): the miner's template-audit write path. A # buffering sink mirroring the RFC 0014 record sink — events buffer on the # request path and flush off the runtime. `flush.outcome` splits a failed