Admit a caller by the subjects it holds, not by its identity alone

crates/connetto-client/src/lib.rs

@@ -27,6 +27,7 @@
 //! [`ConnettoConnection::pump_one`], interleaving [`ConnettoConnection::push`] after local
 //! writes.
 
+use connetto_core::auth::{CapabilityKey, CapabilitySubject};
 pub use connetto_core::messages::{FullResyncReason, Grant, PauseCause, SyncStatus};
 pub use connetto_core::{Custody, NoGate};
 
@@ -606,6 +607,15 @@ pub struct ClientConfig {
     /// does. Fixed for the life of the connection, because a replica belongs
     /// to the identity it was named from.
     caller: Option<(String, String)>,
+    /// What a translated policy means by the caller's subject set: the SQLite
+    /// function name the deployment mapped its subjects setting onto, and the
+    /// packed set it returns.
+    ///
+    /// Packed once here through [`CapabilityKey::pack`], the same rendering
+    /// the server binds into the setting, so the replica's local answer and
+    /// the server's cannot disagree about which keys the caller holds. `None`
+    /// when no policy names the set.
+    subjects: Option<(String, Option<String>)>,
     /// Percentage of `page_count` the freelist must reach before the trimming
     /// pass runs. Local to this device and never sent to the server: trimming
     /// is a whole-replica operation, not a handshake input. Defaults to
@@ -647,6 +657,7 @@ impl ClientConfig {
             policy_tables: PolicyTables::default(),
             unrecorded_tables: HashSet::new(),
             caller: None,
+            subjects: None,
             trim_threshold: DEFAULT_TRIM_THRESHOLD,
             trim_budget: DEFAULT_TRIM_BUDGET,
             rested_statistics_cap: DEFAULT_RESTED_STATISTICS_CAP,
@@ -679,6 +690,31 @@ impl ClientConfig {
         self
     }
 
+    /// What the replica's translated policies mean by the caller's subject
+    /// set: the share keys it holds beside its identity.
+    ///
+    /// `function` is the SQLite function name the build mapped the subjects
+    /// setting onto, and `subjects` are the keys the caller holds, packed
+    /// through their own `Key`. That type is the deployment's and carries
+    /// its separator and its packing, which is what makes the replica unable
+    /// to disagree with the server about which keys are held: both ends call
+    /// one rendering rather than spelling it twice. Holding none is stated by
+    /// passing none, which leaves the function answering `NULL` and every
+    /// membership over the set admitting nothing.
+    ///
+    /// Distinct from [`with_capabilities`](Self::with_capabilities), which is
+    /// what the handshake presents to the server. This is what the replica
+    /// answers its own policies with, and a deployment sets both.
+    #[must_use]
+    pub fn with_subjects<Key: CapabilityKey>(
+        mut self,
+        function: impl Into<String>,
+        subjects: &[CapabilitySubject<Key>],
+    ) -> Self {
+        self.subjects = Some((function.into(), Key::pack(subjects)));
+        self
+    }
+
     /// Capability grants presented alongside the login.
     #[must_use]
     pub fn with_capabilities(mut self, capabilities: impl IntoIterator<Item = Grant>) -> Self {
@@ -1259,6 +1295,27 @@ fn register_caller(
     .map_err(|e| ClientError::Session(format!("registering the caller function: {e}")))
 }
 
+/// Register the no-argument function a translated policy calls for the
+/// caller's subject set.
+///
+/// It answers the packed set rather than one value, because that is what the
+/// server binds into the setting and what the membership test the policy
+/// compiles unpacks. `NULL` for a caller holding no key, which is what makes
+/// an absent capability fail closed: a comparison against `NULL` is `NULL`
+/// rather than true.
+fn register_subjects(
+    db: &mut SqliteConnection,
+    function: &str,
+    packed: Option<String>,
+) -> Result<(), ClientError> {
+    db.register_noarg_sql_function::<diesel::sql_types::Nullable<diesel::sql_types::Text>, _, _>(
+        function,
+        SqliteFunctionBehavior::DETERMINISTIC | SqliteFunctionBehavior::INNOCUOUS,
+        move || packed.clone(),
+    )
+    .map_err(|e| ClientError::Session(format!("registering the subjects function: {e}")))
+}
+
 /// The SQLite function a translated schema's write guards call to let
 /// connetto's own writes land underneath the policy triggers.
 ///
@@ -2323,6 +2380,9 @@ where
         if let Some((function, identity)) = &config.caller {
             register_caller(&mut db, function, identity.clone())?;
         }
+        if let Some((function, packed)) = &config.subjects {
+            register_subjects(&mut db, function, packed.clone())?;
+        }
         // Beside the caller function and for the same reach: the generated
         // write guards call it, so it exists before any trigger can run. A
         // schema that never names it pays nothing.

crates/connetto-client/tests/it/rls_sync_path.rs

@@ -21,6 +21,7 @@ use connetto_client::{
     ClientConfig, ClientError, ClientEvent, ConnettoConnection, PolicyTables, Replica,
 };
 use connetto_core::Cursor;
+use connetto_core::auth::CapabilitySubject;
 use connetto_core::messages::{
     BulkMessage, ControlMessage, FullResyncReason, FullResyncRequired, HandshakeAck, MutationPatch,
     SnapshotBegin, SnapshotEnd, SnapshotPatch, SubscriptionPriority,
@@ -726,3 +727,55 @@ fn an_unaccounted_policy_view_refuses_to_open() {
         "both the policy view and the translation's own view are named",
     );
 }
+
+/// The replica answers the caller's subject set the way the server binds it,
+/// because a policy compiled against the set is evaluated on both ends and
+/// the two renderings have to agree.
+///
+/// Packed through the one `CapabilityKey` rendering rather than spelled
+/// again here: a second spelling is how the replica starts admitting rows the
+/// server does not, or refusing rows it does, with nothing reported.
+#[test]
+fn the_replica_answers_the_packed_subject_set() {
+    use diesel::RunQueryDsl;
+
+    let (ddl, tables) = translation();
+    let held = [
+        CapabilitySubject::<String>::new("key:a"),
+        CapabilitySubject::<String>::new("key:b"),
+    ];
+    let config = client_config(tables).with_subjects("current_app_subjects", &held);
+    let mut connection =
+        ConnettoConnection::<LoopbackTransport>::open(&Replica::in_memory(), &ddl, &config, None)
+            .expect("the replica opens");
+    let answered: String = diesel::select(diesel::dsl::sql::<diesel::sql_types::Text>(
+        "current_app_subjects()",
+    ))
+    .get_result(connection.conn())
+    .expect("the registered function answers");
+    assert_eq!(
+        answered, "key:a,key:b",
+        "the replica packs the set exactly as the server binds it"
+    );
+}
+
+/// A caller holding no key answers NULL rather than the empty string, which
+/// is what makes an absent capability fail closed: a membership over the set
+/// compares against NULL and admits nothing.
+#[test]
+fn a_caller_holding_no_key_answers_null() {
+    use diesel::RunQueryDsl;
+
+    let (ddl, tables) = translation();
+    let config = client_config(tables)
+        .with_subjects("current_app_subjects", &[] as &[CapabilitySubject<String>]);
+    let mut connection =
+        ConnettoConnection::<LoopbackTransport>::open(&Replica::in_memory(), &ddl, &config, None)
+            .expect("the replica opens");
+    let answered: Option<String> = diesel::select(diesel::dsl::sql::<
+        diesel::sql_types::Nullable<diesel::sql_types::Text>,
+    >("current_app_subjects()"))
+    .get_result(connection.conn())
+    .expect("the registered function answers");
+    assert_eq!(answered, None, "no key held is unbound, never empty");
+}

crates/connetto-core/src/auth.rs

@@ -7,6 +7,9 @@
 //! checks go through the authorization model rather than through anything here.
 //! See `docs/architecture/12-identity-session-capability.md`.
 
+use core::fmt::Display;
+
+use serde::de::DeserializeOwned;
 use serde::{Deserialize, Serialize};
 
 use crate::SessionId;
@@ -19,7 +22,7 @@
 pub struct AuthContext<Id = String> {
     /// Stable user identifier resolved at handshake time. A developer-defined
     /// distributed id type. Text appears only at the one Postgres GUC bind,
-    /// through [`Display`](std::fmt::Display).
+    /// through [`Display`].
     pub user_id: Id,
 }
 
@@ -51,7 +54,7 @@
 /// Generic over the deployment's own key type for the same reason
 /// [`AuthContext`] is generic over its user id: text belongs at the edges, not
 /// in the middle. The key's serde encoding is what the signed token carries,
-/// and its [`Display`](core::fmt::Display) rendering is what reaches Postgres.
+/// and its [`Display`] rendering is what reaches Postgres.
 ///
 /// It is not a person and it asserts nothing about what it may do: the
 /// authorization model holds the permission as a relation on this name, so
@@ -175,6 +178,60 @@
     }
 }
 
+/// The deployment's share-key type: how one is minted, and how the keys a
+/// caller holds reach Postgres.
+///
+/// A policy can only compare against a value the transaction bound, and a
+/// caller may hold several keys, so the set travels as one text value under
+/// [`SETTING`](Self::SETTING). The default packing joins the keys with
+/// [`SEPARATOR`](Self::SEPARATOR), which a policy unpacks:
+///
+/// ```sql
+/// viewer = ANY(string_to_array(current_setting('app.subjects', true), ','))
+/// ```
+///
+/// Whatever a deployment chooses is the contract its policies are written
+/// against, so choose before writing policies rather than after. A deployment
+/// wanting its own key type, setting, or packing implements this for that type
+/// and everything downstream follows from
+/// [`Principal`]'s key parameter.
+/// Minting lives beside the issuer rather than here, because a replica needs
+/// the rendering to answer its own policies and never needs to make a key.
+pub trait CapabilityKey:
+    Clone + Display + Serialize + DeserializeOwned + Send + Sync + 'static
+{
+    /// The Postgres setting the packed keys are bound to.
+    const SETTING: &'static str = "app.subjects";
+
+    /// The character joining packed keys. A key whose rendering contains it is
+    /// refused at minting, because one that slipped through would split into
+    /// two and grant a neighbouring key's access.
+    const SEPARATOR: char = ',';
+
+    /// Pack the keys a caller holds, or `None` to leave the setting unbound.
+    ///
+    /// Unbound rather than empty is what makes an absent capability fail
+    /// closed: `current_setting` yields NULL, and a comparison against NULL is
+    /// NULL rather than true.
+    fn pack(keys: &[CapabilitySubject<Self>]) -> Option<String> {
+        if keys.is_empty() {
+            return None;
+        }
+        let mut packed = String::new();
+        for key in keys {
+            if !packed.is_empty() {
+                packed.push(Self::SEPARATOR);
+            }
+            packed.push_str(&key.key().to_string());
+        }
+        Some(packed)
+    }
+}
+
+/// The default share-key: `key:` followed by a version 4 UUID, which no
+/// rendering of can contain the separator.
+impl CapabilityKey for String {}
+
 /// More than one login grant resolved on one handshake.
 ///
 /// The identity is dropped rather than picked, so the caller proceeds

crates/connetto-server/src/auth.rs

@@ -26,8 +26,9 @@ mod rls {
     use subql::visibility::{RowView, RowWrite, Verdict, VisibilityPolicy};
     use subql::{DatabaseLike, ParserDB, TableLike};
 
-    use crate::capability::{CallerBinding, CapabilityKey};
+    use crate::capability::CallerBinding;
     use crate::key_filter::{KeyError, KeyFilter};
+    use connetto_core::auth::CapabilityKey;
     use connetto_core::quote_ident;
 
     /// How long a locking read waits for a conflicting writer.

crates/connetto-server/src/bin/connetto-server.rs

@@ -89,6 +89,7 @@ use std::sync::Arc;
 use std::time::{Duration, SystemTime};
 
 use anyhow::{Context, Result, anyhow};
+use connetto_core::auth::CapabilityKey;
 use connetto_core::env::{read_ddl, var_or};
 use connetto_core::messages::{ContentVerb, FatalErrorReason};
 use connetto_core::traits::{ContentTicketSigner, HandshakeAuthority};
@@ -97,6 +98,7 @@ use connetto_core::{SchemaVersion, SessionId};
 use connetto_file_server::{
     self as files, DbPool, DefaultFileSchema, TicketSigner, TicketVerifier,
 };
+use connetto_server::CallerMappings;
 use connetto_server::audit::pg_audit_hook;
 use connetto_server::capability::DEFAULT_USER_SETTING;
 use connetto_server::openfga::{
@@ -845,10 +847,26 @@ async fn prepare_change_log(
 /// function `CONNETTO_CALLER_FUNCTION` names, paired against the identity
 /// setting. Empty means unset, and without it a subscription naming the
 /// caller's local function is refused at registration.
-fn caller_mapping() -> Option<SessionVariableMapping> {
+///
+/// `CONNETTO_SUBJECTS_FUNCTION` names the second one, paired against the
+/// setting the caller's share keys are bound to. It declares the delimiter
+/// those keys are joined with, so a membership test over the set reverse
+/// translates as one, rather than as a comparison against the joined text
+/// that matches nobody. Empty leaves a deployment with no share keys exactly
+/// as it was.
+fn caller_mapping() -> Option<CallerMappings> {
     let function = var_or("CONNETTO_CALLER_FUNCTION", "");
-    (!function.is_empty())
-        .then(|| SessionVariableMapping::current_setting(DEFAULT_USER_SETTING, function))
+    if function.is_empty() {
+        return None;
+    }
+    let subjects = var_or("CONNETTO_SUBJECTS_FUNCTION", "");
+    Some(CallerMappings {
+        identity: SessionVariableMapping::current_setting(DEFAULT_USER_SETTING, function),
+        subjects: (!subjects.is_empty()).then(|| {
+            SessionVariableMapping::current_setting(<String as CapabilityKey>::SETTING, subjects)
+                .holding_set(<String as CapabilityKey>::SEPARATOR)
+        }),
+    })
 }
 
 /// The concrete manager this binary serves.