Skip to content
Merged
Show file tree
Hide file tree
Changes from 3 commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 3 additions & 2 deletions .github/workflows/ci.yml
Original file line number Diff line number Diff line change
Expand Up @@ -384,8 +384,8 @@ jobs:
set -euo pipefail
fga model transform --file model.fga | diff - model.json

# RFC 0047 §5 (RFC0047.1–.9): the OpenFGA resolver, the layer-2
# visibility two-step and the MCP tool gate on the served binary against a real
# RFC 0047 §5 (RFC0047.1–.11): the OpenFGA resolver, the layer-2
# visibility two-step, the MCP tool gate, the graph emitter and erasure 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`).
Expand All @@ -407,6 +407,7 @@ jobs:
--ignored --exact
rfc0047_openfga::rfc0047_1_to_3_resolver_end_to_end
rfc0047_visibility::rfc0047_4_to_9_visibility_end_to_end
rfc0047_emitter::rfc0047_10_11_emitter_and_erasure_end_to_end

# Emission-time semconv conformance: boot the real `ourios-server`,
# point its OTLP export (metrics + the dogfooded logs signal) at
Expand Down
1 change: 1 addition & 0 deletions crates/ourios-core/src/alias.rs
Original file line number Diff line number Diff line change
Expand Up @@ -402,6 +402,7 @@ impl AliasMap {
| AuditPayload::Compaction { .. }
| AuditPayload::RecordQuarantined { .. }
| AuditPayload::IngestDenied { .. }
| AuditPayload::ConversationErased { .. }
| AuditPayload::Unknown { .. } => {}
}
}
Expand Down
23 changes: 23 additions & 0 deletions crates/ourios-core/src/audit.rs
Original file line number Diff line number Diff line change
Expand Up @@ -245,6 +245,21 @@ pub enum AuditPayload {
/// The rejecting token's audit/metric label (RFC 0026 §3.4).
token_name: String,
},
/// A conversation was erased from the tenant (RFC 0047 §3.6): its rows
/// dropped by the compaction rewrite of every partition, then its
/// graph tuples deleted — in that order; the event is written after
/// both. System-scoped like [`Self::Compaction`]; the event's
/// `tenant_id` is the tenant the conversation lived in.
ConversationErased {
/// The raw conversation id (the promoted-column value).
conversation_id: String,
/// Partitions rewritten by the erasure pass.
partitions_rewritten: u64,
/// Rows dropped across those rewrites.
rows_dropped: u64,
/// Graph tuples deleted for the conversation object.
tuples_deleted: u64,
},
}

/// Stable on-disk `event_kind` ordinals (RFC 0005 §3.7 mapping).
Expand Down Expand Up @@ -272,6 +287,9 @@ pub const EVENT_KIND_RECORD_QUARANTINED: u8 = 7;
/// `ingest_denied` — an authenticated cross-tenant write attempt was
/// rejected pre-WAL (RFC 0026 §3.2).
pub const EVENT_KIND_INGEST_DENIED: u8 = 8;
/// `conversation_erased` — a conversation's rows were dropped by the
/// compaction rewrite and its graph tuples deleted (RFC 0047 §3.6).
pub const EVENT_KIND_CONVERSATION_ERASED: u8 = 9;

/// Canonical `event_type` strings paired with the ordinals above
/// (RFC 0005 §3.7 / RFC 0001 §6.4 / RFC 0009 §3.6).
Expand All @@ -293,6 +311,8 @@ pub const EVENT_TYPE_TEMPLATE_CREATED: &str = "template_created";
pub const EVENT_TYPE_RECORD_QUARANTINED: &str = "record_quarantined";
/// The string form of [`EVENT_KIND_INGEST_DENIED`].
pub const EVENT_TYPE_INGEST_DENIED: &str = "ingest_denied";
/// The string form of [`EVENT_KIND_CONVERSATION_ERASED`].
pub const EVENT_TYPE_CONVERSATION_ERASED: &str = "conversation_erased";

/// The `template_version` a leaf is born at (RFC 0017 §3.1). The
/// [`TemplateChange::Created`] variant omits a version field — the invariant
Expand Down Expand Up @@ -320,6 +340,7 @@ impl AuditPayload {
Self::Compaction { .. } => EVENT_KIND_COMPACTION,
Self::RecordQuarantined { .. } => EVENT_KIND_RECORD_QUARANTINED,
Self::IngestDenied { .. } => EVENT_KIND_INGEST_DENIED,
Self::ConversationErased { .. } => EVENT_KIND_CONVERSATION_ERASED,
Self::Unknown { event_kind, .. } => *event_kind,
}
}
Expand All @@ -344,6 +365,7 @@ impl AuditPayload {
Self::Compaction { .. } => EVENT_TYPE_COMPACTION,
Self::RecordQuarantined { .. } => EVENT_TYPE_RECORD_QUARANTINED,
Self::IngestDenied { .. } => EVENT_TYPE_INGEST_DENIED,
Self::ConversationErased { .. } => EVENT_TYPE_CONVERSATION_ERASED,
Self::Unknown { event_type, .. } => event_type,
}
}
Expand All @@ -365,6 +387,7 @@ impl AuditPayload {
| Self::Compaction { .. }
| Self::RecordQuarantined { .. }
| Self::IngestDenied { .. }
| Self::ConversationErased { .. }
| Self::Unknown { .. } => false,
}
}
Expand Down
85 changes: 84 additions & 1 deletion crates/ourios-core/src/auth/openfga/client.rs
Original file line number Diff line number Diff line change
Expand Up @@ -39,7 +39,7 @@ const MAX_CACHE_ENTRIES: usize = 4096;
const MAX_ERROR_BODY_BYTES: usize = 512;

/// One relationship tuple / tuple key on the wire.
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
pub struct TupleKey {
/// `<type>:<id>` or a userset `<type>:<id>#<relation>`.
pub user: String,
Expand Down Expand Up @@ -225,6 +225,42 @@ struct WriteBody<'a> {
authorization_model_id: Option<&'a str>,
}

/// `Read` filter: every tuple on `object` (optionally one `relation`).
#[derive(Serialize)]
struct ReadTupleKey<'a> {
object: &'a str,
#[serde(skip_serializing_if = "Option::is_none")]
relation: Option<&'a str>,
}

#[derive(Serialize)]
struct ReadBody<'a> {
tuple_key: ReadTupleKey<'a>,
page_size: u32,
#[serde(skip_serializing_if = "Option::is_none")]
continuation_token: Option<&'a str>,
}

#[derive(Deserialize)]
struct ReadResponse {
#[serde(default)]
tuples: Vec<ReadTuple>,
#[serde(default)]
continuation_token: Option<String>,
}

#[derive(Deserialize)]
struct ReadTuple {
key: TupleKey,
}

/// `Read` page size — `OpenFGA`'s maximum.
const READ_PAGE_SIZE: u32 = 100;
/// The bound on tuples one `read_by_object` returns: an object with more
/// is not a conversation the emitter wrote (a few relations per
/// participant), and an unbounded read is not fail-closed.
const MAX_READ_TUPLES: usize = 10_000;

/// The `OpenFGA` HTTP API over one store.
#[derive(Clone)]
pub struct OpenFgaClient {
Expand Down Expand Up @@ -393,6 +429,53 @@ impl OpenFgaClient {
Ok(kept)
}

/// `Read`: every tuple on `object` (RFC 0047 §3.6 — "no wildcard
/// delete exists": erasure reads the object's tuples, then deletes
/// them). Paginated to completion; bounded.
///
/// # Errors
///
/// [`OpenFgaError::Unavailable`] on transport/timeout/non-2xx;
/// [`OpenFgaError::BoundExceeded`] past the read bound.
pub async fn read_by_object(&self, object: &str) -> Result<Vec<TupleKey>, OpenFgaError> {
let mut tuples = Vec::new();
let mut continuation: Option<String> = None;
loop {
let body = ReadBody {
tuple_key: ReadTupleKey {
object,
relation: None,
},
page_size: READ_PAGE_SIZE,
continuation_token: continuation.as_deref(),
};
let response = self
.post("read", &body)?
.send()
.await
.map_err(|e| transport(&e))?;
let response = ok_status(response).await?;
let bytes = response
.bytes()
.await
.map_err(|e| OpenFgaError::Unavailable(format!("read read: {e}")))?;
let page: ReadResponse = serde_json::from_slice(&bytes)
.map_err(|e| OpenFgaError::Unavailable(format!("decode read: {e}")))?;
for tuple in page.tuples {
if tuples.len() >= MAX_READ_TUPLES {
return Err(OpenFgaError::BoundExceeded {
bound: MAX_READ_TUPLES,
});
}
tuples.push(tuple.key);
}
match page.continuation_token {
Some(token) if !token.is_empty() => continuation = Some(token),
_ => return Ok(tuples),
}
}
}

/// `Write`: add `writes` and remove `deletes` in one transactional,
/// **idempotent** call (`on_duplicate` / `on_missing` = `ignore`, so a
/// tuple already present or already gone is not an error). `OpenFGA`
Expand Down
30 changes: 29 additions & 1 deletion crates/ourios-core/src/auth/openfga/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -215,6 +215,9 @@ pub const DEFAULT_LIST_TIMEOUT_MS: u64 = 2_000;
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";
/// The RFC 0027 MCP tools as graph objects (RFC 0047 §3.5):
/// `tool:<T>/<name>`; the emitter writes their `parent` tuples per tenant.
pub const MCP_TOOL_NAMES: [&str; 3] = ["query_logs", "list_templates", "template_drift"];

impl fmt::Debug for OpenFgaConfig {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
Expand Down Expand Up @@ -602,13 +605,19 @@ pub struct TenantObjects {

impl TenantObjects {
/// The graph objects of `tenant`, or `None` when the tenant id cannot
/// form an object id.
/// form an object id — raw, or once its segment is encoded (the
/// encoding can only grow it; a tenant whose encoded segment plus the
/// `/` separator leaves no room for a conversation id has no
/// tenant-scoped objects).
#[must_use]
pub fn new(tenant: &str) -> Option<Self> {
if !is_object_id(tenant) {
return None;
}
let encoded = encode_tenant_segment(tenant);
if encoded.len() + 1 >= MAX_OBJECT_ID_BYTES {
return None;
}
Some(Self {
tenant_object: format!("{TENANT_TYPE}:{tenant}"),
conversation_prefix: format!("{CONVERSATION_TYPE}:{encoded}/"),
Expand All @@ -635,6 +644,16 @@ impl TenantObjects {
format!("{}{id}", self.conversation_prefix)
}

/// Whether `conversation:<enc(T)>/<id>` is a valid object: `id` must be
/// an object id itself and the combined id half must fit `OpenFGA`'s
/// 256-byte limit. The emitter skips ids that do not.
#[must_use]
pub fn conversation_fits(&self, id: &str) -> bool {
is_object_id(id)
&& self.conversation_prefix.len() - CONVERSATION_TYPE.len() - 1 + id.len()
<= MAX_OBJECT_ID_BYTES
}

/// `tool:<enc(T)>/<name>`.
#[must_use]
pub fn tool(&self, name: &str) -> String {
Expand Down Expand Up @@ -908,6 +927,15 @@ mod tests {
for bad in ["", "a b", "a:b", "a#b"] {
assert!(TenantObjects::new(bad).is_none(), "{bad:?}");
}
// The encoding may not push the segment past the object-id limit,
// and a conversation id must fit next to it.
let slashes = "/".repeat(90); // 90 raw bytes → 270 encoded
assert!(TenantObjects::new(&slashes).is_none());
let long = "x".repeat(200);
let t = TenantObjects::new(&long).expect("fits alone");
assert!(t.conversation_fits("c-1"));
assert!(!t.conversation_fits(&"y".repeat(60)), "201 + 60 > 256");
assert!(!t.conversation_fits("a b"));
}

/// The principal vocabulary renders exactly the model's type names.
Expand Down
Loading