Add durable credential registry and secret lifecycle

crates/remora-credentials/Cargo.toml

@@ -0,0 +1,25 @@
+[package]
+name = "remora-credentials"
+edition.workspace = true
+publish = false
+rust-version.workspace = true
+version.workspace = true
+
+[dependencies]
+getrandom = "0.3.4"
+remora-domain.workspace = true
+remora-durable.workspace = true
+reqwest = { version = "0.12.28", default-features = false, features = [
+    "json",
+    "rustls-tls-webpki-roots",
+] }
+serde = { workspace = true, features = ["derive"] }
+serde_json.workspace = true
+sha2.workspace = true
+sqlx.workspace = true
+thiserror.workspace = true
+time.workspace = true
+tokio.workspace = true
+
+[dev-dependencies]
+tokio = { workspace = true, features = ["macros", "rt-multi-thread", "sync", "time"] }

crates/remora-credentials/migrations/0001_credentials.sql

@@ -0,0 +1,53 @@
+-- Source credential registry metadata. Secret material lives in Vault KV v2;
+-- this package only stores owner, kind, status, and opaque version coordinates.
+
+CREATE TABLE source_credentials (
+    credential_id TEXT PRIMARY KEY,
+    owner_id TEXT NOT NULL,
+    label TEXT NOT NULL,
+    kind TEXT NOT NULL,
+    status TEXT NOT NULL,
+    active_version_id TEXT,
+    secret_mount TEXT NOT NULL,
+    secret_path TEXT NOT NULL,
+    created_at TIMESTAMPTZ NOT NULL,
+    updated_at TIMESTAMPTZ NOT NULL,
+    CONSTRAINT source_credentials_kind_check CHECK (
+        kind IN (
+            'basic_auth',
+            'bearer_token',
+            'http_headers',
+            'object_store',
+            'hugging_face'
+        )
+    ),
+    CONSTRAINT source_credentials_status_check CHECK (status IN ('active', 'revoked')),
+    CONSTRAINT source_credentials_label_len CHECK (
+        char_length(label) BETWEEN 1 AND 200
+    ),
+    CONSTRAINT source_credentials_active_status CHECK (
+        (status = 'active' AND active_version_id IS NOT NULL)
+        OR (status = 'revoked')
+    )
+);
+
+CREATE INDEX source_credentials_owner_created_idx
+    ON source_credentials (owner_id, created_at DESC, credential_id DESC);
+
+CREATE TABLE source_credential_versions (
+    version_id TEXT PRIMARY KEY,
+    credential_id TEXT NOT NULL REFERENCES source_credentials (credential_id),
+    vault_kv_version BIGINT NOT NULL,
+    created_at TIMESTAMPTZ NOT NULL,
+    destroyed_at TIMESTAMPTZ,
+    CONSTRAINT source_credential_versions_kv_positive CHECK (vault_kv_version >= 1)
+);
+
+CREATE INDEX source_credential_versions_credential_idx
+    ON source_credential_versions (credential_id, created_at DESC);
+
+ALTER TABLE source_credentials
+    ADD CONSTRAINT source_credentials_active_version_fk
+    FOREIGN KEY (active_version_id)
+    REFERENCES source_credential_versions (version_id)
+    DEFERRABLE INITIALLY DEFERRED;

crates/remora-credentials/migrations/0002_secret_write_intents.sql

@@ -0,0 +1,47 @@
+-- Durable intent for Vault secret writes. Secret material is never stored here.
+-- Intents make create/rotate recoverable across ambiguous PostgreSQL commits
+-- without best-effort Vault destruction that could erase a committed version.
+
+CREATE TABLE secret_write_intents (
+    intent_id TEXT PRIMARY KEY,
+    operation TEXT NOT NULL,
+    credential_id TEXT NOT NULL,
+    version_id TEXT NOT NULL,
+    owner_id TEXT,
+    label TEXT,
+    credential_kind TEXT,
+    secret_mount TEXT NOT NULL,
+    secret_path TEXT NOT NULL,
+    vault_kv_version BIGINT,
+    status TEXT NOT NULL,
+    created_at TIMESTAMPTZ NOT NULL,
+    updated_at TIMESTAMPTZ NOT NULL,
+    CONSTRAINT secret_write_intents_operation_check CHECK (
+        operation IN ('create', 'rotate')
+    ),
+    CONSTRAINT secret_write_intents_status_check CHECK (
+        status IN ('prepared', 'vault_written', 'durable_committed', 'abandoned')
+    ),
+    CONSTRAINT secret_write_intents_kv_check CHECK (
+        vault_kv_version IS NULL OR vault_kv_version >= 1
+    ),
+    CONSTRAINT secret_write_intents_create_fields_check CHECK (
+        (operation = 'create'
+            AND owner_id IS NOT NULL
+            AND label IS NOT NULL
+            AND credential_kind IS NOT NULL)
+        OR (operation = 'rotate'
+            AND owner_id IS NULL
+            AND label IS NULL
+            AND credential_kind IS NULL)
+    ),
+    CONSTRAINT secret_write_intents_written_has_version CHECK (
+        (status = 'prepared' AND vault_kv_version IS NULL)
+        OR (status IN ('vault_written', 'durable_committed')
+            AND vault_kv_version IS NOT NULL)
+        OR status = 'abandoned'
+    )
+);
+
+CREATE INDEX secret_write_intents_status_idx
+    ON secret_write_intents (status, created_at);

crates/remora-credentials/migrations/0003_execution_grant_pins.sql

@@ -0,0 +1,21 @@
+-- Durable execution-grant credential pins. Resolution always re-reads this table:
+-- revoked or missing grants fail closed even if a worker still holds a grant id.
+
+CREATE TABLE execution_grant_credential_pins (
+    grant_id TEXT PRIMARY KEY,
+    credential_id TEXT NOT NULL,
+    version_id TEXT NOT NULL,
+    status TEXT NOT NULL,
+    created_at TIMESTAMPTZ NOT NULL,
+    revoked_at TIMESTAMPTZ,
+    CONSTRAINT execution_grant_credential_pins_status_check CHECK (
+        status IN ('active', 'revoked')
+    ),
+    CONSTRAINT execution_grant_credential_pins_revoked_at_check CHECK (
+        (status = 'active' AND revoked_at IS NULL)
+        OR (status = 'revoked' AND revoked_at IS NOT NULL)
+    )
+);
+
+CREATE INDEX execution_grant_credential_pins_credential_idx
+    ON execution_grant_credential_pins (credential_id, status);

crates/remora-credentials/migrations/0004_intent_abandoning.sql

@@ -0,0 +1,29 @@
+-- Reconciler ownership claim: vault destruction runs only after an intent is
+-- moved to abandoning under the credential lock. Create/rotate materialize
+-- refuses non-vault_written intents, so a concurrent commit cannot land a
+-- durable version that points at material the reconciler is about to destroy.
+
+ALTER TABLE secret_write_intents
+    DROP CONSTRAINT secret_write_intents_status_check;
+
+ALTER TABLE secret_write_intents
+    ADD CONSTRAINT secret_write_intents_status_check CHECK (
+        status IN (
+            'prepared',
+            'vault_written',
+            'durable_committed',
+            'abandoning',
+            'abandoned'
+        )
+    );
+
+ALTER TABLE secret_write_intents
+    DROP CONSTRAINT secret_write_intents_written_has_version;
+
+ALTER TABLE secret_write_intents
+    ADD CONSTRAINT secret_write_intents_written_has_version CHECK (
+        (status = 'prepared' AND vault_kv_version IS NULL)
+        OR (status IN ('vault_written', 'durable_committed', 'abandoning')
+            AND vault_kv_version IS NOT NULL)
+        OR status = 'abandoned'
+    );

crates/remora-credentials/migrations/0005_version_secret_coordinates.sql

@@ -0,0 +1,13 @@
+ALTER TABLE source_credential_versions
+    ADD COLUMN secret_mount TEXT,
+    ADD COLUMN secret_path TEXT;
+
+UPDATE source_credential_versions AS version
+SET secret_mount = credential.secret_mount,
+    secret_path = credential.secret_path
+FROM source_credentials AS credential
+WHERE credential.credential_id = version.credential_id;
+
+ALTER TABLE source_credential_versions
+    ALTER COLUMN secret_mount SET NOT NULL,
+    ALTER COLUMN secret_path SET NOT NULL;

crates/remora-credentials/src/authz.rs

@@ -0,0 +1,28 @@
+use remora_domain::OpaqueId;
+
+use crate::error::CredentialError;
+use crate::types::SourceCredential;
+
+/// Caller identity for credential registry authorization.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct CredentialCaller {
+    pub principal_id: OpaqueId,
+    pub administrator: bool,
+}
+
+/// Owner or administrator may manage a credential; peers see NotFound.
+#[must_use]
+pub fn may_manage(caller: &CredentialCaller, owner: &OpaqueId) -> bool {
+    caller.administrator || caller.principal_id == *owner
+}
+
+/// Map missing and unauthorized credentials to the same NotFound outcome.
+pub fn require_manage(
+    caller: &CredentialCaller,
+    credential: Option<&SourceCredential>,
+) -> Result<SourceCredential, CredentialError> {
+    match credential {
+        Some(record) if may_manage(caller, &record.owner) => Ok(record.clone()),
+        Some(_) | None => Err(CredentialError::NotFound),
+    }
+}

crates/remora-credentials/src/error.rs

@@ -0,0 +1,40 @@
+use remora_domain::IdError;
+use remora_durable::DurableError;
+
+#[derive(Debug, thiserror::Error)]
+pub enum CredentialError {
+    #[error("credential not found")]
+    NotFound,
+    #[error("credential is revoked")]
+    Revoked,
+    #[error("credential has no pinable version")]
+    NoActiveVersion,
+    #[error("invalid credential input: {0}")]
+    InvalidInput(String),
+    #[error("credential material does not match stored kind")]
+    KindMismatch,
+    #[error("execution grant is not active")]
+    GrantInactive,
+    #[error("credential version is not resolvable")]
+    Unresolvable,
+    #[error("secret store unavailable")]
+    SecretStoreUnavailable,
+    #[error("configuration error: {0}")]
+    Config(String),
+    #[error(transparent)]
+    Durable(#[from] DurableError),
+}
+
+impl From<IdError> for CredentialError {
+    fn from(_: IdError) -> Self {
+        Self::InvalidInput("invalid opaque identifier".into())
+    }
+}
+
+impl CredentialError {
+    /// Errors never embed secret material; this helper keeps that invariant explicit.
+    #[must_use]
+    pub fn is_safe_for_logs(&self) -> bool {
+        true
+    }
+}

crates/remora-credentials/src/ids.rs

@@ -0,0 +1,30 @@
+use remora_domain::OpaqueId;
+
+use crate::error::CredentialError;
+
+pub fn new_credential_id() -> Result<OpaqueId, CredentialError> {
+    new_id('c')
+}
+
+pub fn new_version_id() -> Result<OpaqueId, CredentialError> {
+    new_id('v')
+}
+
+pub fn new_intent_id() -> Result<OpaqueId, CredentialError> {
+    new_id('i')
+}
+
+fn new_id(prefix: char) -> Result<OpaqueId, CredentialError> {
+    let mut bytes = [0u8; 16];
+    getrandom::fill(&mut bytes).map_err(|error| {
+        CredentialError::Config(format!("failed to draw opaque identifier entropy: {error}"))
+    })?;
+    let mut hex = String::with_capacity(32);
+    for byte in bytes {
+        hex.push(char::from(HEX[(byte >> 4) as usize]));
+        hex.push(char::from(HEX[(byte & 0x0f) as usize]));
+    }
+    OpaqueId::parse(format!("{prefix}{hex}")).map_err(Into::into)
+}
+
+const HEX: &[u8; 16] = b"0123456789abcdef";

crates/remora-credentials/src/lib.rs

@@ -0,0 +1,29 @@
+//! Source credential registry and grant-bound secret resolution.
+//!
+//! Durable metadata and authorization state live in PostgreSQL. Secret material
+//! is stored in Vault KV v2 behind server-allocated paths and immutable versions.
+//! Submission pins the current authorized version into an execution grant; the
+//! trusted resolver accepts only an ActiveCredentialGrant capability formed
+//! after durable grant activity is verified.
+
+mod authz;
+mod error;
+mod ids;
+mod migrate;
+mod registry;
+mod resolve;
+mod types;
+mod vault;
+
+pub use authz::{CredentialCaller, may_manage, require_manage};
+pub use error::CredentialError;
+pub use migrate::{migrate, migrator};
+pub use registry::CredentialRegistry;
+pub use resolve::GrantBoundResolver;
+pub use types::{
+    CreateCredential, CredentialGrantVersion, SecretMaterial, SourceCredential,
+    SourceCredentialKind, SourceCredentialPage, SourceCredentialStatus, SourceCredentialVersion,
+};
+pub use vault::{LiveVaultVersion, VaultClient, VaultConfig, split_vault_path};
+
+pub use remora_domain::ports::ActiveCredentialGrant;

crates/remora-credentials/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_credentials_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-credentials/src/resolve.rs

@@ -0,0 +1,62 @@
+use remora_domain::ports::{ActiveCredentialGrant, CredentialResolver};
+
+use crate::error::CredentialError;
+use crate::registry::CredentialRegistry;
+use crate::types::SecretMaterial;
+use crate::vault::VaultClient;
+
+/// Grant-bound credential resolver.
+///
+/// Every resolve reloads durable execution-grant pin state. Unknown, revoked,
+/// or destroyed pins fail closed. Callers cannot resolve by forging bare
+/// credential/version IDs.
+#[derive(Clone, Debug)]
+pub struct GrantBoundResolver {
+    registry: CredentialRegistry,
+    vault: VaultClient,
+}
+
+impl GrantBoundResolver {
+    #[must_use]
+    pub fn new(registry: CredentialRegistry, vault: VaultClient) -> Self {
+        Self { registry, vault }
+    }
+
+    pub async fn resolve_access(
+        &self,
+        access: &ActiveCredentialGrant,
+    ) -> Result<SecretMaterial, CredentialError> {
+        let pin = self
+            .registry
+            .load_active_grant_pin(access.grant_id())
+            .await?
+            .ok_or(CredentialError::GrantInactive)?;
+        let coord = self
+            .registry
+            .load_version_coord(&pin.credential_id, &pin.version_id)
+            .await?
+            .ok_or(CredentialError::Unresolvable)?;
+        if coord.destroyed {
+            return Err(CredentialError::Unresolvable);
+        }
+        self.vault
+            .get_version(
+                &coord.secret_mount,
+                &coord.secret_path,
+                coord.vault_kv_version,
+            )
+            .await
+    }
+}
+
+impl CredentialResolver for GrantBoundResolver {
+    type Error = CredentialError;
+    type Secret = SecretMaterial;
+
+    fn resolve(
+        &self,
+        access: &ActiveCredentialGrant,
+    ) -> impl std::future::Future<Output = Result<Self::Secret, Self::Error>> + Send {
+        self.resolve_access(access)
+    }
+}

crates/remora-credentials/src/types.rs

@@ -0,0 +1,263 @@
+use std::collections::BTreeMap;
+use std::fmt;
+
+use remora_domain::OpaqueId;
+use serde::{Deserialize, Serialize};
+use time::OffsetDateTime;
+
+#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, PartialEq, Serialize)]
+#[serde(rename_all = "snake_case")]
+pub enum SourceCredentialKind {
+    BasicAuth,
+    BearerToken,
+    HttpHeaders,
+    ObjectStore,
+    HuggingFace,
+}
+
+impl SourceCredentialKind {
+    #[must_use]
+    pub const fn as_str(self) -> &'static str {
+        match self {
+            Self::BasicAuth => "basic_auth",
+            Self::BearerToken => "bearer_token",
+            Self::HttpHeaders => "http_headers",
+            Self::ObjectStore => "object_store",
+            Self::HuggingFace => "hugging_face",
+        }
+    }
+
+    pub fn parse(value: &str) -> Result<Self, super::error::CredentialError> {
+        match value {
+            "basic_auth" => Ok(Self::BasicAuth),
+            "bearer_token" => Ok(Self::BearerToken),
+            "http_headers" => Ok(Self::HttpHeaders),
+            "object_store" => Ok(Self::ObjectStore),
+            "hugging_face" => Ok(Self::HuggingFace),
+            other => Err(super::error::CredentialError::InvalidInput(format!(
+                "unknown credential kind '{other}'"
+            ))),
+        }
+    }
+}
+
+#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, PartialEq, Serialize)]
+#[serde(rename_all = "snake_case")]
+pub enum SourceCredentialStatus {
+    Active,
+    Revoked,
+}
+
+impl SourceCredentialStatus {
+    #[must_use]
+    pub const fn as_str(self) -> &'static str {
+        match self {
+            Self::Active => "active",
+            Self::Revoked => "revoked",
+        }
+    }
+
+    pub fn parse(value: &str) -> Result<Self, super::error::CredentialError> {
+        match value {
+            "active" => Ok(Self::Active),
+            "revoked" => Ok(Self::Revoked),
+            other => Err(super::error::CredentialError::InvalidInput(format!(
+                "unknown credential status '{other}'"
+            ))),
+        }
+    }
+}
+
+/// Write-only secret material. Debug and Display never include values.
+#[derive(Clone, Deserialize, Eq, PartialEq, Serialize)]
+#[serde(tag = "kind", rename_all = "snake_case")]
+pub enum SecretMaterial {
+    BasicAuth {
+        username: String,
+        password: String,
+    },
+    BearerToken {
+        token: String,
+    },
+    HttpHeaders {
+        headers: BTreeMap<String, String>,
+    },
+    ObjectStore {
+        access_key_id: String,
+        secret_access_key: String,
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        session_token: Option<String>,
+    },
+    HuggingFace {
+        token: String,
+    },
+}
+
+impl SecretMaterial {
+    #[must_use]
+    pub fn kind(&self) -> SourceCredentialKind {
+        match self {
+            Self::BasicAuth { .. } => SourceCredentialKind::BasicAuth,
+            Self::BearerToken { .. } => SourceCredentialKind::BearerToken,
+            Self::HttpHeaders { .. } => SourceCredentialKind::HttpHeaders,
+            Self::ObjectStore { .. } => SourceCredentialKind::ObjectStore,
+            Self::HuggingFace { .. } => SourceCredentialKind::HuggingFace,
+        }
+    }
+
+    pub fn validate(&self) -> Result<(), super::error::CredentialError> {
+        match self {
+            Self::BasicAuth { username, password } => {
+                require_nonempty("username", username, 1024)?;
+                require_nonempty("password", password, 16_384)?;
+            }
+            Self::BearerToken { token } | Self::HuggingFace { token } => {
+                require_nonempty("token", token, 16_384)?;
+            }
+            Self::HttpHeaders { headers } => {
+                if headers.is_empty() || headers.len() > 64 {
+                    return Err(super::error::CredentialError::InvalidInput(
+                        "headers must contain between 1 and 64 entries".into(),
+                    ));
+                }
+                for (name, value) in headers {
+                    validate_header_name(name)?;
+                    require_nonempty("header value", value, 16_384)?;
+                }
+            }
+            Self::ObjectStore {
+                access_key_id,
+                secret_access_key,
+                session_token,
+            } => {
+                require_nonempty("access_key_id", access_key_id, 1024)?;
+                require_nonempty("secret_access_key", secret_access_key, 16_384)?;
+                if let Some(token) = session_token {
+                    require_nonempty("session_token", token, 16_384)?;
+                }
+            }
+        }
+        Ok(())
+    }
+}
+
+impl fmt::Debug for SecretMaterial {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            Self::BasicAuth { .. } => formatter.write_str("SecretMaterial::BasicAuth(REDACTED)"),
+            Self::BearerToken { .. } => {
+                formatter.write_str("SecretMaterial::BearerToken(REDACTED)")
+            }
+            Self::HttpHeaders { headers } => formatter
+                .debug_struct("SecretMaterial::HttpHeaders")
+                .field("count", &headers.len())
+                .field("values", &"REDACTED")
+                .finish(),
+            Self::ObjectStore { .. } => {
+                formatter.write_str("SecretMaterial::ObjectStore(REDACTED)")
+            }
+            Self::HuggingFace { .. } => {
+                formatter.write_str("SecretMaterial::HuggingFace(REDACTED)")
+            }
+        }
+    }
+}
+
+impl fmt::Display for SecretMaterial {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(formatter, "SecretMaterial({})", self.kind().as_str())
+    }
+}
+
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct SourceCredentialVersion {
+    pub id: OpaqueId,
+    pub created_at: OffsetDateTime,
+}
+
+/// Opaque application pin carried by an immutable execution grant.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct CredentialGrantVersion {
+    pub credential_id: OpaqueId,
+    pub version_id: OpaqueId,
+}
+
+/// Metadata projection; never includes secret material or Vault coordinates.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct SourceCredential {
+    pub id: OpaqueId,
+    pub owner: OpaqueId,
+    pub label: String,
+    pub kind: SourceCredentialKind,
+    pub status: SourceCredentialStatus,
+    pub active_version: Option<SourceCredentialVersion>,
+    pub created_at: OffsetDateTime,
+    pub updated_at: OffsetDateTime,
+}
+
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct SourceCredentialPage {
+    pub items: Vec<SourceCredential>,
+    pub next_cursor: Option<String>,
+}
+
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct CreateCredential {
+    pub label: String,
+    pub material: SecretMaterial,
+}
+
+fn require_nonempty(
+    field: &str,
+    value: &str,
+    max: usize,
+) -> Result<(), super::error::CredentialError> {
+    if value.is_empty() || value.len() > max {
+        return Err(super::error::CredentialError::InvalidInput(format!(
+            "{field} length must be between 1 and {max}"
+        )));
+    }
+    Ok(())
+}
+
+fn validate_header_name(name: &str) -> Result<(), super::error::CredentialError> {
+    if name.is_empty()
+        || !name.bytes().all(|b| {
+            matches!(b, b'a'..=b'z' | b'0'..=b'9' | b'!' | b'#' | b'$' | b'%' | b'&'
+                | b'\'' | b'*' | b'+' | b'-' | b'.' | b'^' | b'_' | b'`' | b'|' | b'~')
+        })
+    {
+        return Err(super::error::CredentialError::InvalidInput(
+            "header names must be lowercase token characters".into(),
+        ));
+    }
+    const FORBIDDEN: &[&str] = &[
+        "authorization",
+        "connection",
+        "content-length",
+        "cookie",
+        "host",
+        "keep-alive",
+        "proxy-authenticate",
+        "proxy-authorization",
+        "te",
+        "trailer",
+        "transfer-encoding",
+        "upgrade",
+    ];
+    if FORBIDDEN.contains(&name) {
+        return Err(super::error::CredentialError::InvalidInput(format!(
+            "header '{name}' is not permitted"
+        )));
+    }
+    Ok(())
+}
+
+pub(crate) fn validate_label(label: &str) -> Result<(), super::error::CredentialError> {
+    if label.is_empty() || label.len() > 200 {
+        return Err(super::error::CredentialError::InvalidInput(
+            "label length must be between 1 and 200".into(),
+        ));
+    }
+    Ok(())
+}

crates/remora-credentials/src/vault.rs

@@ -0,0 +1,460 @@
+use std::fmt;
+use std::time::Duration;
+
+use remora_domain::OpaqueId;
+use reqwest::redirect::Policy;
+use reqwest::{Client, StatusCode};
+use serde::{Deserialize, Serialize};
+
+use crate::error::CredentialError;
+use crate::types::SecretMaterial;
+
+/// One non-destroyed KV v2 version observed via secret metadata.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct LiveVaultVersion {
+    pub version: u64,
+}
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub(crate) enum EmptyPathFence {
+    Created(u64),
+    Occupied,
+}
+
+/// Vault KV v2 access for credential secret material.
+///
+/// Topology and auth method are deployment choices. This client accepts a token
+/// (fixture root token or workload token after Kubernetes auth) and never logs
+/// request or response bodies that contain secret material.
+#[derive(Clone)]
+pub struct VaultConfig {
+    pub address: String,
+    pub token: String,
+    /// KV v2 mount, e.g. `secret`.
+    pub mount: String,
+    /// Path prefix under the mount where this deployment may allocate secrets.
+    pub path_prefix: String,
+    pub timeout: Duration,
+}
+
+impl fmt::Debug for VaultConfig {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        formatter
+            .debug_struct("VaultConfig")
+            .field("address", &self.address)
+            .field("mount", &self.mount)
+            .field("path_prefix", &self.path_prefix)
+            .field("timeout", &self.timeout)
+            .field("token", &"REDACTED")
+            .finish()
+    }
+}
+
+impl VaultConfig {
+    pub fn from_env() -> Result<Self, CredentialError> {
+        let address = required_env("REMORA_VAULT_ADDR")?;
+        let token = required_env("REMORA_VAULT_TOKEN")?;
+        let full_path = required_env("REMORA_VAULT_PATH")?;
+        let (mount, path_prefix) = split_vault_path(&full_path)?;
+        Ok(Self {
+            address,
+            token,
+            mount,
+            path_prefix,
+            timeout: Duration::from_secs(10),
+        })
+    }
+
+    #[must_use]
+    pub fn with_timeout(mut self, timeout: Duration) -> Self {
+        self.timeout = timeout;
+        self
+    }
+
+    /// Allocate the opaque secret path for one credential version identity.
+    #[must_use]
+    pub fn secret_path_for(&self, credential_id: &OpaqueId, version_id: &OpaqueId) -> String {
+        format!(
+            "{}/versions/{}",
+            self.credential_path_for(credential_id),
+            version_id.as_str()
+        )
+    }
+
+    #[must_use]
+    pub fn credential_path_for(&self, credential_id: &OpaqueId) -> String {
+        if self.path_prefix.is_empty() {
+            format!("credentials/{}", credential_id.as_str())
+        } else {
+            format!(
+                "{}/credentials/{}",
+                self.path_prefix.trim_matches('/'),
+                credential_id.as_str()
+            )
+        }
+    }
+}
+
+#[derive(Clone)]
+pub struct VaultClient {
+    http: Client,
+    config: VaultConfig,
+}
+
+impl fmt::Debug for VaultClient {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        formatter
+            .debug_struct("VaultClient")
+            .field("config", &self.config)
+            .finish()
+    }
+}
+
+impl VaultClient {
+    pub fn new(config: VaultConfig) -> Result<Self, CredentialError> {
+        // Disable redirects and ambient proxy discovery so X-Vault-Token and
+        // secret write bodies cannot leave the configured Vault endpoint via
+        // hop rewriting or HTTP_PROXY/HTTPS_PROXY injection.
+        let http = Client::builder()
+            .timeout(config.timeout)
+            .redirect(Policy::none())
+            .no_proxy()
+            .build()
+            .map_err(|error| {
+                CredentialError::Config(format!("failed to build vault HTTP client: {error}"))
+            })?;
+        Ok(Self { http, config })
+    }
+
+    #[must_use]
+    pub fn config(&self) -> &VaultConfig {
+        &self.config
+    }
+
+    /// Create secret material and return the immutable KV v2 version number.
+    ///
+    /// The write uses CAS 0 so an existing intent-bound path is never overwritten.
+    pub async fn put_version(
+        &self,
+        mount: &str,
+        secret_path: &str,
+        material: &SecretMaterial,
+    ) -> Result<u64, CredentialError> {
+        let url = format!(
+            "{}/v1/{}/data/{}",
+            self.config.address.trim_end_matches('/'),
+            mount.trim_matches('/'),
+            secret_path.trim_start_matches('/')
+        );
+        let body = KvWriteRequest {
+            data: material.clone(),
+            options: KvWriteOptions { cas: 0 },
+        };
+        let response = self
+            .http
+            .post(&url)
+            .header("X-Vault-Token", &self.config.token)
+            .json(&body)
+            .send()
+            .await
+            .map_err(map_transport)?;
+        let status = response.status();
+        if !status.is_success() {
+            let _ = response.bytes().await;
+            return Err(map_http_status(status, "write secret version"));
+        }
+        let parsed: KvWriteResponse = response.json().await.map_err(map_transport)?;
+        let version = parsed.data.and_then(|data| data.version).ok_or_else(|| {
+            CredentialError::Config("vault write response omitted version".into())
+        })?;
+        if version < 1 {
+            return Err(CredentialError::Config(
+                "vault returned non-positive secret version".into(),
+            ));
+        }
+        Ok(version)
+    }
+
+    pub(crate) async fn fence_empty_path(
+        &self,
+        mount: &str,
+        secret_path: &str,
+    ) -> Result<EmptyPathFence, CredentialError> {
+        let url = format!(
+            "{}/v1/{}/data/{}",
+            self.config.address.trim_end_matches('/'),
+            mount.trim_matches('/'),
+            secret_path.trim_start_matches('/')
+        );
+        let response = self
+            .http
+            .post(&url)
+            .header("X-Vault-Token", &self.config.token)
+            .json(&FenceWriteRequest {
+                data: FenceMarker {
+                    remora_write_fence: true,
+                },
+                options: KvWriteOptions { cas: 0 },
+            })
+            .send()
+            .await
+            .map_err(map_transport)?;
+        let status = response.status();
+        if status == StatusCode::BAD_REQUEST {
+            let _ = response.bytes().await;
+            return Ok(EmptyPathFence::Occupied);
+        }
+        if !status.is_success() {
+            let _ = response.bytes().await;
+            return Err(map_http_status(status, "fence empty secret path"));
+        }
+        let parsed: KvWriteResponse = response.json().await.map_err(map_transport)?;
+        let version = parsed.data.and_then(|data| data.version).ok_or_else(|| {
+            CredentialError::Config("vault fence response omitted version".into())
+        })?;
+        if version < 1 {
+            return Err(CredentialError::Config(
+                "vault returned non-positive fence version".into(),
+            ));
+        }
+        Ok(EmptyPathFence::Created(version))
+    }
+
+    /// Read one exact KV v2 version. Missing or destroyed versions fail closed.
+    pub async fn get_version(
+        &self,
+        mount: &str,
+        secret_path: &str,
+        vault_kv_version: u64,
+    ) -> Result<SecretMaterial, CredentialError> {
+        let url = format!(
+            "{}/v1/{}/data/{}?version={vault_kv_version}",
+            self.config.address.trim_end_matches('/'),
+            mount.trim_matches('/'),
+            secret_path.trim_start_matches('/')
+        );
+        let response = self
+            .http
+            .get(&url)
+            .header("X-Vault-Token", &self.config.token)
+            .send()
+            .await
+            .map_err(map_transport)?;
+        let status = response.status();
+        if status == StatusCode::NOT_FOUND {
+            let _ = response.bytes().await;
+            return Err(CredentialError::Unresolvable);
+        }
+        if !status.is_success() {
+            let _ = response.bytes().await;
+            return Err(map_http_status(status, "read secret version"));
+        }
+        let parsed: KvReadResponse = response.json().await.map_err(map_transport)?;
+        let data = parsed.data.ok_or(CredentialError::Unresolvable)?;
+        let material = data.data.ok_or(CredentialError::Unresolvable)?;
+        let reported = data
+            .metadata
+            .and_then(|meta| meta.version)
+            .ok_or(CredentialError::Unresolvable)?;
+        if reported != vault_kv_version {
+            return Err(CredentialError::Unresolvable);
+        }
+        material.validate()?;
+        Ok(material)
+    }
+
+    /// List non-destroyed KV v2 versions present at a path (for intent recovery).
+    ///
+    pub async fn list_live_versions(
+        &self,
+        mount: &str,
+        secret_path: &str,
+    ) -> Result<Vec<LiveVaultVersion>, CredentialError> {
+        let url = format!(
+            "{}/v1/{}/metadata/{}",
+            self.config.address.trim_end_matches('/'),
+            mount.trim_matches('/'),
+            secret_path.trim_start_matches('/')
+        );
+        let response = self
+            .http
+            .get(&url)
+            .header("X-Vault-Token", &self.config.token)
+            .send()
+            .await
+            .map_err(map_transport)?;
+        let status = response.status();
+        if status == StatusCode::NOT_FOUND {
+            let _ = response.bytes().await;
+            return Ok(Vec::new());
+        }
+        if !status.is_success() {
+            let _ = response.bytes().await;
+            return Err(map_http_status(status, "list secret metadata"));
+        }
+        let parsed: KvMetadataResponse = response.json().await.map_err(map_transport)?;
+        let versions = parsed
+            .data
+            .and_then(|data| data.versions)
+            .unwrap_or_default();
+        let mut live = Vec::new();
+        for (key, meta) in versions {
+            if meta.destroyed.unwrap_or(false) {
+                continue;
+            }
+            let Ok(version) = key.parse::<u64>() else {
+                continue;
+            };
+            if version < 1 {
+                continue;
+            }
+            live.push(LiveVaultVersion { version });
+        }
+        live.sort_by_key(|entry| entry.version);
+        Ok(live)
+    }
+
+    /// Irreversibly destroy one KV v2 version.
+    pub async fn destroy_version(
+        &self,
+        mount: &str,
+        secret_path: &str,
+        vault_kv_version: u64,
+    ) -> Result<(), CredentialError> {
+        let url = format!(
+            "{}/v1/{}/destroy/{}",
+            self.config.address.trim_end_matches('/'),
+            mount.trim_matches('/'),
+            secret_path.trim_start_matches('/')
+        );
+        let body = DestroyRequest {
+            versions: vec![vault_kv_version],
+        };
+        let response = self
+            .http
+            .post(&url)
+            .header("X-Vault-Token", &self.config.token)
+            .json(&body)
+            .send()
+            .await
+            .map_err(map_transport)?;
+        let status = response.status();
+        if !status.is_success() {
+            let _ = response.bytes().await;
+            return Err(map_http_status(status, "destroy secret version"));
+        }
+        let _ = response.bytes().await;
+        Ok(())
+    }
+}
+
+#[derive(Serialize)]
+struct KvWriteRequest {
+    data: SecretMaterial,
+    options: KvWriteOptions,
+}
+
+#[derive(Serialize)]
+struct KvWriteOptions {
+    cas: u8,
+}
+
+#[derive(Serialize)]
+struct FenceWriteRequest {
+    data: FenceMarker,
+    options: KvWriteOptions,
+}
+
+#[derive(Serialize)]
+struct FenceMarker {
+    remora_write_fence: bool,
+}
+
+#[derive(Deserialize)]
+struct KvWriteResponse {
+    data: Option<KvWriteData>,
+}
+
+#[derive(Deserialize)]
+struct KvWriteData {
+    version: Option<u64>,
+}
+
+#[derive(Deserialize)]
+struct KvReadResponse {
+    data: Option<KvReadData>,
+}
+
+#[derive(Deserialize)]
+struct KvReadData {
+    data: Option<SecretMaterial>,
+    metadata: Option<KvMetadata>,
+}
+
+#[derive(Deserialize)]
+struct KvMetadata {
+    version: Option<u64>,
+}
+
+#[derive(Deserialize)]
+struct KvMetadataResponse {
+    data: Option<KvMetadataListData>,
+}
+
+#[derive(Deserialize)]
+struct KvMetadataListData {
+    versions: Option<std::collections::BTreeMap<String, KvVersionMeta>>,
+}
+
+#[derive(Deserialize)]
+struct KvVersionMeta {
+    destroyed: Option<bool>,
+}
+
+#[derive(Serialize)]
+struct DestroyRequest {
+    versions: Vec<u64>,
+}
+
+fn required_env(name: &str) -> Result<String, CredentialError> {
+    std::env::var(name)
+        .ok()
+        .filter(|value| !value.is_empty())
+        .ok_or_else(|| CredentialError::Config(format!("{name} is required")))
+}
+
+/// Split a fixture path such as `secret/remora/lane-0` into mount + prefix.
+pub fn split_vault_path(full: &str) -> Result<(String, String), CredentialError> {
+    let trimmed = full.trim().trim_matches('/');
+    if trimmed.is_empty() {
+        return Err(CredentialError::Config(
+            "REMORA_VAULT_PATH must include a mount".into(),
+        ));
+    }
+    match trimmed.split_once('/') {
+        Some((mount, prefix)) if !mount.is_empty() => {
+            Ok((mount.to_owned(), prefix.trim_matches('/').to_owned()))
+        }
+        Some(_) | None => Ok((trimmed.to_owned(), String::new())),
+    }
+}
+
+fn map_transport(_error: reqwest::Error) -> CredentialError {
+    CredentialError::SecretStoreUnavailable
+}
+
+fn map_http_status(status: StatusCode, operation: &str) -> CredentialError {
+    if status.is_server_error() || status == StatusCode::SERVICE_UNAVAILABLE {
+        CredentialError::SecretStoreUnavailable
+    } else if status == StatusCode::FORBIDDEN || status == StatusCode::UNAUTHORIZED {
+        // Treat policy denials as unresolvable rather than leaking store details.
+        CredentialError::Unresolvable
+    } else if status == StatusCode::NOT_FOUND {
+        CredentialError::Unresolvable
+    } else {
+        CredentialError::Config(format!(
+            "vault {operation} failed with HTTP {}",
+            status.as_u16()
+        ))
+    }
+}

crates/remora-credentials/tests/policy.rs

@@ -0,0 +1,74 @@
+use remora_credentials::{
+    CredentialCaller, SourceCredential, SourceCredentialKind, SourceCredentialStatus, may_manage,
+    require_manage,
+};
+use remora_domain::OpaqueId;
+use time::OffsetDateTime;
+
+fn principal(label: &str) -> OpaqueId {
+    OpaqueId::parse(format!("principal-{label}-aaaa")).unwrap()
+}
+
+fn credential(owner: OpaqueId) -> SourceCredential {
+    SourceCredential {
+        id: OpaqueId::parse("credid01-aaaaaaaa").unwrap(),
+        owner,
+        label: "probe".into(),
+        kind: SourceCredentialKind::BearerToken,
+        status: SourceCredentialStatus::Active,
+        active_version: None,
+        created_at: OffsetDateTime::now_utc(),
+        updated_at: OffsetDateTime::now_utc(),
+    }
+}
+
+#[test]
+fn owner_and_admin_may_manage_peers_may_not() {
+    let owner = principal("owner");
+    let peer = principal("peer0");
+    let admin = CredentialCaller {
+        principal_id: principal("admin"),
+        administrator: true,
+    };
+    let owner_caller = CredentialCaller {
+        principal_id: owner.clone(),
+        administrator: false,
+    };
+    let peer_caller = CredentialCaller {
+        principal_id: peer,
+        administrator: false,
+    };
+
+    assert!(may_manage(&owner_caller, &owner));
+    assert!(may_manage(&admin, &owner));
+    assert!(!may_manage(&peer_caller, &owner));
+}
+
+#[test]
+fn missing_and_unauthorized_are_indistinguishable() {
+    let owner = principal("owner");
+    let peer = CredentialCaller {
+        principal_id: principal("peer1"),
+        administrator: false,
+    };
+    let record = credential(owner);
+
+    let missing = require_manage(&peer, None);
+    let unauthorized = require_manage(&peer, Some(&record));
+    assert!(matches!(
+        missing,
+        Err(remora_credentials::CredentialError::NotFound)
+    ));
+    assert!(matches!(
+        unauthorized,
+        Err(remora_credentials::CredentialError::NotFound)
+    ));
+}
+
+#[test]
+fn split_vault_path_parses_fixture_shape() {
+    let (mount, prefix) =
+        remora_credentials::split_vault_path("secret/remora/lane-0").expect("path");
+    assert_eq!(mount, "secret");
+    assert_eq!(prefix, "remora/lane-0");
+}

crates/remora-credentials/tests/redaction.rs

@@ -0,0 +1,65 @@
+use std::collections::BTreeMap;
+
+use remora_credentials::{SecretMaterial, VaultConfig};
+
+#[test]
+fn secret_material_debug_and_display_redact_values() {
+    let material = SecretMaterial::BearerToken {
+        token: "super-secret-token-value".into(),
+    };
+    let debug = format!("{material:?}");
+    let display = format!("{material}");
+    assert!(!debug.contains("super-secret-token-value"));
+    assert!(debug.contains("REDACTED"));
+    assert!(!display.contains("super-secret-token-value"));
+    assert_eq!(display, "SecretMaterial(bearer_token)");
+
+    let headers = SecretMaterial::HttpHeaders {
+        headers: BTreeMap::from([("x-api-key".into(), "header-secret".into())]),
+    };
+    let debug = format!("{headers:?}");
+    assert!(!debug.contains("header-secret"));
+    assert!(debug.contains("REDACTED"));
+}
+
+#[test]
+fn vault_config_debug_redacts_token() {
+    let config = VaultConfig {
+        address: "http://vault:8200".into(),
+        token: "synthetic-vault-token".into(),
+        mount: "secret".into(),
+        path_prefix: "remora/lane-0".into(),
+        timeout: std::time::Duration::from_secs(1),
+    };
+    let debug = format!("{config:?}");
+    assert!(!debug.contains("synthetic-vault-token"));
+    assert!(debug.contains("REDACTED"));
+}
+
+#[test]
+fn error_messages_never_embed_secret_literals() {
+    let material = SecretMaterial::BasicAuth {
+        username: "alice".into(),
+        password: "p@ssw0rd-should-not-leak".into(),
+    };
+    let err = remora_credentials::CredentialError::InvalidInput(format!(
+        "validated kind {}",
+        material.kind().as_str()
+    ));
+    let text = err.to_string();
+    assert!(!text.contains("p@ssw0rd-should-not-leak"));
+    assert!(err.is_safe_for_logs());
+}
+
+#[test]
+fn material_validation_rejects_empty_and_forbidden_headers() {
+    let empty = SecretMaterial::BearerToken {
+        token: String::new(),
+    };
+    assert!(empty.validate().is_err());
+
+    let forbidden = SecretMaterial::HttpHeaders {
+        headers: BTreeMap::from([("authorization".into(), "Bearer x".into())]),
+    };
+    assert!(forbidden.validate().is_err());
+}

crates/remora-credentials/tests/vault_transport.rs

@@ -0,0 +1,225 @@
+use std::io;
+use std::net::SocketAddr;
+use std::sync::{Arc, Mutex};
+use std::time::Duration;
+
+use remora_credentials::{SecretMaterial, VaultClient, VaultConfig};
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio::net::TcpListener;
+
+/// Vault client must not follow redirects that would re-issue X-Vault-Token
+/// (and secret write bodies) to an alternate origin.
+#[tokio::test]
+async fn vault_client_does_not_follow_redirects_with_token() {
+    let evil_hits: Arc<Mutex<usize>> = Arc::new(Mutex::new(0));
+    let evil_hits_bg = Arc::clone(&evil_hits);
+    let evil = TcpListener::bind("127.0.0.1:0").await.expect("evil bind");
+    let evil_addr = evil.local_addr().expect("evil addr");
+    tokio::spawn(async move {
+        if let Ok((mut stream, _)) = evil.accept().await {
+            let mut buf = vec![0u8; 4096];
+            let _ = stream.read(&mut buf).await;
+            *evil_hits_bg.lock().expect("lock") += 1;
+            let _ = stream
+                .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n")
+                .await;
+        }
+    });
+
+    let redirector = TcpListener::bind("127.0.0.1:0")
+        .await
+        .expect("redirector bind");
+    let redirector_addr = redirector.local_addr().expect("redirector addr");
+    tokio::spawn(async move {
+        if let Ok((mut stream, _)) = redirector.accept().await {
+            let mut buf = vec![0u8; 8192];
+            let _ = read_http_request(&mut stream, &mut buf).await;
+            let location = format!("http://{evil_addr}/v1/secret/data/stolen");
+            let response =
+                format!("HTTP/1.1 302 Found\r\nLocation: {location}\r\nContent-Length: 0\r\n\r\n");
+            let _ = stream.write_all(response.as_bytes()).await;
+        }
+    });
+
+    let client = VaultClient::new(VaultConfig {
+        address: format!("http://{redirector_addr}"),
+        token: "synthetic-vault-token-must-not-exfiltrate".into(),
+        mount: "secret".into(),
+        path_prefix: "remora".into(),
+        timeout: Duration::from_secs(2),
+    })
+    .expect("client");
+
+    let result = client
+        .put_version(
+            "secret",
+            "remora/credentials/probe",
+            &SecretMaterial::BearerToken {
+                token: "secret-body-must-not-exfiltrate".into(),
+            },
+        )
+        .await;
+
+    assert!(
+        result.is_err(),
+        "redirect response must fail closed, got {result:?}"
+    );
+    // Give the evil server a moment if a follow were attempted.
+    tokio::time::sleep(Duration::from_millis(50)).await;
+    assert_eq!(
+        *evil_hits.lock().expect("lock"),
+        0,
+        "Vault client followed a redirect and contacted the evil origin"
+    );
+}
+
+/// Ambient HTTP(S)_PROXY must not be consulted for Vault traffic.
+#[tokio::test]
+async fn vault_client_ignores_ambient_proxy_env() {
+    // Safety: test process only; restores prior values at end of scope via scopeguard pattern.
+    let prior_http = std::env::var("HTTP_PROXY").ok();
+    let prior_https = std::env::var("HTTPS_PROXY").ok();
+    let prior_all = std::env::var("ALL_PROXY").ok();
+
+    let proxy_hits: Arc<Mutex<usize>> = Arc::new(Mutex::new(0));
+    let proxy_hits_bg = Arc::clone(&proxy_hits);
+    let proxy = TcpListener::bind("127.0.0.1:0").await.expect("proxy bind");
+    let proxy_addr = proxy.local_addr().expect("proxy addr");
+    tokio::spawn(async move {
+        if let Ok((mut stream, _)) = proxy.accept().await {
+            let mut buf = vec![0u8; 4096];
+            let _ = stream.read(&mut buf).await;
+            *proxy_hits_bg.lock().expect("lock") += 1;
+            let _ = stream.shutdown().await;
+        }
+    });
+
+    // Point proxies at a listener that records any CONNECT/request attempt.
+    // SAFETY: single-threaded test mutation of process env for this test only.
+    unsafe {
+        std::env::set_var("HTTP_PROXY", format!("http://{proxy_addr}"));
+        std::env::set_var("HTTPS_PROXY", format!("http://{proxy_addr}"));
+        std::env::set_var("ALL_PROXY", format!("http://{proxy_addr}"));
+    }
+
+    let vault = TcpListener::bind("127.0.0.1:0").await.expect("vault bind");
+    let vault_addr = vault.local_addr().expect("vault addr");
+    tokio::spawn(async move {
+        if let Ok((mut stream, _)) = vault.accept().await {
+            let mut buf = vec![0u8; 8192];
+            let _ = read_http_request(&mut stream, &mut buf).await;
+            let body = r#"{"data":{"version":1}}"#;
+            let response = format!(
+                "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\n\r\n{body}",
+                body.len()
+            );
+            let _ = stream.write_all(response.as_bytes()).await;
+        }
+    });
+
+    let client = VaultClient::new(VaultConfig {
+        address: format!("http://{vault_addr}"),
+        token: "synthetic-token".into(),
+        mount: "secret".into(),
+        path_prefix: "remora".into(),
+        timeout: Duration::from_secs(2),
+    })
+    .expect("client");
+
+    let version = client
+        .put_version(
+            "secret",
+            "remora/credentials/probe",
+            &SecretMaterial::BearerToken {
+                token: "secret-value".into(),
+            },
+        )
+        .await
+        .expect("direct vault write without proxy");
+    assert_eq!(version, 1);
+
+    tokio::time::sleep(Duration::from_millis(50)).await;
+    assert_eq!(
+        *proxy_hits.lock().expect("lock"),
+        0,
+        "Vault client used ambient proxy environment"
+    );
+
+    restore_env("HTTP_PROXY", prior_http);
+    restore_env("HTTPS_PROXY", prior_https);
+    restore_env("ALL_PROXY", prior_all);
+}
+
+#[tokio::test]
+async fn exact_read_rejects_missing_or_mismatched_version_metadata() {
+    for metadata in ["{}", r#"{"version":2}"#] {
+        let vault = TcpListener::bind("127.0.0.1:0").await.expect("vault bind");
+        let vault_addr = vault.local_addr().expect("vault addr");
+        let body = format!(
+            r#"{{"data":{{"data":{{"kind":"bearer_token","token":"synthetic"}},"metadata":{metadata}}}}}"#
+        );
+        tokio::spawn(async move {
+            if let Ok((mut stream, _)) = vault.accept().await {
+                let mut buf = vec![0u8; 8192];
+                let _ = read_http_request(&mut stream, &mut buf).await;
+                let response = format!(
+                    "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\n\r\n{body}",
+                    body.len()
+                );
+                let _ = stream.write_all(response.as_bytes()).await;
+            }
+        });
+
+        let client = VaultClient::new(VaultConfig {
+            address: format!("http://{vault_addr}"),
+            token: "synthetic-token".into(),
+            mount: "secret".into(),
+            path_prefix: "remora".into(),
+            timeout: Duration::from_secs(2),
+        })
+        .expect("client");
+
+        assert!(matches!(
+            client
+                .get_version("secret", "remora/credentials/probe", 1)
+                .await,
+            Err(remora_credentials::CredentialError::Unresolvable)
+        ));
+    }
+}
+
+fn restore_env(key: &str, prior: Option<String>) {
+    // SAFETY: restoring process env after test mutation.
+    unsafe {
+        match prior {
+            Some(value) => std::env::set_var(key, value),
+            None => std::env::remove_var(key),
+        }
+    }
+}
+
+async fn read_http_request(
+    stream: &mut tokio::net::TcpStream,
+    buf: &mut [u8],
+) -> io::Result<usize> {
+    let mut total = 0;
+    loop {
+        let n = stream.read(&mut buf[total..]).await?;
+        if n == 0 {
+            break;
+        }
+        total += n;
+        if buf[..total].windows(4).any(|w| w == b"\r\n\r\n") {
+            break;
+        }
+        if total == buf.len() {
+            break;
+        }
+    }
+    Ok(total)
+}
+
+#[allow(dead_code)]
+fn _socket_addr_fmt(addr: SocketAddr) -> String {
+    addr.to_string()
+}

crates/remora-domain/src/ports.rs

@@ -42,14 +42,40 @@ pub trait IdentityProvider: Send + Sync {
     ) -> impl Future<Output = Result<Self::Principal, Self::Error>> + Send;
 }
 
+/// Resolve handle for one execution grant's pinned credential.
+///
+/// Carries only the grant identity. Resolvers must re-validate durable grant
+/// activity on every resolve; this type is not a substitute for that check and
+/// does not embed credential or version IDs that could be forged independently
+/// of grant state.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct ActiveCredentialGrant {
+    grant_id: OpaqueId,
+}
+
+impl ActiveCredentialGrant {
+    /// Wrap an execution-grant identity for resolution.
+    ///
+    /// Security is enforced at resolve time against durable grant pin state, not
+    /// at construction. Unknown, revoked, or mismatched grants fail closed.
+    #[must_use]
+    pub fn for_execution_grant(grant_id: OpaqueId) -> Self {
+        Self { grant_id }
+    }
+
+    #[must_use]
+    pub fn grant_id(&self) -> &OpaqueId {
+        &self.grant_id
+    }
+}
+
 pub trait CredentialResolver: Send + Sync {
     type Error;
     type Secret;
 
     fn resolve(
         &self,
-        credential_id: &OpaqueId,
-        version_id: &OpaqueId,
+        access: &ActiveCredentialGrant,
     ) -> impl Future<Output = Result<Self::Secret, Self::Error>> + Send;
 }