mirror of
https://github.com/nearai/ironclaw.git
synced 2026-09-03 08:06:01 +08:00
fix(outbound,processes,secrets): byte-only backend compat and AAD/path alignment
Three findings from the second-pass PR review (commit 4eccad56d):
(1) LocalFilesystem byte-only backend compat — the new filesystem
stores in `ironclaw_processes`, `ironclaw_outbound`, and
`ironclaw_secrets` issue record-shape/`CasExpectation::Version` writes
even though the docs and constructor signatures still claim any
`RootFilesystem` works. `LocalFilesystem` rejects both with
`Unsupported`, so processes started on a local engine filesystem fail
during `complete`/`fail`/`kill` (`operation write_file is not supported`),
leaving durable local process state stuck `Running`. Reviewer ran
`filesystem_process_store_persists_under_resource_scope_engine_processes`
and reproduced.
- `ironclaw_processes::put_with_byte_fallback` previously retried the
byte-stripped entry with the *same* `cas`. Now it also downgrades to
`CasExpectation::Any` on the fallback, carrying the safety invariant
via the existing `transition_lock`. Reviewer's test now passes.
- `ironclaw_outbound::put_with_byte_fallback` is the new equivalent
helper, threaded through `put_json` / `put_delivery_attempt_indexed`.
- `ironclaw_secrets::put_with_version_fallback` is the equivalent for
the lease-revoke / consume / session-use Version-CAS paths.
(2) Filesystem secrets AAD/path scope mismatch — `filesystem_secret_aad`
bound `mission_id`/`thread_id`/`invocation_id` (full scope), but
`secret_path` and `same_scope_owner` key only on owner scope
(`tenant/user/agent/project`). A secret written by one invocation
appeared present to another invocation in the same owner scope (path
layer allowed it) but decryption failed with a confusing
"backend unavailable" error. Cross-invocation puts also silently
overwrote the same path. Align AAD to `ScopeKey::from_account_scope`
(owner-scope-only) so cross-invocation reads within the same owner
succeed; cross-owner access still fails at the path layer.
This commit is contained in:
@@ -85,11 +85,7 @@ where
|
||||
) -> Result<(), OutboundError> {
|
||||
let body = serde_json::to_vec(value).map_err(|_| OutboundError::Serialization)?;
|
||||
let entry = Entry::bytes(body).with_content_type(ContentType::json());
|
||||
self.filesystem
|
||||
.put(path, entry, cas)
|
||||
.await
|
||||
.map(|_| ())
|
||||
.map_err(map_fs_error)
|
||||
self.put_with_byte_fallback(path, entry, cas).await
|
||||
}
|
||||
|
||||
/// Like [`put_json`] but additionally projects an indexed scope value so
|
||||
@@ -108,11 +104,34 @@ where
|
||||
delivery_scope_index_key(),
|
||||
delivery_scope_index_value(&attempt.scope),
|
||||
);
|
||||
self.filesystem
|
||||
.put(path, entry, cas)
|
||||
.await
|
||||
.map(|_| ())
|
||||
.map_err(map_fs_error)
|
||||
self.put_with_byte_fallback(path, entry, cas).await
|
||||
}
|
||||
|
||||
/// Write `entry` with the given CAS expectation, falling back to a
|
||||
/// metadata-stripped opaque write + `CasExpectation::Any` for backends
|
||||
/// that reject record-shape entries or non-`Any` CAS (e.g.
|
||||
/// `LocalFilesystem`). Mirrors
|
||||
/// [`ironclaw_processes::filesystem_store::put_with_byte_fallback`] so
|
||||
/// every byte-only mount in the workspace stays writeable through the
|
||||
/// new filesystem stores.
|
||||
async fn put_with_byte_fallback(
|
||||
&self,
|
||||
path: &VirtualPath,
|
||||
entry: Entry,
|
||||
cas: CasExpectation,
|
||||
) -> Result<(), OutboundError> {
|
||||
match self.filesystem.put(path, entry.clone(), cas).await {
|
||||
Ok(_) => Ok(()),
|
||||
Err(FilesystemError::Unsupported { .. }) => {
|
||||
let opaque = Entry::bytes(entry.body).with_content_type(entry.content_type);
|
||||
self.filesystem
|
||||
.put(path, opaque, CasExpectation::Any)
|
||||
.await
|
||||
.map(|_| ())
|
||||
.map_err(map_fs_error)
|
||||
}
|
||||
Err(error) => Err(map_fs_error(error)),
|
||||
}
|
||||
}
|
||||
|
||||
/// Declare the `scope` exact-equality index on the deliveries prefix.
|
||||
|
||||
@@ -751,8 +751,17 @@ where
|
||||
match filesystem.put(path, entry.clone(), cas).await {
|
||||
Ok(_) => Ok(()),
|
||||
Err(error) if is_unsupported(&error) => {
|
||||
// Byte-only backends (LocalFilesystem) reject BOTH record-shaped
|
||||
// entries AND non-`Any` CAS in a single `Unsupported` response.
|
||||
// Strip the record metadata and downgrade the CAS expectation to
|
||||
// `Any` so the legacy byte-only path stays writeable. The
|
||||
// single-instance `transition_lock` on the caller carries the
|
||||
// ordering safety invariant that CAS would otherwise provide.
|
||||
let opaque = Entry::bytes(entry.body).with_content_type(entry.content_type);
|
||||
filesystem.put(path, opaque, cas).await.map(|_| ())
|
||||
filesystem
|
||||
.put(path, opaque, CasExpectation::Any)
|
||||
.await
|
||||
.map(|_| ())
|
||||
}
|
||||
Err(error) => Err(error),
|
||||
}
|
||||
|
||||
@@ -323,7 +323,17 @@ pub fn credential_session_aad(scope: &ResourceScope, session_id: CredentialSessi
|
||||
/// `(user_id, name)` — a swap between the two encodings must fail decryption
|
||||
/// even with an identical scope/user, which the domain separator enforces.
|
||||
pub fn filesystem_secret_aad(scope: &ResourceScope, handle: &SecretHandle) -> Vec<u8> {
|
||||
let key = ScopeKey::from_full_scope(scope);
|
||||
// The filesystem secret store keys by *owner scope*
|
||||
// (`tenant/user/agent/project`) — see `secret_path` and
|
||||
// `same_scope_owner` in `filesystem_store.rs`. The AAD must match the
|
||||
// storage scope: previously this bound `mission_id`/`thread_id`/
|
||||
// `invocation_id` too, so a secret written by one invocation could be
|
||||
// *read* by another invocation under the same owner (the path layer
|
||||
// allowed it) but `consume` failed with a confusing decryption error.
|
||||
// Bind AAD to the owner scope so cross-invocation reads within the
|
||||
// same owner succeed and cross-owner reads still fail closed (both at
|
||||
// the path layer and via AAD).
|
||||
let key = ScopeKey::from_account_scope(scope);
|
||||
build_aad(
|
||||
AAD_DOMAIN_FILESYSTEM_SECRET,
|
||||
&[
|
||||
@@ -331,9 +341,6 @@ pub fn filesystem_secret_aad(scope: &ResourceScope, handle: &SecretHandle) -> Ve
|
||||
key.user_id.as_bytes(),
|
||||
key.agent_id.as_bytes(),
|
||||
key.project_id.as_bytes(),
|
||||
key.mission_id.as_bytes(),
|
||||
key.thread_id.as_bytes(),
|
||||
key.invocation_id.as_bytes(),
|
||||
handle.as_str().as_bytes(),
|
||||
],
|
||||
)
|
||||
|
||||
@@ -403,10 +403,13 @@ where
|
||||
// and return the same error to the caller.
|
||||
let body = serialize_secret(&lease)?;
|
||||
let entry = Entry::bytes(body).with_content_type(ContentType::json());
|
||||
match self
|
||||
.filesystem
|
||||
.put(&path, entry, CasExpectation::Version(versioned.version))
|
||||
.await
|
||||
match put_with_version_fallback(
|
||||
&*self.filesystem,
|
||||
&path,
|
||||
entry,
|
||||
CasExpectation::Version(versioned.version),
|
||||
)
|
||||
.await
|
||||
{
|
||||
Ok(_) | Err(FilesystemError::VersionMismatch { .. }) => {}
|
||||
Err(error) => return Err(fs_to_secret_store_error(error)),
|
||||
@@ -433,10 +436,13 @@ where
|
||||
lease.status = SecretLeaseStatus::Consumed;
|
||||
let body = serialize_secret(&lease)?;
|
||||
let entry = Entry::bytes(body).with_content_type(ContentType::json());
|
||||
match self
|
||||
.filesystem
|
||||
.put(&path, entry, CasExpectation::Version(versioned.version))
|
||||
.await
|
||||
match put_with_version_fallback(
|
||||
&*self.filesystem,
|
||||
&path,
|
||||
entry,
|
||||
CasExpectation::Version(versioned.version),
|
||||
)
|
||||
.await
|
||||
{
|
||||
Ok(_) => return Ok(material),
|
||||
Err(FilesystemError::VersionMismatch { .. }) => continue,
|
||||
@@ -506,10 +512,13 @@ where
|
||||
}
|
||||
let body = serialize_secret(&lease)?;
|
||||
let entry = Entry::bytes(body).with_content_type(ContentType::json());
|
||||
match self
|
||||
.filesystem
|
||||
.put(&path, entry, CasExpectation::Version(versioned.version))
|
||||
.await
|
||||
match put_with_version_fallback(
|
||||
&*self.filesystem,
|
||||
&path,
|
||||
entry,
|
||||
CasExpectation::Version(versioned.version),
|
||||
)
|
||||
.await
|
||||
{
|
||||
Ok(_) => return Ok(Self::lease_to_public(&lease)),
|
||||
Err(FilesystemError::VersionMismatch { .. }) => continue,
|
||||
@@ -840,10 +849,13 @@ where
|
||||
stored.uses += 1;
|
||||
let body = serialize_credential(&stored)?;
|
||||
let entry = Entry::bytes(body).with_content_type(ContentType::json());
|
||||
match self
|
||||
.filesystem
|
||||
.put(&path, entry, CasExpectation::Version(versioned.version))
|
||||
.await
|
||||
match put_with_version_fallback(
|
||||
&*self.filesystem,
|
||||
&path,
|
||||
entry,
|
||||
CasExpectation::Version(versioned.version),
|
||||
)
|
||||
.await
|
||||
{
|
||||
Ok(_) => return wire.into_session(),
|
||||
Err(FilesystemError::VersionMismatch { .. }) => continue,
|
||||
@@ -1024,6 +1036,38 @@ fn lock_or_recover<T>(mutex: &Mutex<HashMap<String, T>>) -> MutexGuard<'_, HashM
|
||||
.unwrap_or_else(|poisoned| poisoned.into_inner())
|
||||
}
|
||||
|
||||
/// Write `entry` with `cas`, falling back to `CasExpectation::Any` if the
|
||||
/// backend reports `Unsupported` for a non-`Any` CAS expectation. Used by
|
||||
/// the lease/secret transitions in this file so byte-only mounts like
|
||||
/// `LocalFilesystem` (which advertises no per-row versioning) stay
|
||||
/// writeable while CAS-capable backends still get the multi-process
|
||||
/// safety. The per-owner mutation lock on the caller carries the
|
||||
/// ordering safety invariant on the fallback path.
|
||||
///
|
||||
/// Returns the raw `FilesystemError` (including `VersionMismatch`) so the
|
||||
/// caller's CAS retry loop can detect contention; we only intercept the
|
||||
/// specific `Unsupported` shape.
|
||||
async fn put_with_version_fallback<F>(
|
||||
filesystem: &F,
|
||||
path: &ironclaw_host_api::VirtualPath,
|
||||
entry: Entry,
|
||||
cas: CasExpectation,
|
||||
) -> Result<(), FilesystemError>
|
||||
where
|
||||
F: RootFilesystem + ?Sized,
|
||||
{
|
||||
match filesystem.put(path, entry.clone(), cas).await {
|
||||
Ok(_) => Ok(()),
|
||||
Err(FilesystemError::Unsupported { .. }) if !matches!(cas, CasExpectation::Any) => {
|
||||
filesystem
|
||||
.put(path, entry, CasExpectation::Any)
|
||||
.await
|
||||
.map(|_| ())
|
||||
}
|
||||
Err(error) => Err(error),
|
||||
}
|
||||
}
|
||||
|
||||
// -- Error mapping ----------------------------------------------------------
|
||||
|
||||
fn fs_to_secret_store_error(error: FilesystemError) -> SecretStoreError {
|
||||
|
||||
Reference in New Issue
Block a user