Bind credential pins to immutable grants

crates/remora-admission/migrations/0001_admission.sql

@@ -24,13 +24,24 @@ CREATE TABLE execution_grants (
     CONSTRAINT execution_grants_credential_check CHECK (
         (credential_id IS NULL) = (credential_version_id IS NULL)
     ),
-    CONSTRAINT execution_grants_id_status_unique UNIQUE (grant_id, status)
+    CONSTRAINT execution_grants_authority_tuple_unique UNIQUE (
+        grant_id,
+        status,
+        credential_id,
+        credential_version_id
+    )
 );
 
 ALTER TABLE execution_grant_credential_pins
     ADD CONSTRAINT execution_grant_pins_authoritative_grant_fk
-    FOREIGN KEY (grant_id, status)
+    FOREIGN KEY (grant_id, status, credential_id, version_id)
-    REFERENCES execution_grants (grant_id, status)
+    REFERENCES execution_grants (
+        grant_id,
+        status,
+        credential_id,
+        credential_version_id
+    )
+    MATCH FULL
     DEFERRABLE INITIALLY DEFERRED;
 
 CREATE TABLE jobs (

crates/remora-admission/tests/substrate.rs

@@ -6,8 +6,13 @@ use remora_admission::{
 use remora_contracts::{
     HttpSourceSubmissionDto, JobSubmissionDto, SourceSubmissionDto, TargetOverrideDto,
 };
-use remora_credentials::{CredentialRegistry, migrate as credentials_migrate};
+use remora_credentials::{
-use remora_domain::{BucketName, OpaqueId, StorageScope, TargetPrefix, TargetStoreId};
+    CredentialError, CredentialGrantVersion, CredentialRegistry, GrantBoundResolver, VaultClient,
+    VaultConfig, migrate as credentials_migrate,
+};
+use remora_domain::{
+    BucketName, OpaqueId, StorageScope, TargetPrefix, TargetStoreId, ports::ActiveCredentialGrant,
+};
 use remora_durable::{MutationDisposition, PoolConfig, connect, migrate as durable_migrate};
 use remora_identity::{DefaultTarget, Principal};
 use sqlx::PgPool;
@@ -116,6 +121,65 @@ fn accepted(outcome: SubmitOutcome) -> remora_admission::AcceptedJob {
     }
 }
 
+fn registry(pool: PgPool) -> CredentialRegistry {
+    registry_and_resolver(pool).0
+}
+
+fn registry_and_resolver(pool: PgPool) -> (CredentialRegistry, GrantBoundResolver) {
+    let vault = VaultClient::new(VaultConfig {
+        address: "http://127.0.0.1:1".into(),
+        token: "synthetic-test-token".into(),
+        mount: "secret".into(),
+        path_prefix: "admission-tests".into(),
+        timeout: Duration::from_millis(100),
+    })
+    .unwrap();
+    let registry = CredentialRegistry::new(pool, vault.clone());
+    let resolver = GrantBoundResolver::new(registry.clone(), vault);
+    (registry, resolver)
+}
+
+async fn insert_credential(
+    pool: &PgPool,
+    owner_id: &OpaqueId,
+    kind: &str,
+) -> CredentialGrantVersion {
+    let credential_id = unique_id("cred");
+    let version_id = unique_id("version");
+    let now = time::OffsetDateTime::now_utc();
+    let mut tx = pool.begin().await.unwrap();
+    sqlx::query(
+        "INSERT INTO source_credentials (
+            credential_id, owner_id, label, kind, status, active_version_id,
+            secret_mount, secret_path, created_at, updated_at
+         ) VALUES ($1, $2, 'fixture', $3, 'active', $4, 'secret', 'fixture', $5, $5)",
+    )
+    .bind(credential_id.as_str())
+    .bind(owner_id.as_str())
+    .bind(kind)
+    .bind(version_id.as_str())
+    .bind(now)
+    .execute(&mut *tx)
+    .await
+    .unwrap();
+    sqlx::query(
+        "INSERT INTO source_credential_versions (
+            version_id, credential_id, vault_kv_version, created_at, destroyed_at, secret_mount, secret_path
+         ) VALUES ($1, $2, 1, $3, NULL, 'secret', 'fixture/version')",
+    )
+    .bind(version_id.as_str())
+    .bind(credential_id.as_str())
+    .bind(now)
+    .execute(&mut *tx)
+    .await
+    .unwrap();
+    tx.commit().await.unwrap();
+    CredentialGrantVersion {
+        credential_id,
+        version_id,
+    }
+}
+
 #[tokio::test]
 async fn exact_replay_and_lost_response_recovery_create_one_durable_acceptance() {
     let Some(pool) = pool().await else { return };
@@ -248,35 +312,9 @@ async fn credential_pin_and_accepted_grant_revoke_atomically_without_reactivatio
     let Some(pool) = pool().await else { return };
     let service = service(pool.clone());
     let owner_id = unique_id("owner");
-    let credential_id = unique_id("cred");
+    let pin = insert_credential(&pool, &owner_id, "bearer_token").await;
-    let version_id = unique_id("version");
+    let credential_id = pin.credential_id.clone();
-    let now = time::OffsetDateTime::now_utc();
+    let version_id = pin.version_id.clone();
-    let mut tx = pool.begin().await.unwrap();
-    sqlx::query(
-        "INSERT INTO source_credentials (
-            credential_id, owner_id, label, kind, status, active_version_id,
-            secret_mount, secret_path, created_at, updated_at
-         ) VALUES ($1, $2, 'fixture', 'bearer_token', 'active', $3, 'secret', 'fixture', $4, $4)",
-    )
-    .bind(credential_id.as_str())
-    .bind(owner_id.as_str())
-    .bind(version_id.as_str())
-    .bind(now)
-    .execute(&mut *tx)
-    .await
-    .unwrap();
-    sqlx::query(
-        "INSERT INTO source_credential_versions (
-            version_id, credential_id, vault_kv_version, created_at, destroyed_at, secret_mount, secret_path
-         ) VALUES ($1, $2, 1, $3, NULL, 'secret', 'fixture/version')",
-    )
-    .bind(version_id.as_str())
-    .bind(credential_id.as_str())
-    .bind(now)
-    .execute(&mut *tx)
-    .await
-    .unwrap();
-    tx.commit().await.unwrap();
 
     let mut request = request("https://downloads.example/private".into());
     let SourceSubmissionDto::Http(source) = &mut request.source else {
@@ -330,10 +368,6 @@ async fn credential_pin_and_accepted_grant_revoke_atomically_without_reactivatio
     assert_eq!(statuses, ("revoked".into(), "revoked".into(), 1));
 
     let mut tx = pool.begin().await.unwrap();
-    let pin = remora_credentials::CredentialGrantVersion {
-        credential_id,
-        version_id,
-    };
     assert!(
         CredentialRegistry::bind_execution_grant_in_tx(
             &mut tx,
@@ -346,3 +380,198 @@ async fn credential_pin_and_accepted_grant_revoke_atomically_without_reactivatio
     );
     tx.rollback().await.unwrap();
 }
+
+#[tokio::test]
+async fn authoritative_credential_tuple_rejects_both_binding_substitution_paths() {
+    let Some(pool) = pool().await else { return };
+    let service = service(pool.clone());
+    let owner_id = unique_id("owner");
+    let owner = principal(owner_id.clone(), true, false);
+    let arbitrary_pin = insert_credential(&pool, &owner_id, "bearer_token").await;
+    let credentialless = accepted(
+        service
+            .submit(
+                &owner,
+                unique_id("key").as_str(),
+                request("https://downloads.example/public".into()),
+            )
+            .await
+            .unwrap(),
+    );
+    let (registry, resolver) = registry_and_resolver(pool.clone());
+
+    assert!(
+        registry
+            .bind_execution_grant(&credentialless.execution_grant_id, &arbitrary_pin)
+            .await
+            .is_err()
+    );
+    assert_eq!(
+        registry
+            .load_active_grant_pin(&credentialless.execution_grant_id)
+            .await
+            .unwrap(),
+        None
+    );
+    let access =
+        ActiveCredentialGrant::for_execution_grant(credentialless.execution_grant_id.clone());
+    assert!(matches!(
+        resolver.resolve_access(&access).await,
+        Err(CredentialError::GrantInactive)
+    ));
+
+    let mut tx = pool.begin().await.unwrap();
+    CredentialRegistry::bind_execution_grant_in_tx(
+        &mut tx,
+        &credentialless.execution_grant_id,
+        &arbitrary_pin,
+        time::OffsetDateTime::now_utc(),
+    )
+    .await
+    .unwrap();
+    assert!(tx.commit().await.is_err());
+    assert_eq!(
+        registry
+            .load_active_grant_pin(&credentialless.execution_grant_id)
+            .await
+            .unwrap(),
+        None
+    );
+
+    let accepted_pin = insert_credential(&pool, &owner_id, "bearer_token").await;
+    let substitute_pin = insert_credential(&pool, &owner_id, "bearer_token").await;
+    let mut credentialed_request = request("https://downloads.example/private".into());
+    let SourceSubmissionDto::Http(source) = &mut credentialed_request.source else {
+        unreachable!()
+    };
+    source.credential_id = Some(accepted_pin.credential_id.clone());
+    let credentialed = accepted(
+        service
+            .submit(&owner, unique_id("key").as_str(), credentialed_request)
+            .await
+            .unwrap(),
+    );
+    assert_eq!(
+        registry
+            .load_active_grant_pin(&credentialed.execution_grant_id)
+            .await
+            .unwrap(),
+        Some(accepted_pin.clone())
+    );
+    assert!(
+        registry
+            .bind_execution_grant(&credentialed.execution_grant_id, &accepted_pin)
+            .await
+            .is_ok()
+    );
+    assert!(
+        registry
+            .bind_execution_grant(&credentialed.execution_grant_id, &substitute_pin)
+            .await
+            .is_err()
+    );
+    assert_eq!(
+        registry
+            .load_active_grant_pin(&credentialed.execution_grant_id)
+            .await
+            .unwrap(),
+        Some(accepted_pin)
+    );
+}
+
+#[tokio::test]
+async fn admission_transaction_is_invisible_to_concurrent_revoke_until_binding_commits() {
+    let Some(pool) = pool().await else { return };
+    let service = service(pool.clone());
+    let owner_id = unique_id("owner");
+    let pin = insert_credential(&pool, &owner_id, "bearer_token").await;
+    let job_id = unique_id("job");
+    let grant_id = unique_id("grant");
+    let sensitive_source_ref = unique_id("source");
+    let now = time::OffsetDateTime::now_utc();
+    let mut admission_tx = pool.begin().await.unwrap();
+    sqlx::query(
+        "INSERT INTO sensitive_source_inputs (sensitive_source_ref, source_input, created_at)
+         VALUES ($1, '{\"kind\":\"http\"}'::jsonb, $2)",
+    )
+    .bind(sensitive_source_ref.as_str())
+    .bind(now)
+    .execute(&mut *admission_tx)
+    .await
+    .unwrap();
+    sqlx::query(
+        "INSERT INTO execution_grants (
+            grant_id, owner_id, target_store, target_bucket, target_prefix,
+            policy_version, credential_id, credential_version_id, status, created_at, revoked_at
+         ) VALUES ($1, $2, 'primary', 'remora-results', '[\"users\",\"alice\"]'::jsonb,
+                   'source-policy-v1', $3, $4, 'active', $5, NULL)",
+    )
+    .bind(grant_id.as_str())
+    .bind(owner_id.as_str())
+    .bind(pin.credential_id.as_str())
+    .bind(pin.version_id.as_str())
+    .bind(now)
+    .execute(&mut *admission_tx)
+    .await
+    .unwrap();
+    sqlx::query(
+        "INSERT INTO jobs (
+            job_id, owner_id, label, revision, phase, lifecycle, source_projection,
+            sensitive_source_ref, target_store, target_bucket, target_prefix,
+            execution_grant_id, dispatch_needed, created_at, updated_at
+         ) VALUES ($1, $2, NULL, 1, 'queued', 'visible', '{\"kind\":\"http\"}'::jsonb,
+                   $3, 'primary', 'remora-results', '[\"users\",\"alice\"]'::jsonb,
+                   $4, TRUE, $5, $5)",
+    )
+    .bind(job_id.as_str())
+    .bind(owner_id.as_str())
+    .bind(sensitive_source_ref.as_str())
+    .bind(grant_id.as_str())
+    .bind(now)
+    .execute(&mut *admission_tx)
+    .await
+    .unwrap();
+    CredentialRegistry::bind_execution_grant_in_tx(&mut admission_tx, &grant_id, &pin, now)
+        .await
+        .unwrap();
+
+    let invisible: (i64, i64) = sqlx::query_as(
+        "SELECT
+            (SELECT count(*) FROM execution_grants WHERE grant_id = $1),
+            (SELECT count(*) FROM execution_grant_credential_pins WHERE grant_id = $1)",
+    )
+    .bind(grant_id.as_str())
+    .fetch_one(&pool)
+    .await
+    .unwrap();
+    assert_eq!(invisible, (0, 0));
+    let admin = principal(unique_id("admin"), false, true);
+    assert!(matches!(
+        service.revoke_execution_grant(&admin, &grant_id).await,
+        Err(AdmissionError::GrantNotFound)
+    ));
+
+    admission_tx.commit().await.unwrap();
+    assert_eq!(
+        registry(pool.clone())
+            .load_active_grant_pin(&grant_id)
+            .await
+            .unwrap(),
+        Some(pin)
+    );
+    service
+        .revoke_execution_grant(&admin, &grant_id)
+        .await
+        .unwrap();
+    service
+        .revoke_execution_grant(&admin, &grant_id)
+        .await
+        .unwrap();
+    assert_eq!(
+        registry(pool)
+            .load_active_grant_pin(&grant_id)
+            .await
+            .unwrap(),
+        None
+    );
+}