diff --git a/Cargo.lock b/Cargo.lock index f1fc02f86a9..89d55e68f2a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1088,6 +1088,7 @@ dependencies = [ name = "buzz-backend-kubernetes" version = "0.1.0" dependencies = [ + "buzz-sdk", "chrono", "hex", "http", diff --git a/crates/buzz-backend-kubernetes/Cargo.toml b/crates/buzz-backend-kubernetes/Cargo.toml index 1cd17030db7..bc7854c560c 100644 --- a/crates/buzz-backend-kubernetes/Cargo.toml +++ b/crates/buzz-backend-kubernetes/Cargo.toml @@ -11,6 +11,7 @@ name = "buzz-backend-kubernetes" path = "src/main.rs" [dependencies] +buzz-sdk = { workspace = true } kube = { workspace = true } k8s-openapi = { workspace = true } nostr = { workspace = true } diff --git a/crates/buzz-backend-kubernetes/src/config.rs b/crates/buzz-backend-kubernetes/src/config.rs index 4d96735b7b5..df8e94adc6e 100644 --- a/crates/buzz-backend-kubernetes/src/config.rs +++ b/crates/buzz-backend-kubernetes/src/config.rs @@ -1,12 +1,16 @@ //! `provider_config` parsing and the `info` config schema //! (spec §`provider_config` v1 fields, `docs/remote-agents.md:1384-1389`). //! -//! Nine fields, all optional except `image` (required at parse time; the -//! schema offers the published sprig image as a prefill default — §Image). +//! Nine scalar fields and optional structured Pod controls. `image` is required +//! at parse time (the schema offers the published sprig image as a prefill +//! default — §Image). //! No credential field exists, by I2: cluster auth comes from ambient //! kubeconfig resolution and nothing else (`:196-198`). use crate::image::{self, ImageRef}; +use k8s_openapi::api::core::v1::Toleration; +use serde::{Deserialize, Serialize}; +use std::collections::BTreeMap; /// Resource requests and limits (§Pod shape: 1cpu/2Gi → 2cpu/4Gi, all four /// configurable — `cargo build` in an agent workspace makes 500m/1Gi @@ -61,7 +65,99 @@ pub const TERMINATION_GRACE_SECONDS: i64 = 60; /// the combination rather than shipping against an undefended convention. pub const RESTART_POLICY: &str = "Never"; -#[derive(Debug, Clone, PartialEq, Eq)] +/// Optional placement and resource bounds for a bare Pod. Defaults preserve +/// the original Pod shape; configured fields participate in create intent. +#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct PodOptions { + #[serde(default, skip_serializing_if = "BTreeMap::is_empty")] + /// Exact node labels required for scheduling. + pub node_selector: BTreeMap, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + /// Explicit taints this workload may tolerate. + pub tolerations: Vec, + #[serde(default, skip_serializing_if = "Option::is_none")] + /// Optional emptyDir size limit, in Kubernetes quantity syntax. + pub workspace_size_limit: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + /// Scheduler reservation for writable storage and container logs. + pub ephemeral_storage_request: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + /// Eviction bound for aggregate container ephemeral storage. + pub ephemeral_storage_limit: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + /// Positive absolute Pod lifetime, including startup. + pub active_deadline_seconds: Option, + #[serde(default, skip_serializing_if = "std::ops::Not::not")] + /// Expose the server-assigned Pod UID to a bounded owner setup gate. + pub setup_pending: bool, +} + +impl PodOptions { + /// Preserve original intent serialization when no options are set. + pub fn is_default(&self) -> bool { + self == &Self::default() + } +} + +/// A Pod toleration with Kubernetes field names. k8s-openapi's `Toleration` +/// ignores unknown keys, so a misspelled `effect` would tolerate every effect +/// and a misspelled `tolerationSeconds` would make a NoExecute toleration +/// unbounded. This type refuses both before the Pod is built. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields, rename_all = "camelCase")] +pub struct TolerationSpec { + #[serde(default, skip_serializing_if = "Option::is_none")] + pub key: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub operator: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub value: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub effect: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub toleration_seconds: Option, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +pub enum TolerationOperator { + Exists, + Equal, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +pub enum TaintEffect { + NoSchedule, + PreferNoSchedule, + NoExecute, +} + +impl From<&TolerationSpec> for Toleration { + fn from(spec: &TolerationSpec) -> Self { + Self { + key: spec.key.clone(), + operator: spec.operator.map(|operator| { + match operator { + TolerationOperator::Exists => "Exists", + TolerationOperator::Equal => "Equal", + } + .to_string() + }), + value: spec.value.clone(), + effect: spec.effect.map(|effect| { + match effect { + TaintEffect::NoSchedule => "NoSchedule", + TaintEffect::PreferNoSchedule => "PreferNoSchedule", + TaintEffect::NoExecute => "NoExecute", + } + .to_string() + }), + toleration_seconds: spec.toleration_seconds, + } + } +} + +#[derive(Debug, Clone, PartialEq)] pub struct ProviderConfig { /// kubeconfig context; `None` uses the current context. pub context: Option, @@ -71,6 +167,8 @@ pub struct ProviderConfig { /// `None` when `inactivity_seconds` was 0 — refused in v1, see [`parse`]. pub inactivity_seconds: Option, pub service_account: Option, + /// Optional Pod placement, storage, and lifetime controls. + pub pod_options: PodOptions, } /// Read an optional non-empty string field. Rejects non-string scalars rather @@ -165,6 +263,49 @@ pub fn parse(cfg: &serde_json::Value) -> Result { Some(n) => Some(n), }; + let pod_options: PodOptions = match cfg.get("pod_options") { + None => PodOptions::default(), + Some(value) => serde_json::from_value(value.clone()) + .map_err(|e| format!("provider_config.pod_options: {e}"))?, + }; + if pod_options.active_deadline_seconds.is_some_and(|n| n <= 0) { + return Err("provider_config.pod_options.active_deadline_seconds must be positive".into()); + } + for (field, value) in [ + ("workspace_size_limit", &pod_options.workspace_size_limit), + ( + "ephemeral_storage_request", + &pod_options.ephemeral_storage_request, + ), + ( + "ephemeral_storage_limit", + &pod_options.ephemeral_storage_limit, + ), + ] { + if value + .as_deref() + .is_some_and(|value| !valid_storage_quantity(value)) + { + return Err(format!( + "provider_config.pod_options.{field} must be a non-negative Kubernetes quantity" + )); + } + } + // A zero storage limit evicts the Pod as soon as it writes anything. + for (field, value) in [ + ("workspace_size_limit", &pod_options.workspace_size_limit), + ( + "ephemeral_storage_limit", + &pod_options.ephemeral_storage_limit, + ), + ] { + if value.as_deref().is_some_and(zero_quantity) { + return Err(format!( + "provider_config.pod_options.{field} must be greater than zero" + )); + } + } + Ok(ProviderConfig { context: optional_string(cfg, "context")?, namespace, @@ -172,6 +313,7 @@ pub fn parse(cfg: &serde_json::Value) -> Result { resources, inactivity_seconds, service_account: optional_string(cfg, "service_account")?, + pod_options, }) } @@ -232,16 +374,96 @@ pub fn config_schema() -> serde_json::Value { "description": "The agent exits after this long with no work, and can be started again at any time.", "default": DEFAULT_INACTIVITY_SECONDS }, + "pod_options": { + "type": "object", + "title": "Pod placement and bounds", + "description": "Optional structured provider configuration; omit to preserve the original Pod shape.", + "additionalProperties": false, + "properties": { + "node_selector": {"type": "object", "additionalProperties": {"type": "string"}}, + "tolerations": {"type": "array", "items": { + "type": "object", + "additionalProperties": false, + "properties": { + "key": {"type": "string"}, + "operator": {"type": "string", "enum": ["Exists", "Equal"]}, + "value": {"type": "string"}, + "effect": {"type": "string", "enum": ["NoSchedule", "PreferNoSchedule", "NoExecute"]}, + "tolerationSeconds": {"type": "integer"} + } + }}, + "workspace_size_limit": {"type": "string", "minLength": 1}, + "ephemeral_storage_request": {"type": "string", "minLength": 1}, + "ephemeral_storage_limit": {"type": "string", "minLength": 1}, + "active_deadline_seconds": {"type": "integer", "minimum": 1}, + "setup_pending": {"type": "boolean", "default": false} + } + }, "service_account": { "type": "string", "title": "Service account", "description": "Scheduling/RBAC identity only. No API token is mounted." + }, + "identity_policy": { + "type": "object", + "description": "Optional approved agent and owner public keys, cryptographically checked before cluster access. Also enables local --check-identity preflight.", + "additionalProperties": false, + "required": ["agent_pubkey", "owner_pubkey"], + "properties": { + "agent_pubkey": {"type": "string", "pattern": "^[0-9a-f]{64}$"}, + "owner_pubkey": {"type": "string", "pattern": "^[0-9a-f]{64}$"} + } } }, "required": ["namespace", "image"] }) } +// Quantity is only a serialized string in k8s-openapi. Check Kubernetes' numeric +// grammar and suffixes before creating resources; storage amounts cannot be negative. +fn valid_storage_quantity(value: &str) -> bool { + let value = value.strip_prefix('+').unwrap_or(value); + let end = value + .find(|c: char| !c.is_ascii_digit() && c != '.') + .unwrap_or(value.len()); + let (number, suffix) = value.split_at(end); + if !number.bytes().any(|b| b.is_ascii_digit()) + || number.bytes().filter(|b| *b == b'.').count() > 1 + { + return false; + } + matches!( + suffix, + "" | "n" + | "u" + | "m" + | "k" + | "M" + | "G" + | "T" + | "P" + | "E" + | "Ki" + | "Mi" + | "Gi" + | "Ti" + | "Pi" + | "Ei" + ) || suffix + .strip_prefix(['e', 'E']) + .is_some_and(|exponent| exponent.parse::().is_ok()) +} + +/// Whether a quantity already accepted by [`valid_storage_quantity`] is zero. +/// The suffix only scales the number, so the digits alone decide. +fn zero_quantity(value: &str) -> bool { + value + .trim_start_matches('+') + .chars() + .take_while(|c| c.is_ascii_digit() || *c == '.') + .all(|c| c == '0' || c == '.') +} + #[cfg(test)] mod tests { use super::*; @@ -254,6 +476,134 @@ mod tests { serde_json::json!({"namespace": "buzz-agents-abc123", "image": digest_ref()}) } + #[test] + fn storage_quantities_are_validated_before_deployment() { + for field in [ + "workspace_size_limit", + "ephemeral_storage_request", + "ephemeral_storage_limit", + ] { + for value in [ + "50Gi", "1.5Gi", ".5G", "1.", "+10M", "1e-3", "1E+6", "1m", "1n", "1u", "1Ei", + ] { + let mut cfg = minimal(); + cfg["pod_options"] = serde_json::json!({field: value}); + assert!(parse(&cfg).is_ok(), "{field}={value}"); + } + for value in [ + "", + "abc", + "1GB", + "1K", + "-1Gi", + " 1Gi", + "1Gi ", + "1.2.3Gi", + "1e", + "1e+", + "1e1.2", + "1e9223372036854775808", + "NaN", + "+", + ".", + ] { + let mut cfg = minimal(); + cfg["pod_options"] = serde_json::json!({field: value}); + let error = parse(&cfg).unwrap_err(); + assert!(error.contains(field), "{field}={value}: {error}"); + } + } + } + + #[test] + fn zero_is_a_valid_storage_request_but_not_a_limit() { + for value in ["0", "+0", "0Gi", "0.0", ".0", "0e3"] { + let mut cfg = minimal(); + cfg["pod_options"] = serde_json::json!({"ephemeral_storage_request": value}); + assert!(parse(&cfg).is_ok(), "request {value}"); + for field in ["workspace_size_limit", "ephemeral_storage_limit"] { + let mut cfg = minimal(); + cfg["pod_options"] = serde_json::json!({field: value}); + let error = parse(&cfg).unwrap_err(); + assert!(error.contains(field), "{field}={value}: {error}"); + assert!( + error.contains("greater than zero"), + "{field}={value}: {error}" + ); + } + } + let mut cfg = minimal(); + cfg["pod_options"] = serde_json::json!({"ephemeral_storage_limit": "0.5"}); + assert!(parse(&cfg).is_ok()); + } + + #[test] + fn tolerations_refuse_misspelled_or_unknown_values() { + for bad in [ + serde_json::json!({"key": "gpu", "operator": "Exists", "efect": "NoSchedule"}), + serde_json::json!({"key": "gpu", "operator": "Exists", "effect": "NoExecute", + "tolerationSecond": 60}), + serde_json::json!({"key": "gpu", "operator": "Exists", "effect": "NoExecute", + "toleration_seconds": 60}), + serde_json::json!({"key": "gpu", "operater": "Exists"}), + serde_json::json!({"key": "gpu", "operator": "exists"}), + serde_json::json!({"key": "gpu", "operator": "Exists", "effect": "noexecute"}), + ] { + let mut cfg = minimal(); + cfg["pod_options"] = serde_json::json!({"tolerations": [bad.clone()]}); + let error = parse(&cfg).unwrap_err(); + assert!(error.contains("pod_options"), "{bad}: {error}"); + } + + let mut cfg = minimal(); + cfg["pod_options"] = serde_json::json!({"tolerations": [{ + "key": "gpu", "operator": "Exists", "effect": "NoExecute", "tolerationSeconds": 60 + }]}); + let parsed = parse(&cfg).unwrap(); + let toleration = Toleration::from(&parsed.pod_options.tolerations[0]); + assert_eq!(toleration.effect.as_deref(), Some("NoExecute")); + assert_eq!(toleration.operator.as_deref(), Some("Exists")); + assert_eq!(toleration.toleration_seconds, Some(60)); + assert_eq!( + serde_json::to_value(&parsed.pod_options).unwrap()["tolerations"], + cfg["pod_options"]["tolerations"] + ); + } + + #[test] + fn schema_closes_toleration_objects() { + let schema = config_schema(); + let item = &schema["properties"]["pod_options"]["properties"]["tolerations"]["items"]; + assert_eq!(item["additionalProperties"], false); + let mut keys: Vec<&str> = item["properties"] + .as_object() + .unwrap() + .keys() + .map(String::as_str) + .collect(); + keys.sort_unstable(); + assert_eq!( + keys, + ["effect", "key", "operator", "tolerationSeconds", "value"] + ); + } + + #[test] + fn refuses_invalid_pod_options() { + for bad in [ + serde_json::json!({"active_deadline_seconds": 0}), + serde_json::json!({"active_deadline_seconds": -1}), + serde_json::json!({"workspace_size_limit": " "}), + serde_json::json!({"unknown": true}), + serde_json::json!({"node_selector": "amd64"}), + ] { + let mut cfg = minimal(); + cfg["pod_options"] = bad; + assert!(parse(&cfg).unwrap_err().contains("pod_options")); + } + assert!(parse(&minimal()).unwrap().pod_options.is_default()); + } + #[test] fn applies_spec_defaults() { let c = parse(&minimal()).unwrap(); @@ -423,10 +773,10 @@ mod tests { ); } - /// Nine fields exactly (§`provider_config` v1 fields). The cap is 20; the + /// Eleven fields (§`provider_config` v1 fields). The cap is 20; the /// count is pinned so a field added without a spec change is caught here. #[test] - fn schema_declares_exactly_the_nine_v1_fields() { + fn schema_declares_scalar_fields_and_optional_pod_controls() { let schema = config_schema(); let props = schema["properties"].as_object().unwrap(); let mut keys: Vec<&str> = props.keys().map(String::as_str).collect(); @@ -437,11 +787,13 @@ mod tests { "context", "cpu_limit", "cpu_request", + "identity_policy", "image", "inactivity_seconds", "memory_limit", "memory_request", "namespace", + "pod_options", "service_account" ] ); diff --git a/crates/buzz-backend-kubernetes/src/identity.rs b/crates/buzz-backend-kubernetes/src/identity.rs new file mode 100644 index 00000000000..615985a0546 --- /dev/null +++ b/crates/buzz-backend-kubernetes/src/identity.rs @@ -0,0 +1,142 @@ +//! Optional owner-approved identity policy, checked before cluster access. + +use crate::{naming::AgentIdentity, wire::DeployRequest}; +use serde::Deserialize; + +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +struct IdentityPolicy { + agent_pubkey: String, + owner_pubkey: String, +} + +/// Validate an optional public identity policy without exposing key material. +/// A local preflight requires the policy; ordinary deploys retain compatibility. +pub fn check(request: &DeployRequest, required: bool) -> Result { + let identity = AgentIdentity::from_nsec(&request.agent.private_key_nsec)?; + let Some(value) = request.provider_config.get("identity_policy") else { + return if required { + Err("identity preflight requires an approved identity_policy".into()) + } else { + Ok(identity) + }; + }; + let policy: IdentityPolicy = + serde_json::from_value(value.clone()).map_err(|_| "invalid identity_policy".to_string())?; + let approved_agent = nostr::PublicKey::from_hex(&policy.agent_pubkey) + .map_err(|_| "invalid approved agent public key".to_string())?; + let approved_owner = nostr::PublicKey::from_hex(&policy.owner_pubkey) + .map_err(|_| "invalid approved owner public key".to_string())?; + if identity.pubkey_hex() == approved_owner.to_hex() { + return Err("owner identity cannot be deployed as the agent".into()); + } + if identity.pubkey_hex() != approved_agent.to_hex() { + return Err("agent key does not match the approved agent identity".into()); + } + let tag = request + .agent + .auth_tag + .as_deref() + .ok_or_else(|| "owner attestation is required by identity_policy".to_string())?; + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map_err(|_| "could not establish identity verification time".to_string())? + .as_secs(); + // Check the signature first so an expired or not-yet-valid attestation is + // reported apart from a forged one. Neither message includes the tag. + buzz_sdk::nip_oa::verify_auth_tag(tag, &approved_agent) + .map_err(|_| "owner attestation verification failed".to_string())?; + let owner = buzz_sdk::nip_oa::verify_auth_tag_for_auth_event(tag, &approved_agent, now) + .map_err(|_| "owner attestation is expired or not yet valid".to_string())?; + if owner != approved_owner { + return Err("attestation owner does not match the approved owner".into()); + } + let descriptor_owner = request + .agent + .launch + .as_ref() + .and_then(|launch| launch.owner_pubkey.as_deref()); + if descriptor_owner != Some(approved_owner.to_hex().as_str()) { + return Err("launch owner does not match the verified owner".into()); + } + Ok(identity) +} + +#[cfg(test)] +mod tests { + use super::*; + use nostr::nips::nip19::ToBech32; + + fn request(agent: &nostr::Keys, owner: &nostr::Keys) -> DeployRequest { + serde_json::from_value(serde_json::json!({ + "agent": { + "relay_url": "wss://example.invalid", + "private_key_nsec": agent.secret_key().to_bech32().unwrap(), + "auth_tag": buzz_sdk::nip_oa::compute_auth_tag(owner, &agent.public_key(), "").unwrap(), + "launch": {"owner_pubkey": owner.public_key().to_hex()} + }, + "provider_config": {"identity_policy": { + "agent_pubkey": agent.public_key().to_hex(), + "owner_pubkey": owner.public_key().to_hex() + }} + })).unwrap() + } + + #[test] + fn binds_agent_signature_owner_and_descriptor() { + let agent = nostr::Keys::generate(); + let owner = nostr::Keys::generate(); + let mut request = request(&agent, &owner); + assert_eq!( + check(&request, true).unwrap().pubkey_hex(), + agent.public_key().to_hex() + ); + request.agent.launch.as_mut().unwrap().owner_pubkey = Some(agent.public_key().to_hex()); + assert!(check(&request, true).unwrap_err().contains("launch owner")); + } + + #[test] + fn rejects_wrong_key_human_key_and_forged_or_wrong_owner_tag() { + let agent = nostr::Keys::generate(); + let owner = nostr::Keys::generate(); + let stranger = nostr::Keys::generate(); + let mut wrong = request(&agent, &owner); + wrong.agent.private_key_nsec = stranger.secret_key().to_bech32().unwrap(); + assert!(check(&wrong, true).unwrap_err().contains("approved agent")); + wrong.agent.private_key_nsec = owner.secret_key().to_bech32().unwrap(); + assert!(check(&wrong, true).unwrap_err().contains("owner identity")); + let mut forged = request(&agent, &owner); + forged.agent.auth_tag = Some("synthetic-secret-not-a-tag".into()); + let error = check(&forged, true).unwrap_err(); + assert!(error.contains("verification failed")); + assert!(!error.contains("synthetic-secret")); + forged.agent.auth_tag = + Some(buzz_sdk::nip_oa::compute_auth_tag(&stranger, &agent.public_key(), "").unwrap()); + assert!(check(&forged, true).unwrap_err().contains("approved owner")); + } + + #[test] + fn reports_expired_and_future_attestations_apart_from_forged_ones() { + let agent = nostr::Keys::generate(); + let owner = nostr::Keys::generate(); + let mut request = request(&agent, &owner); + for conditions in ["created_at<1", "created_at>4294967294"] { + let tag = buzz_sdk::nip_oa::compute_auth_tag(&owner, &agent.public_key(), conditions) + .unwrap(); + request.agent.auth_tag = Some(tag.clone()); + let error = check(&request, true).unwrap_err(); + assert_eq!(error, "owner attestation is expired or not yet valid"); + assert!(!error.contains(&tag)); + } + } + + #[test] + fn requires_policy_only_for_preflight_and_refuses_unknown_policy_fields() { + let mut request = request(&nostr::Keys::generate(), &nostr::Keys::generate()); + request.provider_config["identity_policy"]["extra"] = serde_json::json!(true); + assert!(check(&request, true).is_err()); + request.provider_config = serde_json::json!({}); + assert!(check(&request, true).is_err()); + assert!(check(&request, false).is_ok()); + } +} diff --git a/crates/buzz-backend-kubernetes/src/intent.rs b/crates/buzz-backend-kubernetes/src/intent.rs index 73218f83bff..2e9467cd62e 100644 --- a/crates/buzz-backend-kubernetes/src/intent.rs +++ b/crates/buzz-backend-kubernetes/src/intent.rs @@ -60,7 +60,7 @@ impl std::fmt::Display for Fingerprint { /// The non-secret, pre-binding description of the pod this deploy would /// create. Every field is provider-controlled and scheduling-relevant; there /// is deliberately no field for env values, Secret data, or the generation. -#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +#[derive(Debug, Clone, PartialEq, Serialize)] pub struct IntentTemplate { /// Schema version of the template itself. Bumping it re-fingerprints every /// pod, which is the intended way to roll out a pod-shape change. @@ -83,6 +83,8 @@ pub struct IntentTemplate { pub workspace_mount_path: String, pub run_as_user: i64, pub run_as_group: i64, + #[serde(skip_serializing_if = "crate::config::PodOptions::is_default")] + pub pod_options: crate::config::PodOptions, } /// Current template schema version. @@ -130,6 +132,7 @@ impl IntentTemplate { workspace_mount_path: crate::config::WORKSPACE_PATH.to_string(), run_as_user: crate::config::RUN_AS_UID, run_as_group: crate::config::RUN_AS_GID, + pod_options: crate::config::PodOptions::default(), } } } diff --git a/crates/buzz-backend-kubernetes/src/main.rs b/crates/buzz-backend-kubernetes/src/main.rs index 5d521a9d372..a416e2e99e6 100644 --- a/crates/buzz-backend-kubernetes/src/main.rs +++ b/crates/buzz-backend-kubernetes/src/main.rs @@ -14,6 +14,7 @@ mod cluster; mod config; mod env; mod gc; +mod identity; mod image; mod intent; mod naming; @@ -46,7 +47,11 @@ fn main() { std::process::exit(1); } - let response = respond(&input); + let response = match std::env::args().skip(1).collect::>().as_slice() { + [] => respond(&input), + [flag] if flag == "--check-identity" => check_identity(&input), + _ => Response::error("unsupported provider arguments"), + }; println!( "{}", serde_json::to_string(&response).unwrap_or_else(|e| { @@ -57,6 +62,21 @@ fn main() { ); } +/// Local identity preflight; never connects to Kubernetes or forwards secrets. +fn check_identity(input: &str) -> Response { + let request: Request = match serde_json::from_str(input) { + Ok(request) => request, + Err(_) => return Response::error("invalid identity preflight request"), + }; + let Request::Deploy(request) = request else { + return Response::error("identity preflight requires a deploy request"); + }; + match identity::check(&request, true) { + Ok(identity) => Response::identity_checked(identity.pod_name()), + Err(error) => Response::error(error), + } +} + /// Produce the single response for one request. Separated from `main` so the /// whole dispatch is testable without a process. fn respond(input: &str) -> Response { @@ -117,7 +137,7 @@ async fn deploy_agent(request: &wire::DeployRequest) -> Result { let cfg = config::parse(&request.provider_config)?; // Identity before any cluster contact: a malformed nsec is a refusal, not // a failed connection (§Deploy State Machine step 0). - let identity = naming::AgentIdentity::from_nsec(&request.agent.private_key_nsec)?; + let identity = identity::check(request, false)?; // One generation for this operation's first attempt; the reconciler mints // its own per attempt and restamps the correlator to match. @@ -196,4 +216,60 @@ mod tests { assert_eq!(json["protocol_version"], wire::PROTOCOL_VERSION); assert!(json["config_schema"]["properties"]["namespace"].is_object()); } + + #[test] + fn identity_policy_refusal_precedes_cluster_connection_and_hides_secrets() { + use nostr::nips::nip19::ToBech32; + let agent = nostr::Keys::generate(); + let owner = nostr::Keys::generate(); + let request = serde_json::json!({ + "op": "deploy", + "agent": { + "relay_url": "wss://example.invalid", + "private_key_nsec": agent.secret_key().to_bech32().unwrap(), + "auth_tag": "synthetic-secret-not-an-attestation", + "launch": {"owner_pubkey": owner.public_key().to_hex()} + }, + "provider_config": { + "namespace": "test", + "image": format!("example.invalid/pilot@sha256:{}", "a".repeat(64)), + "identity_policy": {"agent_pubkey": agent.public_key().to_hex(), + "owner_pubkey": owner.public_key().to_hex()} + } + }) + .to_string(); + let error = error_of(&respond(&request)); + assert_eq!(error, "owner attestation verification failed"); + assert!(!error.contains("synthetic-secret")); + assert_eq!(error_of(&check_identity(&request)), error); + } + + #[test] + fn local_identity_preflight_returns_only_public_pod_identity() { + use nostr::nips::nip19::ToBech32; + let agent = nostr::Keys::generate(); + let owner = nostr::Keys::generate(); + let request = serde_json::json!({ + "op": "deploy", "agent": { + "relay_url": "wss://example.invalid", + "private_key_nsec": agent.secret_key().to_bech32().unwrap(), + "auth_tag": buzz_sdk::nip_oa::compute_auth_tag(&owner, &agent.public_key(), "").unwrap(), + "launch": {"owner_pubkey": owner.public_key().to_hex()} + }, + "provider_config": {"identity_policy": { + "agent_pubkey": agent.public_key().to_hex(), "owner_pubkey": owner.public_key().to_hex() + }} + }).to_string(); + let response = serde_json::to_value(check_identity(&request)).unwrap(); + assert_eq!(response["ok"], true); + assert_eq!( + response["agent_id"], + format!("buzz-agent-{}", &agent.public_key().to_hex()[..12]) + ); + // A deploy response has no `preflight`; callers rely on its presence + // to detect a provider that ignored the flag and deployed. + assert_eq!(response["preflight"], "identity"); + assert_eq!(response.as_object().unwrap().len(), 3); + assert!(error_of(&check_identity(r#"{"op":"info"}"#)).contains("deploy request")); + } } diff --git a/crates/buzz-backend-kubernetes/src/pod.rs b/crates/buzz-backend-kubernetes/src/pod.rs index 725f98fc7dd..bb78816c8e0 100644 --- a/crates/buzz-backend-kubernetes/src/pod.rs +++ b/crates/buzz-backend-kubernetes/src/pod.rs @@ -12,9 +12,9 @@ use crate::naming::{ AgentIdentity, ANNOTATION_CREATE_INTENT, ANNOTATION_IMAGE, ANNOTATION_PUBKEY_FULL, }; use k8s_openapi::api::core::v1::{ - Capabilities, Container, EmptyDirVolumeSource, EnvFromSource, Pod, PodSecurityContext, PodSpec, - ResourceRequirements, SeccompProfile, Secret, SecretEnvSource, SecurityContext, Volume, - VolumeMount, + Capabilities, Container, EmptyDirVolumeSource, EnvFromSource, EnvVar, EnvVarSource, + ObjectFieldSelector, Pod, PodSecurityContext, PodSpec, ResourceRequirements, SeccompProfile, + Secret, SecretEnvSource, SecurityContext, Volume, VolumeMount, }; use k8s_openapi::apimachinery::pkg::api::resource::Quantity; use k8s_openapi::apimachinery::pkg::apis::meta::v1::ObjectMeta; @@ -83,7 +83,7 @@ pub fn build_pod( .into_iter() .collect(); - let requests: BTreeMap = [ + let mut requests: BTreeMap = [ ( "cpu".to_string(), Quantity(cfg.resources.cpu_request.clone()), @@ -95,7 +95,7 @@ pub fn build_pod( ] .into_iter() .collect(); - let limits: BTreeMap = [ + let mut limits: BTreeMap = [ ("cpu".to_string(), Quantity(cfg.resources.cpu_limit.clone())), ( "memory".to_string(), @@ -105,6 +105,13 @@ pub fn build_pod( .into_iter() .collect(); + if let Some(value) = &cfg.pod_options.ephemeral_storage_request { + requests.insert("ephemeral-storage".into(), Quantity(value.clone())); + } + if let Some(value) = &cfg.pod_options.ephemeral_storage_limit { + limits.insert("ephemeral-storage".into(), Quantity(value.clone())); + } + let container = Container { name: CONTAINER_NAME.to_string(), image: Some(cfg.image.as_str().to_string()), @@ -118,6 +125,20 @@ pub fn build_pod( }), ..Default::default() }]), + // Explicit env wins over envFrom, so a Secret cannot spoof the UID. + env: cfg.pod_options.setup_pending.then(|| { + vec![EnvVar { + name: "BUZZ_SETUP_POD_UID".into(), + value_from: Some(EnvVarSource { + field_ref: Some(ObjectFieldSelector { + api_version: Some("v1".into()), + field_path: "metadata.uid".into(), + }), + ..Default::default() + }), + ..Default::default() + }] + }), resources: Some(ResourceRequirements { requests: Some(requests), limits: Some(limits), @@ -151,6 +172,11 @@ pub fn build_pod( }, spec: Some(PodSpec { containers: vec![container], + node_selector: (!cfg.pod_options.node_selector.is_empty()) + .then(|| cfg.pod_options.node_selector.clone()), + tolerations: (!cfg.pod_options.tolerations.is_empty()) + .then(|| cfg.pod_options.tolerations.iter().map(Into::into).collect()), + active_deadline_seconds: cfg.pod_options.active_deadline_seconds, restart_policy: Some(RESTART_POLICY.to_string()), termination_grace_period_seconds: Some(TERMINATION_GRACE_SECONDS), // The agent runs prompted, untrusted code while holding an nsec; @@ -173,7 +199,10 @@ pub fn build_pod( }), volumes: Some(vec![Volume { name: WORKSPACE_VOLUME.to_string(), - empty_dir: Some(EmptyDirVolumeSource::default()), + empty_dir: Some(EmptyDirVolumeSource { + size_limit: cfg.pod_options.workspace_size_limit.clone().map(Quantity), + ..Default::default() + }), ..Default::default() }]), ..Default::default() @@ -187,13 +216,15 @@ pub fn intent_template( cfg: &ProviderConfig, env_keys: impl IntoIterator, ) -> IntentTemplate { - IntentTemplate::new( + let mut template = IntentTemplate::new( &cfg.namespace, &cfg.image, &cfg.resources, cfg.service_account.as_deref(), env_keys, - ) + ); + template.pod_options = cfg.pod_options.clone(); + template } #[cfg(test)] @@ -318,6 +349,104 @@ mod tests { assert_eq!(mount.mount_path, WORKSPACE_PATH); } + #[test] + fn setup_uid_is_opt_in_and_comes_from_the_api_not_the_secret() { + let mut cfg = provider_config(); + let fp = intent_template(&cfg, ["A".into()]).fingerprint(); + assert!(spec(&build_pod(&identity(), &cfg, "g", &fp)).containers[0] + .env + .is_none()); + cfg.pod_options.setup_pending = true; + let changed = intent_template(&cfg, ["A".into()]).fingerprint(); + assert_ne!(fp, changed); + let pod = build_pod(&identity(), &cfg, "g", &changed); + let env = spec(&pod).containers[0].env.as_ref().unwrap(); + assert_eq!(env.len(), 1); + assert_eq!(env[0].name, "BUZZ_SETUP_POD_UID"); + assert!(env[0].value.is_none()); + assert_eq!( + env[0] + .value_from + .as_ref() + .unwrap() + .field_ref + .as_ref() + .unwrap() + .field_path, + "metadata.uid" + ); + } + + #[test] + fn each_optional_pod_control_changes_intent() { + let cfg = provider_config(); + let baseline = intent_template(&cfg, ["A".into()]).fingerprint(); + for value in [ + serde_json::json!({"node_selector": {"kubernetes.io/arch": "amd64"}}), + serde_json::json!({"tolerations": [{"key": "architecture", "operator": "Equal", "value": "x86", "effect": "NoSchedule"}]}), + serde_json::json!({"workspace_size_limit": "2Gi"}), + serde_json::json!({"ephemeral_storage_request": "3Gi"}), + serde_json::json!({"ephemeral_storage_limit": "4Gi"}), + serde_json::json!({"active_deadline_seconds": 28800}), + serde_json::json!({"setup_pending": true}), + ] { + let mut changed = cfg.clone(); + changed.pod_options = serde_json::from_value(value).unwrap(); + assert_ne!( + intent_template(&changed, ["A".into()]).fingerprint(), + baseline + ); + } + let original = serde_json::to_value(intent_template(&cfg, ["A".into()])).unwrap(); + assert!(original.get("pod_options").is_none()); + } + + #[test] + fn optional_placement_storage_and_lifetime_reach_pod_and_intent() { + let mut cfg = provider_config(); + let baseline = intent_template(&cfg, ["A".into()]).fingerprint(); + cfg.pod_options = serde_json::from_value(serde_json::json!({ + "node_selector": {"kubernetes.io/arch": "amd64"}, + "tolerations": [{"key": "architecture", "operator": "Equal", + "value": "x86", "effect": "NoSchedule"}], + "workspace_size_limit": "2Gi", + "ephemeral_storage_request": "3Gi", + "ephemeral_storage_limit": "4Gi", + "active_deadline_seconds": 28800 + })) + .unwrap(); + let fp = intent_template(&cfg, ["A".into()]).fingerprint(); + assert_ne!(fp, baseline); + let pod = build_pod(&identity(), &cfg, "g", &fp); + let spec = spec(&pod); + assert_eq!( + spec.node_selector.as_ref().unwrap()["kubernetes.io/arch"], + "amd64" + ); + assert_eq!( + spec.tolerations.as_ref().unwrap()[0].value.as_deref(), + Some("x86") + ); + assert_eq!(spec.active_deadline_seconds, Some(28800)); + assert_eq!( + spec.volumes.as_ref().unwrap()[0] + .empty_dir + .as_ref() + .unwrap() + .size_limit, + Some(Quantity("2Gi".into())) + ); + let resources = spec.containers[0].resources.as_ref().unwrap(); + assert_eq!( + resources.requests.as_ref().unwrap()["ephemeral-storage"], + Quantity("3Gi".into()) + ); + assert_eq!( + resources.limits.as_ref().unwrap()["ephemeral-storage"], + Quantity("4Gi".into()) + ); + } + #[test] fn resources_carry_the_configured_requests_and_limits() { let mut cfg = provider_config(); diff --git a/crates/buzz-backend-kubernetes/src/reconcile.rs b/crates/buzz-backend-kubernetes/src/reconcile.rs index df2f99789ff..57672eea7e8 100644 --- a/crates/buzz-backend-kubernetes/src/reconcile.rs +++ b/crates/buzz-backend-kubernetes/src/reconcile.rs @@ -684,6 +684,7 @@ mod tests { resources: Resources::default(), inactivity_seconds: Some(7200), service_account: None, + pod_options: crate::config::PodOptions::default(), } } diff --git a/crates/buzz-backend-kubernetes/src/wire.rs b/crates/buzz-backend-kubernetes/src/wire.rs index 89b78b364df..89a81aece4a 100644 --- a/crates/buzz-backend-kubernetes/src/wire.rs +++ b/crates/buzz-backend-kubernetes/src/wire.rs @@ -92,6 +92,7 @@ pub struct LaunchBlock { pub enum Response { Info(InfoResponse), Deploy(DeployResponse), + IdentityPreflight(IdentityPreflightResponse), Error(ErrorResponse), } @@ -111,6 +112,16 @@ pub struct DeployResponse { pub agent_id: String, } +/// `--check-identity` success. `preflight` distinguishes it from a deploy +/// response, so a caller can fail closed when an older provider ignored the +/// flag and deployed. +#[derive(Debug, Serialize)] +pub struct IdentityPreflightResponse { + pub ok: bool, + pub agent_id: String, + pub preflight: &'static str, +} + #[derive(Debug, Serialize)] pub struct ErrorResponse { pub ok: bool, @@ -145,6 +156,14 @@ impl Response { agent_id: agent_id.into(), }) } + + pub fn identity_checked(agent_id: impl Into) -> Self { + Response::IdentityPreflight(IdentityPreflightResponse { + ok: true, + agent_id: agent_id.into(), + preflight: "identity", + }) + } } #[cfg(test)] @@ -242,6 +261,12 @@ mod tests { assert_eq!(v["ok"], true); assert_eq!(v["agent_id"], "buzz-agent-abc"); assert!(v.get("Deploy").is_none()); + assert!(v.get("preflight").is_none()); + + let v = serde_json::to_value(Response::identity_checked("buzz-agent-abc")).unwrap(); + assert_eq!(v["ok"], true); + assert_eq!(v["agent_id"], "buzz-agent-abc"); + assert_eq!(v["preflight"], "identity"); let v = serde_json::to_value(Response::error("boom")).unwrap(); assert_eq!(v["ok"], false); diff --git a/crates/buzz-backend-kubernetes/tests/wire_fixtures.rs b/crates/buzz-backend-kubernetes/tests/wire_fixtures.rs index b3049a98ef3..c7f09aa84c2 100644 --- a/crates/buzz-backend-kubernetes/tests/wire_fixtures.rs +++ b/crates/buzz-backend-kubernetes/tests/wire_fixtures.rs @@ -17,7 +17,12 @@ fn fixtures() -> PathBuf { /// Feed one request to the binary; return `(stdout, exit code)`. fn run(request: &str) -> (String, i32) { + run_with_args(request, &[]) +} + +fn run_with_args(request: &str, args: &[&str]) -> (String, i32) { let mut child = Command::new(env!("CARGO_BIN_EXE_buzz-backend-kubernetes")) + .args(args) // A kubeconfig that does not exist, so a fixture that accidentally // reaches the cluster fails loudly here instead of depending on // whatever cluster the developer is pointed at. @@ -39,6 +44,46 @@ fn run(request: &str) -> (String, i32) { ) } +#[test] +fn built_binary_preflights_and_repeats_identity_policy_before_cluster_access() { + use nostr::nips::nip19::ToBech32; + let agent = nostr::Keys::generate(); + let owner = nostr::Keys::generate(); + let nsec = agent.secret_key().to_bech32().unwrap(); + let mut request: serde_json::Value = + serde_json::from_str(&read("deploy-full-launch.request.json")).unwrap(); + request["agent"]["private_key_nsec"] = serde_json::json!(nsec); + request["agent"]["auth_tag"] = + serde_json::json!( + buzz_sdk::nip_oa::compute_auth_tag(&owner, &agent.public_key(), "").unwrap() + ); + request["agent"]["launch"]["owner_pubkey"] = serde_json::json!(owner.public_key().to_hex()); + request["provider_config"]["identity_policy"] = serde_json::json!({ + "agent_pubkey": agent.public_key().to_hex(), "owner_pubkey": owner.public_key().to_hex() + }); + let (stdout, code) = run_with_args(&request.to_string(), &["--check-identity"]); + assert_eq!(code, 0); + assert!(!stdout.contains(&nsec)); + assert_eq!( + serde_json::from_str::(&stdout).unwrap(), + serde_json::json!({ + "ok": true, + "agent_id": format!("buzz-agent-{}", &agent.public_key().to_hex()[..12]), + "preflight": "identity" + }) + ); + request["agent"]["auth_tag"] = serde_json::json!("synthetic-secret-invalid-attestation"); + for args in [vec![], vec!["--check-identity"]] { + let (stdout, code) = run_with_args(&request.to_string(), &args); + assert_eq!(code, 0); + assert!(!stdout.contains("synthetic-secret")); + assert_eq!( + serde_json::from_str::(&stdout).unwrap()["error"], + "owner attestation verification failed" + ); + } +} + fn read(name: &str) -> String { std::fs::read_to_string(fixtures().join(name)) .unwrap_or_else(|e| panic!("could not read fixture {name}: {e}")) diff --git a/docs/remote-agents.md b/docs/remote-agents.md index 56de86bade0..595b0b48196 100644 --- a/docs/remote-agents.md +++ b/docs/remote-agents.md @@ -184,9 +184,13 @@ one. - **(I2) No secrets in configuration.** `provider_config` — the persisted, schema-rendered, UI-visible settings object — MUST NOT carry secrets. - Enforced by validation: flat object, scalar values only, ≤20 fields, ≤64KB, - and any key whose word-split contains `secret|password|token|key|credential` - is rejected. The match is against key *names*, so it is a lint with false + Enforced by validation: top-level values are scalars, except for the + named structured objects that a provider schema declares + (`pod_options`, `identity_policy`); ≤20 fields, ≤64KB; and any key whose + word-split contains `secret|password|token|key|credential` is rejected. + Structured objects hold only public, non-secret settings, and their nested + keys are subject to the same lint, except that a nested field named exactly + `key` (for example, a Kubernetes toleration key) is allowed. The match is against key *names*, so it is a lint with false positives: a field like `ssh_key_path` holds a path, not a credential, and is refused anyway — a provider author hitting this MUST rename the field (e.g. `identity_file`), not weaken the validator; the rule's job is @@ -1395,9 +1399,31 @@ removable with `kubectl delete`. ### `provider_config` v1 fields `context`, `namespace`, `image`, `cpu_request`, `memory_request`, -`cpu_limit`, `memory_limit`, `inactivity_seconds`, `service_account` — -9 of the 20-field validation cap. Node selectors, tolerations, and PVCs are -deliberately baked out of v1 to preserve budget. +`cpu_limit`, `memory_limit`, `inactivity_seconds`, `service_account`. +The optional `pod_options` object adds `node_selector`, `tolerations`, +`workspace_size_limit`, `ephemeral_storage_request`, `ephemeral_storage_limit`, +`active_deadline_seconds`, and `setup_pending`. +Configured options participate in the create-intent fingerprint; omitted options preserve existing behavior. +The provider rejects unknown options, nonpositive deadlines, and storage values that are not non-negative Kubernetes quantities. +It also rejects a zero `workspace_size_limit` or `ephemeral_storage_limit`. +Each toleration accepts only `key`, `operator` (`Exists` or `Equal`), `value`, `effect` (`NoSchedule`, `PreferNoSchedule`, or `NoExecute`), and `tolerationSeconds`. +Kubernetes validates the remaining scheduling rules, including a storage request that is larger than its limit. +The absolute deadline includes startup and active work, independently of the harness inactivity timer. + +`setup_pending=true` injects `BUZZ_SETUP_POD_UID` from the downward API `metadata.uid`, overriding Secret environment values. +The image uses this binding to complete owner setup before starting the harness. +Provider success still reports Pod startup; account readiness requires a runtime response. + +The optional `identity_policy` contains approved `agent_pubkey` and `owner_pubkey` values. +Before cluster access, the provider checks the derived agent key, launch owner, owner attestation, and attestation time conditions. +It rejects using the owner's key as the agent key. This admission check does not change Pod create intent. +`--check-identity` performs the same check on a deploy request from stdin without cluster access. +On success it returns `{"ok": true, "agent_id": "", "preflight": "identity"}`; on failure it returns a sanitized refusal. +Older providers ignore command-line arguments and treat the request as a real deploy. +Launchers must confirm that the `info` schema includes `identity_policy` before they invoke `--check-identity`. +Launchers must also treat a success response without `"preflight": "identity"` as a failure, because that response comes from a real deploy. + +Launchers must check the `info` schema for required options before credential handoff, since older providers ignore unknown fields. ### Distribution