From dd08eab845c0c0866580587c0fe99510966d86cd Mon Sep 17 00:00:00 2001 From: Larri Date: Mon, 13 Jul 2026 14:58:27 -0600 Subject: [PATCH 01/10] feat: add Larri socket secrets provider --- Cargo.lock | 3 + Cargo.toml | 2 +- docs/RUNTIME_REFERENCE.md | 16 + docs/STDLIB_REFERENCE.md | 4 +- docs/runtime.toml | 28 + src/secret.rs | 9 +- src/stdlib/secrets.rs | 432 ++++++++++++++-- src/stdlib/secrets/socket.rs | 860 +++++++++++++++++++++++++++++++ tests/language_features_tests.rs | 142 +++++ 9 files changed, 1439 insertions(+), 57 deletions(-) create mode 100644 src/stdlib/secrets/socket.rs diff --git a/Cargo.lock b/Cargo.lock index 38073e1..d9637f8 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4207,6 +4207,9 @@ name = "zeroize" version = "1.8.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b97154e67e32c85465826e8bcc1c59429aaaf107c1e4a9e53c8d8ccd5eff88d0" +dependencies = [ + "serde", +] [[package]] name = "zerotrie" diff --git a/Cargo.toml b/Cargo.toml index d2886eb..a126104 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -99,7 +99,7 @@ jsonwebtoken = "9" pulldown-cmark = "0.13.0" hickory-resolver = "0.24.4" socket2 = "0.6" -zeroize = "1.8.2" +zeroize = { version = "1.8.2", features = ["serde"] } [build-dependencies] # Doc comment scanner (build.rs extracts /// @ntnt blocks) diff --git a/docs/RUNTIME_REFERENCE.md b/docs/RUNTIME_REFERENCE.md index 54eeede..cfb5869 100644 --- a/docs/RUNTIME_REFERENCE.md +++ b/docs/RUNTIME_REFERENCE.md @@ -29,6 +29,10 @@ Environment variables that control NTNT runtime behavior | `NTNT_LINT_MODE` | `default`, `warn`, `strict` | default | Controls lint strictness for type annotations. `default`: only check annotated code. `warn`: also warn about missing annotations (non-fatal). `strict`: missing annotations are errors. CLI flags (`--strict`, `--warn-untyped`) override this. | | `NTNT_MAX_RECURSION` | integer | 256 | Maximum recursion depth for function calls. Prevents stack overflow from runaway recursion. | | `NTNT_NET_ALLOW_PRIVATE` | `1`, `true`, `yes` | unset (disabled — private targets blocked) | Process-level opt-in for `std/net` probes (`ping`, `tcp_connect`, `reachable`, `port_scan`, `tls_info`, `traceroute`) against private/internal targets. Each call must also pass `allow_private: true`. Special-purpose targets such as cloud metadata, multicast, broadcast, unspecified, and documentation ranges remain blocked. This is separate from `NTNT_ALLOW_PRIVATE_IPS`, which only applies to `fetch()`. | +| `NTNT_SECRETS_AUTHORIZATION_SCOPE` | opaque ASCII deployment identifier | unset (required for larri-socket) | Trusted non-credential deployment scope expected in every Larri agent response. It detects endpoint miswiring but is not authentication and is never rendered in diagnostics. Leading or trailing whitespace is rejected. | +| `NTNT_SECRETS_PROVIDER` | `env`, `larri-socket` | env | Selects the provider behind `std/secrets`. The exact-name environment provider is development-only and is rejected when `NTNT_ENV` is `production` or `prod`. `larri-socket` is Unix-only and talks to a local Larri host agent without depending directly on a secrets backend. | +| `NTNT_SECRETS_SOCKET_ENDPOINTS` | comma-separated absolute Unix socket paths | unset (required for larri-socket) | Ordered Larri host-agent endpoints for secret lookup. One through eight unique paths are allowed. Each endpoint is tried twice, and failover occurs only for bounded unavailable results. Production paths must be beneath the host-controlled `/run/larri-secrets` root; raw paths are never included in runtime diagnostics. | +| `NTNT_SECRETS_TIMEOUT_MS` | integer (milliseconds) | 1000 | Per-attempt Larri socket connect, write, and read timeout. Values must be between 10 and 10000 milliseconds. | | `NTNT_STRICT` | `1`, `true` | unset (disabled) | **Deprecated — use `NTNT_LINT_MODE=strict` instead.** Enable strict type checking. Still works but emits a deprecation warning. | | `NTNT_TIMEOUT` | integer (seconds) | 30 | Request timeout for HTTP server in seconds. | | `NTNT_TYPE_MODE` | `strict`, `warn`, `forgiving` | warn | Controls runtime behavior for type mismatches. `strict`: type mismatches crash (fail-closed, recommended for auth/payments). `warn`: log `[WARN]` and continue (default). `forgiving`: silent degradation (pre-v0.4 behavior). See [Type Safety Modes](#type-safety-modes). | @@ -54,6 +58,18 @@ NTNT_MAX_RECURSION=512 ntnt run server.tnt # Process-level opt-in for `std/net` probes (`ping`, `tcp_connect`, `reachable`, `port_scan`, `tls_info`, `traceroute`) against private/internal targets NTNT_NET_ALLOW_PRIVATE=1 ntnt run monitor.tnt +# Trusted non-credential deployment scope expected in every Larri agent response +NTNT_SECRETS_AUTHORIZATION_SCOPE=deployment-a + +# Selects the provider behind `std/secrets` +NTNT_SECRETS_PROVIDER=larri-socket ntnt run server.tnt + +# Ordered Larri host-agent endpoints for secret lookup +NTNT_SECRETS_SOCKET_ENDPOINTS=/run/larri-secrets/secrets.sock + +# Per-attempt Larri socket connect, write, and read timeout +NTNT_SECRETS_TIMEOUT_MS=500 + # **Deprecated — use `NTNT_LINT_MODE=strict` instead.** Enable strict type checking NTNT_STRICT=1 ntnt run server.tnt diff --git a/docs/STDLIB_REFERENCE.md b/docs/STDLIB_REFERENCE.md index 726f13c..d66f7b9 100644 --- a/docs/STDLIB_REFERENCE.md +++ b/docs/STDLIB_REFERENCE.md @@ -10905,7 +10905,7 @@ get_secret(name: String) -> Option Looks up a secret by its provider-neutral logical name. -The v0.5.1 development-only environment provider reads the exact environment variable name and is disabled when `NTNT_ENV` is `production` or `prod`. Projects with an `ntnt.toml` must declare accessible names under `[secrets.]`; undeclared lookups fail before contacting the provider. Declaration metadata may contain `label`, `description`, `required`, and `environments`; secret values are never accepted in the manifest. Secret values remain opaque and redact themselves in output and diagnostics. +The environment provider reads the exact environment variable name and is disabled when `NTNT_ENV` is `production` or `prod`. Unix deployments can instead select the Larri host-agent provider with `NTNT_SECRETS_PROVIDER=larri-socket`, a comma-separated list of absolute `NTNT_SECRETS_SOCKET_ENDPOINTS`, and the expected non-credential deployment identifier in `NTNT_SECRETS_AUTHORIZATION_SCOPE`. Optional `NTNT_SECRETS_TIMEOUT_MS` is bounded from 10 through 10000 milliseconds. Production socket paths must be beneath `/run/larri-secrets`; endpoints are retried twice and failed over only for bounded unavailable results. Plaintext caching remains in the host agent rather than ntnt. Projects with an `ntnt.toml` must declare accessible names under `[secrets.]`; undeclared lookups fail before contacting the provider. Declaration metadata may contain `label`, `description`, `required`, and `environments`; secret values are never accepted in the manifest. Secret values remain opaque and redact themselves in output and diagnostics. **Parameters:** @@ -10921,7 +10921,7 @@ get_secret("STRIPE_SECRET_KEY") // => Some([REDACTED]) // Optional lookup **Errors:** -- **RuntimeError**: Unsupported secrets provider — *Fix: Set NTNT_SECRETS_PROVIDER=env for v0.5.1* +- **RuntimeError**: Unsupported secrets provider — *Fix: Select env for development or larri-socket with deployment-scoped socket configuration* **See also:** `require_secret` diff --git a/docs/runtime.toml b/docs/runtime.toml index 8aa5bcf..236a971 100644 --- a/docs/runtime.toml +++ b/docs/runtime.toml @@ -20,6 +20,34 @@ description = "Controls runtime mode. Production mode disables hot-reload for be example = "NTNT_ENV=production ntnt run server.tnt" affects = ["hot-reload"] +[env_vars.NTNT_SECRETS_PROVIDER] +values = ["env", "larri-socket"] +default = "env" +description = "Selects the provider behind `std/secrets`. The exact-name environment provider is development-only and is rejected when `NTNT_ENV` is `production` or `prod`. `larri-socket` is Unix-only and talks to a local Larri host agent without depending directly on a secrets backend." +example = "NTNT_SECRETS_PROVIDER=larri-socket ntnt run server.tnt" +affects = ["std-secrets", "security"] + +[env_vars.NTNT_SECRETS_SOCKET_ENDPOINTS] +type = "comma-separated absolute Unix socket paths" +default = "unset (required for larri-socket)" +description = "Ordered Larri host-agent endpoints for secret lookup. One through eight unique paths are allowed. Each endpoint is tried twice, and failover occurs only for bounded unavailable results. Production paths must be beneath the host-controlled `/run/larri-secrets` root; raw paths are never included in runtime diagnostics." +example = "NTNT_SECRETS_SOCKET_ENDPOINTS=/run/larri-secrets/secrets.sock" +affects = ["std-secrets", "security"] + +[env_vars.NTNT_SECRETS_AUTHORIZATION_SCOPE] +type = "opaque ASCII deployment identifier" +default = "unset (required for larri-socket)" +description = "Trusted non-credential deployment scope expected in every Larri agent response. It detects endpoint miswiring but is not authentication and is never rendered in diagnostics. Leading or trailing whitespace is rejected." +example = "NTNT_SECRETS_AUTHORIZATION_SCOPE=deployment-a" +affects = ["std-secrets", "security"] + +[env_vars.NTNT_SECRETS_TIMEOUT_MS] +type = "integer (milliseconds)" +default = "1000" +description = "Per-attempt Larri socket connect, write, and read timeout. Values must be between 10 and 10000 milliseconds." +example = "NTNT_SECRETS_TIMEOUT_MS=500" +affects = ["std-secrets", "security"] + [env_vars.NTNT_TIMEOUT] type = "integer (seconds)" default = "30" diff --git a/src/secret.rs b/src/secret.rs index 3c7db5f..b9127a1 100644 --- a/src/secret.rs +++ b/src/secret.rs @@ -20,12 +20,19 @@ pub struct SecretValue { impl SecretValue { /// Construct a secret after validating its logical name. + #[cfg(test)] pub(crate) fn new(name: impl Into, value: impl Into) -> Result { + Self::new_zeroizing(name, Zeroizing::new(value.into())) + } + + /// Construct a secret from provider-owned zeroizing storage without copying + /// plaintext through an ordinary String at the provider boundary. + pub(crate) fn new_zeroizing(name: impl Into, value: Zeroizing) -> Result { let name = name.into(); validate_secret_name(&name)?; Ok(Self { name: Arc::from(name), - value: Arc::new(Zeroizing::new(value.into())), + value: Arc::new(value), }) } diff --git a/src/stdlib/secrets.rs b/src/stdlib/secrets.rs index 683ca0b..4badd4e 100644 --- a/src/stdlib/secrets.rs +++ b/src/stdlib/secrets.rs @@ -4,11 +4,28 @@ use crate::error::{IntentError, Result}; use crate::interpreter::{SecretValue, Value}; use crate::secret::validate_secret_name; use std::collections::{HashMap, HashSet}; -use std::path::{Path, PathBuf}; +use std::fmt; +use std::path::{Component, Path, PathBuf}; use std::sync::{Arc, OnceLock, RwLock}; +use std::time::Duration; +use zeroize::Zeroizing; + +#[cfg(unix)] +#[path = "secrets/socket.rs"] +mod socket; const PROVIDER_ENV: &str = "NTNT_SECRETS_PROVIDER"; +const SOCKET_ENDPOINTS_ENV: &str = "NTNT_SECRETS_SOCKET_ENDPOINTS"; +const SOCKET_SCOPE_ENV: &str = "NTNT_SECRETS_AUTHORIZATION_SCOPE"; +const SOCKET_TIMEOUT_ENV: &str = "NTNT_SECRETS_TIMEOUT_MS"; const DEFAULT_ATTEMPTS_PER_ENDPOINT: usize = 2; +const DEFAULT_SOCKET_TIMEOUT_MS: u64 = 1_000; +const MIN_SOCKET_TIMEOUT_MS: u64 = 10; +const MAX_SOCKET_TIMEOUT_MS: u64 = 10_000; +const MAX_SOCKET_ENDPOINTS: usize = 8; +const MAX_SOCKET_PATH_BYTES: usize = 96; +const MAX_AUTHORIZATION_SCOPE_BYTES: usize = 128; +const PRODUCTION_SOCKET_ROOT: &str = "/run/larri-secrets"; #[cfg_attr(test, allow(dead_code))] #[derive(Default)] @@ -160,37 +177,66 @@ fn enforce_declared(name: &str) -> Result<()> { /// A provider result that is deliberately not `Debug`: the found payload is plaintext. enum ProviderLookup { - Found(String), + Found(Zeroizing), Missing, } +/// A non-sensitive provider identity safe for diagnostics. +#[derive(Debug, Clone, PartialEq, Eq)] +struct ProviderEndpointLabel(String); + +impl ProviderEndpointLabel { + fn env() -> Self { + Self("env".to_string()) + } + + #[cfg(unix)] + fn socket(index: usize) -> Self { + Self(format!("larri-socket-{index}")) + } + + #[cfg(test)] + fn fixture(value: &str) -> Self { + assert!(value + .chars() + .all(|character| character.is_ascii_alphanumeric() || character == '-')); + Self(value.to_string()) + } +} + +impl fmt::Display for ProviderEndpointLabel { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter.write_str(&self.0) + } +} + /// Stable failure classes for HA failover. The v0.5.1 environment provider /// can only emit invalid configuration; socket providers use the remaining classes. -#[cfg_attr(not(test), allow(dead_code))] #[derive(Debug, Clone, Copy, PartialEq, Eq)] enum ProviderErrorKind { Unavailable, AccessDenied, + InvalidRequest, InvalidConfiguration, } #[derive(Debug, Clone)] struct ProviderError { kind: ProviderErrorKind, - endpoint: String, + endpoint: ProviderEndpointLabel, } impl ProviderError { - fn new(kind: ProviderErrorKind, endpoint: impl Into) -> Self { + fn new(kind: ProviderErrorKind, endpoint: &ProviderEndpointLabel) -> Self { Self { kind, - endpoint: sanitize_endpoint(&endpoint.into()), + endpoint: endpoint.clone(), } } } trait SecretProvider: Send + Sync { - fn endpoint(&self) -> &str; + fn endpoint(&self) -> &ProviderEndpointLabel; fn authorization_scope(&self) -> &str; /// Perform one endpoint lookup. Implementations must apply a bounded timeout @@ -200,8 +246,9 @@ trait SecretProvider: Send + Sync { /// Ordered equivalent provider endpoints with classified failover. /// -/// Only `Unavailable` advances to the next endpoint. Missing, denied, and invalid -/// configuration results are terminal so failover cannot change authorization scope. +/// Only `Unavailable` advances to the next endpoint. Missing, denied, invalid +/// request, and invalid configuration results are terminal so failover cannot +/// change authorization scope. struct ProviderGroup { providers: Vec>, attempts_per_endpoint: usize, @@ -239,7 +286,7 @@ impl ProviderGroup { attempts += 1; match provider.lookup(name) { Ok(ProviderLookup::Found(value)) => { - return SecretValue::new(name, value).map(Some) + return SecretValue::new_zeroizing(name, value).map(Some) } Ok(ProviderLookup::Missing) => return Ok(None), Err(error) => match error.kind { @@ -258,6 +305,12 @@ impl ProviderGroup { error.endpoint ))) } + ProviderErrorKind::InvalidRequest => { + return Err(IntentError::runtime_error(format!( + "Secret provider rejected the request for '{name}' at endpoint '{}'", + error.endpoint + ))) + } ProviderErrorKind::InvalidConfiguration => { return Err(IntentError::runtime_error(format!( "Secret provider configuration is invalid for endpoint '{}'", @@ -272,7 +325,11 @@ impl ProviderGroup { Err(IntentError::runtime_error(format!( "Secret provider unavailable after {attempts} bounded attempt(s) across {} endpoint(s): {}", unavailable.len(), - unavailable.join(", ") + unavailable + .iter() + .map(ToString::to_string) + .collect::>() + .join(", ") ))) } } @@ -280,8 +337,9 @@ impl ProviderGroup { struct EnvSecretProvider; impl SecretProvider for EnvSecretProvider { - fn endpoint(&self) -> &str { - "env" + fn endpoint(&self) -> &ProviderEndpointLabel { + static ENDPOINT: OnceLock = OnceLock::new(); + ENDPOINT.get_or_init(ProviderEndpointLabel::env) } fn authorization_scope(&self) -> &str { @@ -291,7 +349,7 @@ impl SecretProvider for EnvSecretProvider { fn lookup(&self, name: &str) -> std::result::Result { match std::env::var(name) { Ok(value) if value.is_empty() => Ok(ProviderLookup::Missing), - Ok(value) => Ok(ProviderLookup::Found(value)), + Ok(value) => Ok(ProviderLookup::Found(Zeroizing::new(value))), Err(std::env::VarError::NotPresent) => Ok(ProviderLookup::Missing), Err(std::env::VarError::NotUnicode(_)) => Err(ProviderError::new( ProviderErrorKind::InvalidConfiguration, @@ -301,16 +359,121 @@ impl SecretProvider for EnvSecretProvider { } } -fn sanitize_endpoint(endpoint: &str) -> String { - if endpoint.is_empty() - || endpoint.len() > 64 - || !endpoint - .chars() - .all(|c| c.is_ascii_alphanumeric() || matches!(c, '_' | '.' | ':' | '/' | '-')) +struct SocketProviderConfig { + endpoints: Vec, + authorization_scope: String, + timeout: Duration, +} + +fn socket_config_error() -> IntentError { + IntentError::runtime_error("Larri socket secrets provider configuration is invalid".to_string()) +} + +#[cfg(unix)] +fn validate_no_symlink_components(path: &Path) -> Result<()> { + let mut current = PathBuf::new(); + for component in path.components() { + current.push(component.as_os_str()); + match std::fs::symlink_metadata(¤t) { + Ok(metadata) if metadata.file_type().is_symlink() => { + return Err(socket_config_error()); + } + Ok(_) => {} + Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(()), + Err(_) => return Err(socket_config_error()), + } + } + Ok(()) +} + +#[cfg(not(unix))] +fn validate_no_symlink_components(_path: &Path) -> Result<()> { + Ok(()) +} + +#[cfg(unix)] +fn validate_existing_socket_path(path: &Path) -> Result<()> { + use std::os::unix::fs::FileTypeExt; + + match std::fs::symlink_metadata(path) { + Ok(metadata) if metadata.file_type().is_symlink() || !metadata.file_type().is_socket() => { + Err(socket_config_error()) + } + Ok(_) => Ok(()), + Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()), + Err(_) => Err(socket_config_error()), + } +} + +#[cfg(not(unix))] +fn validate_existing_socket_path(_path: &Path) -> Result<()> { + Ok(()) +} + +fn parse_socket_provider_config( + endpoints: Option<&str>, + authorization_scope: Option<&str>, + timeout_ms: Option<&str>, + production: bool, +) -> Result { + let raw_endpoints = endpoints.ok_or_else(socket_config_error)?; + let parts: Vec<&str> = raw_endpoints.split(',').map(str::trim).collect(); + if parts.is_empty() || parts.len() > MAX_SOCKET_ENDPOINTS { + return Err(socket_config_error()); + } + + let mut seen = HashSet::with_capacity(parts.len()); + let mut parsed_endpoints = Vec::with_capacity(parts.len()); + for endpoint in parts { + if endpoint.is_empty() + || endpoint.len() > MAX_SOCKET_PATH_BYTES + || endpoint.chars().any(char::is_control) + { + return Err(socket_config_error()); + } + let path = PathBuf::from(endpoint); + if !path.is_absolute() + || path + .components() + .any(|component| matches!(component, Component::ParentDir | Component::CurDir)) + || (production + && (path == Path::new(PRODUCTION_SOCKET_ROOT) + || !path.starts_with(PRODUCTION_SOCKET_ROOT))) + || !seen.insert(path.clone()) + { + return Err(socket_config_error()); + } + if production { + validate_no_symlink_components(&path)?; + } + validate_existing_socket_path(&path)?; + parsed_endpoints.push(path); + } + + let authorization_scope = authorization_scope.ok_or_else(socket_config_error)?; + if authorization_scope != authorization_scope.trim() + || authorization_scope.is_empty() + || authorization_scope.len() > MAX_AUTHORIZATION_SCOPE_BYTES + || !authorization_scope.chars().all(|character| { + character.is_ascii_alphanumeric() || matches!(character, '_' | '.' | ':' | '-') + }) { - return "unknown".to_string(); + return Err(socket_config_error()); + } + + let timeout_ms = match timeout_ms { + Some(value) => value.parse::().map_err(|_| socket_config_error())?, + None => DEFAULT_SOCKET_TIMEOUT_MS, + }; + if !(MIN_SOCKET_TIMEOUT_MS..=MAX_SOCKET_TIMEOUT_MS).contains(&timeout_ms) { + return Err(socket_config_error()); } - endpoint.to_string() + + Ok(SocketProviderConfig { + endpoints: parsed_endpoints, + authorization_scope: authorization_scope.to_string(), + timeout: Duration::from_millis(timeout_ms), + }) } fn is_production_mode() -> bool { @@ -319,20 +482,85 @@ fn is_production_mode() -> bool { .unwrap_or(false) } -fn configured_provider_group() -> Result { - let provider = std::env::var(PROVIDER_ENV).unwrap_or_else(|_| "env".to_string()); - match provider.as_str() { - "env" if is_production_mode() => Err(IntentError::runtime_error( +fn configured_provider_group_from_values( + provider: &str, + socket_endpoints: Option<&str>, + authorization_scope: Option<&str>, + timeout_ms: Option<&str>, + production: bool, +) -> Result { + match provider { + "env" if production => Err(IntentError::runtime_error( "The environment secrets provider is development-only and is disabled in production" .to_string(), )), "env" => ProviderGroup::new(vec![Arc::new(EnvSecretProvider)]), + "larri-socket" => { + let config = parse_socket_provider_config( + socket_endpoints, + authorization_scope, + timeout_ms, + production, + )?; + #[cfg(unix)] + { + let providers = config + .endpoints + .into_iter() + .enumerate() + .map(|(index, path)| { + let endpoint = ProviderEndpointLabel::socket(index + 1); + let provider = if production { + socket::SocketSecretProvider::new_with_trusted_root( + path, + endpoint, + config.authorization_scope.clone(), + config.timeout, + PathBuf::from(PRODUCTION_SOCKET_ROOT), + ) + } else { + socket::SocketSecretProvider::new( + path, + endpoint, + config.authorization_scope.clone(), + config.timeout, + ) + }; + Arc::new(provider) as Arc + }) + .collect(); + ProviderGroup::new(providers) + } + #[cfg(not(unix))] + { + let _ = config; + Err(IntentError::runtime_error( + "The Larri socket secrets provider is supported only on Unix platforms" + .to_string(), + )) + } + } _ => Err(IntentError::runtime_error( - "Unsupported secrets provider; ntnt v0.5.1 supports 'env' in development".to_string(), + "Unsupported secrets provider; supported providers are 'env' and 'larri-socket'" + .to_string(), )), } } +fn configured_provider_group() -> Result { + let provider = std::env::var(PROVIDER_ENV).unwrap_or_else(|_| "env".to_string()); + let socket_endpoints = std::env::var(SOCKET_ENDPOINTS_ENV).ok(); + let authorization_scope = std::env::var(SOCKET_SCOPE_ENV).ok(); + let timeout_ms = std::env::var(SOCKET_TIMEOUT_ENV).ok(); + configured_provider_group_from_values( + &provider, + socket_endpoints.as_deref(), + authorization_scope.as_deref(), + timeout_ms.as_deref(), + is_production_mode(), + ) +} + fn lookup_secret(name: &str) -> Result> { // Validate before the name can reach declaration diagnostics or a provider. // SecretValue validates again only to preserve its own construction invariant. @@ -360,8 +588,16 @@ pub fn init() -> HashMap { // @signature get_secret(name: String) -> Option // Looks up a secret by its provider-neutral logical name. // - // The v0.5.1 development-only environment provider reads the exact environment - // variable name and is disabled when `NTNT_ENV` is `production` or `prod`. + // The environment provider reads the exact environment variable name and is + // disabled when `NTNT_ENV` is `production` or `prod`. Unix deployments can + // instead select the Larri host-agent provider with + // `NTNT_SECRETS_PROVIDER=larri-socket`, a comma-separated list of absolute + // `NTNT_SECRETS_SOCKET_ENDPOINTS`, and the expected non-credential deployment + // identifier in `NTNT_SECRETS_AUTHORIZATION_SCOPE`. Optional + // `NTNT_SECRETS_TIMEOUT_MS` is bounded from 10 through 10000 milliseconds. + // Production socket paths must be beneath `/run/larri-secrets`; endpoints are + // retried twice and failed over only for bounded unavailable results. Plaintext + // caching remains in the host agent rather than ntnt. // Projects with an `ntnt.toml` must declare accessible names under // `[secrets.]`; undeclared lookups fail before contacting the provider. // Declaration metadata may contain `label`, `description`, `required`, and @@ -373,7 +609,7 @@ pub fn init() -> HashMap { // @since v0.5.1 // @tags #security, #secrets // @example get_secret("STRIPE_SECRET_KEY") => Some([REDACTED]) ~ "Optional lookup" - // @error RuntimeError ~ "Unsupported secrets provider" fix: "Set NTNT_SECRETS_PROVIDER=env for v0.5.1" + // @error RuntimeError ~ "Unsupported secrets provider" fix: "Select env for development or larri-socket with deployment-scoped socket configuration" module.insert( "get_secret".to_string(), Value::NativeFunction { @@ -442,6 +678,97 @@ mod tests { use std::sync::Mutex; use std::time::{SystemTime, UNIX_EPOCH}; + #[test] + fn socket_provider_config_is_bounded_and_production_scoped() { + let config = parse_socket_provider_config( + Some("/tmp/one.sock,/tmp/two.sock"), + Some("deployment-a"), + Some("250"), + false, + ) + .expect("valid development socket config"); + assert_eq!(config.endpoints.len(), 2); + assert_eq!(config.authorization_scope, "deployment-a"); + assert_eq!(config.timeout, std::time::Duration::from_millis(250)); + parse_socket_provider_config( + Some("/run/larri-secrets/agent.sock"), + Some("deployment-a"), + None, + true, + ) + .expect("valid production socket root"); + + for (endpoints, scope, timeout, production) in [ + (None, Some("deployment-a"), None, false), + (Some("/tmp/a.sock"), None, None, false), + (Some("/tmp/a.sock"), Some(" deployment-a "), None, false), + (Some("relative.sock"), Some("deployment-a"), None, false), + (Some("/tmp/bad\n.sock"), Some("deployment-a"), None, false), + (Some("/run/larri-secrets"), Some("deployment-a"), None, true), + ( + Some("/tmp/a.sock,/tmp/a.sock"), + Some("deployment-a"), + None, + false, + ), + ( + Some("/run/larri-secrets/../other.sock"), + Some("deployment-a"), + None, + true, + ), + (Some("/tmp/a.sock"), Some("deployment-a"), None, true), + (Some("/tmp/a.sock"), Some("deployment-a"), Some("9"), false), + ( + Some("/tmp/a.sock"), + Some("deployment-a"), + Some("10001"), + false, + ), + (Some("/tmp/a.sock"), Some("deployment\nscope"), None, false), + ] { + assert!( + parse_socket_provider_config(endpoints, scope, timeout, production).is_err(), + "invalid socket configuration must fail closed" + ); + } + + let too_many = (0..9) + .map(|index| format!("/tmp/{index}.sock")) + .collect::>() + .join(","); + assert!( + parse_socket_provider_config(Some(&too_many), Some("deployment-a"), None, false,) + .is_err() + ); + + let overlong = format!("/tmp/{}.sock", "a".repeat(96)); + assert!( + parse_socket_provider_config(Some(&overlong), Some("deployment-a"), None, false,) + .is_err() + ); + } + + #[cfg(unix)] + #[test] + fn socket_provider_rejects_symlinked_parent_components() { + let suffix = SystemTime::now() + .duration_since(UNIX_EPOCH) + .expect("clock") + .as_nanos(); + let root = std::env::temp_dir().join(format!( + "ntnt-socket-parent-{}-{suffix}", + std::process::id() + )); + let real = root.join("real"); + std::fs::create_dir_all(&real).expect("create real parent"); + let link = root.join("link"); + std::os::unix::fs::symlink(&real, &link).expect("create parent symlink"); + + assert!(validate_no_symlink_components(&link.join("agent.sock")).is_err()); + std::fs::remove_dir_all(root).ok(); + } + #[test] fn declaration_identity_is_stable_when_manifest_appears() { let suffix = SystemTime::now() @@ -470,7 +797,7 @@ mod tests { } struct MockProvider { - endpoint: String, + endpoint: ProviderEndpointLabel, authorization_scope: String, calls: AtomicUsize, results: Mutex>>, @@ -482,7 +809,7 @@ mod tests { results: Vec>, ) -> Self { Self { - endpoint: endpoint.to_string(), + endpoint: ProviderEndpointLabel::fixture(endpoint), authorization_scope: "deployment-a".to_string(), calls: AtomicUsize::new(0), results: Mutex::new(results.into()), @@ -500,7 +827,7 @@ mod tests { } impl SecretProvider for MockProvider { - fn endpoint(&self) -> &str { + fn endpoint(&self) -> &ProviderEndpointLabel { &self.endpoint } @@ -518,6 +845,14 @@ mod tests { } } + fn found(value: &str) -> ProviderLookup { + ProviderLookup::Found(Zeroizing::new(value.to_string())) + } + + fn provider_error(kind: ProviderErrorKind, endpoint: &str) -> ProviderError { + ProviderError::new(kind, &ProviderEndpointLabel::fixture(endpoint)) + } + #[test] fn provider_group_rejects_mixed_authorization_scopes() { let first = Arc::new( @@ -541,19 +876,13 @@ mod tests { let first = Arc::new(MockProvider::new( "socket-a", vec![ - Err(ProviderError::new( - ProviderErrorKind::Unavailable, - "socket-a", - )), - Err(ProviderError::new( - ProviderErrorKind::Unavailable, - "socket-a", - )), + Err(provider_error(ProviderErrorKind::Unavailable, "socket-a")), + Err(provider_error(ProviderErrorKind::Unavailable, "socket-a")), ], )); let second = Arc::new(MockProvider::new( "socket-b", - vec![Ok(ProviderLookup::Found("secret-value".to_string()))], + vec![Ok(found("secret-value"))], )); let group = ProviderGroup::new(vec![first.clone(), second.clone()]).expect("group"); @@ -568,16 +897,13 @@ mod tests { let first = Arc::new(MockProvider::new( "socket-a", vec![ - Err(ProviderError::new( - ProviderErrorKind::Unavailable, - "socket-a", - )), - Ok(ProviderLookup::Found("secret-value".to_string())), + Err(provider_error(ProviderErrorKind::Unavailable, "socket-a")), + Ok(found("secret-value")), ], )); let second = Arc::new(MockProvider::new( "socket-b", - vec![Ok(ProviderLookup::Found("must-not-read".to_string()))], + vec![Ok(found("must-not-read"))], )); let group = ProviderGroup::new(vec![first.clone(), second.clone()]).expect("group"); @@ -591,14 +917,14 @@ mod tests { fn provider_group_stops_on_access_denied() { let first = Arc::new(MockProvider::new( "socket-a", - vec![Err(ProviderError::new( + vec![Err(provider_error( ProviderErrorKind::AccessDenied, "socket-a", ))], )); let second = Arc::new(MockProvider::new( "socket-b", - vec![Ok(ProviderLookup::Found("must-not-read".to_string()))], + vec![Ok(found("must-not-read"))], )); let group = ProviderGroup::new(vec![first, second.clone()]).expect("group"); @@ -615,7 +941,7 @@ mod tests { )); let second = Arc::new(MockProvider::new( "socket-b", - vec![Ok(ProviderLookup::Found("must-not-read".to_string()))], + vec![Ok(found("must-not-read"))], )); let group = ProviderGroup::new(vec![first, second.clone()]).expect("group"); @@ -631,8 +957,8 @@ mod tests { Arc::new(MockProvider::new( endpoint, vec![ - Err(ProviderError::new(ProviderErrorKind::Unavailable, endpoint)), - Err(ProviderError::new(ProviderErrorKind::Unavailable, endpoint)), + Err(provider_error(ProviderErrorKind::Unavailable, endpoint)), + Err(provider_error(ProviderErrorKind::Unavailable, endpoint)), ], )) as Arc }) diff --git a/src/stdlib/secrets/socket.rs b/src/stdlib/secrets/socket.rs new file mode 100644 index 0000000..7ca1747 --- /dev/null +++ b/src/stdlib/secrets/socket.rs @@ -0,0 +1,860 @@ +//! Larri host-agent secret protocol v1. +//! +//! Each lookup opens one Unix stream, writes exactly one newline-delimited JSON +//! request, half-closes the write side, reads one bounded newline-delimited JSON +//! response through EOF, and closes the stream. Requests contain only +//! `protocol`, a non-credential numeric `request_id`, `op = "get"`, and the +//! validated logical secret `name`. Responses echo `protocol` and `request_id`, +//! include the expected deployment authorization `scope`, and use one of: +//! `found`, `missing`, `access_denied`, `unavailable`, `invalid_request`, or +//! `invalid_configuration`. Only `found` includes `value`. +//! +//! Frames are limited to 64 KiB, values to 32 KiB, and every connect/read/write +//! operation is bounded by the configured timeout. Unknown fields, extra frames, +//! version/request/scope mismatches, empty values, and malformed responses fail +//! closed without rendering response bytes. + +use super::{ + validate_no_symlink_components, ProviderEndpointLabel, ProviderError, ProviderErrorKind, + ProviderLookup, SecretProvider, +}; +use serde::{Deserialize, Serialize}; +use socket2::{Domain, SockAddr, Socket, Type}; +use std::io::{Read, Write}; +use std::net::Shutdown; +use std::os::fd::{FromRawFd, IntoRawFd}; +use std::os::unix::fs::FileTypeExt; +use std::os::unix::net::UnixStream; +use std::path::PathBuf; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::Duration; +use zeroize::Zeroizing; + +const PROTOCOL_VERSION: u8 = 1; +const MAX_RESPONSE_SIZE: usize = 65_536; +const MAX_SECRET_SIZE: usize = 32_768; +static NEXT_REQUEST_ID: AtomicU64 = AtomicU64::new(1); + +pub(super) struct SocketSecretProvider { + path: PathBuf, + endpoint: ProviderEndpointLabel, + authorization_scope: String, + timeout: Duration, + trusted_root: Option, +} + +impl SocketSecretProvider { + pub(super) fn new( + path: PathBuf, + endpoint: ProviderEndpointLabel, + authorization_scope: String, + timeout: Duration, + ) -> Self { + Self { + path, + endpoint, + authorization_scope, + timeout, + trusted_root: None, + } + } + + pub(super) fn new_with_trusted_root( + path: PathBuf, + endpoint: ProviderEndpointLabel, + authorization_scope: String, + timeout: Duration, + trusted_root: PathBuf, + ) -> Self { + Self { + path, + endpoint, + authorization_scope, + timeout, + trusted_root: Some(trusted_root), + } + } + + fn error(&self, kind: ProviderErrorKind) -> ProviderError { + ProviderError::new(kind, &self.endpoint) + } + + fn connect(&self) -> std::result::Result { + if let Some(trusted_root) = &self.trusted_root { + if self.path == *trusted_root + || !self.path.starts_with(trusted_root) + || validate_no_symlink_components(&self.path).is_err() + { + return Err(self.error(ProviderErrorKind::InvalidConfiguration)); + } + } + match std::fs::symlink_metadata(&self.path) { + Ok(metadata) + if metadata.file_type().is_symlink() || !metadata.file_type().is_socket() => + { + return Err(self.error(ProviderErrorKind::InvalidConfiguration)); + } + Ok(_) => {} + Err(error) if error.kind() == std::io::ErrorKind::NotFound => { + return Err(self.error(ProviderErrorKind::Unavailable)); + } + Err(_) => return Err(self.error(ProviderErrorKind::InvalidConfiguration)), + } + let socket = Socket::new(Domain::UNIX, Type::STREAM, None) + .map_err(|_| self.error(ProviderErrorKind::Unavailable))?; + let address = SockAddr::unix(&self.path) + .map_err(|_| self.error(ProviderErrorKind::InvalidConfiguration))?; + socket + .connect_timeout(&address, self.timeout) + .map_err(|_| self.error(ProviderErrorKind::Unavailable))?; + + // SAFETY: ownership of the live stream descriptor moves from socket2 to + // UnixStream exactly once; `socket` cannot close it after `into_raw_fd`. + let stream = unsafe { UnixStream::from_raw_fd(socket.into_raw_fd()) }; + stream + .set_nonblocking(false) + .and_then(|_| stream.set_read_timeout(Some(self.timeout))) + .and_then(|_| stream.set_write_timeout(Some(self.timeout))) + .map_err(|_| self.error(ProviderErrorKind::Unavailable))?; + Ok(stream) + } + + fn validate_response_header( + &self, + protocol: u8, + response_request_id: u64, + expected_request_id: u64, + scope: &str, + ) -> std::result::Result<(), ProviderError> { + if protocol != PROTOCOL_VERSION || response_request_id != expected_request_id { + return Err(self.error(ProviderErrorKind::InvalidConfiguration)); + } + if scope != self.authorization_scope { + return Err(self.error(ProviderErrorKind::AccessDenied)); + } + Ok(()) + } + + fn write_request( + &self, + writer: &mut impl Write, + name: &str, + ) -> std::result::Result { + let request_id = NEXT_REQUEST_ID.fetch_add(1, Ordering::Relaxed); + let request = SocketRequest { + protocol: PROTOCOL_VERSION, + request_id, + op: "get", + name, + }; + serde_json::to_writer(&mut *writer, &request) + .map_err(|_| self.error(ProviderErrorKind::Unavailable))?; + writer + .write_all(b"\n") + .and_then(|_| writer.flush()) + .map_err(|_| self.error(ProviderErrorKind::Unavailable))?; + Ok(request_id) + } + + fn read_response_frame( + &self, + stream: &mut UnixStream, + ) -> std::result::Result>, ProviderError> { + let mut response = Zeroizing::new(Vec::with_capacity(1024)); + let mut chunk = Zeroizing::new([0_u8; 4096]); + + loop { + let remaining = MAX_RESPONSE_SIZE + 1 - response.len(); + let read_size = remaining.min(chunk.len()); + match stream.read(&mut chunk[..read_size]) { + Ok(0) => return Err(self.error(ProviderErrorKind::InvalidConfiguration)), + Ok(bytes_read) => { + let bytes = &chunk[..bytes_read]; + if bytes.contains(&b'\r') { + return Err(self.error(ProviderErrorKind::InvalidConfiguration)); + } + if let Some(newline) = bytes.iter().position(|byte| *byte == b'\n') { + response.extend_from_slice(&bytes[..=newline]); + if newline + 1 != bytes_read || response.len() > MAX_RESPONSE_SIZE { + return Err(self.error(ProviderErrorKind::InvalidConfiguration)); + } + + let mut trailing = Zeroizing::new([0_u8; 1]); + return match stream.read(&mut trailing[..]) { + Ok(0) => Ok(response), + Ok(_) | Err(_) => { + Err(self.error(ProviderErrorKind::InvalidConfiguration)) + } + }; + } + + response.extend_from_slice(bytes); + if response.len() > MAX_RESPONSE_SIZE { + return Err(self.error(ProviderErrorKind::InvalidConfiguration)); + } + } + Err(_) => return Err(self.error(ProviderErrorKind::Unavailable)), + } + } + } +} + +#[derive(Serialize)] +struct SocketRequest<'a> { + protocol: u8, + request_id: u64, + op: &'static str, + name: &'a str, +} + +#[derive(Deserialize)] +#[serde(tag = "status", rename_all = "snake_case", deny_unknown_fields)] +enum SocketResponse { + Found { + protocol: u8, + request_id: u64, + scope: String, + value: Zeroizing, + }, + Missing { + protocol: u8, + request_id: u64, + scope: String, + }, + AccessDenied { + protocol: u8, + request_id: u64, + scope: String, + }, + Unavailable { + protocol: u8, + request_id: u64, + scope: String, + }, + InvalidRequest { + protocol: u8, + request_id: u64, + scope: String, + }, + InvalidConfiguration { + protocol: u8, + request_id: u64, + scope: String, + }, +} + +impl SecretProvider for SocketSecretProvider { + fn endpoint(&self) -> &ProviderEndpointLabel { + &self.endpoint + } + + fn authorization_scope(&self) -> &str { + &self.authorization_scope + } + + fn lookup(&self, name: &str) -> std::result::Result { + let mut stream = self.connect()?; + let request_id = self.write_request(&mut stream, name)?; + stream + .shutdown(Shutdown::Write) + .map_err(|_| self.error(ProviderErrorKind::Unavailable))?; + + let response = self.read_response_frame(&mut stream)?; + + let body = &response[..response.len() - 1]; + let parsed: SocketResponse = serde_json::from_slice(body) + .map_err(|_| self.error(ProviderErrorKind::InvalidConfiguration))?; + match parsed { + SocketResponse::Found { + protocol, + request_id: response_request_id, + scope, + value, + } => { + self.validate_response_header(protocol, response_request_id, request_id, &scope)?; + if value.is_empty() || value.len() > MAX_SECRET_SIZE { + return Err(self.error(ProviderErrorKind::InvalidConfiguration)); + } + Ok(ProviderLookup::Found(value)) + } + SocketResponse::Missing { + protocol, + request_id: response_request_id, + scope, + } => { + self.validate_response_header(protocol, response_request_id, request_id, &scope)?; + Ok(ProviderLookup::Missing) + } + SocketResponse::AccessDenied { + protocol, + request_id: response_request_id, + scope, + } => { + self.validate_response_header(protocol, response_request_id, request_id, &scope)?; + Err(self.error(ProviderErrorKind::AccessDenied)) + } + SocketResponse::Unavailable { + protocol, + request_id: response_request_id, + scope, + } => { + self.validate_response_header(protocol, response_request_id, request_id, &scope)?; + Err(self.error(ProviderErrorKind::Unavailable)) + } + SocketResponse::InvalidRequest { + protocol, + request_id: response_request_id, + scope, + } => { + self.validate_response_header(protocol, response_request_id, request_id, &scope)?; + Err(self.error(ProviderErrorKind::InvalidRequest)) + } + SocketResponse::InvalidConfiguration { + protocol, + request_id: response_request_id, + scope, + } => { + self.validate_response_header(protocol, response_request_id, request_id, &scope)?; + Err(self.error(ProviderErrorKind::InvalidConfiguration)) + } + } + } +} + +#[cfg(test)] +mod tests { + use super::super::{ + configured_provider_group_from_values, ProviderEndpointLabel, ProviderErrorKind, + ProviderLookup, SecretProvider, + }; + use super::SocketSecretProvider; + use serde_json::Value as JsonValue; + use std::io::{BufRead, BufReader, Write}; + use std::os::unix::net::UnixListener; + use std::path::PathBuf; + use std::sync::atomic::{AtomicU64, Ordering}; + use std::sync::mpsc; + use std::time::Duration; + + const SECRET_CANARY: &str = "SOCKET_SECRET_CANARY_DO_NOT_DISCLOSE"; + const REQUEST_ID_PLACEHOLDER: &str = "__REQUEST_ID__"; + static SOCKET_TEST_COUNTER: AtomicU64 = AtomicU64::new(1); + + fn socket_path(_label: &str) -> PathBuf { + let counter = SOCKET_TEST_COUNTER.fetch_add(1, Ordering::Relaxed); + PathBuf::from("/tmp").join(format!("ntnt-sp-{}-{counter}.sock", std::process::id())) + } + + fn serve_responses( + label: &str, + responses: Vec>, + ) -> (PathBuf, mpsc::Receiver, std::thread::JoinHandle<()>) { + let path = socket_path(label); + let listener = UnixListener::bind(&path).expect("bind fixture socket"); + let (request_tx, request_rx) = mpsc::channel(); + let server = std::thread::spawn(move || { + for mut response in responses { + let (mut stream, _) = listener.accept().expect("accept provider request"); + stream + .set_read_timeout(Some(Duration::from_secs(1))) + .expect("set fixture read timeout"); + stream + .set_write_timeout(Some(Duration::from_secs(1))) + .expect("set fixture write timeout"); + + let mut request = String::new(); + BufReader::new(&stream) + .read_line(&mut request) + .expect("read provider request"); + if let Ok(text) = String::from_utf8(response.clone()) { + if text.contains(REQUEST_ID_PLACEHOLDER) { + let parsed: JsonValue = + serde_json::from_str(&request).expect("valid provider request"); + let request_id = parsed["request_id"].as_u64().expect("numeric request id"); + response = text + .replace( + &format!("\"{REQUEST_ID_PLACEHOLDER}\""), + &request_id.to_string(), + ) + .into_bytes(); + } + } + request_tx.send(request).expect("capture request"); + stream + .write_all(&response) + .expect("write provider response"); + } + }); + (path, request_rx, server) + } + + fn serve_response( + label: &str, + response: Vec, + ) -> (PathBuf, mpsc::Receiver, std::thread::JoinHandle<()>) { + serve_responses(label, vec![response]) + } + + fn provider(path: PathBuf) -> SocketSecretProvider { + SocketSecretProvider::new( + path, + ProviderEndpointLabel::socket(1), + "deployment-a".to_string(), + Duration::from_millis(250), + ) + } + + #[test] + fn socket_provider_rechecks_trusted_root_parent_symlinks_at_connect_time() { + let root = socket_path("trusted-root"); + std::fs::create_dir_all(&root).expect("create trusted root fixture"); + let actual = root.join("actual"); + std::fs::create_dir_all(&actual).expect("create actual socket directory"); + let actual_socket = actual.join("agent.sock"); + let _listener = UnixListener::bind(&actual_socket).expect("bind actual socket"); + let symlinked_parent = root.join("link"); + std::os::unix::fs::symlink(&actual, &symlinked_parent).expect("create parent symlink"); + + let Err(error) = SocketSecretProvider::new_with_trusted_root( + symlinked_parent.join("agent.sock"), + ProviderEndpointLabel::socket(1), + "deployment-a".to_string(), + Duration::from_millis(50), + root.clone(), + ) + .lookup("API_KEY") else { + panic!("symlinked parent must fail closed"); + }; + assert_eq!(error.kind, ProviderErrorKind::InvalidConfiguration); + + std::fs::remove_dir_all(root).ok(); + } + + #[test] + fn socket_provider_rejects_regular_files_and_symlinks_at_connect_time() { + let regular_path = socket_path("regular-file"); + std::fs::write(®ular_path, b"not a socket").expect("write regular fixture"); + let Err(regular_error) = provider(regular_path.clone()).lookup("API_KEY") else { + panic!("regular files must not be used as provider sockets"); + }; + assert_eq!(regular_error.kind, ProviderErrorKind::InvalidConfiguration); + std::fs::remove_file(regular_path).ok(); + + let target_path = socket_path("symlink-target"); + let _listener = UnixListener::bind(&target_path).expect("bind target socket"); + let symlink_path = socket_path("symlink-path"); + std::os::unix::fs::symlink(&target_path, &symlink_path).expect("create socket symlink"); + let Err(symlink_error) = provider(symlink_path.clone()).lookup("API_KEY") else { + panic!("symlink socket path must fail closed"); + }; + assert_eq!(symlink_error.kind, ProviderErrorKind::InvalidConfiguration); + std::fs::remove_file(symlink_path).ok(); + std::fs::remove_file(target_path).ok(); + } + + #[test] + fn configured_socket_provider_retries_transient_failures_then_fails_over() { + let unavailable = + b"{\"protocol\":1,\"request_id\":\"__REQUEST_ID__\",\"status\":\"unavailable\",\"scope\":\"deployment-a\"}\n".to_vec(); + let found = format!( + "{}\n", + serde_json::json!({ + "protocol": 1, + "request_id": REQUEST_ID_PLACEHOLDER, + "status": "found", + "scope": "deployment-a", + "value": SECRET_CANARY, + }) + ) + .into_bytes(); + let (first_path, first_requests, first_server) = + serve_responses("failover-first", vec![unavailable.clone(), unavailable]); + let (second_path, second_requests, second_server) = + serve_response("failover-second", found); + let endpoints = format!("{},{}", first_path.display(), second_path.display()); + + let group = configured_provider_group_from_values( + "larri-socket", + Some(&endpoints), + Some("deployment-a"), + Some("100"), + false, + ) + .expect("valid socket provider group"); + let secret = group + .lookup("API_KEY") + .expect("failover lookup") + .expect("secret found"); + assert_eq!(secret.expose(), SECRET_CANARY); + assert_eq!(first_requests.iter().count(), 2); + assert_eq!(second_requests.iter().count(), 1); + + first_server.join().expect("first fixture server"); + second_server.join().expect("second fixture server"); + std::fs::remove_file(first_path).ok(); + std::fs::remove_file(second_path).ok(); + } + + #[test] + fn socket_provider_treats_complete_frame_without_eof_as_terminal_malformed() { + let path = socket_path("complete-without-eof"); + let listener = UnixListener::bind(&path).expect("bind fixture socket"); + let server = std::thread::spawn(move || { + let (mut stream, _) = listener.accept().expect("accept fixture request"); + stream + .set_read_timeout(Some(Duration::from_secs(1))) + .expect("set fixture timeout"); + let mut request = String::new(); + BufReader::new(&stream) + .read_line(&mut request) + .expect("read request"); + let request: JsonValue = + serde_json::from_str(request.trim_end()).expect("request JSON"); + let response = serde_json::json!({ + "protocol": 1, + "request_id": request["request_id"], + "status": "found", + "scope": "deployment-a", + "value": SECRET_CANARY, + }); + writeln!(stream, "{response}").expect("write complete response"); + stream.flush().expect("flush response"); + std::thread::sleep(Duration::from_millis(150)); + }); + + let Err(error) = SocketSecretProvider::new( + path.clone(), + ProviderEndpointLabel::socket(1), + "deployment-a".to_string(), + Duration::from_millis(30), + ) + .lookup("API_KEY") else { + panic!("complete frame without EOF must fail"); + }; + assert_eq!(error.kind, ProviderErrorKind::InvalidConfiguration); + + server.join().expect("fixture server"); + std::fs::remove_file(path).ok(); + } + + #[test] + fn socket_provider_bounds_connect_and_read_failures() { + let missing_path = socket_path("missing-endpoint"); + let started = std::time::Instant::now(); + let Err(connect_error) = SocketSecretProvider::new( + missing_path, + ProviderEndpointLabel::socket(1), + "deployment-a".to_string(), + Duration::from_millis(30), + ) + .lookup("API_KEY") else { + panic!("missing endpoint must be unavailable"); + }; + assert_eq!(connect_error.kind, ProviderErrorKind::Unavailable); + assert!(started.elapsed() < Duration::from_secs(1)); + + let path = socket_path("silent-endpoint"); + let listener = UnixListener::bind(&path).expect("bind silent fixture"); + let server = std::thread::spawn(move || { + let (stream, _) = listener.accept().expect("accept silent request"); + stream + .set_read_timeout(Some(Duration::from_secs(1))) + .expect("set fixture timeout"); + let mut request = String::new(); + BufReader::new(&stream) + .read_line(&mut request) + .expect("read request"); + std::thread::sleep(Duration::from_millis(150)); + }); + + let started = std::time::Instant::now(); + let Err(read_error) = SocketSecretProvider::new( + path.clone(), + ProviderEndpointLabel::socket(1), + "deployment-a".to_string(), + Duration::from_millis(30), + ) + .lookup("API_KEY") else { + panic!("silent endpoint must time out"); + }; + assert_eq!(read_error.kind, ProviderErrorKind::Unavailable); + assert!(started.elapsed() < Duration::from_secs(1)); + + server.join().expect("silent fixture server"); + std::fs::remove_file(path).ok(); + } + + struct PartialWriteFailure { + wrote_once: bool, + } + + impl Write for PartialWriteFailure { + fn write(&mut self, buffer: &[u8]) -> std::io::Result { + if self.wrote_once { + Err(std::io::Error::new( + std::io::ErrorKind::BrokenPipe, + "fixture write failure", + )) + } else { + self.wrote_once = true; + Ok(buffer.len().min(1)) + } + } + + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } + } + + #[test] + fn socket_provider_classifies_partial_write_failure_as_unavailable() { + let mut writer = PartialWriteFailure { wrote_once: false }; + let error = provider(PathBuf::from("/unused")) + .write_request(&mut writer, "API_KEY") + .expect_err("partial write must fail"); + assert_eq!(error.kind, ProviderErrorKind::Unavailable); + assert!(!format!("{error:?}").contains("fixture write failure")); + } + + #[test] + fn socket_provider_rejects_malformed_or_ambiguous_frames() { + let valid = format!( + "{}", + serde_json::json!({ + "protocol": 1, + "request_id": REQUEST_ID_PLACEHOLDER, + "status": "found", + "scope": "deployment-a", + "value": SECRET_CANARY, + }) + ); + let frames = vec![ + ("invalid-json", b"not json\n".to_vec()), + ( + "unknown-field", + format!( + "{{\"protocol\":1,\"request_id\":\"__REQUEST_ID__\",\"status\":\"found\",\"scope\":\"deployment-a\",\"value\":\"{SECRET_CANARY}\",\"extra\":true}}\n" + ) + .into_bytes(), + ), + ( + "wrong-protocol", + format!( + "{{\"protocol\":2,\"request_id\":\"__REQUEST_ID__\",\"status\":\"found\",\"scope\":\"deployment-a\",\"value\":\"{SECRET_CANARY}\"}}\n" + ) + .into_bytes(), + ), + ("no-newline", valid.clone().into_bytes()), + ("carriage-return", format!("{valid}\r\n").into_bytes()), + ("extra-newline", format!("{valid}\n\n").into_bytes()), + ( + "multiple-messages", + format!("{valid}\n{{\"status\":\"missing\"}}\n").into_bytes(), + ), + ("oversized", vec![b'x'; 65_537]), + ]; + + for (label, frame) in frames { + let (path, _request_rx, server) = serve_response(label, frame); + let Err(error) = provider(path.clone()).lookup("API_KEY") else { + panic!("malformed frame '{label}' must fail closed"); + }; + assert_eq!(error.kind, ProviderErrorKind::InvalidConfiguration); + let rendered = format!("{error:?}"); + assert!(!rendered.contains(SECRET_CANARY)); + assert!(!rendered.contains(label)); + + server.join().expect("fixture server"); + std::fs::remove_file(path).ok(); + } + } + + #[test] + fn socket_provider_classifies_agent_failures_without_backend_text() { + for (status, expected) in [ + ("unavailable", ProviderErrorKind::Unavailable), + ("invalid_request", ProviderErrorKind::InvalidRequest), + ( + "invalid_configuration", + ProviderErrorKind::InvalidConfiguration, + ), + ] { + let response = format!( + "{}\n", + serde_json::json!({ + "protocol": 1, + "request_id": REQUEST_ID_PLACEHOLDER, + "status": status, + "scope": "deployment-a", + }) + ); + let (path, _request_rx, server) = serve_response(status, response.into_bytes()); + + let Err(error) = provider(path.clone()).lookup("API_KEY") else { + panic!("agent failure must remain an error"); + }; + assert_eq!(error.kind, expected); + assert!(!format!("{error:?}").contains(SECRET_CANARY)); + + server.join().expect("fixture server"); + std::fs::remove_file(path).ok(); + } + } + + #[test] + fn socket_provider_rejects_empty_or_oversized_values() { + for (label, value) in [ + ("empty-value", String::new()), + ("oversized-value", "x".repeat(32_769)), + ] { + let response = format!( + "{}\n", + serde_json::json!({ + "protocol": 1, + "request_id": REQUEST_ID_PLACEHOLDER, + "status": "found", + "scope": "deployment-a", + "value": value, + }) + ); + let (path, _request_rx, server) = serve_response(label, response.into_bytes()); + + let Err(error) = provider(path.clone()).lookup("API_KEY") else { + panic!("invalid value size must fail closed"); + }; + assert_eq!(error.kind, ProviderErrorKind::InvalidConfiguration); + assert!(!format!("{error:?}").contains(SECRET_CANARY)); + + server.join().expect("fixture server"); + std::fs::remove_file(path).ok(); + } + } + + #[test] + fn socket_provider_rejects_mismatched_request_id() { + let response = format!( + "{}\n", + serde_json::json!({ + "protocol": 1, + "request_id": 0, + "status": "found", + "scope": "deployment-a", + "value": SECRET_CANARY, + }) + ); + let (path, _request_rx, server) = serve_response("wrong-request-id", response.into_bytes()); + + let Err(error) = provider(path.clone()).lookup("API_KEY") else { + panic!("mismatched request id must fail closed"); + }; + assert_eq!(error.kind, ProviderErrorKind::InvalidConfiguration); + assert!(!format!("{error:?}").contains(SECRET_CANARY)); + + server.join().expect("fixture server"); + std::fs::remove_file(path).ok(); + } + + #[test] + fn socket_provider_rejects_wrong_authorization_scope_before_returning_value() { + let response = format!( + "{}\n", + serde_json::json!({ + "protocol": 1, + "request_id": REQUEST_ID_PLACEHOLDER, + "status": "found", + "scope": "deployment-b", + "value": SECRET_CANARY, + }) + ); + let (path, _request_rx, server) = serve_response("wrong-scope", response.into_bytes()); + + let Err(error) = provider(path.clone()).lookup("API_KEY") else { + panic!("wrong authorization scope must fail closed"); + }; + assert_eq!(error.kind, ProviderErrorKind::AccessDenied); + let rendered = format!("{error:?}"); + assert!(!rendered.contains("deployment-b")); + assert!(!rendered.contains(SECRET_CANARY)); + + server.join().expect("fixture server"); + std::fs::remove_file(path).ok(); + } + + #[test] + fn socket_provider_maps_access_denied_to_terminal_class() { + let response = + b"{\"protocol\":1,\"request_id\":\"__REQUEST_ID__\",\"status\":\"access_denied\",\"scope\":\"deployment-a\"}\n".to_vec(); + let (path, _request_rx, server) = serve_response("secret-denied", response); + + let Err(error) = provider(path.clone()).lookup("API_KEY") else { + panic!("access denial must be terminal"); + }; + assert_eq!(error.kind, ProviderErrorKind::AccessDenied); + assert!(!format!("{error:?}").contains(SECRET_CANARY)); + + server.join().expect("fixture server"); + std::fs::remove_file(path).ok(); + } + + #[test] + fn socket_provider_maps_missing_without_exposing_backend_details() { + let response = + b"{\"protocol\":1,\"request_id\":\"__REQUEST_ID__\",\"status\":\"missing\",\"scope\":\"deployment-a\"}\n".to_vec(); + let (path, _request_rx, server) = serve_response("secret-missing", response); + + let lookup = provider(path.clone()) + .lookup("API_KEY") + .expect("missing is a successful optional lookup"); + assert!(matches!(lookup, ProviderLookup::Missing)); + + server.join().expect("fixture server"); + std::fs::remove_file(path).ok(); + } + + #[test] + fn socket_provider_returns_matching_scope_secret() { + let response = format!( + "{}\n", + serde_json::json!({ + "protocol": 1, + "request_id": REQUEST_ID_PLACEHOLDER, + "status": "found", + "scope": "deployment-a", + "value": SECRET_CANARY, + }) + ); + let (path, request_rx, server) = serve_response("secret-found", response.into_bytes()); + + let lookup = provider(path.clone()) + .lookup("API_KEY") + .expect("socket lookup"); + match lookup { + ProviderLookup::Found(value) => assert_eq!(value.as_str(), SECRET_CANARY), + ProviderLookup::Missing => panic!("expected found secret"), + } + + let mut request: JsonValue = serde_json::from_str( + request_rx + .recv_timeout(Duration::from_secs(1)) + .expect("captured request") + .trim_end(), + ) + .expect("valid request JSON"); + assert!(request["request_id"].as_u64().is_some()); + request + .as_object_mut() + .expect("request object") + .remove("request_id"); + assert_eq!( + request, + serde_json::json!({ + "protocol": 1, + "op": "get", + "name": "API_KEY", + }) + ); + + server.join().expect("fixture server"); + std::fs::remove_file(path).ok(); + } +} diff --git a/tests/language_features_tests.rs b/tests/language_features_tests.rs index 4dd11f4..dacd246 100644 --- a/tests/language_features_tests.rs +++ b/tests/language_features_tests.rs @@ -11,6 +11,11 @@ use std::net::TcpListener; use std::process::Command; use std::sync::atomic::{AtomicU64, Ordering}; +#[cfg(unix)] +use std::io::{BufRead, BufReader}; +#[cfg(unix)] +use std::os::unix::net::UnixListener; + static TEST_COUNTER: AtomicU64 = AtomicU64::new(0); fn capture_http_request(listener: TcpListener) -> std::thread::JoinHandle> { @@ -53,6 +58,52 @@ fn capture_http_request(listener: TcpListener) -> std::thread::JoinHandle