diff --git a/Cargo.lock b/Cargo.lock index 3205365c3..f0520a9bf 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1722,6 +1722,7 @@ dependencies = [ "opentelemetry_sdk", "prost", "reqwest 0.12.28", + "rustix", "schemars", "semver", "serde", @@ -1742,6 +1743,7 @@ dependencies = [ "typed-builder", "unicode-general-category", "uuid", + "windows-sys 0.61.2", ] [[package]] diff --git a/crates/adaptive/README.md b/crates/adaptive/README.md index e96211c4e..5d938e03e 100644 --- a/crates/adaptive/README.md +++ b/crates/adaptive/README.md @@ -68,6 +68,12 @@ shared persistence: cargo add nemo-relay-adaptive --features redis-backend ``` +The response cache is opt-in. Its required `namespace` identifies one trusted +cache-sharing domain and must not be reused across mutually untrusted tenants +or upstream authorities. When tenants share a backend, include every +answer-affecting tenant or routing header in `header_allowlist`; Relay cannot +infer application-owned tenant identity. + For local source development: ```bash diff --git a/crates/adaptive/tests/unit/response_cache/key_tests.rs b/crates/adaptive/tests/unit/response_cache/key_tests.rs index b7b5a47b5..5eca5fbdb 100644 --- a/crates/adaptive/tests/unit/response_cache/key_tests.rs +++ b/crates/adaptive/tests/unit/response_cache/key_tests.rs @@ -1092,7 +1092,7 @@ fn cache_error_policy_partitions_tool_keys() { #[test] fn header_allowlist_policy_partitions_keys_and_normalizes_case() { - let request = request(json!({ + let mut request = request(json!({ "model": "gpt-4o", "messages": [{"role": "user", "content": "hi"}], "temperature": 0.0, @@ -1117,6 +1117,19 @@ fn header_allowlist_policy_partitions_keys_and_normalizes_case() { key_of("openai", &request, &duplicate_spelling), "case-only and duplicate policy spellings are equivalent" ); + + request + .headers + .insert("x-tenant".to_string(), json!("tenant-a")); + let tenant_a = key_of("openai", &request, &tenant_partitioned); + request + .headers + .insert("x-tenant".to_string(), json!("tenant-b")); + assert_ne!( + tenant_a, + key_of("openai", &request, &tenant_partitioned), + "different allowlisted tenant identities must not share entries" + ); } #[test] diff --git a/crates/cli/README.md b/crates/cli/README.md index dc78c9519..81ecaa5ee 100644 --- a/crates/cli/README.md +++ b/crates/cli/README.md @@ -190,6 +190,13 @@ the dynamic plugin references in the selected physical `plugins.toml`. Dynamic plugins with a manifest-declared JSON Schema provide structured field controls. Other dynamic plugins use a raw JSON object editor. +At runtime, dynamic plugins that do not have an explicit host-policy override +are classified as required and must carry a valid signature from a configured +trusted public key. Installation and inspection remain available for unsigned +plugins, but Relay refuses to activate them. Worker processes inherit only the +small environment allowlist needed for process startup and TLS. Native plugins +run in-process and must be treated as trusted host code after verification. + The canonical plugin file is `plugins.toml`; user config lives at `~/.config/nemo-relay/plugins.toml` or `$XDG_CONFIG_HOME/nemo-relay/plugins.toml`. Use diff --git a/crates/cli/src/agents/claude/adapter.rs b/crates/cli/src/agents/claude/adapter.rs index ed1cdde98..7ff1d29d3 100644 --- a/crates/cli/src/agents/claude/adapter.rs +++ b/crates/cli/src/agents/claude/adapter.rs @@ -6,16 +6,16 @@ use serde_json::{Value, json}; use crate::agents::shared::adapters::{ AdapterOutcome, CLAUDE_CODE_PAYLOAD_EXTRACTOR, ClassificationRules, classify, + permission_request, }; -use crate::events::{AgentKind, NormalizedEvent}; +use crate::events::AgentKind; /// Normalizes Claude Code hook payloads and returns the hook response Claude expects. /// -/// Claude Code uses permission-bearing tool hooks, so pre-tool events are explicitly allowed -/// instead of returning the generic `{ continue: true }` shape. All other hooks acknowledge with -/// `{ continue: true }` so the gateway remains observational and never blocks Claude's lifecycle -/// by default. Note: Claude's hook output schema rejects `null` for optional string fields like -/// `stopReason`; omit them entirely instead. +/// Claude Code uses permission-bearing tool hooks. Pre-tool events acknowledge guardrail success +/// without granting host permission; the later `PermissionRequest` receives the final decision. +/// Note: Claude's hook output schema rejects `null` for optional string fields like `stopReason`; +/// omit them entirely instead. pub(crate) fn adapt(payload: Value, headers: &HeaderMap) -> AdapterOutcome { let events = classify( &payload, @@ -40,17 +40,15 @@ pub(crate) fn adapt(payload: Value, headers: &HeaderMap) -> AdapterOutcome { ], }, ); - // Response shape is decided by the primary event (first in the vec); secondary events like - // `TurnEnded` are observability-only and don't influence the hook response Claude gets back. - let response = match events.first() { - Some(NormalizedEvent::ToolStarted(_)) => json!({ - "continue": true, - "hookSpecificOutput": { - "hookEventName": "PreToolUse", - "permissionDecision": "allow" - } - }), - _ => json!({ "continue": true }), - }; - AdapterOutcome { events, response } + let response = json!({ "continue": true }); + AdapterOutcome { + events, + response, + permission: permission_request( + &payload, + headers, + AgentKind::ClaudeCode, + &CLAUDE_CODE_PAYLOAD_EXTRACTOR, + ), + } } diff --git a/crates/cli/src/agents/claude/launch.rs b/crates/cli/src/agents/claude/launch.rs index 89ab79e32..4c67e89cc 100644 --- a/crates/cli/src/agents/claude/launch.rs +++ b/crates/cli/src/agents/claude/launch.rs @@ -36,10 +36,13 @@ pub(crate) fn prepare( [ "--plugin-dir".into(), "".into(), - "--settings".into(), - "".into(), ], ); + insert_before_argument_boundary( + &mut launch.argv, + launch.host_index, + ["--settings".into(), "".into()], + ); launch .env .push(("ANTHROPIC_BASE_URL".into(), gateway_url.to_string())); @@ -80,12 +83,12 @@ pub(crate) fn prepare( insert_after_host( &mut launch.argv, launch.host_index, - [ - "--plugin-dir".into(), - root.display().to_string(), - "--settings".into(), - settings_path.display().to_string(), - ], + ["--plugin-dir".into(), root.display().to_string()], + ); + insert_before_argument_boundary( + &mut launch.argv, + launch.host_index, + ["--settings".into(), settings_path.display().to_string()], ); launch .env @@ -94,6 +97,19 @@ pub(crate) fn prepare( Ok(()) } +fn insert_before_argument_boundary( + argv: &mut Vec, + host_index: usize, + values: impl IntoIterator, +) { + let boundary = argv + .iter() + .skip(host_index + 1) + .position(|argument| argument == "--") + .map_or(argv.len(), |offset| host_index + 1 + offset); + argv.splice(boundary..boundary, values); +} + fn replace_custom_header(existing: &str, replacement: &str) -> String { let replacement_name = replacement .split_once(':') diff --git a/crates/cli/src/agents/codex/adapter.rs b/crates/cli/src/agents/codex/adapter.rs index 536f778ba..ed7ee1601 100644 --- a/crates/cli/src/agents/codex/adapter.rs +++ b/crates/cli/src/agents/codex/adapter.rs @@ -5,7 +5,7 @@ use axum::http::HeaderMap; use serde_json::{Value, json}; use crate::agents::shared::adapters::{ - AdapterOutcome, CODEX_PAYLOAD_EXTRACTOR, ClassificationRules, classify, + AdapterOutcome, CODEX_PAYLOAD_EXTRACTOR, ClassificationRules, classify, permission_request, }; use crate::events::AgentKind; @@ -32,5 +32,11 @@ pub(crate) fn adapt(payload: Value, headers: &HeaderMap) -> AdapterOutcome { AdapterOutcome { events, response: json!({}), + permission: permission_request( + &payload, + headers, + AgentKind::Codex, + &CODEX_PAYLOAD_EXTRACTOR, + ), } } diff --git a/crates/cli/src/agents/shared/adapters.rs b/crates/cli/src/agents/shared/adapters.rs index 8d5b3aff5..40488933e 100644 --- a/crates/cli/src/agents/shared/adapters.rs +++ b/crates/cli/src/agents/shared/adapters.rs @@ -28,6 +28,8 @@ pub(crate) struct AdapterOutcome { pub(crate) events: Vec, /// Hook response body returned to the invoking agent process. pub(crate) response: Value, + /// Final permission request evaluated separately so its observable hook event stays unchanged. + pub(crate) permission: Option>, } pub(super) struct ClassificationRules<'a> { @@ -221,6 +223,57 @@ pub(crate) struct ToolPathSet { status: &'static [&'static [&'static str]], } +pub(super) fn permission_request( + payload: &Value, + headers: &HeaderMap, + kind: AgentKind, + extractor: &dyn AgentPayloadExtractor, +) -> Option> { + let event_name = extractor.event_name(payload)?; + if normalize_name(&event_name) != "permissionrequest" { + return None; + } + let session_id = match extractor.session_id(payload, headers) { + Some(value) if !value.trim().is_empty() => value, + _ => { + return Some(Err( + "permission request is missing a session identifier".into() + )); + } + }; + let tool = extractor.tool_call(payload, headers, &event_name); + let tool_call_id = match tool.tool_call_id { + Some(value) if !value.trim().is_empty() => value, + None if kind == AgentKind::ClaudeCode => String::new(), + _ => { + return Some(Err( + "permission request is missing a tool-call identifier".into() + )); + } + }; + let tool_name = match tool.tool_name { + Some(value) if !value.trim().is_empty() => value, + _ => return Some(Err("permission request is missing a tool name".into())), + }; + let arguments = match tool.arguments { + Some(value) => value, + None => return Some(Err("permission request is missing tool arguments".into())), + }; + Some(Ok(ToolEvent { + session_id, + agent_kind: kind, + event_name: event_name.clone(), + tool_call_id, + tool_name, + subagent_id: tool.subagent_id, + arguments, + result: Value::Null, + status: tool.status, + payload: payload.clone(), + metadata: extractor.metadata(payload, headers, kind, &event_name), + })) +} + /// Whether an extractor accepts the Claude installed-mode session header. #[derive(Clone, Copy)] pub(crate) enum SessionHeaderPolicy { diff --git a/crates/cli/src/agents/shared/alignment.rs b/crates/cli/src/agents/shared/alignment.rs index 40e49322f..8c1c2c858 100644 --- a/crates/cli/src/agents/shared/alignment.rs +++ b/crates/cli/src/agents/shared/alignment.rs @@ -216,6 +216,7 @@ impl ProviderRequestExtractor for AnthropicCountTokensRequestExtractor { pub(crate) struct SessionAlias { pub(crate) parent_session_id: String, pub(crate) subagent_id: String, + authenticated_owner: Option, // Metadata explains why this alias exists and is stamped on rewritten events. Phoenix traces // then stay filterable/debuggable even after the event has been moved under its parent scope. metadata: Value, @@ -228,6 +229,7 @@ impl SessionAlias { Self { parent_session_id, subagent_id, + authenticated_owner: None, metadata, } } @@ -237,6 +239,14 @@ impl SessionAlias { pub(crate) fn metadata(&self) -> Value { self.metadata.clone() } + + pub(crate) fn set_authenticated_owner(&mut self, owner: Option) { + self.authenticated_owner = owner; + } + + pub(crate) fn authenticated_owner(&self) -> Option<&str> { + self.authenticated_owner.as_deref() + } } #[derive(Debug, Clone)] @@ -245,6 +255,7 @@ pub(crate) struct PendingSubagentStart { // hook or gateway request, after this hook request has already returned. pub(crate) event: SessionEvent, context: SubagentSessionContext, + authenticated_owner: Option, } impl PendingSubagentStart { @@ -259,6 +270,14 @@ impl PendingSubagentStart { pub(crate) fn alias_for_child_session(&self, child_session_id: String) -> SessionAlias { alias_for_child_session(child_session_id, &self.context) } + + pub(crate) fn set_authenticated_owner(&mut self, owner: Option) { + self.authenticated_owner = owner; + } + + pub(crate) fn authenticated_owner(&self) -> Option<&str> { + self.authenticated_owner.as_deref() + } } // Owns all cross-session correlation state used by the session manager. Keeping aliases and @@ -551,6 +570,7 @@ pub(crate) async fn pending_subagent_start( PendingSubagentStart { event: session_event.clone(), context, + authenticated_owner: None, }, )) } diff --git a/crates/cli/src/configuration/mod.rs b/crates/cli/src/configuration/mod.rs index 04f143a25..677a8d5bd 100644 --- a/crates/cli/src/configuration/mod.rs +++ b/crates/cli/src/configuration/mod.rs @@ -475,12 +475,14 @@ const BOOTSTRAP_HMAC_KEY_BYTES: usize = 32; const BOOTSTRAP_HMAC_LOCK_TIMEOUT: Duration = Duration::from_secs(5); const BOOTSTRAP_CHALLENGE_DOMAIN: &[u8] = b"nemo-relay/bootstrap-health/v1\0"; const BOOTSTRAP_CLIENT_TOKEN_DOMAIN: &[u8] = b"nemo-relay/bootstrap-client/v1\0"; +const HOOK_CLIENT_TOKEN_DOMAIN: &[u8] = b"nemo-relay/hook-client/v1\0"; const TRANSPARENT_GATEWAY_DOMAIN: &[u8] = b"nemo-relay/transparent-gateway/v1\0"; const PYTHON_ENVIRONMENT_ATTESTATION_DOMAIN: &[u8] = b"nemo-relay/python-environment-attestation/v1\0"; /// Private proof installed into supported coding-agent provider configuration. pub(crate) const BOOTSTRAP_CLIENT_TOKEN_HEADER: &str = "x-nemo-relay-client-token"; +pub(crate) const HOOK_CLIENT_TOKEN_HEADER: &str = "x-nemo-relay-hook-client"; /// Stable health-proof context shared by a transparent wrapper and plugin-owned MCP client. pub(crate) fn transparent_gateway_fingerprint(gateway_url: &str) -> String { @@ -558,6 +560,33 @@ impl BootstrapChallengeKey { hmac::verify(&self.0, BOOTSTRAP_CLIENT_TOKEN_DOMAIN, &tag).is_ok() } + pub(crate) fn hook_client_token(&self, identity: &str) -> String { + let identity = digest::digest(&digest::SHA256, identity.as_bytes()) + .as_ref() + .iter() + .map(|byte| format!("{byte:02x}")) + .collect::(); + let mut context = hmac::Context::with_key(&self.0); + context.update(HOOK_CLIENT_TOKEN_DOMAIN); + context.update(identity.as_bytes()); + format!("{identity}.{}", encode_hmac_tag(context.sign())) + } + + pub(crate) fn verify_hook_client_token(&self, token: &str) -> Option { + let (identity, signature) = token.split_once('.')?; + if identity.len() != 64 || !identity.bytes().all(|byte| byte.is_ascii_hexdigit()) { + return None; + } + let encoded = signature.strip_prefix("hmac-sha256:")?; + let tag = decode_fixed_hex::<32>(encoded)?; + let mut message = Vec::with_capacity(HOOK_CLIENT_TOKEN_DOMAIN.len() + identity.len()); + message.extend_from_slice(HOOK_CLIENT_TOKEN_DOMAIN); + message.extend_from_slice(identity.as_bytes()); + hmac::verify(&self.0, &message, &tag) + .is_ok() + .then(|| format!("hook-client:{identity}")) + } + #[cfg(test)] pub(crate) fn from_bytes(bytes: &[u8]) -> Self { Self(hmac::Key::new(hmac::HMAC_SHA256, bytes)) diff --git a/crates/cli/src/gateway/client.rs b/crates/cli/src/gateway/client.rs index c26a822cb..dae33aca1 100644 --- a/crates/cli/src/gateway/client.rs +++ b/crates/cli/src/gateway/client.rs @@ -14,6 +14,7 @@ use reqwest::Url; use ring::rand::{SecureRandom, SystemRandom}; use serde_json::Value; +use crate::configuration::BOOTSTRAP_CLIENT_TOKEN_HEADER; use crate::configuration::BootstrapChallengeKey; use crate::bootstrap::{BOOTSTRAP_PROTOCOL_VERSION, HEALTHZ_TIMEOUT}; @@ -200,11 +201,18 @@ pub(crate) fn post_verified( let mut request = format!("POST {path} HTTP/1.1\r\nHost: {authority}\r\n"); for (name, value) in headers { + if name.eq_ignore_ascii_case(BOOTSTRAP_CLIENT_TOKEN_HEADER) { + continue; + } request.push_str(name); request.push_str(": "); request.push_str(value); request.push_str("\r\n"); } + request.push_str(BOOTSTRAP_CLIENT_TOKEN_HEADER); + request.push_str(": "); + request.push_str(&key.client_token()); + request.push_str("\r\n"); request.push_str(&format!( "Content-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", body.len() diff --git a/crates/cli/src/hooks/delivery.rs b/crates/cli/src/hooks/delivery.rs index cb9cf566f..232c362c8 100644 --- a/crates/cli/src/hooks/delivery.rs +++ b/crates/cli/src/hooks/delivery.rs @@ -187,23 +187,25 @@ pub(crate) async fn send_verified_hook_forward_request( Result, CliError, > { - let headers = gateway_headers( + let mut headers = gateway_headers( command.profile.as_deref(), command.session_metadata.as_deref(), command.gateway_mode, - )? - .iter() - .map(|(name, value)| { - value - .to_str() - .map(|value| (name.as_str().to_string(), value.to_string())) - .map_err(|error| { - CliError::Install(format!( - "hook header {name} is not valid HTTP text: {error}" - )) - }) - }) - .collect::, _>>()?; + )?; + attach_internal_hook_credentials(command, &mut headers)?; + let headers = headers + .iter() + .map(|(name, value)| { + value + .to_str() + .map(|value| (name.as_str().to_string(), value.to_string())) + .map_err(|error| { + CliError::Install(format!( + "hook header {name} is not valid HTTP text: {error}" + )) + }) + }) + .collect::, _>>()?; let gateway = gateway.clone(); let gateway_url = gateway_url.to_string(); let path = command.agent.hook_path().to_string(); @@ -228,23 +230,60 @@ async fn send_hook_forward_request( url: &str, input: String, ) -> Result, CliError> { + let mut headers = gateway_headers( + command.profile.as_deref(), + command.session_metadata.as_deref(), + command.gateway_mode, + )?; + attach_internal_hook_credentials(command, &mut headers)?; Ok(reqwest::Client::builder() .no_proxy() .redirect(reqwest::redirect::Policy::none()) .timeout(HOOK_FORWARD_TIMEOUT) .build()? .post(url) - .headers(gateway_headers( - command.profile.as_deref(), - command.session_metadata.as_deref(), - command.gateway_mode, - )?) + .headers(headers) .header(CONTENT_TYPE, "application/json") .body(input) .send() .await) } +fn attach_internal_hook_credentials( + command: &HookForwardRequest, + headers: &mut HeaderMap, +) -> Result<(), CliError> { + let key = crate::configuration::BootstrapChallengeKey::load()?; + let client_token = key.client_token(); + insert_header( + headers, + crate::configuration::BOOTSTRAP_CLIENT_TOKEN_HEADER, + Some(&client_token), + )?; + if command.transparent_run { + let credential = std::env::var(crate::provider_auth::TRANSPARENT_PROXY_CREDENTIAL_ENV) + .map_err(|_| { + CliError::Launch( + "transparent hook forwarding is missing its invocation credential".into(), + ) + })?; + insert_header( + headers, + crate::provider_auth::TRANSPARENT_PROXY_CREDENTIAL_HEADER, + Some(&credential), + )?; + } + if let Some(generation) = command.generation_token.as_deref() { + let token = key.hook_client_token(generation); + insert_header( + headers, + crate::configuration::HOOK_CLIENT_TOKEN_HEADER, + Some(&token), + )?; + } + Ok(()) +} + // Handles hook delivery results without changing agent control flow unless `--fail-closed` was // requested. Successful non-empty endpoint bodies are printed verbatim for the invoking hook API. fn validate_optional_json(name: &str, value: Option<&str>) -> Result<(), CliError> { diff --git a/crates/cli/src/plugins/lifecycle/mod.rs b/crates/cli/src/plugins/lifecycle/mod.rs index 45e7a135f..97e2ed877 100644 --- a/crates/cli/src/plugins/lifecycle/mod.rs +++ b/crates/cli/src/plugins/lifecycle/mod.rs @@ -1849,8 +1849,10 @@ pub(crate) fn active_dynamic_plugin_components_for_identity( explicit_plugin_config: Option<&PathBuf>, resolved: &ResolvedConfig, ) -> Result, CliError> { + let mut resolved = resolved.clone(); + crate::plugins::policy::apply_secure_runtime_defaults(&mut resolved.dynamic_plugin_policy); let scopes = load_scoped_registries(explicit_plugin_config)?; - active_dynamic_plugin_components_from_scopes(&scopes, resolved, false) + active_dynamic_plugin_components_from_scopes(&scopes, &resolved, false) } fn active_dynamic_plugin_components_inner( @@ -1858,8 +1860,10 @@ fn active_dynamic_plugin_components_inner( resolved: &ResolvedConfig, create_activation_snapshots: bool, ) -> Result, CliError> { - let scopes = load_and_hydrate_scopes(explicit_plugin_config, resolved)?; - active_dynamic_plugin_components_from_scopes(&scopes, resolved, create_activation_snapshots) + let mut resolved = resolved.clone(); + crate::plugins::policy::apply_secure_runtime_defaults(&mut resolved.dynamic_plugin_policy); + let scopes = load_and_hydrate_scopes(explicit_plugin_config, &resolved)?; + active_dynamic_plugin_components_from_scopes(&scopes, &resolved, create_activation_snapshots) } fn active_dynamic_plugin_components_from_scopes( diff --git a/crates/cli/src/plugins/policy.rs b/crates/cli/src/plugins/policy.rs index 4a43d38e9..67ba7e59c 100644 --- a/crates/cli/src/plugins/policy.rs +++ b/crates/cli/src/plugins/policy.rs @@ -177,6 +177,17 @@ pub(crate) fn evaluate_dynamic_plugin_host_policy( } } +pub(crate) fn apply_secure_runtime_defaults(policy: &mut DynamicPluginHostPolicy) { + policy + .defaults + .startup + .get_or_insert(DynamicPluginStartupClass::Required); + policy + .defaults + .attestation + .get_or_insert(DynamicPluginAttestationMode::SignatureRequired); +} + fn policy_rule_matches( rule: &DynamicPluginHostPolicyRule, manifest: &DynamicPluginManifest, diff --git a/crates/cli/src/provider_auth.rs b/crates/cli/src/provider_auth.rs index 8bffa95a7..65a0d9114 100644 --- a/crates/cli/src/provider_auth.rs +++ b/crates/cli/src/provider_auth.rs @@ -10,6 +10,7 @@ use std::sync::Arc; use axum::http::{HeaderMap, HeaderValue, header}; +use ring::digest; use ring::rand::{SecureRandom, SystemRandom}; use subtle::ConstantTimeEq; @@ -41,6 +42,11 @@ impl TransparentProxyCredential { &self.0 } + /// Stable non-secret identity for binding invocation-owned gateway state. + pub(crate) fn identity(&self) -> String { + credential_identity(&self.0) + } + /// Verify and consume this invocation's proxy credential without disturbing an independent /// provider credential carried by a dedicated header. pub(crate) fn consume( @@ -106,6 +112,18 @@ impl TransparentProxyCredential { } } +pub(crate) fn credential_identity(value: &str) -> String { + let digest = digest::digest(&digest::SHA256, value.as_bytes()); + format!( + "sha256:{}", + digest + .as_ref() + .iter() + .map(|byte| format!("{byte:02x}")) + .collect::() + ) +} + #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub(crate) enum SourceCredentialDisposition { RelayProxyCredential { provider_credential_present: bool }, diff --git a/crates/cli/src/server/mod.rs b/crates/cli/src/server/mod.rs index 0a20996cd..3dfef0f2e 100644 --- a/crates/cli/src/server/mod.rs +++ b/crates/cli/src/server/mod.rs @@ -15,7 +15,7 @@ use std::time::{Duration, Instant}; use axum::body::Body; use axum::extract::rejection::JsonRejection; use axum::extract::{DefaultBodyLimit, State}; -use axum::http::{HeaderMap, HeaderValue, Request, StatusCode}; +use axum::http::{HeaderMap, HeaderValue, Request, StatusCode, header}; use axum::response::{IntoResponse, Response}; use axum::routing::{get, post}; use axum::{Json, Router}; @@ -37,7 +37,8 @@ use tokio::sync::oneshot; use crate::agents::shared::adapters::{claude_code, codex}; use crate::configuration::{ - BOOTSTRAP_CLIENT_TOKEN_HEADER, BootstrapChallengeKey, GatewayConfig, ManagedBootstrapIdentity, + BOOTSTRAP_CLIENT_TOKEN_HEADER, BootstrapChallengeKey, GatewayConfig, HOOK_CLIENT_TOKEN_HEADER, + ManagedBootstrapIdentity, }; use crate::error::CliError; use crate::gateway; @@ -88,6 +89,12 @@ pub(crate) async fn serve_with_dynamic( ready_file: Option<&Path>, bootstrap_shutdown_token: Option, ) -> Result<(), CliError> { + if !config.bind.ip().is_loopback() { + return Err(CliError::Config(format!( + "explicit Relay gateways require a loopback bind address, got {}", + config.bind + ))); + } let bind = config.bind.to_string(); log::info!( target: "nemo_relay.server", @@ -265,10 +272,7 @@ async fn serve_listener_with_dynamic_inner( shutdown_token: bootstrap_shutdown_token, transparent_proxy_credential, } = bootstrap; - let bootstrap_challenge_key = bootstrap_fingerprint - .as_ref() - .map(|_| BootstrapChallengeKey::load()) - .transpose()?; + let bootstrap_challenge_key = Some(BootstrapChallengeKey::load()?); let bootstrap_tls = bootstrap_fingerprint .as_ref() .map(|_| crate::gateway::tls::RelayTlsIdentity::load_or_create()) @@ -277,7 +281,8 @@ async fn serve_listener_with_dynamic_inner( .map(|identity| identity.server_config()) .transpose() .map_err(CliError::Launch)?; - let require_provider_client_token = managed_bootstrap.is_some(); + let require_provider_client_token = + bootstrap_fingerprint.is_some() && transparent_proxy_credential.is_none(); let plugin_activation = initialize_plugin_host(config.plugin_config.clone(), dynamic_plugins).await?; let (bootstrap_shutdown, bootstrap_shutdown_rx) = @@ -536,6 +541,11 @@ impl AppState { &self, headers: &mut HeaderMap, ) -> Result { + if headers.contains_key(header::ORIGIN) { + return Err(CliError::Unauthorized( + "browser-originated Relay provider requests are not accepted".into(), + )); + } if let Some(proxy) = &self.transparent_proxy_credential { let source_credential = proxy.consume(headers).inspect_err(|error| { log::warn!( @@ -570,6 +580,53 @@ impl AppState { allow_environment_provider_auth, }) } + + fn authorize_hook_request(&self, headers: &mut HeaderMap) -> Result { + #[cfg(test)] + if self.bootstrap_challenge_key.is_none() && self.transparent_proxy_credential.is_none() { + return Ok("test-unauthenticated-hook-client".into()); + } + if headers.contains_key(header::ORIGIN) { + return Err(CliError::Unauthorized( + "browser-originated Relay hook requests are not accepted".into(), + )); + } + if let Some(proxy) = &self.transparent_proxy_credential + && headers + .get(crate::provider_auth::TRANSPARENT_PROXY_CREDENTIAL_HEADER) + .is_some() + { + let identity = proxy.identity(); + proxy.consume(headers)?; + headers.remove(BOOTSTRAP_CLIENT_TOKEN_HEADER); + headers.remove(HOOK_CLIENT_TOKEN_HEADER); + return Ok(identity); + } + let token = headers + .get(BOOTSTRAP_CLIENT_TOKEN_HEADER) + .and_then(|value| value.to_str().ok()) + .ok_or_else(|| { + CliError::Unauthorized( + "Relay hook request did not present an internal client credential".into(), + ) + })?; + let key = self.bootstrap_challenge_key.as_ref().ok_or_else(|| { + CliError::Unauthorized("Relay hook authentication is unavailable".into()) + })?; + if !key.verify_client_token(token) { + return Err(CliError::Unauthorized( + "Relay hook client credential was invalid".into(), + )); + } + let identity = headers + .get(HOOK_CLIENT_TOKEN_HEADER) + .and_then(|value| value.to_str().ok()) + .and_then(|value| key.verify_hook_client_token(value)) + .unwrap_or_else(|| crate::provider_auth::credential_identity(token)); + headers.remove(BOOTSTRAP_CLIENT_TOKEN_HEADER); + headers.remove(HOOK_CLIENT_TOKEN_HEADER); + Ok(identity) + } } fn router_with_state(state: AppState) -> Router { @@ -1127,16 +1184,25 @@ impl Drop for PluginActivation { // adapter's pass-through response body so hook delivery stays causally ordered with observability. async fn codex_hook( State(state): State, - headers: HeaderMap, + mut headers: HeaderMap, payload: Result, JsonRejection>, ) -> Result, CliError> { state.touch(); + let owner = state.authorize_hook_request(&mut headers)?; let Json(payload) = payload.map_err(hook_payload_rejection)?; let outcome = codex::adapt(payload, &headers); state .sessions - .apply_events(&headers, outcome.events) + .apply_authenticated_events(&headers, outcome.events, &owner) .await?; + if let Some(permission) = outcome.permission + && let Err(error) = authorize_hook_permission(&state, permission, &owner).await + { + return Ok(Json(serde_json::json!({ + "decision": "deny", + "reason": permission_denial_reason(error), + }))); + } Ok(Json(outcome.response)) } @@ -1144,19 +1210,69 @@ async fn codex_hook( // are committed before the response so Claude lifecycle hooks can close scopes deterministically. async fn claude_code_hook( State(state): State, - headers: HeaderMap, + mut headers: HeaderMap, payload: Result, JsonRejection>, ) -> Result, CliError> { state.touch(); + let owner = state.authorize_hook_request(&mut headers)?; let Json(payload) = payload.map_err(hook_payload_rejection)?; let outcome = claude_code::adapt(payload, &headers); state .sessions - .apply_events(&headers, outcome.events) + .apply_authenticated_events(&headers, outcome.events, &owner) .await?; + if let Some(permission) = outcome.permission { + let result = authorize_hook_permission(&state, permission, &owner).await; + return Ok(Json(match result { + Ok(()) => serde_json::json!({ + "continue": true, + "hookSpecificOutput": { + "hookEventName": "PermissionRequest", + "decision": { + "behavior": "allow", + } + } + }), + Err(error) => { + serde_json::json!({ + "continue": true, + "hookSpecificOutput": { + "hookEventName": "PermissionRequest", + "decision": { + "behavior": "deny", + "message": permission_denial_reason(error), + } + } + }) + } + })); + } Ok(Json(outcome.response)) } +async fn authorize_hook_permission( + state: &AppState, + permission: Result, + owner: &str, +) -> Result<(), CliError> { + match permission { + Ok(permission) => { + state + .sessions + .authorize_tool_permission(&permission, owner) + .await + } + Err(reason) => Err(CliError::InvalidPayload(reason)), + } +} + +fn permission_denial_reason(error: CliError) -> String { + error + .guardrail_rejection_reason() + .map(ToOwned::to_owned) + .unwrap_or_else(|| error.to_string()) +} + fn hook_payload_rejection(rejection: JsonRejection) -> CliError { if rejection.status() == axum::http::StatusCode::PAYLOAD_TOO_LARGE { CliError::PayloadTooLarge(rejection.to_string()) diff --git a/crates/cli/src/sessions/idle.rs b/crates/cli/src/sessions/idle.rs index 7833ef124..44c3a150d 100644 --- a/crates/cli/src/sessions/idle.rs +++ b/crates/cli/src/sessions/idle.rs @@ -35,6 +35,7 @@ pub(super) async fn close_sessions_for_shutdown( pub(super) async fn close_idle_sessions_from_parts( inner: &Arc>>, + authenticated_owners: &Arc>>, alignment: &Arc>, now: Instant, timeout: Duration, @@ -44,14 +45,42 @@ pub(super) async fn close_idle_sessions_from_parts( if idle_sessions.is_empty() { return Ok(0); } + let idle_session_ids = idle_sessions + .iter() + .map(|(session_id, _)| session_id.clone()) + .collect::>(); let (closed_turns, closed_subagents, retained_sessions, first_error) = close_idle_turns(idle_sessions, reason).await; + let retained_session_ids = retained_sessions + .iter() + .map(|(session_id, _)| session_id.clone()) + .collect::>(); + let released_owner_ids = idle_session_ids + .difference(&retained_session_ids) + .cloned() + .collect::>(); let cleanup_sessions = restore_retained_sessions(inner, retained_sessions, &closed_subagents).await; clear_closed_subagents(alignment, closed_subagents, &cleanup_sessions).await; + release_closed_owner_ids(inner, authenticated_owners, &released_owner_ids).await; first_error.map_or(Ok(closed_turns), Err) } +pub(super) async fn release_closed_owner_ids( + inner: &Arc>>, + authenticated_owners: &Arc>>, + released_owner_ids: &HashSet, +) { + if released_owner_ids.is_empty() { + return; + } + let mut owners = authenticated_owners.lock().await; + let sessions = inner.lock().await; + owners.retain(|session_id, _| { + !released_owner_ids.contains(session_id) || sessions.contains_key(session_id) + }); +} + async fn take_idle_sessions( inner: &Arc>>, now: Instant, diff --git a/crates/cli/src/sessions/mod.rs b/crates/cli/src/sessions/mod.rs index 1c92e4435..ab1116185 100644 --- a/crates/cli/src/sessions/mod.rs +++ b/crates/cli/src/sessions/mod.rs @@ -63,6 +63,7 @@ const ROUTING_IDENTITY_HEADERS: &[&str] = &[ #[derive(Clone)] pub(crate) struct SessionManager { inner: Arc>>, + authenticated_owners: Arc>>, // Cross-session alignment state owns child-session aliases and child-first SessionStart hooks. // Applies to Codex child threads today; the generic state lives in `alignment` so session code // only orchestrates when promotion is safe. @@ -279,11 +280,95 @@ impl SessionManager { pub(crate) fn new(default_config: GatewayConfig) -> Self { Self { inner: Arc::new(Mutex::new(HashMap::new())), + authenticated_owners: Arc::new(Mutex::new(HashMap::new())), alignment: Arc::new(Mutex::new(SessionAlignmentState::default())), default_config, } } + /// Applies authenticated hook events after binding every named session to its first client. + pub(crate) async fn apply_authenticated_events( + &self, + headers: &HeaderMap, + events: Vec, + owner: &str, + ) -> Result<(), CliError> { + let mut owners = self.authenticated_owners.lock().await; + let mut prospective_owners = owners.clone(); + for event in &events { + let session_id = event.session_id(); + if prospective_owners + .get(session_id) + .is_some_and(|existing| existing != owner) + { + return Err(CliError::Unauthorized(format!( + "Relay hook client does not own session '{session_id}'" + ))); + } + } + for event in &events { + prospective_owners + .entry(event.session_id().to_string()) + .or_insert_with(|| owner.to_string()); + } + let released_owner_ids = self + .apply_events_inner(headers, events, Some(&prospective_owners), Some(owner)) + .await?; + *owners = prospective_owners; + drop(owners); + release_closed_owner_ids(&self.inner, &self.authenticated_owners, &released_owner_ids) + .await; + Ok(()) + } + + /// Evaluates a final host permission request inside its existing session scope. + pub(crate) async fn authorize_tool_permission( + &self, + event: &ToolEvent, + owner: &str, + ) -> Result<(), CliError> { + let owners = self.authenticated_owners.lock().await; + match owners.get(&event.session_id) { + Some(existing) if existing == owner => {} + Some(_) => { + return Err(CliError::Unauthorized(format!( + "Relay hook client does not own session '{}'", + event.session_id + ))); + } + None => { + return Err(CliError::InvalidPayload(format!( + "permission request names unknown session '{}'", + event.session_id + ))); + } + } + drop(owners); + let sessions = self.inner.lock().await; + let session = sessions.get(&event.session_id).ok_or_else(|| { + CliError::InvalidPayload(format!( + "permission request names unknown session '{}'", + event.session_id + )) + })?; + if !session.permission_request_matches(event) { + return Err(CliError::InvalidPayload(format!( + "permission request does not match the recorded tool call '{}'", + event.tool_call_id + ))); + } + let stack = session.scope_stack.clone(); + let name = event.tool_name.clone(); + let arguments = normalize_tool_arguments(event.arguments.clone()); + drop(sessions); + TASK_SCOPE_STACK + .scope(stack, async move { + tool_conditional_execution(&name, &arguments).await + }) + .await + .map_err(CliError::from) + } + /// Starts the fail-safe idle closer used by the HTTP gateway. /// /// Some coding agents, notably Codex child threads, do not always emit native agent-end hooks. @@ -293,17 +378,23 @@ impl SessionManager { /// shutdown paths. pub(crate) fn start_idle_sweeper(&self) { let inner = Arc::downgrade(&self.inner); + let authenticated_owners = Arc::downgrade(&self.authenticated_owners); let alignment = Arc::downgrade(&self.alignment); tokio::spawn(async move { let mut interval = tokio::time::interval(AGENT_IDLE_SWEEP_INTERVAL); interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); loop { interval.tick().await; - let (Some(inner), Some(alignment)) = (inner.upgrade(), alignment.upgrade()) else { + let (Some(inner), Some(authenticated_owners), Some(alignment)) = ( + inner.upgrade(), + authenticated_owners.upgrade(), + alignment.upgrade(), + ) else { break; }; if let Err(error) = close_idle_sessions_from_parts( &inner, + &authenticated_owners, &alignment, Instant::now(), AGENT_IDLE_TIMEOUT, @@ -335,15 +426,31 @@ impl SessionManager { /// metadata reflects the actual agent. Note: agent-scope and observer identities are baked at /// scope-open time, so this upgrade applies to session metadata only — the /// provider-inferred kind set in `start_llm` is the primary defense. + #[cfg(test)] pub(crate) async fn apply_events( &self, headers: &HeaderMap, events: Vec, ) -> Result<(), CliError> { + self.apply_events_inner(headers, events, None, None) + .await + .map(|_| ()) + } + + async fn apply_events_inner( + &self, + headers: &HeaderMap, + events: Vec, + authenticated_owners: Option<&HashMap>, + authenticated_owner: Option<&str>, + ) -> Result, CliError> { let mut subscriber_deliveries = Vec::new(); + let mut released_owner_ids = HashSet::new(); let mut alignment_state = self.alignment.lock().await; let mut sessions = self.inner.lock().await; for event in events { + let original_session_id = event.session_id().to_string(); + let original_was_terminal = event.is_terminal(); let mut event = event; let config = self.default_config.session_config_from_headers(headers); if queue_or_promote_child_start( @@ -351,17 +458,38 @@ impl SessionManager { &mut sessions, &mut alignment_state, config.clone(), + authenticated_owners, + authenticated_owner, ) .await? { continue; } + if let Some(owners) = authenticated_owners + && let Some(alias) = alignment_state.alias_for_session(&original_session_id) + && let Some(alias_owner) = alias.authenticated_owner() + && owners + .get(&alias.parent_session_id) + .is_none_or(|existing| existing != alias_owner) + { + return Err(CliError::Unauthorized(format!( + "Relay hook client does not own session '{}'", + alias.parent_session_id + ))); + } + let Some((event, session_id, is_agent_started)) = route_event_for_session(event, &mut sessions, &mut alignment_state) else { + if original_was_terminal { + released_owner_ids.insert(original_session_id); + } continue; }; + if original_was_terminal && original_session_id != session_id { + released_owner_ids.insert(original_session_id); + } let event_kind = event_agent_kind(&event); let (should_remove_session, subscriber_delivery) = apply_event_to_session( &mut sessions, @@ -383,11 +511,13 @@ impl SessionManager { &mut alignment_state, &session_id, config.clone(), + authenticated_owners, ) .await?; } if should_remove_session { sessions.remove(&session_id); + released_owner_ids.insert(session_id); } } drop(sessions); @@ -395,7 +525,7 @@ impl SessionManager { for subscriber_delivery in subscriber_deliveries { subscriber_delivery.wait().await?; } - Ok(()) + Ok(released_owner_ids) } /// Legacy manual-lifecycle entry point retained for tests that drive correlation behavior @@ -514,6 +644,15 @@ impl SessionManager { let mut closing = completed.then(|| sessions.remove(session_id)).flatten(); drop(sessions); + if completed { + release_closed_owner_ids( + &self.inner, + &self.authenticated_owners, + &HashSet::from([session_id.to_string()]), + ) + .await; + } + if finish == GatewaySessionFinish::Close && let Some(session) = closing.as_mut() { @@ -638,6 +777,7 @@ impl SessionManager { /// observability plugins are still active. Applies to Codex transparent runs today. pub(crate) async fn close_all(&self, reason: &str) -> Result<(), CliError> { self.alignment.lock().await.clear(); + self.authenticated_owners.lock().await.clear(); let mut sessions = { let mut guard = self.inner.lock().await; guard @@ -655,7 +795,15 @@ impl SessionManager { timeout: Duration, reason: &str, ) -> Result { - close_idle_sessions_from_parts(&self.inner, &self.alignment, now, timeout, reason).await + close_idle_sessions_from_parts( + &self.inner, + &self.authenticated_owners, + &self.alignment, + now, + timeout, + reason, + ) + .await } // Applies known or pending child-session aliases before the gateway chooses a session. This is @@ -670,14 +818,39 @@ impl SessionManager { let Some(session_id) = start.session_id.clone() else { return Ok(None); }; + let mut owners = self.authenticated_owners.lock().await; let mut alignment_state = self.alignment.lock().await; if let Some(alias) = alignment_state.alias_for_session(&session_id) { + if let Some(alias_owner) = alias.authenticated_owner() + && owners + .get(&alias.parent_session_id) + .is_none_or(|existing| existing != alias_owner) + { + return Err(CliError::Unauthorized(format!( + "Relay gateway request does not own session '{}'", + alias.parent_session_id + ))); + } apply_start_alias(start, &alias); return Ok(Some(alias)); } let Some(pending) = alignment_state.pending_for_session(&session_id) else { return Ok(None); }; + if let Some(owner) = pending.authenticated_owner() { + match owners.get(pending.parent_session_id()) { + Some(existing) if existing != owner => { + return Err(CliError::Unauthorized(format!( + "Relay gateway request does not own session '{}'", + pending.parent_session_id() + ))); + } + Some(_) => {} + None => { + owners.insert(pending.parent_session_id().to_string(), owner.to_string()); + } + } + } let mut sessions = self.inner.lock().await; let alias = promote_pending_subagent( &mut sessions, @@ -685,6 +858,7 @@ impl SessionManager { session_id, pending, config, + Some(&owners), ) .await?; if let Some(alias) = alias.as_ref() { @@ -726,6 +900,29 @@ impl Session { } } + fn permission_request_matches(&self, event: &ToolEvent) -> bool { + let arguments = normalize_tool_arguments(event.arguments.clone()); + if event.tool_call_id.is_empty() { + return self + .tools + .values() + .filter(|active| active.name == event.tool_name && active.arguments == arguments) + .count() + == 1; + } + let active_matches = self + .tools + .get(&event.tool_call_id) + .is_some_and(|active| active.name == event.tool_name && active.arguments == arguments); + active_matches + || self.pending_tool_hints.iter().any(|pending| { + pending.hint.tool_call_id.as_deref() == Some(event.tool_call_id.as_str()) + && pending.hint.tool_name.as_deref() == Some(event.tool_name.as_str()) + && !pending.hint.arguments.is_null() + && pending.hint.arguments == arguments + }) + } + // A child session can only be converted into a subagent before any real scope, LLM, or tool // state has been opened for it. Once work exists under the child, reparenting would move only // future events and leave an inconsistent trace. diff --git a/crates/cli/src/sessions/routing.rs b/crates/cli/src/sessions/routing.rs index 4a2aa6d81..e6037a5be 100644 --- a/crates/cli/src/sessions/routing.rs +++ b/crates/cli/src/sessions/routing.rs @@ -28,10 +28,14 @@ pub(super) async fn queue_or_promote_child_start( sessions: &mut HashMap, alignment_state: &mut SessionAlignmentState, config: SessionConfig, + authenticated_owners: Option<&HashMap>, + authenticated_owner: Option<&str>, ) -> Result { - let Some((child_session_id, pending)) = alignment::pending_subagent_start(event).await else { + let Some((child_session_id, mut pending)) = alignment::pending_subagent_start(event).await + else { return Ok(false); }; + pending.set_authenticated_owner(authenticated_owner.map(ToOwned::to_owned)); if sessions .get(&child_session_id) .is_some_and(|session| !session.can_reparent_as_subagent_alias()) @@ -39,9 +43,26 @@ pub(super) async fn queue_or_promote_child_start( return Ok(false); } if sessions.contains_key(pending.parent_session_id()) { + if !parent_owner_matches( + authenticated_owners, + pending.parent_session_id(), + pending.authenticated_owner(), + ) { + return Err(CliError::Unauthorized(format!( + "Relay hook client does not own session '{}'", + pending.parent_session_id() + ))); + } alignment_state.remove_pending(&child_session_id); - promote_pending_subagent(sessions, alignment_state, child_session_id, pending, config) - .await?; + promote_pending_subagent( + sessions, + alignment_state, + child_session_id, + pending, + config, + authenticated_owners, + ) + .await?; } else { sessions.remove(&child_session_id); alignment_state.insert_pending(child_session_id, pending); @@ -75,14 +96,23 @@ pub(super) async fn promote_pending_subagents_for_parent( alignment_state: &mut SessionAlignmentState, parent_session_id: &str, config: SessionConfig, + authenticated_owners: Option<&HashMap>, ) -> Result<(), CliError> { for (child_session_id, pending) in alignment_state.pending_for_parent(parent_session_id) { + if !parent_owner_matches( + authenticated_owners, + parent_session_id, + pending.authenticated_owner(), + ) { + continue; + } promote_pending_subagent( sessions, alignment_state, child_session_id, pending, config.clone(), + authenticated_owners, ) .await?; } @@ -95,6 +125,7 @@ pub(super) async fn promote_pending_subagent( child_session_id: String, pending: PendingSubagentStart, config: SessionConfig, + authenticated_owners: Option<&HashMap>, ) -> Result, CliError> { if sessions .get(&child_session_id) @@ -104,6 +135,13 @@ pub(super) async fn promote_pending_subagent( } sessions.remove(&child_session_id); let parent_session_id = pending.parent_session_id().to_string(); + if !parent_owner_matches( + authenticated_owners, + &parent_session_id, + pending.authenticated_owner(), + ) { + return Ok(None); + } let parent_session = sessions .entry(parent_session_id.clone()) .or_insert_with(|| { @@ -125,11 +163,25 @@ pub(super) async fn promote_pending_subagent( pending.subagent_start_event(), )) .await?; - let alias = pending.alias_for_child_session(child_session_id.clone()); + let mut alias = pending.alias_for_child_session(child_session_id.clone()); + alias.set_authenticated_owner(pending.authenticated_owner().map(ToOwned::to_owned)); alignment_state.insert_alias(child_session_id, alias.clone()); Ok(Some(alias)) } +fn parent_owner_matches( + authenticated_owners: Option<&HashMap>, + parent_session_id: &str, + pending_owner: Option<&str>, +) -> bool { + match (authenticated_owners, pending_owner) { + (Some(owners), Some(owner)) => owners + .get(parent_session_id) + .is_some_and(|existing| existing == owner), + _ => true, + } +} + pub(super) fn route_event_for_session( event: NormalizedEvent, sessions: &mut HashMap, diff --git a/crates/cli/tests/cli_tests.rs b/crates/cli/tests/cli_tests.rs index c1253a0f4..cd5ab8c8b 100644 --- a/crates/cli/tests/cli_tests.rs +++ b/crates/cli/tests/cli_tests.rs @@ -3957,7 +3957,7 @@ fn invocation_diagnostic_cli_warns_for_agent_shortcut() { .find(|line| line.starts_with("argv = ")) .expect("dry run should print the resolved argv"); assert!( - argv.ends_with(" claude -p private synthetic value"), + argv.ends_with(" claude -p private synthetic value --settings "), "{argv}" ); @@ -4812,6 +4812,7 @@ fn cli_transparent_run_suppresses_persistent_hooks_and_rejects_a_foreign_gateway ]) .env("NEMO_RELAY_TRANSPARENT_RUN", "1") .env("NEMO_RELAY_GATEWAY_URL", &server_url) + .env("NEMO_RELAY_PROXY_CREDENTIAL", "nrp_foreign-test-credential") .stdin(Stdio::piped()) .stdout(Stdio::piped()) .stderr(Stdio::piped()) diff --git a/crates/cli/tests/coverage/agents/adapters_tests.rs b/crates/cli/tests/coverage/agents/adapters_tests.rs index 7dc2cc576..470249e74 100644 --- a/crates/cli/tests/coverage/agents/adapters_tests.rs +++ b/crates/cli/tests/coverage/agents/adapters_tests.rs @@ -38,14 +38,7 @@ fn maps_claude_canonical_tool_payload() { } event => panic!("unexpected event: {event:?}"), } - assert_eq!(outcome.response["continue"], json!(true)); - assert_eq!( - outcome.response["hookSpecificOutput"], - json!({ - "hookEventName": "PreToolUse", - "permissionDecision": "allow" - }) - ); + assert_eq!(outcome.response, json!({"continue": true})); } #[test] @@ -580,6 +573,70 @@ fn keeps_codex_response_unwrapped() { assert_eq!(outcome.response, json!({})); } +#[test] +fn permission_requests_keep_hook_marks_and_require_exact_tool_identity() { + let outcome = claude_code::adapt( + json!({ + "session_id": "claude-session", + "hook_event_name": "PermissionRequest", + "tool_use_id": "toolu-1", + "tool_name": "Write", + "tool_input": {"file_path": "README.md"} + }), + &HeaderMap::new(), + ); + assert!(matches!(outcome.events[0], NormalizedEvent::HookMark(_))); + let permission = outcome.permission.unwrap().unwrap(); + assert_eq!(permission.session_id, "claude-session"); + assert_eq!(permission.tool_call_id, "toolu-1"); + assert_eq!(permission.tool_name, "Write"); + assert_eq!(permission.arguments, json!({"file_path": "README.md"})); + + let claude_without_id = claude_code::adapt( + json!({ + "session_id": "claude-session", + "hook_event_name": "PermissionRequest", + "tool_name": "Write", + "tool_input": {"file_path": "README.md"} + }), + &HeaderMap::new(), + ); + assert_eq!( + claude_without_id.permission.unwrap().unwrap().tool_call_id, + "" + ); + + let codex_permission = codex::adapt( + json!({ + "session_id": "codex-session", + "hook_event_name": "PermissionRequest", + "tool_call_id": "call-1", + "tool_name": "shell", + "arguments": {"cmd": "pwd"} + }), + &HeaderMap::new(), + ) + .permission + .unwrap() + .unwrap(); + assert_eq!(codex_permission.session_id, "codex-session"); + assert_eq!(codex_permission.tool_call_id, "call-1"); + assert_eq!(codex_permission.tool_name, "shell"); + assert_eq!(codex_permission.arguments, json!({"cmd": "pwd"})); + assert_eq!(codex_permission.agent_kind, AgentKind::Codex); + + let missing_id = codex::adapt( + json!({ + "session_id": "codex-session", + "hook_event_name": "PermissionRequest", + "tool_name": "shell", + "arguments": {"cmd": "pwd"} + }), + &HeaderMap::new(), + ); + assert!(missing_id.permission.unwrap().is_err()); +} + #[test] fn normalizes_mark_style_events_and_header_session_ids() { let mut headers = HeaderMap::new(); diff --git a/crates/cli/tests/coverage/agents/launcher_tests.rs b/crates/cli/tests/coverage/agents/launcher_tests.rs index 3b00de60d..4ba9ffce0 100644 --- a/crates/cli/tests/coverage/agents/launcher_tests.rs +++ b/crates/cli/tests/coverage/agents/launcher_tests.rs @@ -906,7 +906,21 @@ fn prepares_claude_dry_inserts_plugin_dir_after_authoritative_agent_executable() prepared.argv[plugin_index + 1], "" ); - assert_eq!(prepared.argv.last().map(String::as_str), Some("--resume")); + let resume_index = prepared + .argv + .iter() + .position(|argument| argument == "--resume") + .unwrap(); + let settings_index = prepared + .argv + .iter() + .rposition(|argument| argument == "--settings") + .unwrap(); + assert!(settings_index > resume_index); + assert_eq!( + prepared.argv[settings_index + 1], + "" + ); assert!(prepared.temp_dirs.is_empty()); } @@ -994,9 +1008,14 @@ fn claude_transparent_run_preserves_user_settings_and_prompt_boundary() { .unwrap(); assert_eq!(prepared.argv[1], "--plugin-dir"); - assert_eq!(prepared.argv[3], "--settings"); + let separator = prepared.argv.iter().position(|arg| arg == "--").unwrap(); + let settings_index = prepared.argv[..separator] + .iter() + .rposition(|arg| arg == "--settings") + .unwrap(); let overlay: serde_json::Value = - serde_json::from_slice(&std::fs::read(&prepared.argv[4]).unwrap()).unwrap(); + serde_json::from_slice(&std::fs::read(&prepared.argv[settings_index + 1]).unwrap()) + .unwrap(); assert_eq!(overlay["model"], "claude-user-setting-sentinel"); assert_eq!(overlay["enabledPlugins"]["other@market"], true); assert_eq!( @@ -1021,7 +1040,6 @@ fn claude_transparent_run_preserves_user_settings_and_prompt_boundary() { .iter() .any(|arg| arg.contains("ignored-second-source")) ); - let separator = prepared.argv.iter().position(|arg| arg == "--").unwrap(); assert_eq!( &prepared.argv[separator..], &["--", "--settings", "literal-prompt-value"] diff --git a/crates/cli/tests/coverage/shared/config_tests.rs b/crates/cli/tests/coverage/shared/config_tests.rs index cb9f3edd3..66f6cb1de 100644 --- a/crates/cli/tests/coverage/shared/config_tests.rs +++ b/crates/cli/tests/coverage/shared/config_tests.rs @@ -2930,7 +2930,7 @@ fn persistent_hook_identity_authenticates_python_marker_without_rehashing_enviro std::fs::write( &plugins_toml, format!( - "version = 1\n\n[[plugins.dynamic]]\nmanifest = {:?}\n", + "version = 1\n\n[plugins.policy.defaults]\nstartup = \"optional\"\nattestation = \"integrity_only\"\n\n[[plugins.dynamic]]\nmanifest = {:?}\n", manifest_path.to_string_lossy() ), ) diff --git a/crates/cli/tests/coverage/shared/installer_tests.rs b/crates/cli/tests/coverage/shared/installer_tests.rs index 1064373a0..f7ae88d74 100644 --- a/crates/cli/tests/coverage/shared/installer_tests.rs +++ b/crates/cli/tests/coverage/shared/installer_tests.rs @@ -43,6 +43,32 @@ impl Drop for BootstrapConfigHome { } } +struct ScopedEnvVar { + key: &'static str, + previous: Option, +} + +impl ScopedEnvVar { + fn set(key: &'static str, value: &std::ffi::OsStr) -> Self { + let previous = std::env::var_os(key); + // SAFETY: The caller holds the process-wide environment mutex through BootstrapConfigHome. + unsafe { std::env::set_var(key, value) }; + Self { key, previous } + } +} + +impl Drop for ScopedEnvVar { + fn drop(&mut self) { + // SAFETY: BootstrapConfigHome outlives this guard and still holds the environment mutex. + unsafe { + match self.previous.take() { + Some(previous) => std::env::set_var(self.key, previous), + None => std::env::remove_var(self.key), + } + } + } +} + #[tokio::test] async fn transparent_hook_delivery_authenticates_the_wrapper_gateway() { let _plugin_guard = crate::test_support::PLUGIN_CONFIG_TEST_LOCK.lock().await; @@ -57,12 +83,17 @@ async fn transparent_hook_delivery_authenticates_the_wrapper_gateway() { ..crate::configuration::GatewayConfig::default() }; let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel(); + let proxy_credential = crate::provider_auth::TransparentProxyCredential::generate().unwrap(); + let _proxy_credential = ScopedEnvVar::set( + crate::provider_auth::TRANSPARENT_PROXY_CREDENTIAL_ENV, + proxy_credential.expose().as_ref(), + ); let server = tokio::spawn(crate::server::serve_transparent_listener_with_dynamic( listener, config, Vec::new(), fingerprint.clone(), - crate::provider_auth::TransparentProxyCredential::generate().unwrap(), + proxy_credential, Some(shutdown_rx), )); tokio::time::timeout(Duration::from_secs(5), async { diff --git a/crates/cli/tests/coverage/shared/plugins_lifecycle_tests.rs b/crates/cli/tests/coverage/shared/plugins_lifecycle_tests.rs index eeb06b909..de070d683 100644 --- a/crates/cli/tests/coverage/shared/plugins_lifecycle_tests.rs +++ b/crates/cli/tests/coverage/shared/plugins_lifecycle_tests.rs @@ -22,6 +22,13 @@ use ring::rand::SystemRandom; use ring::signature::{Ed25519KeyPair, KeyPair}; use sha2::{Digest, Sha256}; +fn allow_unsigned_test_plugins(resolved: &mut ResolvedConfig) { + resolved.dynamic_plugin_policy.defaults.startup = + Some(nemo_relay::plugin::dynamic::DynamicPluginStartupClass::Optional); + resolved.dynamic_plugin_policy.defaults.attestation = + Some(nemo_relay::plugin::dynamic::DynamicPluginAttestationMode::IntegrityOnly); +} + #[test] fn activation_snapshots_use_a_short_directory_prefix() { let _env = EnvScope::set(&[(ACTIVATION_SNAPSHOT_DIR_ENV, None)]); @@ -2190,7 +2197,8 @@ fn add_provisions_persists_and_removes_managed_python_environment() { &server, ) .unwrap(); - let resolved = resolve_plugins_config(None).unwrap(); + let mut resolved = resolve_plugins_config(None).unwrap(); + allow_unsigned_test_plugins(&mut resolved); let active = active_dynamic_plugin_components(None, &resolved).unwrap(); assert_eq!(active.len(), 1); assert_eq!(active[0].environment_ref.as_deref(), Some(environment_ref)); @@ -2715,7 +2723,8 @@ fn active_dynamic_plugin_components_user_enabled_native_records_only() { &server, ) .unwrap(); - let resolved = resolve_plugins_config(None).unwrap(); + let mut resolved = resolve_plugins_config(None).unwrap(); + allow_unsigned_test_plugins(&mut resolved); let active = active_dynamic_plugin_components(None, &resolved).unwrap(); assert_eq!(active.len(), 1); assert_eq!(active[0].plugin_id, "acme.native"); @@ -2755,7 +2764,8 @@ fn active_dynamic_plugin_components_accept_enabled_worker_records() { ) .unwrap(); - let resolved = resolve_plugins_config(None).unwrap(); + let mut resolved = resolve_plugins_config(None).unwrap(); + allow_unsigned_test_plugins(&mut resolved); let active = active_dynamic_plugin_components(None, &resolved).unwrap(); assert_eq!(active.len(), 1); assert_eq!(active[0].plugin_id, "acme.worker"); diff --git a/crates/cli/tests/coverage/shared/server_tests.rs b/crates/cli/tests/coverage/shared/server_tests.rs index 4c679d9ea..65968230f 100644 --- a/crates/cli/tests/coverage/shared/server_tests.rs +++ b/crates/cli/tests/coverage/shared/server_tests.rs @@ -152,7 +152,17 @@ impl Drop for PluginKindCleanup { } fn test_http_client() -> reqwest::Client { - reqwest::Client::new() + let key = BootstrapChallengeKey::load().expect("test hook credential should load"); + let mut headers = reqwest::header::HeaderMap::new(); + headers.insert( + crate::configuration::BOOTSTRAP_CLIENT_TOKEN_HEADER, + reqwest::header::HeaderValue::from_str(&key.client_token()) + .expect("test hook credential should be a valid header"), + ); + reqwest::Client::builder() + .default_headers(headers) + .build() + .expect("test HTTP client should build") } struct GenericTestPlugin; @@ -551,6 +561,15 @@ async fn managed_sidecar_requires_private_client_proof_for_forwarded_credentials .allow_environment_provider_auth ); + let explicit_daemon = + AppState::new_with_bootstrap(test_config(), None, Some(key.clone()), false, None, None); + assert!( + explicit_daemon + .authorize_provider_request(&mut HeaderMap::new()) + .unwrap() + .allow_environment_provider_auth + ); + let transparent = AppState::new_with_bootstrap( test_config(), Some("transparent-fingerprint".into()), @@ -579,6 +598,41 @@ async fn managed_sidecar_requires_private_client_proof_for_forwarded_credentials ); } +#[tokio::test] +async fn explicit_daemon_hook_authentication_is_internal_and_rejects_bad_tokens() { + let key = BootstrapChallengeKey::from_bytes(b"test challenge key"); + let state = + AppState::new_with_bootstrap(test_config(), None, Some(key.clone()), true, None, None); + assert!(state.authorize_hook_request(&mut HeaderMap::new()).is_err()); + + let mut headers = HeaderMap::new(); + headers.insert( + crate::configuration::BOOTSTRAP_CLIENT_TOKEN_HEADER, + HeaderValue::from_str(&key.client_token()).unwrap(), + ); + let hook_client_token = key.hook_client_token("test-hook-installation"); + let expected_hook_owner = key.verify_hook_client_token(&hook_client_token).unwrap(); + headers.insert( + crate::configuration::HOOK_CLIENT_TOKEN_HEADER, + HeaderValue::from_str(&hook_client_token).unwrap(), + ); + let owner = state.authorize_hook_request(&mut headers).unwrap(); + assert_eq!(owner, expected_hook_owner); + assert!(!headers.contains_key(crate::configuration::BOOTSTRAP_CLIENT_TOKEN_HEADER)); + assert!(!headers.contains_key(crate::configuration::HOOK_CLIENT_TOKEN_HEADER)); + + let mut browser_headers = HeaderMap::new(); + browser_headers.insert( + header::ORIGIN, + HeaderValue::from_static("https://example.test"), + ); + browser_headers.insert( + crate::configuration::BOOTSTRAP_CLIENT_TOKEN_HEADER, + HeaderValue::from_str(&key.client_token()).unwrap(), + ); + assert!(state.authorize_hook_request(&mut browser_headers).is_err()); +} + #[tokio::test] async fn healthz_only_refreshes_idle_activity_for_an_authenticated_heartbeat() { let challenge_key = BootstrapChallengeKey::from_bytes(b"test challenge key"); @@ -2284,6 +2338,124 @@ async fn claude_code_hook_returns_continue_shape() { assert_eq!(body["continue"], json!(true)); } +#[tokio::test] +async fn claude_permission_request_allows_an_exact_active_tool() { + let app = router(test_config()); + let pre_tool = app + .clone() + .oneshot( + Request::builder() + .method("POST") + .uri("/hooks/claude-code") + .header("content-type", "application/json") + .body(Body::from( + json!({ + "session_id": "claude-permission", + "hook_event_name": "PreToolUse", + "tool_use_id": "tool-1", + "tool_name": "Read", + "tool_input": {"file_path": "README.md"} + }) + .to_string(), + )) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(pre_tool.status(), StatusCode::OK); + let bytes = pre_tool.into_body().collect().await.unwrap().to_bytes(); + let body: Value = serde_json::from_slice(&bytes).unwrap(); + assert_eq!(body, json!({"continue": true})); + + let permission = app + .clone() + .oneshot( + Request::builder() + .method("POST") + .uri("/hooks/claude-code") + .header("content-type", "application/json") + .body(Body::from( + json!({ + "session_id": "claude-permission", + "hook_event_name": "PermissionRequest", + "tool_name": "Read", + "tool_input": {"file_path": "README.md"} + }) + .to_string(), + )) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(permission.status(), StatusCode::OK); + let bytes = permission.into_body().collect().await.unwrap().to_bytes(); + let body: Value = serde_json::from_slice(&bytes).unwrap(); + assert_eq!( + body["hookSpecificOutput"]["hookEventName"], + json!("PermissionRequest") + ); + assert_eq!( + body["hookSpecificOutput"]["decision"]["behavior"], + json!("allow") + ); + + let second_pre_tool = app + .clone() + .oneshot( + Request::builder() + .method("POST") + .uri("/hooks/claude-code") + .header("content-type", "application/json") + .body(Body::from( + json!({ + "session_id": "claude-permission", + "hook_event_name": "PreToolUse", + "tool_use_id": "tool-2", + "tool_name": "Read", + "tool_input": {"file_path": "README.md"} + }) + .to_string(), + )) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(second_pre_tool.status(), StatusCode::OK); + + let ambiguous = app + .oneshot( + Request::builder() + .method("POST") + .uri("/hooks/claude-code") + .header("content-type", "application/json") + .body(Body::from( + json!({ + "session_id": "claude-permission", + "hook_event_name": "PermissionRequest", + "tool_name": "Read", + "tool_input": {"file_path": "README.md"} + }) + .to_string(), + )) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(ambiguous.status(), StatusCode::OK); + let bytes = ambiguous.into_body().collect().await.unwrap().to_bytes(); + let body: Value = serde_json::from_slice(&bytes).unwrap(); + assert_eq!( + body["hookSpecificOutput"]["decision"]["behavior"], + json!("deny") + ); + assert!( + body["hookSpecificOutput"]["decision"]["message"] + .as_str() + .unwrap() + .contains("does not match") + ); +} + #[tokio::test] async fn pre_tool_hook_rejects_when_conditional_guardrail_blocks() { let _guard = PLUGIN_CONFIG_TEST_LOCK.lock().await; diff --git a/crates/cli/tests/coverage/shared/session_tests.rs b/crates/cli/tests/coverage/shared/session_tests.rs index 8531803f4..c5abf0628 100644 --- a/crates/cli/tests/coverage/shared/session_tests.rs +++ b/crates/cli/tests/coverage/shared/session_tests.rs @@ -20,6 +20,345 @@ use super::*; use crate::events::{LlmHintEvent, SessionEvent, ToolEvent}; use crate::test_support::PLUGIN_CONFIG_TEST_LOCK; +#[tokio::test] +async fn authenticated_hook_clients_cannot_take_over_existing_sessions() { + let manager = SessionManager::new(session_test_config()); + let event = || NormalizedEvent::AgentStarted(session_event("owned-session", "SessionStart")); + manager + .apply_authenticated_events(&HeaderMap::new(), vec![event()], "client-a") + .await + .unwrap(); + manager + .apply_authenticated_events(&HeaderMap::new(), vec![event()], "client-a") + .await + .unwrap(); + let error = manager + .apply_authenticated_events(&HeaderMap::new(), vec![event()], "client-b") + .await + .unwrap_err(); + assert!(matches!(error, CliError::Unauthorized(_))); +} + +#[tokio::test] +async fn rejected_authenticated_batch_does_not_claim_new_sessions() { + let manager = SessionManager::new(session_test_config()); + let event = + |session_id| NormalizedEvent::AgentStarted(session_event(session_id, "SessionStart")); + manager + .apply_authenticated_events(&HeaderMap::new(), vec![event("owned-session")], "client-a") + .await + .unwrap(); + + let error = manager + .apply_authenticated_events( + &HeaderMap::new(), + vec![event("new-session"), event("owned-session")], + "client-b", + ) + .await + .unwrap_err(); + assert!(matches!(error, CliError::Unauthorized(_))); + + manager + .apply_authenticated_events(&HeaderMap::new(), vec![event("new-session")], "client-a") + .await + .unwrap(); +} + +#[tokio::test] +async fn ended_authenticated_sessions_can_be_reused_by_another_client() { + let manager = SessionManager::new(session_test_config()); + manager + .apply_authenticated_events( + &HeaderMap::new(), + vec![ + NormalizedEvent::AgentStarted(session_event("reused-session", "SessionStart")), + NormalizedEvent::AgentEnded(session_event("reused-session", "SessionEnd")), + ], + "client-a", + ) + .await + .unwrap(); + + manager + .apply_authenticated_events( + &HeaderMap::new(), + vec![NormalizedEvent::AgentStarted(session_event( + "reused-session", + "SessionStart", + ))], + "client-b", + ) + .await + .unwrap(); +} + +#[tokio::test] +async fn authenticated_child_cannot_promote_into_another_clients_parent() { + let manager = SessionManager::new(session_test_config()); + manager + .apply_authenticated_events( + &HeaderMap::new(), + vec![NormalizedEvent::AgentStarted(codex_session_event( + "parent-thread", + "SessionStart", + json!({}), + ))], + "client-a", + ) + .await + .unwrap(); + + let error = manager + .apply_authenticated_events( + &HeaderMap::new(), + vec![NormalizedEvent::AgentStarted(SessionEvent { + session_id: "child-thread".into(), + agent_kind: AgentKind::Codex, + event_name: "SessionStart".into(), + payload: json!({ + "source": {"subagent": {"thread_spawn": {"parent_thread_id": "parent-thread"}}} + }), + metadata: json!({}), + })], + "client-b", + ) + .await + .unwrap_err(); + + assert!(matches!(error, CliError::Unauthorized(_))); + assert!(!has_alignment_alias(&manager, "child-thread").await); + assert!( + !manager + .inner + .lock() + .await + .get("parent-thread") + .unwrap() + .subagents + .contains_key("child-thread") + ); +} + +#[tokio::test] +async fn foreign_pending_child_is_discarded_when_its_parent_starts() { + let manager = SessionManager::new(session_test_config()); + manager + .apply_authenticated_events( + &HeaderMap::new(), + vec![NormalizedEvent::AgentStarted(SessionEvent { + session_id: "child-thread".into(), + agent_kind: AgentKind::Codex, + event_name: "SessionStart".into(), + payload: json!({ + "source": {"subagent": {"thread_spawn": {"parent_thread_id": "parent-thread"}}} + }), + metadata: json!({}), + })], + "client-a", + ) + .await + .unwrap(); + assert!(has_pending_alignment(&manager, "child-thread").await); + + manager + .apply_authenticated_events( + &HeaderMap::new(), + vec![NormalizedEvent::AgentStarted(codex_session_event( + "parent-thread", + "SessionStart", + json!({}), + ))], + "client-b", + ) + .await + .unwrap(); + + assert!(!has_pending_alignment(&manager, "child-thread").await); + assert!(!has_alignment_alias(&manager, "child-thread").await); + assert!( + !manager + .inner + .lock() + .await + .get("parent-thread") + .unwrap() + .subagents + .contains_key("child-thread") + ); +} + +#[tokio::test] +async fn authenticated_alias_rejects_events_from_a_different_client() { + let manager = SessionManager::new(session_test_config()); + let child = || { + NormalizedEvent::AgentStarted(SessionEvent { + session_id: "child-thread".into(), + agent_kind: AgentKind::Codex, + event_name: "SessionStart".into(), + payload: json!({ + "source": {"subagent": {"thread_spawn": {"parent_thread_id": "parent-thread"}}} + }), + metadata: json!({}), + }) + }; + manager + .apply_authenticated_events( + &HeaderMap::new(), + vec![ + NormalizedEvent::AgentStarted(codex_session_event( + "parent-thread", + "SessionStart", + json!({}), + )), + child(), + ], + "client-a", + ) + .await + .unwrap(); + assert!(has_alignment_alias(&manager, "child-thread").await); + + let error = manager + .apply_authenticated_events( + &HeaderMap::new(), + vec![NormalizedEvent::AgentEnded(codex_session_event( + "child-thread", + "SessionEnd", + json!({}), + ))], + "client-b", + ) + .await + .unwrap_err(); + assert!(matches!(error, CliError::Unauthorized(_))); + assert!(has_alignment_alias(&manager, "child-thread").await); +} + +#[tokio::test] +async fn pending_child_gateway_promotion_claims_its_authenticated_parent() { + let manager = SessionManager::new(session_test_config()); + manager + .apply_authenticated_events( + &HeaderMap::new(), + vec![NormalizedEvent::AgentStarted(SessionEvent { + session_id: "child-thread".into(), + agent_kind: AgentKind::Codex, + event_name: "SessionStart".into(), + payload: json!({ + "source": {"subagent": {"thread_spawn": {"parent_thread_id": "parent-thread"}}} + }), + metadata: json!({}), + })], + "client-a", + ) + .await + .unwrap(); + + let active = manager + .start_llm( + &HeaderMap::new(), + LlmGatewayStart { + session_id: Some("child-thread".into()), + ..llm_start() + }, + ) + .await + .unwrap(); + assert_eq!(active.session_id, "parent-thread"); + assert_eq!( + manager + .authenticated_owners + .lock() + .await + .get("parent-thread"), + Some(&"client-a".to_string()) + ); + manager.end_llm(active, json!({}), json!({})).await.unwrap(); +} + +#[tokio::test] +async fn permission_requests_require_an_exact_recorded_tool_call() { + let manager = SessionManager::new(session_test_config()); + let mut session = Session::new( + "permission-session".into(), + AgentKind::ClaudeCode, + SessionConfig::default(), + ); + session.pending_tool_hints.push(PendingToolHint { + hint: ToolHint { + tool_call_id: Some("call-1".into()), + tool_name: Some("Read".into()), + subagent_id: None, + arguments: json!({"path": "README.md"}), + source: "test".into(), + }, + inserted_at: Instant::now(), + }); + manager + .inner + .lock() + .await + .insert("permission-session".into(), session); + + let request = ToolEvent { + session_id: "permission-session".into(), + agent_kind: AgentKind::ClaudeCode, + event_name: "PermissionRequest".into(), + tool_call_id: "call-1".into(), + tool_name: "Read".into(), + subagent_id: None, + arguments: json!({"path": "README.md"}), + result: Value::Null, + status: None, + payload: json!({}), + metadata: json!({}), + }; + manager + .authenticated_owners + .lock() + .await + .insert("permission-session".into(), "client-a".into()); + manager + .authorize_tool_permission(&request, "client-a") + .await + .unwrap(); + + assert!( + manager + .authorize_tool_permission(&request, "client-b") + .await + .is_err() + ); + + let mut changed = request.clone(); + changed.arguments = json!({"path": "secrets.txt"}); + assert!( + manager + .authorize_tool_permission(&changed, "client-a") + .await + .is_err() + ); + + let mut changed = request.clone(); + changed.tool_call_id = "call-2".into(); + assert!( + manager + .authorize_tool_permission(&changed, "client-a") + .await + .is_err() + ); + + let mut changed = request; + changed.tool_name = "Write".into(); + assert!( + manager + .authorize_tool_permission(&changed, "client-a") + .await + .is_err() + ); +} + #[test] fn routing_identity_enrichment_replaces_untrusted_reserved_headers() { let mut request = LlmRequest { diff --git a/crates/core/Cargo.toml b/crates/core/Cargo.toml index 680e7c44e..4ca4d7e67 100644 --- a/crates/core/Cargo.toml +++ b/crates/core/Cargo.toml @@ -85,6 +85,12 @@ log = { version = "0.4", features = ["kv"] } spdlog-rs = { version = "0.5", features = ["log", "multi-thread"] } unicode-general-category = "1.1" +[target.'cfg(unix)'.dependencies] +rustix = { version = "1.1.4", features = ["fs"] } + +[target.'cfg(windows)'.dependencies] +windows-sys = { version = "0.61", features = ["Wdk_Foundation", "Wdk_Storage_FileSystem", "Win32_Foundation", "Win32_Security", "Win32_Security_Authorization", "Win32_Storage_FileSystem", "Win32_System_IO"] } + [dev-dependencies] tokio = { version = "1", features = ["rt", "macros", "sync", "test-util", "rt-multi-thread", "time"] } futures = "0.3" diff --git a/crates/core/src/observability/atof.rs b/crates/core/src/observability/atof.rs index b81bf8df3..03fc022a9 100644 --- a/crates/core/src/observability/atof.rs +++ b/crates/core/src/observability/atof.rs @@ -9,7 +9,7 @@ //! one JSON object per JSONL line. use std::collections::HashMap; -use std::fs::{File, OpenOptions, create_dir_all}; +use std::fs::File; use std::io::{BufWriter, Write}; use std::path::{Path, PathBuf}; use std::sync::{Arc, Mutex, mpsc as std_mpsc}; @@ -23,6 +23,7 @@ use serde_json::Value as Json; #[cfg(feature = "atof-streaming")] use tokio_tungstenite::tungstenite::client::IntoClientRequest; +use super::private_file::{create_private_dir_all, open_private}; use crate::api::event::Event; use crate::api::runtime::EventSubscriberFn; use crate::api::subscriber::{deregister_subscriber, flush_subscribers, register_subscriber}; @@ -380,13 +381,13 @@ impl AtofExporter { let (path, writer, endpoints) = match config.sink { AtofSinkConfig::File(file_sink) => { let path = file_sink.path(); - create_dir_all(&file_sink.output_directory).map_err(|source| { + create_private_dir_all(&file_sink.output_directory).map_err(|source| { AtofExporterError::OpenFile { path: path.clone(), source, } })?; - let file = open_file(&path, file_sink.mode)?; + let file = open_file(&file_sink.output_directory, &path, file_sink.mode)?; log::info!( target: "nemo_relay.observability", event = "storage_access_validated", @@ -588,23 +589,13 @@ fn default_endpoint_timeout_millis() -> u64 { 3_000 } -fn open_file(path: &Path, mode: AtofExporterMode) -> Result { - let mut options = OpenOptions::new(); - options.create(true); - match mode { - AtofExporterMode::Append => { - options.append(true); - } - AtofExporterMode::Overwrite => { - options.write(true).truncate(true); - } - } - options - .open(path) - .map_err(|source| AtofExporterError::OpenFile { +fn open_file(root: &Path, path: &Path, mode: AtofExporterMode) -> Result { + open_private(root, path, matches!(mode, AtofExporterMode::Append)).map_err(|source| { + AtofExporterError::OpenFile { path: path.to_path_buf(), source, - }) + } + }) } fn write_json_value(writer: &mut BufWriter, value: &Json) -> std::result::Result<(), String> { diff --git a/crates/core/src/observability/confined_fs.rs b/crates/core/src/observability/confined_fs.rs new file mode 100644 index 000000000..a89b3b340 --- /dev/null +++ b/crates/core/src/observability/confined_fs.rs @@ -0,0 +1,14 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +#[cfg(unix)] +#[path = "confined_fs/unix.rs"] +mod platform; +#[cfg(windows)] +#[path = "confined_fs/windows.rs"] +mod platform; + +#[cfg(not(any(unix, windows)))] +compile_error!("private observability files require Unix or Windows filesystem primitives"); + +pub(in crate::observability) use platform::ConfinedDir; diff --git a/crates/core/src/observability/confined_fs/unix.rs b/crates/core/src/observability/confined_fs/unix.rs new file mode 100644 index 000000000..b009811fe --- /dev/null +++ b/crates/core/src/observability/confined_fs/unix.rs @@ -0,0 +1,131 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +use rustix::fd::OwnedFd; +use rustix::fs::{ + AtFlags, FileType, Mode, OFlags, fchmod, fcntl_getfl, fcntl_setfl, fstat, mkdirat, openat, + renameat, statat, unlinkat, +}; +use std::ffi::OsStr; +use std::fs::File; +use std::io; +use std::path::Path; + +pub(in crate::observability) struct ConfinedDir(OwnedFd); + +impl ConfinedDir { + pub(in crate::observability) fn open_anchor(path: &Path) -> io::Result { + Ok(Self(openat( + rustix::fs::CWD, + path, + OFlags::RDONLY | OFlags::DIRECTORY | OFlags::CLOEXEC, + Mode::empty(), + )?)) + } + + pub(in crate::observability) fn open_or_create_child(&self, name: &OsStr) -> io::Result { + match self.open_child(name) { + Ok(directory) => Ok(directory), + Err(error) if error.kind() == io::ErrorKind::NotFound => { + match mkdirat(&self.0, name, Mode::RUSR | Mode::WUSR | Mode::XUSR) { + Ok(()) => {} + Err(error) if error.kind() == io::ErrorKind::AlreadyExists => {} + Err(error) => return Err(error.into()), + } + self.open_child(name) + } + Err(error) => Err(io::Error::new( + error.kind(), + format!( + "refusing unsafe observability directory component '{}': {error}", + name.to_string_lossy() + ), + )), + } + } + + fn open_child(&self, name: &OsStr) -> io::Result { + Ok(Self(openat( + &self.0, + name, + OFlags::RDONLY | OFlags::DIRECTORY | OFlags::CLOEXEC | OFlags::NOFOLLOW, + Mode::empty(), + )?)) + } + + pub(in crate::observability) fn open_private_file( + &self, + name: &OsStr, + append: bool, + ) -> io::Result { + self.reject_unsafe_target(name)?; + let mut flags = + OFlags::WRONLY | OFlags::CREATE | OFlags::CLOEXEC | OFlags::NOFOLLOW | OFlags::NONBLOCK; + flags |= if append { + OFlags::APPEND + } else { + OFlags::TRUNC + }; + let descriptor = openat(&self.0, name, flags, Mode::RUSR | Mode::WUSR)?; + let metadata = fstat(&descriptor)?; + if !FileType::from_raw_mode(metadata.st_mode).is_file() { + return Err(io::Error::other(format!( + "observability output '{}' is not a regular file", + name.to_string_lossy() + ))); + } + let mut status_flags = fcntl_getfl(&descriptor)?; + status_flags.remove(OFlags::NONBLOCK); + fcntl_setfl(&descriptor, status_flags)?; + fchmod(&descriptor, Mode::RUSR | Mode::WUSR)?; + Ok(owned_fd_into_file(descriptor)) + } + + pub(in crate::observability) fn create_private_new(&self, name: &OsStr) -> io::Result { + let descriptor = openat( + &self.0, + name, + OFlags::WRONLY | OFlags::CREATE | OFlags::EXCL | OFlags::CLOEXEC | OFlags::NOFOLLOW, + Mode::RUSR | Mode::WUSR, + )?; + fchmod(&descriptor, Mode::RUSR | Mode::WUSR)?; + Ok(owned_fd_into_file(descriptor)) + } + + pub(in crate::observability) fn reject_unsafe_target(&self, name: &OsStr) -> io::Result<()> { + match statat(&self.0, name, AtFlags::SYMLINK_NOFOLLOW) { + Ok(metadata) if FileType::from_raw_mode(metadata.st_mode).is_symlink() => { + Err(io::Error::other(format!( + "refusing symlinked observability file '{}'", + name.to_string_lossy() + ))) + } + Ok(metadata) if !FileType::from_raw_mode(metadata.st_mode).is_file() => { + Err(io::Error::other(format!( + "observability output '{}' is not a regular file", + name.to_string_lossy() + ))) + } + Ok(_) => Ok(()), + Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(()), + Err(error) => Err(error.into()), + } + } + + pub(in crate::observability) fn rename_file( + &self, + _file: &File, + source: &OsStr, + target: &OsStr, + ) -> io::Result<()> { + Ok(renameat(&self.0, source, &self.0, target)?) + } + + pub(in crate::observability) fn remove_file(&self, name: &OsStr) -> io::Result<()> { + Ok(unlinkat(&self.0, name, AtFlags::empty())?) + } +} + +fn owned_fd_into_file(descriptor: OwnedFd) -> File { + File::from(descriptor) +} diff --git a/crates/core/src/observability/confined_fs/windows.rs b/crates/core/src/observability/confined_fs/windows.rs new file mode 100644 index 000000000..0572ea597 --- /dev/null +++ b/crates/core/src/observability/confined_fs/windows.rs @@ -0,0 +1,351 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +use std::ffi::OsStr; +use std::fs::File; +use std::io; +use std::os::windows::ffi::OsStrExt; +use std::os::windows::io::{AsRawHandle, FromRawHandle}; +use std::path::{Path, PathBuf}; +use windows_sys::Wdk::Foundation::OBJECT_ATTRIBUTES; +use windows_sys::Wdk::Storage::FileSystem::{ + FILE_CREATE, FILE_DIRECTORY_FILE, FILE_NON_DIRECTORY_FILE, FILE_OPEN, FILE_OPEN_IF, + FILE_OPEN_REPARSE_POINT, FILE_OVERWRITE_IF, FILE_SYNCHRONOUS_IO_NONALERT, NtCreateFile, +}; +use windows_sys::Win32::Foundation::{ + HANDLE, INVALID_HANDLE_VALUE, LocalFree, OBJ_CASE_INSENSITIVE, RtlNtStatusToDosError, + UNICODE_STRING, +}; +use windows_sys::Win32::Security::Authorization::{ + ConvertStringSecurityDescriptorToSecurityDescriptorW, SDDL_REVISION_1, +}; +use windows_sys::Win32::Security::{ + DACL_SECURITY_INFORMATION, PROTECTED_DACL_SECURITY_INFORMATION, PSECURITY_DESCRIPTOR, + SetKernelObjectSecurity, +}; +use windows_sys::Win32::Storage::FileSystem::{ + BY_HANDLE_FILE_INFORMATION, CreateFileW, DELETE, FILE_ADD_FILE, FILE_ADD_SUBDIRECTORY, + FILE_APPEND_DATA, FILE_ATTRIBUTE_DIRECTORY, FILE_ATTRIBUTE_NORMAL, + FILE_ATTRIBUTE_REPARSE_POINT, FILE_DISPOSITION_INFO, FILE_FLAG_BACKUP_SEMANTICS, + FILE_GENERIC_READ, FILE_GENERIC_WRITE, FILE_READ_ATTRIBUTES, FILE_RENAME_INFO, + FILE_SHARE_DELETE, FILE_SHARE_READ, FILE_SHARE_WRITE, FileDispositionInfo, FileRenameInfo, + GetFileInformationByHandle, OPEN_EXISTING, SYNCHRONIZE, SetFileInformationByHandle, WRITE_DAC, +}; +use windows_sys::Win32::System::IO::IO_STATUS_BLOCK; + +const SHARE_ALL: u32 = FILE_SHARE_READ | FILE_SHARE_WRITE | FILE_SHARE_DELETE; +const PRIVATE_SECURITY_SDDL: &str = "D:P(A;;GA;;;SY)(A;;GA;;;OW)"; + +pub(in crate::observability) struct ConfinedDir { + file: File, + path: PathBuf, +} + +impl ConfinedDir { + pub(in crate::observability) fn open_anchor(path: &Path) -> io::Result { + let anchor_path = path.to_path_buf(); + let path = wide_null(path.as_os_str()); + // SAFETY: `path` is NUL-terminated, and a successful owned handle is transferred to File. + let handle = unsafe { + CreateFileW( + path.as_ptr(), + FILE_GENERIC_READ | FILE_ADD_FILE | FILE_ADD_SUBDIRECTORY, + SHARE_ALL, + std::ptr::null(), + OPEN_EXISTING, + FILE_FLAG_BACKUP_SEMANTICS, + std::ptr::null_mut(), + ) + }; + if handle == INVALID_HANDLE_VALUE { + return Err(io::Error::last_os_error()); + } + // SAFETY: `handle` is newly owned and valid. + let file = unsafe { File::from_raw_handle(handle) }; + validate_handle(&file, true)?; + Ok(Self { + file, + path: anchor_path, + }) + } + + pub(in crate::observability) fn open_or_create_child(&self, name: &OsStr) -> io::Result { + let file = open_relative( + &self.file, + name, + FILE_GENERIC_READ | FILE_ADD_FILE | FILE_ADD_SUBDIRECTORY, + FILE_OPEN_IF, + FILE_DIRECTORY_FILE, + FILE_ATTRIBUTE_DIRECTORY, + true, + )?; + validate_handle(&file, true)?; + restrict_private_handle(&file)?; + Ok(Self { + file, + path: self.path.join(name), + }) + } + + pub(in crate::observability) fn open_private_file( + &self, + name: &OsStr, + append: bool, + ) -> io::Result { + let file = open_relative( + &self.file, + name, + if append { + FILE_APPEND_DATA | FILE_READ_ATTRIBUTES + } else { + FILE_GENERIC_WRITE | FILE_READ_ATTRIBUTES + }, + if append { + FILE_OPEN_IF + } else { + FILE_OVERWRITE_IF + }, + FILE_NON_DIRECTORY_FILE, + FILE_ATTRIBUTE_NORMAL, + true, + )?; + validate_handle(&file, false)?; + restrict_private_handle(&file)?; + Ok(file) + } + + pub(in crate::observability) fn create_private_new(&self, name: &OsStr) -> io::Result { + let file = open_relative( + &self.file, + name, + FILE_GENERIC_WRITE | FILE_READ_ATTRIBUTES | DELETE, + FILE_CREATE, + FILE_NON_DIRECTORY_FILE, + FILE_ATTRIBUTE_NORMAL, + true, + )?; + validate_handle(&file, false)?; + restrict_private_handle(&file)?; + Ok(file) + } + + pub(in crate::observability) fn reject_unsafe_target(&self, name: &OsStr) -> io::Result<()> { + match open_relative( + &self.file, + name, + FILE_READ_ATTRIBUTES, + FILE_OPEN, + FILE_NON_DIRECTORY_FILE, + FILE_ATTRIBUTE_NORMAL, + false, + ) { + Ok(file) => validate_handle(&file, false), + Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(()), + Err(error) => Err(error), + } + } + + pub(in crate::observability) fn rename_file( + &self, + file: &File, + _source: &OsStr, + target: &OsStr, + ) -> io::Result<()> { + let target = wide_null(self.path.join(target).as_os_str()); + let target_name_len = target + .len() + .checked_sub(1) + .ok_or_else(|| io::Error::other("observability output filename is empty"))?; + // `SetFileInformationByHandle` expects the complete fixed-size record plus the + // UTF-16 target name. The one-element `FileName` array remains part of the + // fixed-size Rust representation. + let byte_len = + std::mem::size_of::() + target.len() * std::mem::size_of::(); + let mut storage = vec![0usize; byte_len.div_ceil(std::mem::size_of::())]; + let info = storage.as_mut_ptr().cast::(); + // SAFETY: `storage` is aligned and sized for the header plus the complete UTF-16 name. + unsafe { + (*info).Anonymous.ReplaceIfExists = true; + (*info).RootDirectory = std::ptr::null_mut(); + (*info).FileNameLength = (target_name_len * std::mem::size_of::()) as u32; + std::ptr::copy_nonoverlapping( + target.as_ptr(), + (*info).FileName.as_mut_ptr(), + target.len(), + ); + } + // SAFETY: `info` points to the initialized buffer described above for the duration of the call. + if unsafe { + SetFileInformationByHandle( + file.as_raw_handle(), + FileRenameInfo, + info.cast(), + byte_len as u32, + ) + } == 0 + { + return Err(io::Error::last_os_error()); + } + Ok(()) + } + + pub(in crate::observability) fn remove_file(&self, name: &OsStr) -> io::Result<()> { + let file = open_relative( + &self.file, + name, + DELETE, + FILE_OPEN, + FILE_NON_DIRECTORY_FILE, + FILE_ATTRIBUTE_NORMAL, + false, + )?; + let delete = FILE_DISPOSITION_INFO { DeleteFile: true }; + // SAFETY: `delete` has the required layout and remains valid for the call. + if unsafe { + SetFileInformationByHandle( + file.as_raw_handle(), + FileDispositionInfo, + (&raw const delete).cast(), + std::mem::size_of_val(&delete) as u32, + ) + } == 0 + { + return Err(io::Error::last_os_error()); + } + Ok(()) + } +} + +fn open_relative( + parent: &File, + name: &OsStr, + desired_access: u32, + disposition: u32, + type_option: u32, + attributes: u32, + private: bool, +) -> io::Result { + let mut name = wide(name); + let byte_len = name + .len() + .checked_mul(std::mem::size_of::()) + .and_then(|length| u16::try_from(length).ok()) + .ok_or_else(|| io::Error::other("observability path component is too long"))?; + let unicode = UNICODE_STRING { + Length: byte_len, + MaximumLength: byte_len, + Buffer: name.as_mut_ptr(), + }; + let security_descriptor = private.then(PrivateSecurityDescriptor::new).transpose()?; + let object = OBJECT_ATTRIBUTES { + Length: std::mem::size_of::() as u32, + RootDirectory: parent.as_raw_handle(), + ObjectName: &raw const unicode, + Attributes: OBJ_CASE_INSENSITIVE, + SecurityDescriptor: security_descriptor + .as_ref() + .map_or(std::ptr::null(), |descriptor| descriptor.0.cast()), + SecurityQualityOfService: std::ptr::null(), + }; + let mut status = IO_STATUS_BLOCK::default(); + let mut handle: HANDLE = std::ptr::null_mut(); + // SAFETY: all pointers reference initialized values for the duration of the synchronous call. + let result = unsafe { + NtCreateFile( + &mut handle, + desired_access | SYNCHRONIZE | if private { WRITE_DAC } else { 0 }, + &object, + &mut status, + std::ptr::null(), + attributes, + SHARE_ALL, + disposition, + type_option | FILE_OPEN_REPARSE_POINT | FILE_SYNCHRONOUS_IO_NONALERT, + std::ptr::null(), + 0, + ) + }; + if result < 0 { + // SAFETY: status conversion has no preconditions. + let code = unsafe { RtlNtStatusToDosError(result) }; + return Err(io::Error::from_raw_os_error(code as i32)); + } + // SAFETY: a successful NtCreateFile returns a newly owned handle. + Ok(unsafe { File::from_raw_handle(handle) }) +} + +struct PrivateSecurityDescriptor(PSECURITY_DESCRIPTOR); + +impl PrivateSecurityDescriptor { + fn new() -> io::Result { + let sddl = wide_null(OsStr::new(PRIVATE_SECURITY_SDDL)); + let mut descriptor = std::ptr::null_mut(); + // SAFETY: `sddl` is NUL-terminated and `descriptor` is a valid output pointer. + if unsafe { + ConvertStringSecurityDescriptorToSecurityDescriptorW( + sddl.as_ptr(), + SDDL_REVISION_1, + &mut descriptor, + std::ptr::null_mut(), + ) + } == 0 + { + return Err(io::Error::last_os_error()); + } + Ok(Self(descriptor)) + } +} + +impl Drop for PrivateSecurityDescriptor { + fn drop(&mut self) { + // SAFETY: the descriptor was allocated by + // `ConvertStringSecurityDescriptorToSecurityDescriptorW` and is freed once here. + unsafe { LocalFree(self.0) }; + } +} + +fn restrict_private_handle(file: &File) -> io::Result<()> { + let descriptor = PrivateSecurityDescriptor::new()?; + // SAFETY: the handle is valid and the descriptor remains allocated for the call. + if unsafe { + SetKernelObjectSecurity( + file.as_raw_handle(), + DACL_SECURITY_INFORMATION | PROTECTED_DACL_SECURITY_INFORMATION, + descriptor.0, + ) + } == 0 + { + return Err(io::Error::last_os_error()); + } + Ok(()) +} + +fn validate_handle(file: &File, expect_directory: bool) -> io::Result<()> { + let mut information = BY_HANDLE_FILE_INFORMATION::default(); + // SAFETY: `information` is a writable output buffer with the required layout. + if unsafe { GetFileInformationByHandle(file.as_raw_handle(), &mut information) } == 0 { + return Err(io::Error::last_os_error()); + } + if information.dwFileAttributes & FILE_ATTRIBUTE_REPARSE_POINT != 0 { + return Err(io::Error::other( + "refusing reparse-point observability path component", + )); + } + let is_directory = information.dwFileAttributes & FILE_ATTRIBUTE_DIRECTORY != 0; + if is_directory != expect_directory { + return Err(io::Error::other(if expect_directory { + "observability path component is not a directory" + } else { + "observability output is not a regular file" + })); + } + Ok(()) +} + +fn wide(value: &OsStr) -> Vec { + value.encode_wide().collect() +} + +fn wide_null(value: &OsStr) -> Vec { + value.encode_wide().chain(Some(0)).collect() +} diff --git a/crates/core/src/observability/mod.rs b/crates/core/src/observability/mod.rs index 99236ec78..b36042193 100644 --- a/crates/core/src/observability/mod.rs +++ b/crates/core/src/observability/mod.rs @@ -40,6 +40,7 @@ pub(crate) fn test_mutex() -> &'static Mutex<()> { pub mod atif; pub mod atof; +mod confined_fs; pub(crate) mod manual; pub(crate) mod openinference; pub mod otel; @@ -48,6 +49,7 @@ pub mod otel_logs; pub mod otel_metrics; mod otel_signal; pub mod plugin_component; +mod private_file; pub use otel_signal::{OpenTelemetryRuntimeDiagnostic, OpenTelemetryRuntimeDiagnostics}; diff --git a/crates/core/src/observability/plugin_component.rs b/crates/core/src/observability/plugin_component.rs index 9c435d208..7446f55de 100644 --- a/crates/core/src/observability/plugin_component.rs +++ b/crates/core/src/observability/plugin_component.rs @@ -35,6 +35,7 @@ use serde::{Deserialize, Serialize}; use serde_json::{Map, Value as Json}; use uuid::Uuid; +use super::private_file::atomic_private_write; use crate::api::event::{Event, LogSeverity, ScopeCategory, ValidatedMetricMeasurement}; use crate::api::runtime::{EventSubscriberFn, current_scope_stack, global_context}; use crate::api::scope::ScopeType; @@ -485,9 +486,9 @@ pub struct AtifSectionConfig { #[serde(default, skip_serializing_if = "Option::is_none")] pub output_directory: Option, /// Filename template. `{session_id}` is replaced with the top-level trajectory scope UUID, and - /// `{metadata.:-fallback}` placeholders use path-safe strings from the top-level scope - /// metadata or the optional literal fallback. When [`storage`] is non-empty, the rendered - /// filename is appended to each backend's key prefix. + /// `{metadata.:-fallback}` placeholders sanitize strings from the top-level scope + /// metadata into path-safe filename fragments or use the optional literal fallback. When + /// [`storage`] is non-empty, the rendered filename is appended to each backend's key prefix. /// /// [`storage`]: Self::storage #[serde(default = "default_atif_filename_template")] @@ -2437,6 +2438,7 @@ struct AtifDispatcher { struct ManagedAtifExporter { exporter: AtifExporter, filename: String, + local_root: Option, local_path: Option, correlation: AtifCorrelation, observed_events: Vec, @@ -2452,6 +2454,7 @@ struct PendingAtifWrite { // object-store feature; without it, only the local sink reads `local_path`. #[cfg_attr(not(feature = "object-store"), allow(dead_code))] filename: String, + local_root: Option, local_path: Option, payload: Vec, } @@ -2465,6 +2468,7 @@ struct PendingAtifExport { agent_uuid: Uuid, exporter: AtifExporter, filename: String, + local_root: Option, local_path: Option, correlation: AtifCorrelation, } @@ -2559,26 +2563,27 @@ impl AtifDispatcher { // subscriber is attached after that start event has already been // emitted. let session_id = event.uuid().to_string(); - let (filename, local_path) = match self.prepare_destination(&session_id, event.metadata()) { - Ok(destination) => destination, - Err(error) => { - self.record_runtime_failure( - "atif.destination_render_failed", - Some("filename_template".into()), - error.clone(), - Some(session_id.clone()), - ); - log::warn!( - target: "nemo_relay.observability", - event = "atif_destination_render_failed", - plugin_kind = OBSERVABILITY_PLUGIN_KIND, - exporter = "atif", - session_id = session_id.as_str(); - "ATIF destination rendering failed: {error}" - ); - return None; - } - }; + let (filename, local_root, local_path) = + match self.prepare_destination(&session_id, event.metadata()) { + Ok(destination) => destination, + Err(error) => { + self.record_runtime_failure( + "atif.destination_render_failed", + Some("filename_template".into()), + error.clone(), + Some(session_id.clone()), + ); + log::warn!( + target: "nemo_relay.observability", + event = "atif_destination_render_failed", + plugin_kind = OBSERVABILITY_PLUGIN_KIND, + exporter = "atif", + session_id = session_id.as_str(); + "ATIF destination rendering failed: {error}" + ); + return None; + } + }; let exporter = AtifExporter::new(session_id.clone(), self.agent_info()); (exporter.subscriber())(event); let correlation = AtifCorrelation::from_event(event); @@ -2588,6 +2593,7 @@ impl AtifDispatcher { ManagedAtifExporter { exporter, filename, + local_root, local_path, correlation, observed_events: vec![event.clone()], @@ -2755,6 +2761,7 @@ impl AtifDispatcher { agent_uuid, exporter: agent.exporter.clone(), filename: agent.filename.clone(), + local_root: agent.local_root.clone(), local_path: agent.local_path.clone(), correlation: agent.correlation.clone(), }); @@ -2804,16 +2811,33 @@ impl AtifDispatcher { &self, session_id: &str, metadata: Option<&Json>, - ) -> Result<(String, Option), String> { + ) -> Result<(String, Option, Option), String> { validate_atif_filename_template(&self.config.filename_template)?; + if !is_safe_atif_session_id(session_id) { + return Err("ATIF session_id must be a non-empty path-safe filename component".into()); + } let filename = render_atif_filename(&self.config.filename_template, session_id, metadata)?; + let rendered_path = Path::new(&filename); + if rendered_path.is_absolute() + || rendered_path.components().any(|component| { + matches!( + component, + Component::ParentDir + | Component::CurDir + | Component::RootDir + | Component::Prefix(_) + ) + }) + { + return Err("rendered ATIF filename must remain a path-safe relative path".into()); + } let directory = self .config .output_directory .clone() .unwrap_or_else(default_output_directory); let path = directory.join(&filename); - Ok((filename, Some(path))) + Ok((filename, Some(directory), Some(path))) } fn sink_targets(&self) -> Vec { @@ -2856,6 +2880,13 @@ impl AtifDispatcher { } } +fn is_safe_atif_session_id(value: &str) -> bool { + !matches!(value, "" | "." | "..") + && value + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_' | b'.' | b'~')) +} + fn is_valid_atif_metadata_selector(selector: &str) -> bool { !selector.is_empty() && selector.split('.').all(|segment| { @@ -2957,8 +2988,8 @@ fn render_atif_filename( }; } let value = match resolved { - Some(Json::String(value)) => value.as_str(), - None | Some(Json::Null) => fallback.ok_or_else(|| { + Some(Json::String(value)) => sanitize_atif_metadata_fragment(value), + None | Some(Json::Null) => fallback.map(str::to_string).ok_or_else(|| { format!( "filename_template placeholder '{{metadata.{selector}}}' must resolve to a string" ) @@ -2969,17 +3000,45 @@ fn render_atif_filename( )); } }; - if !is_safe_atif_metadata_path(value) { + if !is_safe_atif_metadata_path(&value) { return Err(format!( "metadata path '{selector}' must be a path-safe relative fragment" )); } - rendered.replace_range(start..=end, value); + rendered.replace_range(start..=end, &value); cursor = start + value.len(); } Ok(rendered) } +fn sanitize_atif_metadata_fragment(value: &str) -> String { + let mut characters = value.chars().peekable(); + let mut sanitized = String::with_capacity(value.len()); + let mut replacing = false; + + while let Some(character) = characters.next() { + let safe = if character == '.' { + let mut count = 1; + while characters.next_if_eq(&'.').is_some() { + count += 1; + } + count == 1 && !(sanitized.is_empty() && characters.peek().is_none()) + } else { + character.is_ascii_alphanumeric() || matches!(character, '-' | '_' | '~') + }; + + if safe { + sanitized.push(character); + replacing = false; + } else if !replacing { + sanitized.push('-'); + replacing = true; + } + } + + sanitized +} + fn is_safe_atif_metadata_path(value: &str) -> bool { !value.is_empty() && value.split('/').all(|segment| { @@ -3064,6 +3123,7 @@ fn prepare_atif_file( prepare_atif_payload( agent_uuid, agent.filename.clone(), + agent.local_root.clone(), agent.local_path.clone(), trajectory, observed_events, @@ -3088,6 +3148,7 @@ fn prepare_atif_shutdown_file( prepare_atif_payload( export.agent_uuid, export.filename.clone(), + export.local_root.clone(), export.local_path.clone(), trajectory, observed_events, @@ -3098,6 +3159,7 @@ fn prepare_atif_shutdown_file( fn prepare_atif_payload( agent_uuid: Uuid, filename: String, + local_root: Option, local_path: Option, trajectory: crate::observability::atif::AtifTrajectory, observed_events: Vec, @@ -3131,6 +3193,7 @@ fn prepare_atif_payload( agent_uuid, session_id: agent_uuid.to_string(), filename, + local_root, local_path, payload, }) @@ -3145,9 +3208,9 @@ fn write_atif( .iter() .map(|label| { let result = match label { - SinkLabel::Local => match &write.local_path { - Some(path) => write_atif_local(path, &write.payload), - None => Err(std::io::Error::other( + SinkLabel::Local => match (&write.local_root, &write.local_path) { + (Some(root), Some(path)) => write_atif_local(root, path, &write.payload), + _ => Err(std::io::Error::other( "ATIF local destination has no output path", )), }, @@ -3162,9 +3225,9 @@ fn write_atif( .all(|label| matches!(label, SinkLabel::Remote(_))) && results.iter().all(|(_, result)| result.is_err()) { - let fallback = match &write.local_path { - Some(path) => write_atif_local(path, &write.payload), - None => Err(std::io::Error::other( + let fallback = match (&write.local_root, &write.local_path) { + (Some(root), Some(path)) => write_atif_local(root, path, &write.payload), + _ => Err(std::io::Error::other( "ATIF local fallback has no output path", )), }; @@ -3173,11 +3236,8 @@ fn write_atif( results } -fn write_atif_local(path: &PathBuf, payload: &[u8]) -> std::io::Result<()> { - if let Some(parent) = path.parent() { - std::fs::create_dir_all(parent)?; - } - std::fs::write(path, payload) +fn write_atif_local(root: &Path, path: &Path, payload: &[u8]) -> std::io::Result<()> { + atomic_private_write(root, path, payload) } #[cfg(feature = "object-store")] diff --git a/crates/core/src/observability/private_file.rs b/crates/core/src/observability/private_file.rs new file mode 100644 index 000000000..ad15c6e35 --- /dev/null +++ b/crates/core/src/observability/private_file.rs @@ -0,0 +1,173 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +use super::confined_fs::ConfinedDir; +use std::ffi::OsString; +use std::fs::{self, File}; +use std::io::{self, Write}; +use std::path::{Component, Path, PathBuf}; + +pub(super) fn create_private_dir_all(path: &Path) -> io::Result<()> { + open_or_create_private_dir(path).map(drop) +} + +fn open_or_create_private_dir(path: &Path) -> io::Result { + let absolute = if path.is_absolute() { + path.to_path_buf() + } else { + std::env::current_dir()?.join(path) + }; + let mut anchor = PathBuf::new(); + for component in absolute.components() { + match component { + Component::Prefix(_) if anchor.as_os_str().is_empty() => { + anchor.push(component.as_os_str()); + } + // Windows absolute paths are represented as a drive prefix followed by a root + // component (`C:` then `\\`). Both components make up the trusted filesystem anchor. + Component::RootDir + if anchor.as_os_str().is_empty() + || matches!(anchor.components().next(), Some(Component::Prefix(_))) + && anchor.components().nth(1).is_none() => + { + anchor.push(component.as_os_str()); + } + Component::Normal(name) => anchor.push(name), + Component::CurDir => {} + Component::ParentDir | Component::Prefix(_) | Component::RootDir => { + return Err(io::Error::other(format!( + "observability directory '{}' contains unsafe traversal", + path.display() + ))); + } + } + } + + if anchor.as_os_str().is_empty() { + return Err(io::Error::other(format!( + "observability directory '{}' has no filesystem anchor", + path.display() + ))); + } + + let mut existing = anchor.as_path(); + let mut missing = Vec::new(); + while !fs::metadata(existing).is_ok_and(|metadata| metadata.is_dir()) { + let name = existing.file_name().ok_or_else(|| { + io::Error::other(format!( + "observability directory '{}' has no existing filesystem anchor", + path.display() + )) + })?; + missing.push(name.to_owned()); + existing = existing.parent().ok_or_else(|| { + io::Error::other(format!( + "observability directory '{}' has no existing filesystem anchor", + path.display() + )) + })?; + } + + // The configured root is trusted and may include platform filesystem aliases such as + // macOS `/var`. Descendants are still opened relative to this stable directory handle + // without following symlinks. + let mut current = ConfinedDir::open_anchor(existing)?; + for name in missing.into_iter().rev() { + current = current.open_or_create_child(&name)?; + } + Ok(current) +} + +#[cfg(all(test, windows))] +mod tests { + use super::{atomic_private_write, create_private_dir_all}; + + #[test] + fn absolute_temp_directory_is_accepted() { + let temporary = tempfile::tempdir().expect("temporary directory should be created"); + let output = temporary.path().join("atof"); + + create_private_dir_all(&output).expect("absolute Windows output directory should open"); + atomic_private_write(&output, &output.join("trajectory.json"), b"{}") + .expect("absolute Windows output file should write atomically"); + + assert_eq!( + std::fs::read(output.join("trajectory.json")).unwrap(), + b"{}" + ); + } +} + +pub(super) fn open_private(root: &Path, path: &Path, append: bool) -> io::Result { + let (parent, filename) = prepare_confined_parent(root, path)?; + parent.open_private_file(&filename, append) +} + +pub(super) fn atomic_private_write(root: &Path, path: &Path, payload: &[u8]) -> io::Result<()> { + let (parent, filename) = prepare_confined_parent(root, path)?; + parent.reject_unsafe_target(&filename)?; + let filename_text = filename + .to_str() + .ok_or_else(|| io::Error::other("observability output filename is not valid text"))?; + let mut last_collision = None; + for _ in 0..16 { + let temporary = format!(".{filename_text}.{}.tmp", uuid::Uuid::now_v7()); + match parent.create_private_new(std::ffi::OsStr::new(&temporary)) { + Ok(mut file) => { + let result = (|| { + file.write_all(payload)?; + file.sync_all()?; + parent.reject_unsafe_target(&filename)?; + parent.rename_file(&file, std::ffi::OsStr::new(&temporary), &filename) + })(); + if result.is_err() { + let _ = parent.remove_file(std::ffi::OsStr::new(&temporary)); + } + return result; + } + Err(error) if error.kind() == io::ErrorKind::AlreadyExists => { + last_collision = Some(error); + } + Err(error) => return Err(error), + } + } + Err(last_collision.unwrap_or_else(|| { + io::Error::other("failed to allocate a private observability temporary file") + })) +} + +fn prepare_confined_parent(root: &Path, path: &Path) -> io::Result<(ConfinedDir, OsString)> { + let mut current = open_or_create_private_dir(root)?; + let relative = path.strip_prefix(root).map_err(|_| { + io::Error::other(format!( + "observability output '{}' is outside configured directory '{}'", + path.display(), + root.display() + )) + })?; + if relative.as_os_str().is_empty() + || relative.components().any(|component| { + matches!( + component, + Component::ParentDir + | Component::CurDir + | Component::RootDir + | Component::Prefix(_) + ) + }) + { + return Err(io::Error::other( + "observability output must be a confined relative file path", + )); + } + let filename = relative + .file_name() + .ok_or_else(|| io::Error::other("observability output path has no filename"))? + .to_owned(); + let relative_parent = relative.parent().unwrap_or_else(|| Path::new("")); + for component in relative_parent.components() { + let component = component.as_os_str(); + current = current.open_or_create_child(component)?; + } + Ok((current, filename)) +} diff --git a/crates/core/src/plugin/dynamic/worker.rs b/crates/core/src/plugin/dynamic/worker.rs index 92e713471..48ac660e1 100644 --- a/crates/core/src/plugin/dynamic/worker.rs +++ b/crates/core/src/plugin/dynamic/worker.rs @@ -1033,6 +1033,7 @@ fn spawn_worker_process(spec: WorkerProcessLaunch<'_>) -> crate::plugin::Result< (Command::new(entrypoint), command_display) } }; + minimize_worker_environment(&mut command); command .current_dir(manifest_dir) .env("NEMO_RELAY_WORKER_ID", spec.activation_id) @@ -1054,6 +1055,25 @@ fn spawn_worker_process(spec: WorkerProcessLaunch<'_>) -> crate::plugin::Result< }) } +fn minimize_worker_environment(command: &mut Command) { + const ALLOWLIST: &[&str] = &[ + "PATH", + "SYSTEMROOT", + "WINDIR", + "TMPDIR", + "TEMP", + "TMP", + "SSL_CERT_FILE", + "SSL_CERT_DIR", + ]; + let retained = ALLOWLIST + .iter() + .filter_map(|name| std::env::var_os(name).map(|value| (*name, value))) + .collect::>(); + command.env_clear(); + command.envs(retained); +} + fn resolve_python_executable( plugin_id: &str, environment_ref: Option<&str>, diff --git a/crates/core/tests/fixtures/worker_plugin/src/main.rs b/crates/core/tests/fixtures/worker_plugin/src/main.rs index a15869d17..b6de2a52d 100644 --- a/crates/core/tests/fixtures/worker_plugin/src/main.rs +++ b/crates/core/tests/fixtures/worker_plugin/src/main.rs @@ -18,9 +18,6 @@ struct FixtureWorkerPlugin; impl WorkerPlugin for FixtureWorkerPlugin { fn plugin_id(&self) -> &str { - if std::env::var("FIXTURE_WORKER_PLUGIN_ID").as_deref() == Ok("other_worker") { - return "other_worker"; - } "fixture_worker" } diff --git a/crates/core/tests/integration/worker_plugin_tests.rs b/crates/core/tests/integration/worker_plugin_tests.rs index 8609c71f0..5874093ae 100644 --- a/crates/core/tests/integration/worker_plugin_tests.rs +++ b/crates/core/tests/integration/worker_plugin_tests.rs @@ -1175,12 +1175,11 @@ async fn worker_invalid_registration_plan_fails_activation() { #[tokio::test] async fn worker_handshake_plugin_id_mismatch_reports_config_error() { let _guard = WORKER_PLUGIN_TEST_LOCK.lock().await; - let _env = EnvVarGuard::set("FIXTURE_WORKER_PLUGIN_ID", "other_worker"); let fixture = build_fixture_worker(); let (_manifest_dir, manifest_ref) = write_manifest(fixture.binary_path()); let error = match load_worker_plugins([WorkerPluginLoadSpec { - plugin_id: "fixture_worker".into(), + plugin_id: "other_worker".into(), manifest_ref: manifest_ref.to_string_lossy().into_owned(), environment_ref: None, config: Map::new(), @@ -1191,7 +1190,10 @@ async fn worker_handshake_plugin_id_mismatch_reports_config_error() { } Err(error) => error.to_string(), }; - assert!(error.contains("returned id 'other_worker'"), "{error}"); + assert!( + error.contains("manifest id 'fixture_worker' does not match expected id 'other_worker'"), + "{error}" + ); } #[tokio::test] @@ -1844,35 +1846,6 @@ fn assert_error_mentions_manifest_relative_entrypoint( ); } -struct EnvVarGuard { - key: &'static str, - previous: Option, -} - -impl EnvVarGuard { - fn set(key: &'static str, value: &str) -> Self { - let previous = std::env::var(key).ok(); - // SAFETY: this module serializes worker tests with WORKER_PLUGIN_TEST_LOCK. - unsafe { - std::env::set_var(key, value); - } - Self { key, previous } - } -} - -impl Drop for EnvVarGuard { - fn drop(&mut self) { - // SAFETY: this module serializes worker tests with WORKER_PLUGIN_TEST_LOCK. - unsafe { - if let Some(previous) = &self.previous { - std::env::set_var(self.key, previous); - } else { - std::env::remove_var(self.key); - } - } - } -} - fn find_event<'a>( events: &'a [Event], name: &str, diff --git a/crates/core/tests/unit/observability/atof_tests.rs b/crates/core/tests/unit/observability/atof_tests.rs index 87c398c86..6b7ff62c1 100644 --- a/crates/core/tests/unit/observability/atof_tests.rs +++ b/crates/core/tests/unit/observability/atof_tests.rs @@ -38,7 +38,7 @@ fn temp_dir(prefix: &str) -> PathBuf { .as_nanos(); let path = std::env::temp_dir().join(format!("nemo-relay-{prefix}-{id}")); fs::create_dir_all(&path).unwrap(); - path + path.canonicalize().unwrap() } fn reset_global() { @@ -1334,22 +1334,86 @@ fn missing_output_directory_is_created() { assert!(output_path.exists()); } +#[cfg(unix)] #[test] -fn invalid_filename_errors_cleanly() { - let dir = temp_dir("atof-invalid-filename"); +fn output_directory_resolves_symlinked_configured_root() { + use std::os::unix::fs::symlink; + + let parent = temp_dir("atof-symlinked-output-parent"); + let outside = temp_dir("atof-symlinked-output-outside"); + let link = parent.join("linked"); + symlink(&outside, &link).unwrap(); + let output_dir = link.join("atof"); + + let exporter = AtofExporter::new( + AtofExporterConfig::new() + .with_output_directory(&output_dir) + .with_filename("events.jsonl"), + ) + .unwrap(); + + assert_eq!( + exporter.path(), + Some(output_dir.join("events.jsonl").as_path()) + ); + assert!(outside.join("atof/events.jsonl").exists()); +} + +#[cfg(unix)] +#[test] +fn existing_non_regular_output_is_rejected() { + use std::os::unix::net::UnixListener; + + let temp = tempfile::tempdir_in("/tmp").unwrap(); + let path = temp.path().join("events.jsonl"); + let _listener = UnixListener::bind(&path).unwrap(); let error = match AtofExporter::new( AtofExporterConfig::new() - .with_output_directory(&dir) - .with_filename("missing-parent/events.jsonl"), + .with_output_directory(temp.path()) + .with_filename("events.jsonl"), ) { - Ok(_) => panic!("expected invalid filename path error"), + Ok(_) => panic!("expected a non-regular output target to be rejected"), Err(error) => error, }; assert!(matches!(error, AtofExporterError::OpenFile { .. })); } +#[test] +fn nested_filename_creates_private_parent_directories() { + let dir = temp_dir("atof-invalid-filename"); + + let exporter = AtofExporter::new( + AtofExporterConfig::new() + .with_output_directory(&dir) + .with_filename("missing-parent/events.jsonl"), + ) + .unwrap(); + + assert_eq!( + exporter.path(), + Some(dir.join("missing-parent/events.jsonl").as_path()) + ); + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + + let parent_mode = std::fs::metadata(dir.join("missing-parent")) + .unwrap() + .permissions() + .mode() + & 0o777; + let file_mode = std::fs::metadata(dir.join("missing-parent/events.jsonl")) + .unwrap() + .permissions() + .mode() + & 0o777; + assert_eq!(parent_mode, 0o700); + assert_eq!(file_mode, 0o600); + } +} + #[test] #[cfg(feature = "atof-streaming")] fn invalid_endpoint_config_errors_cleanly() { @@ -1727,16 +1791,13 @@ fn atof_config_helpers_cover_file_path_and_replace_dots_policy() { #[cfg(target_os = "linux")] #[test] -fn atof_file_sink_reports_deferred_dev_full_write_failures() { - let exporter = AtofExporter::new( +fn atof_file_sink_rejects_dev_full_as_an_unsafe_output_target() { + let result = AtofExporter::new( AtofExporterConfig::new() .with_output_directory("/dev") .with_filename("full"), - ) - .unwrap(); - exporter.subscriber()(&make_mark_event("write-failure")); - assert!(exporter.force_flush().is_err()); - assert!(exporter.shutdown().is_err()); + ); + assert!(matches!(result, Err(AtofExporterError::OpenFile { .. }))); } #[test] diff --git a/crates/core/tests/unit/observability/plugin_component_tests.rs b/crates/core/tests/unit/observability/plugin_component_tests.rs index 8ae135a95..6e1093d0a 100644 --- a/crates/core/tests/unit/observability/plugin_component_tests.rs +++ b/crates/core/tests/unit/observability/plugin_component_tests.rs @@ -60,7 +60,7 @@ fn temp_dir(prefix: &str) -> PathBuf { .as_nanos(); let path = std::env::temp_dir().join(format!("nemo-relay-{prefix}-{id}")); fs::create_dir_all(&path).unwrap(); - path + path.canonicalize().unwrap() } #[cfg(feature = "atof-streaming")] @@ -2825,7 +2825,7 @@ fn atif_defaults_create_one_file_per_top_level_agent() { } #[test] -fn atif_filename_template_routes_by_metadata_and_skips_invalid_paths() { +fn atif_filename_template_sanitizes_metadata_paths() { let _guard = crate::observability::test_mutex().lock().unwrap(); reset_runtime(); let dir = temp_dir("observability-atif-metadata-template"); @@ -2839,15 +2839,15 @@ fn atif_filename_template_routes_by_metadata_and_skips_invalid_paths() { })); futures::executor::block_on(initialize_plugins_exact(config)).unwrap(); - let invalid = crate::api::scope::push_scope( + let sanitized = crate::api::scope::push_scope( PushScopeParams::builder() - .name("invalid-metadata-path-agent") + .name("sanitized-metadata-path-agent") .scope_type(ScopeType::Agent) .metadata(json!({"routing": {"artifact_path": "../escape"}})) .build(), ) .unwrap(); - pop(&invalid); + pop(&sanitized); let valid = crate::api::scope::push_scope( PushScopeParams::builder() @@ -2861,32 +2861,20 @@ fn atif_filename_template_routes_by_metadata_and_skips_invalid_paths() { flush_subscribers().unwrap(); assert!( - crate::plugin::active_plugin_report() - .unwrap() - .runtime_diagnostics - .iter() - .any(|diagnostic| diagnostic.code == "atif.destination_render_failed") - ); - let teardown = clear_plugin_configuration().unwrap_err(); - assert!( - teardown - .to_string() - .contains("atif.destination_render_failed") - ); - let invalid_filename = format!("trajectory-{}.json", invalid.uuid); - assert!( - !dir.join(&invalid_filename).exists() - && !dir.join("../escape").join(&invalid_filename).exists(), - "unsafe metadata path should not produce a trajectory file" + dir.join(format!("-escape/trajectory-{}.json", sanitized.uuid)) + .exists(), + "unsafe metadata groups should be replaced with a dash" ); assert!( dir.join(format!( - "tenant-a/session-123/trajectory-{}.json", + "tenant-a-session-123/trajectory-{}.json", valid.uuid )) .exists() ); + clear_plugin_configuration().unwrap(); + futures::executor::block_on(initialize_plugins_exact(plugin_config(json!({ "atif": { "enabled": true, @@ -3341,6 +3329,7 @@ fn write_atif_reports_missing_local_path_and_unregistered_remote_sink() { agent_uuid, session_id: agent_uuid.to_string(), filename: "trajectory.json".into(), + local_root: None, local_path: None, payload: b"{}".to_vec(), }; @@ -3372,6 +3361,7 @@ fn write_atif_spills_to_local_when_all_remote_sinks_fail() { agent_uuid, session_id: agent_uuid.to_string(), filename: "trajectory.json".into(), + local_root: Some(dir.clone()), local_path: Some(path.clone()), payload: b"{}".to_vec(), }; @@ -3388,8 +3378,10 @@ fn write_atif_spills_to_local_when_all_remote_sinks_fail() { #[test] fn atif_dispatcher_default_output_path_uses_current_directory() { let dispatcher = AtifDispatcher::new(AtifSectionConfig::default()); - let (filename, local_path) = dispatcher.prepare_destination("session-1", None).unwrap(); + let (filename, local_root, local_path) = + dispatcher.prepare_destination("session-1", None).unwrap(); assert_eq!(filename, "nemo-relay-atif-session-1.json"); + assert_eq!(local_root.unwrap(), std::env::current_dir().unwrap()); assert_eq!( local_path.unwrap(), std::env::current_dir() @@ -3398,6 +3390,87 @@ fn atif_dispatcher_default_output_path_uses_current_directory() { ); } +#[test] +fn atif_session_id_cannot_escape_the_output_directory() { + let dispatcher = AtifDispatcher::new(AtifSectionConfig::default()); + for session_id in ["../escape", "/absolute", "nested/../../escape"] { + assert!( + dispatcher.prepare_destination(session_id, None).is_err(), + "unsafe session id should be rejected: {session_id:?}" + ); + } +} + +#[cfg(unix)] +#[test] +fn atif_local_write_replaces_regular_files_privately_and_rejects_symlinks() { + use std::os::unix::fs::{PermissionsExt, symlink}; + + let dir = tempfile::tempdir().unwrap(); + let dir_path = dir.path().canonicalize().unwrap(); + let path = dir_path.join("trajectory.json"); + write_atif_local(&dir_path, &path, b"first").unwrap(); + write_atif_local(&dir_path, &path, b"second").unwrap(); + assert_eq!(fs::read(&path).unwrap(), b"second"); + assert_eq!( + fs::metadata(&path).unwrap().permissions().mode() & 0o777, + 0o600 + ); + + let target = dir_path.join("target.json"); + fs::write(&target, b"preserve").unwrap(); + let link = dir_path.join("link.json"); + symlink(&target, &link).unwrap(); + assert!(write_atif_local(&dir_path, &link, b"overwrite").is_err()); + assert_eq!(fs::read(target).unwrap(), b"preserve"); + + let outside = tempfile::tempdir().unwrap(); + let outside_path = outside.path().canonicalize().unwrap(); + let linked_parent = dir_path.join("linked-parent"); + symlink(&outside_path, &linked_parent).unwrap(); + assert!(write_atif_local(&dir_path, &linked_parent.join("escaped.json"), b"escape").is_err()); + assert!(!outside_path.join("escaped.json").exists()); +} + +#[cfg(unix)] +#[test] +fn atif_local_write_resolves_symlinked_configured_root() { + use std::os::unix::fs::symlink; + + let parent = tempfile::tempdir().unwrap(); + let outside = tempfile::tempdir().unwrap(); + let parent_path = parent.path().canonicalize().unwrap(); + let outside_path = outside.path().canonicalize().unwrap(); + let link = parent_path.join("linked"); + symlink(&outside_path, &link).unwrap(); + let root = link.join("atif"); + let path = root.join("trajectory.json"); + + write_atif_local(&root, &path, b"trajectory").unwrap(); + assert_eq!( + fs::read(outside_path.join("atif/trajectory.json")).unwrap(), + b"trajectory" + ); +} + +#[test] +fn atif_metadata_template_values_are_sanitized() { + assert_eq!(sanitize_atif_metadata_fragment("tenant-a"), "tenant-a"); + assert_eq!(sanitize_atif_metadata_fragment("../tenant/a"), "-tenant-a"); + assert_eq!( + sanitize_atif_metadata_fragment(r"tenant\project"), + "tenant-project" + ); + assert_eq!( + sanitize_atif_metadata_fragment("tenant / : project"), + "tenant-project" + ); + assert_eq!(sanitize_atif_metadata_fragment("run...name"), "run-name"); + assert_eq!(sanitize_atif_metadata_fragment("ténant///项目"), "t-nant-"); + assert_eq!(sanitize_atif_metadata_fragment("."), "-"); + assert_eq!(sanitize_atif_metadata_fragment(".."), "-"); +} + #[test] fn atif_metadata_template_values_must_be_safe_path_fragments() { assert!( @@ -3477,7 +3550,7 @@ fn atif_metadata_template_values_must_be_safe_path_fragments() { let destination = nested_dispatcher .prepare_destination("session-1", Some(&nested_string)) .unwrap(); - assert_eq!(destination.0, "tenant-a/team_1/trajectory-session-1.json"); + assert_eq!(destination.0, "tenant-a-team_1/trajectory-session-1.json"); for template in [ "/tmp/trajectory-{session_id}.json", @@ -3536,6 +3609,7 @@ fn atif_payload_merges_correlation_with_existing_trajectory_extra() { agent_uuid, format!("trajectory-{agent_uuid}.json"), None, + None, trajectory, Vec::new(), AtifCorrelation { diff --git a/examples/language-binding-plugin/rust/Cargo.lock b/examples/language-binding-plugin/rust/Cargo.lock index 27bdb866a..e8e1973e5 100644 --- a/examples/language-binding-plugin/rust/Cargo.lock +++ b/examples/language-binding-plugin/rust/Cargo.lock @@ -415,6 +415,16 @@ dependencies = [ "typeid", ] +[[package]] +name = "errno" +version = "0.3.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" +dependencies = [ + "libc", + "windows-sys 0.61.2", +] + [[package]] name = "find-msvc-tools" version = "0.1.10" @@ -990,6 +1000,12 @@ dependencies = [ "windows-link", ] +[[package]] +name = "linux-raw-sys" +version = "0.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "32a66949e030da00e8c7d4434b251670a91556f4144941d37452769c25d58a53" + [[package]] name = "litemap" version = "0.8.2" @@ -1067,6 +1083,7 @@ dependencies = [ "opentelemetry-semantic-conventions", "opentelemetry_sdk", "reqwest 0.12.28", + "rustix", "semver", "serde", "serde_json", @@ -1083,6 +1100,7 @@ dependencies = [ "typed-builder", "unicode-general-category", "uuid", + "windows-sys 0.61.2", ] [[package]] @@ -1690,6 +1708,19 @@ dependencies = [ "semver", ] +[[package]] +name = "rustix" +version = "1.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6fe4565b9518b83ef4f91bb47ce29620ca828bd32cb7e408f0062e9930ba190" +dependencies = [ + "bitflags", + "errno", + "libc", + "linux-raw-sys", + "windows-sys 0.61.2", +] + [[package]] name = "rustls" version = "0.23.43" diff --git a/justfile b/justfile index f3583c317..72cbb27a3 100644 --- a/justfile +++ b/justfile @@ -1393,8 +1393,9 @@ test-python-plugin-e2e: cargo build -p nemo-relay-cli cli="$NEMO_RELAY_REPO_ROOT/target/debug/nemo-relay" config="$tmp/gateway.toml" - # Explicit --config paths must exist; plugin state is written to sibling files. + # Explicit --config paths must exist; plugin policy and state are sibling files. : > "$config" + printf '[plugins.policy.defaults]\nattestation = "integrity_only"\n' > "$tmp/plugins.toml" manifest="$NEMO_RELAY_REPO_ROOT/examples/python-grpc-worker-plugin/relay-plugin.toml" PIP_FIND_LINKS="$tmp/wheels" NEMO_RELAY_PYTHON="$python_executable" \ "$cli" --config "$config" plugins add "$manifest" diff --git a/scripts/test-claude-plugin-e2e.sh b/scripts/test-claude-plugin-e2e.sh index 037e0a587..fbadfe462 100755 --- a/scripts/test-claude-plugin-e2e.sh +++ b/scripts/test-claude-plugin-e2e.sh @@ -51,7 +51,7 @@ export XDG_DATA_HOME="$work/data" export XDG_RUNTIME_DIR="$work/runtime" export TMPDIR="$work/tmp" export PATH="$repo_root/target/debug:$PATH" -export ANTHROPIC_API_KEY="relay-claude-e2e-key" +export ANTHROPIC_AUTH_TOKEN="relay-claude-e2e-token" export CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC=1 export DISABLE_AUTOUPDATER=1 export NEMO_RELAY_GATEWAY_URL="http://127.0.0.1:1" @@ -69,6 +69,13 @@ mkdir -p \ "$work/provider-barrier" \ "$work/workspace" +cat >"$HOME/.claude.json" <<'EOF' +{ + "hasCompletedOnboarding": true, + "theme": "dark" +} +EOF + provider_ready="$work/provider-ready.json" provider_log="$work/provider-requests.jsonl" python3 "$repo_root/scripts/test-support/codex_mock_provider.py" \ @@ -97,7 +104,7 @@ kind = "observability" enabled = true [components.config] -version = 2 +version = 4 [components.config.atof] enabled = true @@ -190,36 +197,104 @@ PY } run_transparent_claude() { - output="$work/claude-transparent.json" - stderr="$work/claude-transparent.stderr" + output="$work/claude-transparent.terminal" debug="$work/claude-transparent.debug.log" - ( - cd "$work/workspace" - nemo-relay run \ - --config "$XDG_CONFIG_HOME/nemo-relay/config.toml" \ - -- \ - claude \ - --settings "$work/claude-user-settings.json" \ - -p "ping" \ - --output-format json \ - --no-session-persistence \ - --tools "" \ - --debug-file "$debug" - ) >"$output" 2>"$stderr" - python3 - "$output" "$stderr" "$debug" <<'PY' -import json + python3 - \ + "$output" \ + "$debug" \ + "$work/workspace" \ + "$XDG_CONFIG_HOME/nemo-relay/config.toml" \ + "$work/claude-user-settings.json" <<'PY' +import os +import pty +import select +import signal +import shutil +import subprocess import sys +import termios +import time +import fcntl from pathlib import Path -output, stderr, debug = map(Path, sys.argv[1:]) -result = json.loads(output.read_text()) -assert result["subtype"] == "success", (result, stderr.read_text()) -assert result["result"] == "pong", result +output, debug, workspace, config, settings = map(Path, sys.argv[1:]) +relay = shutil.which("nemo-relay") +claude = shutil.which("claude") +assert relay and claude, (relay, claude) +master, slave = pty.openpty() + + +def make_controlling_terminal(): + os.setsid() + fcntl.ioctl(slave, termios.TIOCSCTTY, 0) + + +process = subprocess.Popen( + [ + relay, + "run", + "--config", + str(config), + "--", + claude, + "--settings", + str(settings), + "--permission-mode", + "manual", + "--debug-file", + str(debug), + "relay-e2e-tool", + ], + cwd=workspace, + stdin=slave, + stdout=slave, + stderr=slave, + preexec_fn=make_controlling_terminal, +) +os.close(slave) +terminal = bytearray() +deadline = time.monotonic() + 30 +sent_exit = False +trusted_workspace = False +confirmed_api_key = False +try: + while process.poll() is None and time.monotonic() < deadline: + readable, _, _ = select.select([master], [], [], 0.2) + if readable: + try: + terminal.extend(os.read(master, 65536)) + except OSError: + break + if not trusted_workspace and b"Accessing" in terminal and b"Quick" in terminal: + os.write(master, b"\r") + trusted_workspace = True + if not confirmed_api_key and b"Detected" in terminal and b"ANTHROPIC_API_KEY" in terminal: + os.write(master, b"\x1b[A\r") + confirmed_api_key = True + if not sent_exit and b"pong" in terminal.lower(): + os.write(master, b"/exit\r") + sent_exit = True + if process.poll() is None: + os.killpg(process.pid, signal.SIGTERM) + try: + process.wait(timeout=5) + except subprocess.TimeoutExpired: + process.kill() + process.wait(timeout=5) +finally: + os.close(master) + output.write_bytes(terminal) +assert sent_exit, terminal.decode(errors="replace") +assert process.returncode == 0, (process.returncode, terminal.decode(errors="replace")) + log = debug.read_text() assert 1 <= log.count("Hook SessionStart:startup") <= 2, log assert 1 <= log.count("Hook UserPromptSubmit") <= 2, log assert 1 <= log.count('Hook Stop (Stop) success') <= 2, log -assert 1 <= log.count("SessionEnd:other") <= 2, log +assert 1 <= log.count("SessionEnd:") <= 2, log +assert "Hook PreToolUse" in log, log +assert "Hook PermissionRequest" in log, log +assert "Hook PostToolUse" in log, log assert log.count('MCP server "plugin:nemo-relay-plugin:nemo-relay": Successfully connected') == 1, log PY return 0 @@ -251,9 +326,15 @@ import sys from urllib.parse import urlparse requests = [json.loads(line) for line in open(sys.argv[1], encoding="utf-8") if line.strip()] -messages = [row for row in requests if urlparse(row["path"]).path.endswith("/messages")] -assert len(messages) == 1, requests +messages = [ + row + for row in requests + if urlparse(row["path"]).path.endswith("/messages") and row["tools"] +] +assert len(messages) == 2, requests assert messages[0]["model"] == "claude-haiku-4-5", messages +assert [message["has_tool_result"] for message in messages] == [False, True], messages +assert any(tool["name"] == "Bash" for tool in messages[0]["tools"]), messages events = [json.loads(line) for line in open(sys.argv[2], encoding="utf-8") if line.strip()] turn_starts = [ event for event in events @@ -268,9 +349,26 @@ turn_ends = [ and event.get("scope_category") == "end" ] assert len(turn_starts) == len(turn_ends) == 1, (turn_starts, turn_ends) +tool_starts = [ + event for event in events + if event.get("kind") == "scope" + and event.get("category") == "tool" + and event.get("scope_category") == "start" +] +tool_ends = [ + event for event in events + if event.get("kind") == "scope" + and event.get("category") == "tool" + and event.get("scope_category") == "end" +] +assert len(tool_starts) == len(tool_ends) == 1, (tool_starts, tool_ends) PY nemo-relay doctor --plugin claude-code --install-dir "$work/install" +if [[ "${RELAY_E2E_TRANSPARENT_ONLY:-0}" == "1" ]]; then + exit 0 +fi + wait_for_relay_port_release : >"$provider_log" rm -f "$events" diff --git a/scripts/test-codex-plugin-e2e.sh b/scripts/test-codex-plugin-e2e.sh index 3a1fffc91..765182e97 100755 --- a/scripts/test-codex-plugin-e2e.sh +++ b/scripts/test-codex-plugin-e2e.sh @@ -145,7 +145,7 @@ kind = "observability" enabled = true [components.config] -version = 2 +version = 4 [components.config.atof] enabled = true @@ -418,9 +418,13 @@ with open(stdout_path, "wb") as stdout, open(stderr_path, "wb") as stderr: "codex", "--profile", "relay-user-profile", + "--config", + 'approval_policy="on-request"', "exec", + "--sandbox", + "workspace-write", "--skip-git-repo-check", - "ping", + "relay-e2e-tool", ], stdout=stdout, stderr=stderr, @@ -480,7 +484,32 @@ rm -f "$events" wait_for_relay_port_release run_transparent_codex_ping wait_for_relay_port_release -cmp "$CODEX_HOME/config.toml" "$work/codex-config-before-transparent.toml" +python3 - "$work/codex-config-before-transparent.toml" "$CODEX_HOME/config.toml" "$transparent_project" <<'PY' +import sys +import tomllib +from pathlib import Path + + +def load(path): + with open(path, "rb") as config: + return tomllib.load(config) + + +before = load(sys.argv[1]) +after = load(sys.argv[2]) +project = str(Path(sys.argv[3]).resolve()) + +# Codex records trust for a workspace the first time it executes a tool there. Relay must preserve +# every other user setting, including the provider and plugin configuration. +expected_projects = dict(before.get("projects", {})) +expected_projects[project] = {"trust_level": "trusted"} +assert after.get("projects", {}) == expected_projects, (before, after) +if "projects" in before: + after["projects"] = before["projects"] +else: + after.pop("projects", None) +assert after == before, (before, after) +PY cmp "$CODEX_HOME/relay-user-profile.config.toml" "$work/codex-profile-before-transparent.toml" [[ -z "$(find_sidecar_file 'sidecar-*.owner.json')" ]] python3 - "$provider_log" "$events" <<'PY' @@ -489,8 +518,10 @@ import sys requests = [json.loads(line) for line in open(sys.argv[1], encoding="utf-8") if line.strip()] responses = [row for row in requests if row["method"] == "POST" and row["path"].endswith("/responses")] -assert len(responses) == 1, requests +assert len(responses) == 2, requests assert responses[0]["model"] == "gpt-5.1-codex", responses +assert [response["has_tool_result"] for response in responses] == [False, True], responses +assert any(tool["name"] == "exec_command" for tool in responses[0]["tools"]), responses events = [json.loads(line) for line in open(sys.argv[2], encoding="utf-8") if line.strip()] turn_starts = [ event for event in events @@ -507,9 +538,26 @@ turn_ends = [ assert len(turn_starts) == len(turn_ends) == 1, (turn_starts, turn_ends) assert turn_starts[0].get("data", {}).get("hook_event_name", "").lower() == "userpromptsubmit", turn_starts assert turn_ends[0].get("metadata", {}).get("hook_event_name", "").lower() == "stop", turn_ends +tool_starts = [ + event for event in events + if event.get("kind") == "scope" + and event.get("category") == "tool" + and event.get("scope_category") == "start" +] +tool_ends = [ + event for event in events + if event.get("kind") == "scope" + and event.get("category") == "tool" + and event.get("scope_category") == "end" +] +assert len(tool_starts) == len(tool_ends) == 1, (tool_starts, tool_ends) PY nemo-relay doctor --plugin codex --install-dir "$install_dir" +if [[ "${RELAY_E2E_TRANSPARENT_ONLY:-0}" == "1" ]]; then + exit 0 +fi + # Exercise incompatible configuration handling before collecting acceptance events. export NEMO_RELAY_PLUGIN_IDLE_TIMEOUT_SECS=300 holder_fifo="$work/mcp-holder.stdin" diff --git a/scripts/test-support/codex_mock_provider.py b/scripts/test-support/codex_mock_provider.py index 3a4aefbf5..314d2bd6b 100644 --- a/scripts/test-support/codex_mock_provider.py +++ b/scripts/test-support/codex_mock_provider.py @@ -126,6 +126,99 @@ def response_events(request: dict[str, Any]) -> list[dict[str, Any]]: ] +def response_tool_events(request: dict[str, Any]) -> list[dict[str, Any]]: + response_id = f"resp_{uuid.uuid4().hex}" + item_id = f"fc_{uuid.uuid4().hex}" + call_id = f"call_{uuid.uuid4().hex}" + model = request.get("model", "gpt-5-codex") + created_at = int(time.time()) + arguments = json.dumps( + { + "cmd": "printf relay-e2e-tool-ok", + "sandbox_permissions": "require_escalated", + "justification": "Verify Relay permission hook handling.", + }, + separators=(",", ":"), + ) + item = { + "id": item_id, + "type": "function_call", + "status": "completed", + "call_id": call_id, + "name": "exec_command", + "arguments": arguments, + } + response = { + "id": response_id, + "object": "response", + "created_at": created_at, + "completed_at": created_at, + "status": "completed", + "background": False, + "error": None, + "incomplete_details": None, + "instructions": None, + "max_output_tokens": None, + "max_tool_calls": None, + "model": model, + "output": [item], + "parallel_tool_calls": True, + "previous_response_id": None, + "prompt_cache_key": None, + "reasoning": {"effort": "medium", "summary": None}, + "safety_identifier": None, + "service_tier": "default", + "store": False, + "temperature": None, + "text": {"format": {"type": "text"}, "verbosity": "medium"}, + "tool_choice": "auto", + "tools": [], + "top_logprobs": 0, + "top_p": None, + "truncation": "disabled", + "usage": { + "input_tokens": 1, + "input_tokens_details": {"cached_tokens": 0}, + "output_tokens": 1, + "output_tokens_details": {"reasoning_tokens": 0}, + "total_tokens": 2, + }, + "user": None, + "metadata": {}, + } + in_progress = {**response, "completed_at": None, "status": "in_progress", "output": []} + return [ + {"type": "response.created", "response": in_progress}, + { + "type": "response.output_item.added", + "response_id": response_id, + "output_index": 0, + "item": {**item, "status": "in_progress", "arguments": ""}, + }, + { + "type": "response.function_call_arguments.delta", + "response_id": response_id, + "item_id": item_id, + "output_index": 0, + "delta": arguments, + }, + { + "type": "response.function_call_arguments.done", + "response_id": response_id, + "item_id": item_id, + "output_index": 0, + "arguments": arguments, + }, + { + "type": "response.output_item.done", + "response_id": response_id, + "output_index": 0, + "item": item, + }, + {"type": "response.completed", "response": response}, + ] + + def anthropic_events(request: dict[str, Any]) -> list[tuple[str, dict[str, Any]]]: message_id = f"msg_{uuid.uuid4().hex}" model = request.get("model", "claude-sonnet-4-5") @@ -175,6 +268,83 @@ def anthropic_events(request: dict[str, Any]) -> list[tuple[str, dict[str, Any]] ] +def anthropic_tool_events(request: dict[str, Any]) -> list[tuple[str, dict[str, Any]]]: + message_id = f"msg_{uuid.uuid4().hex}" + tool_use_id = f"toolu_{uuid.uuid4().hex}" + model = request.get("model", "claude-sonnet-4-5") + arguments = json.dumps( + { + "command": "printf relay-e2e-tool-ok > relay-e2e-permission.txt", + "description": "Verify Relay permission hooks", + }, + separators=(",", ":"), + ) + return [ + ( + "message_start", + { + "type": "message_start", + "message": { + "id": message_id, + "type": "message", + "role": "assistant", + "content": [], + "model": model, + "stop_reason": None, + "stop_sequence": None, + "usage": {"input_tokens": 1, "output_tokens": 0}, + }, + }, + ), + ( + "content_block_start", + { + "type": "content_block_start", + "index": 0, + "content_block": { + "type": "tool_use", + "id": tool_use_id, + "name": "Bash", + "input": {}, + }, + }, + ), + ( + "content_block_delta", + { + "type": "content_block_delta", + "index": 0, + "delta": {"type": "input_json_delta", "partial_json": arguments}, + }, + ), + ("content_block_stop", {"type": "content_block_stop", "index": 0}), + ( + "message_delta", + { + "type": "message_delta", + "delta": {"stop_reason": "tool_use", "stop_sequence": None}, + "usage": {"output_tokens": 1}, + }, + ), + ("message_stop", {"type": "message_stop"}), + ] + + +def contains_type(value: Any, expected: set[str]) -> bool: + if isinstance(value, dict): + value_type = value.get("type") + if isinstance(value_type, str) and value_type in expected: + return True + return any(contains_type(child, expected) for child in value.values()) + if isinstance(value, list): + return any(contains_type(child, expected) for child in value) + return False + + +def requests_tool_scenario(request: dict[str, Any]) -> bool: + return "relay-e2e-tool" in json.dumps(request, sort_keys=True) + + def chat_completion_chunks(request: dict[str, Any]) -> list[dict[str, Any]]: completion_id = f"chatcmpl_{uuid.uuid4().hex}" model = request.get("model", "gpt-4o-mini") @@ -270,8 +440,18 @@ def do_POST(self) -> None: # noqa: N802 raw = self.rfile.read(length) request = json.loads(raw or b"{}") path = urlparse(self.path).path - response_stream = response_events(request) if path.endswith("/responses") else None - anthropic_stream = anthropic_events(request) if path.endswith("/messages") else None + tool_scenario = requests_tool_scenario(request) + has_tool_result = contains_type(request, {"function_call_output", "tool_result"}) + response_stream = ( + (response_tool_events(request) if tool_scenario and not has_tool_result else response_events(request)) + if path.endswith("/responses") + else None + ) + anthropic_stream = ( + (anthropic_tool_events(request) if tool_scenario and not has_tool_result else anthropic_events(request)) + if path.endswith("/messages") + else None + ) chat_stream = chat_completion_chunks(request) if path.endswith("/chat/completions") else None self.server.log_request_record( { @@ -281,6 +461,12 @@ def do_POST(self) -> None: # noqa: N802 "x_api_key": self.headers.get("x-api-key"), "relay_client_token": self.headers.get("x-nemo-relay-client-token"), "model": request.get("model"), + "tools": [ + {"type": tool.get("type"), "name": tool.get("name")} + for tool in request.get("tools", []) + if isinstance(tool, dict) + ], + "has_tool_result": has_tool_result, "response_id": (response_stream[-1]["response"]["id"] if response_stream else None), } )