feat!: make products safe to compose under deferred task grants, fresh grant status, and original receipts

.github/workflows/ci.yml

@@ -245,7 +245,7 @@ jobs:
     # which fail rather than skip without their disposable database.
     services:
       postgres:
-        image: postgres:17.11@sha256:67f41722b7a8cbdb868a44a4995c846eddfdc2973bccb291ce937dce88ad5675
+        image: public.ecr.aws/docker/library/postgres:17.11@sha256:67f41722b7a8cbdb868a44a4995c846eddfdc2973bccb291ce937dce88ad5675
         env:
           POSTGRES_DB: dispatch
           POSTGRES_USER: dispatch
@@ -760,7 +760,7 @@ jobs:
     timeout-minutes: 30
     services:
       postgres:
-        image: postgres:17.11@sha256:67f41722b7a8cbdb868a44a4995c846eddfdc2973bccb291ce937dce88ad5675
+        image: public.ecr.aws/docker/library/postgres:17.11@sha256:67f41722b7a8cbdb868a44a4995c846eddfdc2973bccb291ce937dce88ad5675
         env:
           POSTGRES_DB: casework
           POSTGRES_USER: casework
@@ -849,11 +849,11 @@ jobs:
           docker exec "$POSTGRES_CONTAINER_ID" createdb -U casework casework_source_retention
           docker exec "$POSTGRES_CONTAINER_ID" createdb -U casework casework_review
           docker exec "$POSTGRES_CONTAINER_ID" createdb -U casework casework_review_migration
-      - name: Verify assignment and source routing
+      - name: Verify assignment, source routing, and task-grant invalidation
         env:
           CASEWORK_ASSIGNMENT_TEST_DATABASE_URL: postgresql://casework:casework_test@localhost:${{ job.services.postgres.ports['5432'] }}/casework_review
           CASEWORK_ROUTING_TEST_DATABASE_URL: postgresql://casework:casework_test@localhost:${{ job.services.postgres.ports['5432'] }}/casework_review
-        run: cargo test --locked -p registry-casework --features postgres-test --test assignment_postgres --test routing_postgres
+        run: cargo test --locked -p registry-casework --features postgres-test --test assignment_postgres --test routing_postgres --test task_grant_invalidation
       - name: Verify persisted source clocks, restart, and recompute
         # An inexact name filter that selects nothing still exits 0, so the
         # step names the test exactly and fails unless it reports one pass.
@@ -957,7 +957,7 @@ jobs:
     timeout-minutes: 30
     services:
       postgres:
-        image: postgres:17.11@sha256:67f41722b7a8cbdb868a44a4995c846eddfdc2973bccb291ce937dce88ad5675
+        image: public.ecr.aws/docker/library/postgres:17.11@sha256:67f41722b7a8cbdb868a44a4995c846eddfdc2973bccb291ce937dce88ad5675
         env:
           POSTGRES_DB: scheduling
           POSTGRES_USER: scheduling
@@ -1091,7 +1091,7 @@ jobs:
     timeout-minutes: 30
     services:
       postgres:
-        image: postgres:17.11@sha256:67f41722b7a8cbdb868a44a4995c846eddfdc2973bccb291ce937dce88ad5675
+        image: public.ecr.aws/docker/library/postgres:17.11@sha256:67f41722b7a8cbdb868a44a4995c846eddfdc2973bccb291ce937dce88ad5675
         env:
           POSTGRES_DB: messaging
           POSTGRES_USER: messaging
@@ -1167,7 +1167,7 @@ jobs:
     timeout-minutes: 20
     services:
       mailpit:
-        image: axllent/mailpit:v1.31.2@sha256:74d609a42ec279aa63c6b4622a6fa9b5408d1ad5b1d76a1c4be40a265ce0863d
+        image: ghcr.io/axllent/mailpit:v1.31.2@sha256:74d609a42ec279aa63c6b4622a6fa9b5408d1ad5b1d76a1c4be40a265ce0863d
         options: >-
           --health-cmd "/mailpit readyz"
           --health-interval 5s
@@ -2068,7 +2068,7 @@ jobs:
     timeout-minutes: 30
     services:
       postgres:
-        image: postgres:17.11@sha256:67f41722b7a8cbdb868a44a4995c846eddfdc2973bccb291ce937dce88ad5675
+        image: public.ecr.aws/docker/library/postgres:17.11@sha256:67f41722b7a8cbdb868a44a4995c846eddfdc2973bccb291ce937dce88ad5675
         env:
           POSTGRES_DB: casework
           POSTGRES_USER: casework

.github/workflows/nightly-rust-coverage.yml

@@ -60,7 +60,7 @@ jobs:
     # ci.yml, which .github/scripts/test_ci_changes.py holds.
     services:
       postgres:
-        image: postgres:17.11@sha256:67f41722b7a8cbdb868a44a4995c846eddfdc2973bccb291ce937dce88ad5675
+        image: public.ecr.aws/docker/library/postgres:17.11@sha256:67f41722b7a8cbdb868a44a4995c846eddfdc2973bccb291ce937dce88ad5675
         env:
           POSTGRES_DB: dispatch
           POSTGRES_USER: dispatch

AGENTS.md

@@ -209,7 +209,11 @@ assertion as an ordinary relying party through `registry-evidence-client` and
 already does. Adding such a crate relaxes this sentence here and in the gate,
 in the same change; until then the sentence stands as written and no local
 exception may be taken to it. The runtime, the source-neutral core, the client,
-and adopter tooling stay on the strict side of it in every case.
+and adopter tooling stay on the strict side of it in every case. Checking a
+task grant's current status at the operator-configured task authority is not
+such a reach: the runtime calls that endpoint over HTTP through the
+authority-neutral status client in `registry-platform-oidc`, a protocol call
+and not a crate dependency, so the gate is unchanged by it.
 
 Registry Messaging is implemented by `registry-messaging`,
 `registry-messaging-core`, `registry-messagingctl`, and its Rust client crate.

crates/registry-breg/src/runtime_config.rs

@@ -17,7 +17,8 @@ use registry_platform_audit::{AuditDestination, AuditDestinationKind, AuditProfi
 use registry_platform_config::{
     AuditKeyConfig, JwksSource, ListenerBind, OidcIssuerConfig,
     PackageConfig as SharedPackageConfig, RemovedKey, RuntimeConfigLoader, RuntimeEnvelope,
-    SecretError, SecretReference, SecretResolver, REMOVED_OIDC_JWKS_SOURCE_KIND,
+    SecretError, SecretReference, SecretResolver, TaskGrantStatusConfig,
+    REMOVED_OIDC_JWKS_SOURCE_KIND,
 };
 use registry_platform_crypto::{parse_json_strict, PublicJwk, SigningAlgorithm};
 #[cfg(feature = "schema")]
@@ -264,6 +265,8 @@ pub enum RuntimeConfigError {
     Reader(ReaderRefusal),
     #[error("runtime configuration contains an invalid deployment binding")]
     InvalidBinding,
+    #[error("runtime configuration taskGrantStatus must contain at most 32 unique absolute source issuers; configure safe service/token URLs, non-empty client and assertion audience, absolute resource identifiers, one status scope token, and a status path with one {{grantId}} segment")]
+    InvalidTaskGrantStatus,
     #[error(
         "runtime configuration identity.instanceId must start with a lowercase letter and hold at most 64 lowercase letters, digits, `-`, or `_`; set identity.instanceId to such a value"
     )]
@@ -324,6 +327,7 @@ impl RuntimeConfigError {
         match self {
             Self::Reader(refusal) => &refusal.deciding_diagnostic().code,
             Self::InvalidBinding => "breg.runtime.invalid-binding",
+            Self::InvalidTaskGrantStatus => "breg.runtime.invalid-task-grant-status",
             Self::InvalidInstanceId => "breg.runtime.invalid-instance-id",
             Self::EnvironmentIdentityConflict => "breg.runtime.environment-identity-conflict",
             Self::InvalidListener => "breg.runtime.invalid-listener",
@@ -356,6 +360,7 @@ impl RuntimeConfigError {
         match self {
             Self::Reader(refusal) => &refusal.deciding_diagnostic().path,
             Self::InvalidBinding | Self::Secret => "",
+            Self::InvalidTaskGrantStatus => "/taskGrantStatus",
             Self::InvalidInstanceId => "/identity/instanceId",
             Self::EnvironmentIdentityConflict => "/identity/databaseInitializationEnvironment",
             Self::InvalidListener => "/listener",
@@ -387,6 +392,7 @@ impl RuntimeConfigError {
     fn suggested_action(&self) -> &'static str {
         match self {
             Self::Reader(_) => "",
+            Self::InvalidTaskGrantStatus => "Configure at most 32 distinct taskGrantStatus source issuers, safe service/token URLs, a client and assertion audience, absolute resource identifiers, one statusScope token, and a statusPath that starts with / and holds exactly one {grantId} segment.",
             Self::InvalidBinding => {
                 "Give every identity value as trimmed text without control characters, and declare in the runtime file, with a valid endpoint and credential, every task grant status source, Evidence provider, review authority, and review executor the package relies on."
             }
@@ -687,10 +693,15 @@ fn runtime_stand_in(pointer: &str) -> &'static str {
         ["audit", "path"]
         | ["fieldEncryption", "provider", "unixSocketPath"]
         | ["eventDestinations", _, "path"] => "/deferred",
+        ["taskGrantStatus", _, "statusPath"] => "/{grantId}",
         ["attachmentStorage" | "attachmentVerification", "endpoint"]
         | ["eventDestinations", _, "origin"]
         | ["evidenceProviders", _, "baseUrl"]
-        | ["taskGrantStatus", _, "baseUrl" | "sourceIssuer"]
+        | ["taskGrantStatus", _, "baseUrl"
+        | "sourceIssuer"
+        | "tokenEndpoint"
+        | "clientAssertionAudience"
+        | "caseworkResource"]
         | ["reviewAuthorities" | "reviewExecutors", _, "endpoint"]
         | ["reviewAuthorities" | "reviewExecutors", _, "privateKeyJwt", "tokenEndpoint" | "assertionAudience" | "resource"] => {
             "https://deferred.invalid/"
@@ -725,7 +736,7 @@ pub struct RuntimeConfig {
     field_encryption: crate::field_encryption::FieldEncryptionConfig,
     package: PackageConfig,
     authentication: AuthenticationConfig,
-    task_grant_status: Vec<crate::task_grant::TaskGrantStatusConfig>,
+    task_grant_status: Vec<TaskGrantStatusConfig>,
     audit: AuditConfig,
     cursor: CursorConfig,
     event_destinations: EventDestinationConfigs,
@@ -759,6 +770,11 @@ impl RuntimeConfig {
                 .map_err(|_| RuntimeConfigError::InvalidFieldEncryption)?;
         let package = PackageConfig::from_raw(raw.package)?;
         let authentication = AuthenticationConfig::from_raw(raw.authentication)?;
+        crate::task_grant::TaskGrantStatusRegistry::validate_configuration(
+            &raw.task_grant_status,
+            &authentication.oidc.audience,
+        )
+        .map_err(|_| RuntimeConfigError::InvalidTaskGrantStatus)?;
         let audit = AuditConfig::from_raw(raw.audit)?;
         let cursor = CursorConfig::from_raw(raw.cursor)?;
         let event_destinations = EventDestinationConfigs::from_raw(raw.event_destinations)
@@ -2868,7 +2884,7 @@ struct RawRuntimeConfig {
     package: RawPackageConfig,
     authentication: RawAuthenticationConfig,
     #[serde(default)]
-    task_grant_status: Vec<crate::task_grant::TaskGrantStatusConfig>,
+    task_grant_status: Vec<TaskGrantStatusConfig>,
     audit: RawAuditConfig,
     cursor: RawCursorConfig,
     #[serde(default, deserialize_with = "local_id_map")]

crates/registry-breg/src/task_grant.rs

@@ -1,17 +1,15 @@
 // SPDX-License-Identifier: Apache-2.0
 //! Immutable task authority retained separately from a later human reviewer.
-use registry_platform_httputil::{
-    client::{build_client, OutboundOptions, PrivateKeyJwt, ServiceBaseUrl, TokenProvider},
-    read_bounded, validate_response_headers,
-};
+use registry_platform_httputil::client::PrivateKeyJwt;
+pub use registry_platform_oidc::task_grant::TaskGrantError;
 use registry_platform_oidc::{GrantBounds, GrantClaims};
 use serde::{Deserialize, Serialize};
 use serde_json::Value;
-use std::{collections::BTreeMap, fmt, sync::Arc, time::Duration};
-use thiserror::Error;
+use std::{collections::BTreeMap, fmt, sync::Arc};
 use uuid::Uuid;
 mod config;
-pub use config::{TaskGrantStatusConfig, TaskGrantStatusRegistry};
+pub use config::TaskGrantStatusRegistry;
+pub use registry_platform_config::TaskGrantStatusConfig;
 
 #[derive(Clone, Eq, PartialEq, Deserialize, Serialize)]
 #[serde(rename_all = "camelCase", deny_unknown_fields)]
@@ -112,16 +110,6 @@ impl TaskGrantBinding {
     }
 }
 
-#[derive(Clone, Copy, Debug, Eq, PartialEq, Error)]
-pub enum TaskGrantError {
-    #[error("task authority was refused")]
-    Refused,
-    #[error("task authority status is unavailable")]
-    Unavailable,
-    #[error("task authority status configuration is invalid")]
-    Configuration,
-}
-
 pub trait TaskGrantStatusChecker: Send + Sync {
     /// Each invocation must perform a fresh check. Never cache positive status.
     fn check<'a>(
@@ -130,13 +118,8 @@ pub trait TaskGrantStatusChecker: Send + Sync {
     ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<(), TaskGrantError>> + Send + 'a>>;
 }
 
-pub struct TaskGrantStatusClient {
-    source_issuer: String,
-    resource: String,
-    base: ServiceBaseUrl,
-    http: reqwest::Client,
-    token: Arc<PrivateKeyJwt>,
-}
+/// BReg retains its product-specific binding policy around the shared transport.
+pub struct TaskGrantStatusClient(registry_platform_oidc::task_grant::TaskGrantStatusClient);
 impl fmt::Debug for TaskGrantStatusClient {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         f.write_str("TaskGrantStatusClient(<configured>)")
@@ -150,97 +133,35 @@ impl TaskGrantStatusClient {
         token: Arc<PrivateKeyJwt>,
         trusted_roots: Option<&[u8]>,
     ) -> Result<Self, TaskGrantError> {
-        if !registry_platform_httputil::valid_resource_uri(&source_issuer)
-            || !registry_platform_httputil::valid_resource_uri(&resource)
-        {
-            return Err(TaskGrantError::Configuration);
-        }
-        let base = ServiceBaseUrl::new(base).map_err(|_| TaskGrantError::Configuration)?;
-        let http = build_client(OutboundOptions {
-            request_timeout: Duration::from_secs(5),
-            connect_timeout: Duration::from_secs(3),
-            user_agent: Some("registry-breg-task-status"),
-            trusted_root_certificates: trusted_roots,
-        })
-        .map_err(|_| TaskGrantError::Configuration)?;
-        Ok(Self {
+        registry_platform_oidc::task_grant::TaskGrantStatusClient::new(
             source_issuer,
             resource,
             base,
-            http,
+            registry_platform_config::DEFAULT_TASK_GRANT_STATUS_PATH,
             token,
-        })
+            trusted_roots,
+        )
+        .map(Self)
     }
 }
-#[derive(Deserialize)]
-#[serde(deny_unknown_fields)]
-struct StatusResponse {
-    active: bool,
-    #[serde(default)]
-    grant: Option<TaskGrantBinding>,
-}
 impl TaskGrantStatusChecker for TaskGrantStatusClient {
     fn check<'a>(
         &'a self,
         binding: &'a TaskGrantBinding,
     ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<(), TaskGrantError>> + Send + 'a>>
     {
-        Box::pin(async move {
-            binding.validate()?;
-            if !binding.is_current()
-                || binding.source_issuer != self.source_issuer
-                || binding.resource != self.resource
-            {
-                return Err(TaskGrantError::Refused);
-            }
-            let token = self
-                .token
-                .bearer_token()
-                .await
-                .map_err(|_| TaskGrantError::Unavailable)?;
-            let url = registry_platform_httputil::url::append_path_segments(
-                self.base.as_url(),
-                &["v1", "task-grants", &binding.grant_id, "status"],
-            )
-            .map_err(|_| TaskGrantError::Configuration)?;
-            let response = self
-                .http
-                .get(url)
-                .header(
-                    reqwest::header::AUTHORIZATION,
-                    token.authorization_header_value(),
-                )
-                .header(reqwest::header::ACCEPT, "application/json")
-                .send()
-                .await
-                .map_err(|_| TaskGrantError::Unavailable)?;
-            validate_response_headers(response.headers())
-                .map_err(|_| TaskGrantError::Unavailable)?;
-            if matches!(response.status().as_u16(), 401 | 403 | 404) {
-                return Err(TaskGrantError::Refused);
-            }
-            if response.status() != reqwest::StatusCode::OK
-                || response
-                    .headers()
-                    .get(reqwest::header::CONTENT_TYPE)
-                    .and_then(|value| value.to_str().ok())
-                    .and_then(|value| value.split(';').next())
-                    .is_none_or(|value| !value.trim().eq_ignore_ascii_case("application/json"))
-            {
-                return Err(TaskGrantError::Unavailable);
-            }
-            let bytes = read_bounded(response, 64 * 1024)
-                .await
-                .map_err(|_| TaskGrantError::Unavailable)?;
-            let json = registry_platform_crypto::parse_json_strict(&bytes)
-                .map_err(|_| TaskGrantError::Unavailable)?;
-            let status: StatusResponse =
-                serde_json::from_value(json).map_err(|_| TaskGrantError::Unavailable)?;
-            if !status.active || status.grant.as_ref() != Some(binding) || !binding.is_current() {
-                return Err(TaskGrantError::Refused);
-            }
-            Ok(())
-        })
+        Box::pin(async move { self.0.check(&binding.status_binding()?).await })
+    }
+}
+impl TaskGrantBinding {
+    fn status_binding(
+        &self,
+    ) -> Result<registry_platform_oidc::task_grant::TaskGrantStatusBinding, TaskGrantError> {
+        self.validate()?;
+        // Retain the existing persisted BReg shape and product-specific bounds
+        // check; the shared transport consumes the identical closed wire model.
+        serde_json::from_value(serde_json::to_value(self).map_err(|_| TaskGrantError::Refused)?)
+            .map_err(|_| TaskGrantError::Refused)
     }
 }