Add identity and target authorization package

OIDC validation, principal mapping, remora:v1 roles, and current-authz
filtering with kind acceptance on Keycloak and ZITADEL.

crates/remora-identity/src/authz.rs

@@ -0,0 +1,109 @@
+use remora_domain::{OpaqueId, StorageScope, Target};
+
+use crate::error::IdentityError;
+use crate::principal::Principal;
+use crate::target::{DefaultTarget, TargetOverride, resolve_target};
+
+/// Management operations governed by current authorization (architecture A-3).
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub enum ManagementOperation {
+    List,
+    Read,
+    Cancel,
+    SoftDelete,
+    Restore,
+    HardDelete,
+}
+
+/// Indistinguishable missing-or-unauthorized outcome for single-resource reads.
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub enum ResourceVisibility {
+    Visible,
+    NotFound,
+}
+
+/// Minimal durable job projection needed for current-authorization decisions.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct AuthorizedResource {
+    pub owner: OpaqueId,
+    pub target: Target,
+}
+
+#[must_use]
+pub fn can_target(principal: &Principal, target: &Target) -> bool {
+    if principal.administrator {
+        return true;
+    }
+    principal.scopes.iter().any(|scope| scope.covers(target))
+}
+
+/// Canonicalize defaults/overrides then authorize the result for the principal.
+///
+/// Submission and other admission paths should use this rather than calling
+/// [`resolve_target`] without a subsequent authorization check.
+pub fn resolve_authorized_target(
+    principal: &Principal,
+    defaults: &DefaultTarget,
+    override_: &TargetOverride,
+    reserved_segments: &[&str],
+) -> Result<Target, IdentityError> {
+    let target = resolve_target(defaults, override_, reserved_segments)?;
+    if can_target(principal, &target) {
+        Ok(target)
+    } else {
+        Err(IdentityError::UnauthorizedTarget)
+    }
+}
+
+#[must_use]
+pub fn may_manage(
+    principal: &Principal,
+    resource: &AuthorizedResource,
+    _operation: ManagementOperation,
+) -> bool {
+    if principal.administrator {
+        return true;
+    }
+    principal.id == resource.owner && can_target(principal, &resource.target)
+}
+
+/// Whether the principal is an administrator eligible to *request* revocation
+/// of an accepted execution grant. Does not mutate grants; durable revoke and
+/// cancellation mapping belong to admission/lifecycle packages.
+#[must_use]
+pub fn may_request_execution_grant_revocation(principal: &Principal) -> bool {
+    principal.administrator
+}
+
+/// Single-resource access: missing and unauthorized are the same response class.
+#[must_use]
+pub fn resource_visibility(
+    principal: &Principal,
+    resource: Option<&AuthorizedResource>,
+    operation: ManagementOperation,
+) -> ResourceVisibility {
+    match resource {
+        Some(resource) if may_manage(principal, resource, operation) => ResourceVisibility::Visible,
+        _ => ResourceVisibility::NotFound,
+    }
+}
+
+/// List filtering never reveals unauthorized existence.
+pub fn filter_visible<'a>(
+    principal: &Principal,
+    resources: impl IntoIterator<Item = &'a AuthorizedResource>,
+) -> Vec<&'a AuthorizedResource> {
+    resources
+        .into_iter()
+        .filter(|resource| may_manage(principal, resource, ManagementOperation::List))
+        .collect()
+}
+
+#[must_use]
+pub fn covering_scopes<'a>(principal: &'a Principal, target: &Target) -> Vec<&'a StorageScope> {
+    principal
+        .scopes
+        .iter()
+        .filter(|scope| scope.covers(target))
+        .collect()
+}

crates/remora-identity/src/error.rs

@@ -0,0 +1,40 @@
+use remora_domain::{IdError, TargetStoreIdError, ValidationError};
+use remora_durable::DurableError;
+
+#[derive(Debug, thiserror::Error)]
+pub enum IdentityError {
+    #[error("invalid or untrusted bearer token")]
+    InvalidToken,
+    #[error("identity provider unavailable: {0}")]
+    ProviderUnavailable(String),
+    #[error("invalid role claim: {0}")]
+    InvalidRole(String),
+    #[error("invalid target: {0}")]
+    InvalidTarget(String),
+    #[error("target not authorized for principal")]
+    UnauthorizedTarget,
+    #[error("invalid principal identifier")]
+    InvalidPrincipal,
+    #[error("configuration error: {0}")]
+    Config(String),
+    #[error(transparent)]
+    Durable(#[from] DurableError),
+}
+
+impl From<ValidationError> for IdentityError {
+    fn from(error: ValidationError) -> Self {
+        Self::InvalidTarget(error.to_string())
+    }
+}
+
+impl From<TargetStoreIdError> for IdentityError {
+    fn from(error: TargetStoreIdError) -> Self {
+        Self::InvalidTarget(error.to_string())
+    }
+}
+
+impl From<IdError> for IdentityError {
+    fn from(_: IdError) -> Self {
+        Self::InvalidPrincipal
+    }
+}

crates/remora-identity/src/lib.rs

@@ -0,0 +1,33 @@
+//! Identity and current target authorization.
+//!
+//! Validates OIDC bearer tokens, maps issuer+subject pairs to stable principal
+//! identifiers, decodes administrator and storage-scope roles, canonicalizes
+//! concrete targets from deployment defaults, and applies current-authorization
+//! filtering for management operations. Execution grants and job admission are
+//! owned by later packages; this crate supplies the identity and policy surface
+//! those packages compose.
+
+mod authz;
+mod error;
+mod migrate;
+mod oidc;
+mod principal;
+mod roles;
+mod store;
+mod target;
+
+pub use authz::{
+    AuthorizedResource, ManagementOperation, ResourceVisibility, can_target, covering_scopes,
+    filter_visible, may_manage, may_request_execution_grant_revocation, resolve_authorized_target,
+    resource_visibility,
+};
+pub use error::IdentityError;
+pub use migrate::{migrate, migrator};
+pub use oidc::{OidcConfig, OidcIdentityProvider, ZITADEL_PROJECT_ROLES_CLAIM, roles_from_names};
+pub use principal::{ExternalIdentity, Principal, principal_id_for};
+pub use roles::{
+    ADMIN_ROLE, DecodedRoles, ROLE_GRAMMAR_VERSION, decode_roles, encode_admin_role,
+    encode_storage_role,
+};
+pub use store::{PrincipalRecord, load_by_external, load_by_id, resolve_principal};
+pub use target::{DefaultTarget, TargetOverride, parse_default_target, resolve_target};

crates/remora-identity/src/migrate.rs

@@ -0,0 +1,19 @@
+use remora_durable::DurableError;
+use sqlx::PgPool;
+
+/// Separate migration history from the DurableStore substrate so feature-package
+/// versions do not collide with `remora-durable` in `_sqlx_migrations`.
+const MIGRATION_TABLE: &str = "_remora_identity_migrations";
+
+pub async fn migrate(pool: &PgPool) -> Result<(), DurableError> {
+    let mut migrator = sqlx::migrate!("./migrations");
+    migrator.dangerous_set_table_name(MIGRATION_TABLE);
+    migrator.run(pool).await?;
+    Ok(())
+}
+
+pub fn migrator() -> sqlx::migrate::Migrator {
+    let mut migrator = sqlx::migrate!("./migrations");
+    migrator.dangerous_set_table_name(MIGRATION_TABLE);
+    migrator
+}

crates/remora-identity/src/oidc.rs

@@ -0,0 +1,498 @@
+//! OIDC bearer-token validation against discovery + JWKS.
+//!
+//! Validates signature, `iss`, `exp`, and a mandatory expected `aud`. Role
+//! claims are taken only from explicitly configured sources: Keycloak
+//! `resource_access[<client>]` and exact ZITADEL project-role claim names.
+//! Remora authority is encoded only in namespaced `remora:v1:…` roles.
+
+use std::collections::HashMap;
+use std::sync::Arc;
+use std::time::{Duration, Instant};
+
+use jsonwebtoken::{Algorithm, DecodingKey, Validation, decode, decode_header};
+use remora_domain::ports::IdentityProvider;
+use serde::Deserialize;
+use serde_json::Value;
+use tokio::sync::RwLock;
+
+use crate::error::IdentityError;
+use crate::principal::ExternalIdentity;
+use crate::roles::{DecodedRoles, decode_roles};
+
+/// Default JWKS cache lifetime before re-fetch (key rotation / removal).
+const DEFAULT_JWKS_CACHE_TTL: Duration = Duration::from_secs(300);
+
+/// Well-known ZITADEL claim that carries project role keys when role assertion
+/// is enabled for the application.
+pub const ZITADEL_PROJECT_ROLES_CLAIM: &str = "urn:zitadel:iam:org:project:roles";
+
+#[derive(Clone, Debug)]
+pub struct OidcConfig {
+    pub issuer: String,
+    /// Required expected `aud` claim (typically the OIDC client id).
+    pub audience: String,
+    /// Keycloak client id whose `resource_access` roles are trusted.
+    /// Defaults to `audience` when not overridden.
+    pub resource_access_client: String,
+    /// Exact JWT claim names that carry ZITADEL (or similar) project role maps.
+    pub project_role_claims: Vec<String>,
+    pub reserved_segments: Vec<String>,
+    pub http_timeout: Duration,
+    /// How long a successful JWKS fetch remains valid before revalidation.
+    pub jwks_cache_ttl: Duration,
+}
+
+impl OidcConfig {
+    /// Production constructor: non-empty issuer and audience are mandatory.
+    pub fn new(
+        issuer: impl Into<String>,
+        audience: impl Into<String>,
+    ) -> Result<Self, IdentityError> {
+        let issuer = issuer.into();
+        let audience = audience.into();
+        if issuer.is_empty() {
+            return Err(IdentityError::Config("issuer must be non-empty".into()));
+        }
+        if audience.is_empty() {
+            return Err(IdentityError::Config("audience must be non-empty".into()));
+        }
+        Ok(Self {
+            resource_access_client: audience.clone(),
+            issuer,
+            audience,
+            project_role_claims: Vec::new(),
+            reserved_segments: Vec::new(),
+            http_timeout: Duration::from_secs(10),
+            jwks_cache_ttl: DEFAULT_JWKS_CACHE_TTL,
+        })
+    }
+
+    /// Trust Keycloak client roles under a specific client id (defaults to audience).
+    #[must_use]
+    pub fn with_resource_access_client(mut self, client_id: impl Into<String>) -> Self {
+        self.resource_access_client = client_id.into();
+        self
+    }
+
+    /// Trust an exact ZITADEL project-role claim name (repeatable).
+    #[must_use]
+    pub fn with_project_role_claim(mut self, claim: impl Into<String>) -> Self {
+        let claim = claim.into();
+        if !self
+            .project_role_claims
+            .iter()
+            .any(|existing| existing == &claim)
+        {
+            self.project_role_claims.push(claim);
+        }
+        self
+    }
+
+    /// Convenience: trust the standard ZITADEL project roles claim.
+    #[must_use]
+    pub fn with_zitadel_project_roles(self) -> Self {
+        self.with_project_role_claim(ZITADEL_PROJECT_ROLES_CLAIM)
+    }
+
+    #[must_use]
+    pub fn with_reserved_segments(mut self, segments: impl IntoIterator<Item = String>) -> Self {
+        self.reserved_segments = segments.into_iter().collect();
+        self
+    }
+
+    #[must_use]
+    pub fn with_jwks_cache_ttl(mut self, ttl: Duration) -> Self {
+        self.jwks_cache_ttl = ttl;
+        self
+    }
+}
+
+#[derive(Clone)]
+pub struct OidcIdentityProvider {
+    config: OidcConfig,
+    client: reqwest::Client,
+    state: Arc<RwLock<ProviderState>>,
+}
+
+#[derive(Default)]
+struct ProviderState {
+    jwks_uri: Option<String>,
+    keys: HashMap<String, DecodingKey>,
+    keys_fetched_at: Option<Instant>,
+    static_keys: bool,
+}
+
+impl ProviderState {
+    fn is_stale(&self, ttl: Duration) -> bool {
+        if self.static_keys {
+            return false;
+        }
+        match self.keys_fetched_at {
+            Some(at) => at.elapsed() >= ttl,
+            None => true,
+        }
+    }
+}
+
+#[derive(Debug, Deserialize)]
+struct DiscoveryDocument {
+    issuer: String,
+    jwks_uri: String,
+}
+
+#[derive(Debug, Deserialize)]
+struct JwksDocument {
+    keys: Vec<Jwk>,
+}
+
+#[derive(Debug, Deserialize)]
+struct Jwk {
+    kid: Option<String>,
+    kty: String,
+    #[serde(rename = "use")]
+    use_: Option<String>,
+    alg: Option<String>,
+    n: Option<String>,
+    e: Option<String>,
+}
+
+#[derive(Debug, Deserialize)]
+struct TokenClaims {
+    iss: String,
+    sub: String,
+    #[serde(default)]
+    aud: Audience,
+    #[allow(dead_code)]
+    exp: i64,
+    #[serde(flatten)]
+    extra: HashMap<String, Value>,
+}
+
+#[derive(Debug, Default, Deserialize)]
+#[serde(untagged)]
+enum Audience {
+    String(String),
+    List(Vec<String>),
+    #[default]
+    Missing,
+}
+
+impl OidcIdentityProvider {
+    pub fn new(config: OidcConfig) -> Result<Self, IdentityError> {
+        if config.audience.is_empty() {
+            return Err(IdentityError::Config("audience must be non-empty".into()));
+        }
+        let client = reqwest::Client::builder()
+            .timeout(config.http_timeout)
+            .build()
+            .map_err(|error| IdentityError::Config(error.to_string()))?;
+        Ok(Self {
+            config,
+            client,
+            state: Arc::new(RwLock::new(ProviderState::default())),
+        })
+    }
+
+    /// Test and offline harness helper: install fixed RS256 keys without discovery.
+    pub fn with_static_keys(
+        config: OidcConfig,
+        keys: HashMap<String, DecodingKey>,
+    ) -> Result<Self, IdentityError> {
+        let provider = Self::new(config)?;
+        let mut state = provider
+            .state
+            .try_write()
+            .map_err(|_| IdentityError::Config("provider state lock busy".into()))?;
+        state.keys = keys;
+        state.jwks_uri = Some("static://test-jwks".into());
+        state.keys_fetched_at = Some(Instant::now());
+        state.static_keys = true;
+        drop(state);
+        Ok(provider)
+    }
+
+    pub async fn authenticate_token(
+        &self,
+        bearer_token: &str,
+    ) -> Result<ExternalIdentity, IdentityError> {
+        let token = bearer_token
+            .strip_prefix("Bearer ")
+            .or_else(|| bearer_token.strip_prefix("bearer "))
+            .unwrap_or(bearer_token)
+            .trim();
+        if token.is_empty() {
+            return Err(IdentityError::InvalidToken);
+        }
+
+        let header = decode_header(token).map_err(|_| IdentityError::InvalidToken)?;
+        let kid = header.kid.ok_or(IdentityError::InvalidToken)?;
+        let key = self.decoding_key(&kid).await?;
+
+        let mut validation = Validation::new(Algorithm::RS256);
+        validation.set_issuer(std::slice::from_ref(&self.config.issuer));
+        validation.set_audience(std::slice::from_ref(&self.config.audience));
+        validation.leeway = 30;
+
+        let data = decode::<TokenClaims>(token, &key, &validation).map_err(|error| {
+            if is_provider_side(error.kind()) {
+                IdentityError::ProviderUnavailable(error.to_string())
+            } else {
+                IdentityError::InvalidToken
+            }
+        })?;
+        let claims = data.claims;
+        if claims.iss != self.config.issuer {
+            return Err(IdentityError::InvalidToken);
+        }
+        if !audience_matches(&claims.aud, &self.config.audience) {
+            return Err(IdentityError::InvalidToken);
+        }
+        if claims.sub.is_empty() {
+            return Err(IdentityError::InvalidToken);
+        }
+
+        let role_names = extract_role_names(
+            &claims.extra,
+            &self.config.resource_access_client,
+            &self.config.project_role_claims,
+        );
+        let reserved: Vec<&str> = self
+            .config
+            .reserved_segments
+            .iter()
+            .map(String::as_str)
+            .collect();
+        let roles = decode_roles(role_names.iter().map(String::as_str), &reserved)?;
+        Ok(ExternalIdentity {
+            issuer: claims.iss,
+            subject: claims.sub,
+            roles,
+        })
+    }
+
+    async fn decoding_key(&self, kid: &str) -> Result<DecodingKey, IdentityError> {
+        {
+            let state = self.state.read().await;
+            if state.static_keys {
+                return state
+                    .keys
+                    .get(kid)
+                    .cloned()
+                    .ok_or(IdentityError::InvalidToken);
+            }
+            if !state.is_stale(self.config.jwks_cache_ttl)
+                && let Some(key) = state.keys.get(kid)
+            {
+                return Ok(key.clone());
+            }
+        }
+        self.refresh_jwks().await?;
+        let state = self.state.read().await;
+        state
+            .keys
+            .get(kid)
+            .cloned()
+            .ok_or(IdentityError::InvalidToken)
+    }
+
+    async fn refresh_jwks(&self) -> Result<(), IdentityError> {
+        let jwks_uri = {
+            let state = self.state.read().await;
+            if state.static_keys {
+                return Ok(());
+            }
+            state.jwks_uri.clone()
+        };
+        let jwks_uri = match jwks_uri {
+            Some(uri) => uri,
+            None => self.discover().await?,
+        };
+        let document = self
+            .client
+            .get(&jwks_uri)
+            .send()
+            .await
+            .map_err(|error| IdentityError::ProviderUnavailable(error.to_string()))?
+            .error_for_status()
+            .map_err(|error| IdentityError::ProviderUnavailable(error.to_string()))?
+            .json::<JwksDocument>()
+            .await
+            .map_err(|error| IdentityError::ProviderUnavailable(error.to_string()))?;
+
+        let mut keys = HashMap::new();
+        for jwk in document.keys {
+            if jwk.kty != "RSA" {
+                continue;
+            }
+            if let Some(use_) = &jwk.use_
+                && use_ != "sig"
+            {
+                continue;
+            }
+            if let Some(alg) = &jwk.alg
+                && alg != "RS256"
+            {
+                continue;
+            }
+            let (Some(n), Some(e), Some(kid)) = (jwk.n, jwk.e, jwk.kid) else {
+                continue;
+            };
+            let key = DecodingKey::from_rsa_components(&n, &e)
+                .map_err(|error| IdentityError::ProviderUnavailable(error.to_string()))?;
+            keys.insert(kid, key);
+        }
+        if keys.is_empty() {
+            return Err(IdentityError::ProviderUnavailable(
+                "JWKS contained no usable RS256 signing keys".into(),
+            ));
+        }
+        let mut state = self.state.write().await;
+        state.jwks_uri = Some(jwks_uri);
+        state.keys = keys;
+        state.keys_fetched_at = Some(Instant::now());
+        state.static_keys = false;
+        Ok(())
+    }
+
+    async fn discover(&self) -> Result<String, IdentityError> {
+        let url = format!(
+            "{}/.well-known/openid-configuration",
+            self.config.issuer.trim_end_matches('/')
+        );
+        let document = self
+            .client
+            .get(url)
+            .send()
+            .await
+            .map_err(|error| IdentityError::ProviderUnavailable(error.to_string()))?
+            .error_for_status()
+            .map_err(|error| IdentityError::ProviderUnavailable(error.to_string()))?
+            .json::<DiscoveryDocument>()
+            .await
+            .map_err(|error| IdentityError::ProviderUnavailable(error.to_string()))?;
+        if document.issuer != self.config.issuer {
+            return Err(IdentityError::ProviderUnavailable(format!(
+                "discovery issuer {} does not match configured {}",
+                document.issuer, self.config.issuer
+            )));
+        }
+        let mut state = self.state.write().await;
+        state.jwks_uri = Some(document.jwks_uri.clone());
+        Ok(document.jwks_uri)
+    }
+}
+
+impl IdentityProvider for OidcIdentityProvider {
+    type Error = IdentityError;
+    type Principal = ExternalIdentity;
+
+    async fn authenticate(&self, bearer_token: &str) -> Result<Self::Principal, Self::Error> {
+        self.authenticate_token(bearer_token).await
+    }
+}
+
+fn audience_matches(aud: &Audience, expected: &str) -> bool {
+    match aud {
+        Audience::String(value) => value == expected,
+        Audience::List(values) => values.iter().any(|value| value == expected),
+        Audience::Missing => false,
+    }
+}
+
+/// Extract Remora role *names* only from configured Keycloak client roles and
+/// exact project-role claim maps. Plain `roles`, realm roles, and unlisted
+/// project claims are ignored.
+fn extract_role_names(
+    extra: &HashMap<String, Value>,
+    resource_access_client: &str,
+    project_role_claims: &[String],
+) -> Vec<String> {
+    let mut roles = Vec::new();
+    if let Some(Value::Object(resource_access)) = extra.get("resource_access")
+        && let Some(Value::Object(client_roles)) = resource_access.get(resource_access_client)
+        && let Some(Value::Array(values)) = client_roles.get("roles")
+    {
+        push_string_array(&mut roles, values);
+    }
+    for claim in project_role_claims {
+        if let Some(Value::Object(map)) = extra.get(claim) {
+            for role_key in map.keys() {
+                roles.push(role_key.clone());
+            }
+        }
+    }
+    roles.sort();
+    roles.dedup();
+    roles
+}
+
+fn push_string_array(roles: &mut Vec<String>, values: &[Value]) {
+    for value in values {
+        if let Value::String(role) = value {
+            roles.push(role.clone());
+        }
+    }
+}
+
+fn is_provider_side(kind: &jsonwebtoken::errors::ErrorKind) -> bool {
+    matches!(
+        kind,
+        jsonwebtoken::errors::ErrorKind::InvalidKeyFormat
+            | jsonwebtoken::errors::ErrorKind::InvalidRsaKey(_)
+            | jsonwebtoken::errors::ErrorKind::InvalidEcdsaKey
+    )
+}
+
+pub fn roles_from_names(
+    roles: &[&str],
+    reserved_segments: &[&str],
+) -> Result<DecodedRoles, IdentityError> {
+    decode_roles(roles.iter().copied(), reserved_segments)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use serde_json::json;
+
+    fn claim_map(value: Value) -> HashMap<String, Value> {
+        match value {
+            Value::Object(map) => map.into_iter().collect(),
+            _ => panic!("expected object"),
+        }
+    }
+
+    #[test]
+    fn only_configured_client_resource_access_is_trusted() {
+        let map = claim_map(json!({
+            "resource_access": {
+                "remora-client": { "roles": ["remora:v1:admin"] },
+                "evil-client": { "roles": ["remora:v1:admin"] }
+            },
+            "realm_access": { "roles": ["admin", "remora:v1:admin"] },
+            "roles": ["remora:v1:admin"]
+        }));
+        let roles = extract_role_names(&map, "remora-client", &[]);
+        assert_eq!(roles, vec!["remora:v1:admin".to_owned()]);
+    }
+
+    #[test]
+    fn only_configured_project_role_claims_are_trusted() {
+        let map = claim_map(json!({
+            "urn:zitadel:iam:org:project:roles": {
+                "remora:v1:admin": { "org": "x" }
+            },
+            "urn:zitadel:iam:org:project:999:roles": {
+                "remora:v1:admin": { "org": "x" }
+            }
+        }));
+        let allowed = vec![ZITADEL_PROJECT_ROLES_CLAIM.to_owned()];
+        let roles = extract_role_names(&map, "unused", &allowed);
+        assert_eq!(roles, vec!["remora:v1:admin".to_owned()]);
+    }
+
+    #[test]
+    fn config_rejects_empty_audience() {
+        assert!(OidcConfig::new("https://issuer.example", "").is_err());
+    }
+}

crates/remora-identity/src/principal.rs

@@ -0,0 +1,51 @@
+use remora_domain::{OpaqueId, StorageScope};
+use sha2::{Digest, Sha256};
+
+use crate::error::IdentityError;
+use crate::roles::DecodedRoles;
+
+/// Authenticated caller after token validation and role decoding.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct Principal {
+    pub id: OpaqueId,
+    pub issuer: String,
+    pub subject: String,
+    pub administrator: bool,
+    pub scopes: Vec<StorageScope>,
+}
+
+/// OIDC identity pair before durable principal-id binding.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct ExternalIdentity {
+    pub issuer: String,
+    pub subject: String,
+    pub roles: DecodedRoles,
+}
+
+impl Principal {
+    pub fn from_external(id: OpaqueId, external: ExternalIdentity) -> Self {
+        Self {
+            id,
+            issuer: external.issuer,
+            subject: external.subject,
+            administrator: external.roles.administrator,
+            scopes: external.roles.scopes,
+        }
+    }
+}
+
+/// Deterministic stable principal id for one exact issuer+subject pair.
+pub fn principal_id_for(issuer: &str, subject: &str) -> Result<OpaqueId, IdentityError> {
+    let mut hasher = Sha256::new();
+    hasher.update(b"remora-principal-v1\0");
+    hasher.update(issuer.as_bytes());
+    hasher.update([0]);
+    hasher.update(subject.as_bytes());
+    let digest = hasher.finalize();
+    let hex = digest
+        .iter()
+        .take(16)
+        .map(|byte| format!("{byte:02x}"))
+        .collect::<String>();
+    OpaqueId::parse(format!("p{hex}")).map_err(Into::into)
+}

crates/remora-identity/src/roles.rs

@@ -0,0 +1,178 @@
+//! Remora OIDC project-role grammar (`remora:v1`).
+//!
+//! - `remora:v1:admin` — deployment administrator
+//! - `remora:v1:storage|<store_b64url>|<bucket_b64url>|<prefix_b64url>` — one storage scope
+//!
+//! The versioned `remora:v1:` prefix is part of every role key so generic IdP
+//! roles such as `admin` never grant Remora authority. Store, bucket, and
+//! prefix are unpadded base64url encodings of the UTF-8 canonical values. An
+//! empty prefix means the complete bucket. Malformed roles are ignored so an
+//! unrelated IdP role cannot fail authentication closed.
+
+use base64::{Engine, engine::general_purpose::URL_SAFE_NO_PAD};
+use remora_domain::{BucketName, StorageScope, TargetPrefix, TargetStoreId};
+
+use crate::error::IdentityError;
+
+pub const ROLE_GRAMMAR_VERSION: &str = "v1";
+const ROLE_NAMESPACE: &str = "remora:v1:";
+pub const ADMIN_ROLE: &str = "remora:v1:admin";
+const STORAGE_PREFIX: &str = "remora:v1:storage|";
+
+#[derive(Clone, Debug, Default, Eq, PartialEq)]
+pub struct DecodedRoles {
+    pub administrator: bool,
+    pub scopes: Vec<StorageScope>,
+}
+
+pub fn decode_roles<'a>(
+    roles: impl IntoIterator<Item = &'a str>,
+    reserved_segments: &[&str],
+) -> Result<DecodedRoles, IdentityError> {
+    let mut administrator = false;
+    let mut scopes = Vec::new();
+    for role in roles {
+        if role == ADMIN_ROLE {
+            administrator = true;
+            continue;
+        }
+        if let Some(scope) = try_decode_storage_role(role, reserved_segments)?
+            && !scopes.iter().any(|existing| existing == &scope)
+        {
+            scopes.push(scope);
+        }
+    }
+    scopes.sort_by(|left, right| {
+        (
+            left.store.as_str(),
+            left.bucket.as_str(),
+            left.prefix.canonical(),
+        )
+            .cmp(&(
+                right.store.as_str(),
+                right.bucket.as_str(),
+                right.prefix.canonical(),
+            ))
+    });
+    Ok(DecodedRoles {
+        administrator,
+        scopes,
+    })
+}
+
+pub fn encode_admin_role() -> String {
+    ADMIN_ROLE.to_owned()
+}
+
+pub fn encode_storage_role(scope: &StorageScope) -> Result<String, IdentityError> {
+    Ok(format!(
+        "{STORAGE_PREFIX}{}|{}|{}",
+        encode_component(scope.store.as_str()),
+        encode_component(scope.bucket.as_str()),
+        encode_component(&scope.prefix.canonical()),
+    ))
+}
+
+fn try_decode_storage_role(
+    role: &str,
+    reserved_segments: &[&str],
+) -> Result<Option<StorageScope>, IdentityError> {
+    let Some(rest) = role.strip_prefix(STORAGE_PREFIX) else {
+        // Reject unversioned `storage|…` and other non-Remora names.
+        let _ = ROLE_NAMESPACE;
+        return Ok(None);
+    };
+    let parts: Vec<&str> = rest.splitn(3, '|').collect();
+    if parts.len() != 3 {
+        return Ok(None);
+    }
+    let store = match decode_component(parts[0]) {
+        Ok(value) => value,
+        Err(_) => return Ok(None),
+    };
+    let bucket = match decode_component(parts[1]) {
+        Ok(value) => value,
+        Err(_) => return Ok(None),
+    };
+    let prefix = match decode_component(parts[2]) {
+        Ok(value) => value,
+        Err(_) => return Ok(None),
+    };
+    let store = match TargetStoreId::parse(store) {
+        Ok(value) => value,
+        Err(_) => return Ok(None),
+    };
+    let bucket = match BucketName::parse(bucket) {
+        Ok(value) => value,
+        Err(_) => return Ok(None),
+    };
+    let prefix = match TargetPrefix::parse(&prefix, reserved_segments.iter().copied()) {
+        Ok(value) => value,
+        Err(_) => return Ok(None),
+    };
+    Ok(Some(StorageScope {
+        store,
+        bucket,
+        prefix,
+    }))
+}
+
+fn encode_component(value: &str) -> String {
+    URL_SAFE_NO_PAD.encode(value.as_bytes())
+}
+
+fn decode_component(value: &str) -> Result<String, IdentityError> {
+    let bytes = URL_SAFE_NO_PAD
+        .decode(value.as_bytes())
+        .map_err(|_| IdentityError::InvalidRole("invalid base64url component".into()))?;
+    String::from_utf8(bytes)
+        .map_err(|_| IdentityError::InvalidRole("role component is not UTF-8".into()))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn scope(store: &str, bucket: &str, prefix: &str) -> StorageScope {
+        StorageScope {
+            store: TargetStoreId::parse(store).unwrap(),
+            bucket: BucketName::parse(bucket).unwrap(),
+            prefix: TargetPrefix::parse(prefix, []).unwrap(),
+        }
+    }
+
+    #[test]
+    fn round_trips_storage_roles_including_unicode_prefix() {
+        let original = scope("target-main", "target-bucket", "caf\u{e9}/data");
+        let encoded = encode_storage_role(&original).unwrap();
+        assert!(encoded.starts_with("remora:v1:storage|"));
+        let decoded = decode_roles([encoded.as_str()], &[]).unwrap();
+        assert_eq!(decoded.scopes, vec![original]);
+        assert!(!decoded.administrator);
+    }
+
+    #[test]
+    fn namespaced_admin_required_foreign_admin_ignored() {
+        let storage = encode_storage_role(&scope("s1", "bucket-one", "models")).unwrap();
+        let decoded = decode_roles(["admin", "test-user", storage.as_str(), "other"], &[]).unwrap();
+        assert!(!decoded.administrator);
+        assert_eq!(decoded.scopes.len(), 1);
+
+        let with_admin = decode_roles([ADMIN_ROLE, storage.as_str()], &[]).unwrap();
+        assert!(with_admin.administrator);
+    }
+
+    #[test]
+    fn empty_prefix_means_whole_bucket() {
+        let original = scope("target-main", "target-bucket", "");
+        let encoded = encode_storage_role(&original).unwrap();
+        let decoded = decode_roles([encoded.as_str()], &[]).unwrap();
+        assert_eq!(decoded.scopes[0].prefix.segments(), &[] as &[String]);
+    }
+
+    #[test]
+    fn unversioned_storage_role_is_ignored() {
+        let decoded = decode_roles(["storage|YQ|YnVja2V0|"], &[]).unwrap();
+        assert!(decoded.scopes.is_empty());
+    }
+}

crates/remora-identity/src/store.rs

@@ -0,0 +1,96 @@
+use remora_domain::OpaqueId;
+use remora_durable::DurableError;
+use sqlx::PgPool;
+use time::OffsetDateTime;
+
+use crate::error::IdentityError;
+use crate::principal::{ExternalIdentity, Principal, principal_id_for};
+
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct PrincipalRecord {
+    pub principal_id: OpaqueId,
+    pub issuer: String,
+    pub subject: String,
+    pub created_at: OffsetDateTime,
+}
+
+/// Resolve or create the durable principal mapping for an external identity.
+pub async fn resolve_principal(
+    pool: &PgPool,
+    external: ExternalIdentity,
+) -> Result<Principal, IdentityError> {
+    let principal_id = principal_id_for(&external.issuer, &external.subject)?;
+    sqlx::query(
+        "INSERT INTO principal_identities (principal_id, issuer, subject)
+         VALUES ($1, $2, $3)
+         ON CONFLICT (issuer, subject) DO NOTHING",
+    )
+    .bind(principal_id.as_str())
+    .bind(&external.issuer)
+    .bind(&external.subject)
+    .execute(pool)
+    .await
+    .map_err(DurableError::from)?;
+
+    let record = load_by_external(pool, &external.issuer, &external.subject)
+        .await?
+        .ok_or_else(|| {
+            IdentityError::Config("principal insert succeeded but row is missing".into())
+        })?;
+
+    // Prefer the stored id so a future non-deterministic scheme remains stable.
+    Ok(Principal::from_external(record.principal_id, external))
+}
+
+pub async fn load_by_id(
+    pool: &PgPool,
+    principal_id: &OpaqueId,
+) -> Result<Option<PrincipalRecord>, IdentityError> {
+    let row = sqlx::query_as::<_, PrincipalRow>(
+        "SELECT principal_id, issuer, subject, created_at
+         FROM principal_identities
+         WHERE principal_id = $1",
+    )
+    .bind(principal_id.as_str())
+    .fetch_optional(pool)
+    .await
+    .map_err(DurableError::from)?;
+    row.map(PrincipalRow::into_record).transpose()
+}
+
+pub async fn load_by_external(
+    pool: &PgPool,
+    issuer: &str,
+    subject: &str,
+) -> Result<Option<PrincipalRecord>, IdentityError> {
+    let row = sqlx::query_as::<_, PrincipalRow>(
+        "SELECT principal_id, issuer, subject, created_at
+         FROM principal_identities
+         WHERE issuer = $1 AND subject = $2",
+    )
+    .bind(issuer)
+    .bind(subject)
+    .fetch_optional(pool)
+    .await
+    .map_err(DurableError::from)?;
+    row.map(PrincipalRow::into_record).transpose()
+}
+
+#[derive(sqlx::FromRow)]
+struct PrincipalRow {
+    principal_id: String,
+    issuer: String,
+    subject: String,
+    created_at: OffsetDateTime,
+}
+
+impl PrincipalRow {
+    fn into_record(self) -> Result<PrincipalRecord, IdentityError> {
+        Ok(PrincipalRecord {
+            principal_id: OpaqueId::parse(self.principal_id)?,
+            issuer: self.issuer,
+            subject: self.subject,
+            created_at: self.created_at,
+        })
+    }
+}

crates/remora-identity/src/target.rs

@@ -0,0 +1,56 @@
+use remora_domain::{BucketName, Target, TargetPrefix, TargetStoreId};
+
+use crate::error::IdentityError;
+
+/// Configured deployment default concrete target used when a submission omits
+/// bucket and/or prefix overrides.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct DefaultTarget {
+    pub store: TargetStoreId,
+    pub bucket: BucketName,
+    pub prefix: TargetPrefix,
+}
+
+#[derive(Clone, Debug, Default, Eq, PartialEq)]
+pub struct TargetOverride {
+    pub bucket: Option<String>,
+    pub prefix: Option<String>,
+}
+
+/// Canonicalize a concrete target from deployment defaults and optional
+/// request overrides. Store identity is never caller-selected.
+///
+/// Validation only. For submission, use [`crate::resolve_authorized_target`]
+/// so overrides cannot be accepted without a covering scope.
+pub fn resolve_target(
+    defaults: &DefaultTarget,
+    override_: &TargetOverride,
+    reserved_segments: &[&str],
+) -> Result<Target, IdentityError> {
+    let bucket = match override_.bucket.as_deref() {
+        Some(value) => BucketName::parse(value)?,
+        None => defaults.bucket.clone(),
+    };
+    let prefix = match override_.prefix.as_deref() {
+        Some(value) => TargetPrefix::parse(value, reserved_segments.iter().copied())?,
+        None => defaults.prefix.clone(),
+    };
+    Ok(Target {
+        store: defaults.store.clone(),
+        bucket,
+        prefix,
+    })
+}
+
+pub fn parse_default_target(
+    store: &str,
+    bucket: &str,
+    prefix: &str,
+    reserved_segments: &[&str],
+) -> Result<DefaultTarget, IdentityError> {
+    Ok(DefaultTarget {
+        store: TargetStoreId::parse(store)?,
+        bucket: BucketName::parse(bucket)?,
+        prefix: TargetPrefix::parse(prefix, reserved_segments.iter().copied())?,
+    })
+}