diff --git a/.github/workflows/branch-e2e.yml b/.github/workflows/branch-e2e.yml index 2e316f88bd..10320c5558 100644 --- a/.github/workflows/branch-e2e.yml +++ b/.github/workflows/branch-e2e.yml @@ -379,6 +379,10 @@ jobs: agent_sandbox_version: v0.5.0 topology: sidecar extra_helm_values: deploy/helm/openshell/ci/values-sidecar.yaml + - agent_sandbox_api: v1beta1 + agent_sandbox_version: v0.5.0 + topology: proxy-pod + extra_helm_values: deploy/helm/openshell/ci/values-proxy-pod.yaml permissions: actions: read contents: read diff --git a/Cargo.lock b/Cargo.lock index 4c38a39778..b6cfb05c97 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3959,6 +3959,7 @@ dependencies = [ "miette", "notify", "openshell-core", + "openshell-isolation-interface", "openshell-otel", "openshell-otel-test-support", "openshell-policy", @@ -3966,6 +3967,8 @@ dependencies = [ "opentelemetry_sdk", "prost", "prost-types", + "rand 0.9.4", + "rcgen", "serde", "serde_json", "temp-env", diff --git a/crates/openshell-driver-kubernetes/Cargo.toml b/crates/openshell-driver-kubernetes/Cargo.toml index 3a3d843f1a..b1816de367 100644 --- a/crates/openshell-driver-kubernetes/Cargo.toml +++ b/crates/openshell-driver-kubernetes/Cargo.toml @@ -16,6 +16,7 @@ path = "src/main.rs" [dependencies] openshell-core = { path = "../openshell-core", default-features = false } +openshell-isolation-interface = { path = "../openshell-isolation-interface" } openshell-otel = { path = "../openshell-otel" } openshell-policy = { path = "../openshell-policy" } @@ -37,6 +38,8 @@ tracing-subscriber = { workspace = true } thiserror = { workspace = true } miette = { workspace = true } notify = "8" +rand = { workspace = true } +rcgen = { workspace = true } [dev-dependencies] openshell-otel-test-support = { path = "../openshell-otel-test-support" } diff --git a/crates/openshell-driver-kubernetes/README.md b/crates/openshell-driver-kubernetes/README.md index 02dcfe5e87..f98e92a26a 100644 --- a/crates/openshell-driver-kubernetes/README.md +++ b/crates/openshell-driver-kubernetes/README.md @@ -49,6 +49,59 @@ this driver. Kubernetes owns scheduling and pod lifecycle. The `openshell-sandbox` supervisor inside each workload owns agent isolation, credential injection, policy polling, logs, and the gateway relay. +### RFC 0012 proxy-pod topology + +Set `topology = "proxy-pod"` and explicitly set +`proxy_pod.network_policy_enforced = true` after verifying that the cluster CNI +enforces `networking.k8s.io/v1` `NetworkPolicy`. The workload pod runs +`openshell-sandbox --mode=boundary`; a separately scheduled Deployment runs +`openshell-sandbox --mode=control` and owns gateway, provider, policy, and +upstream credentials. + +The driver creates the workload egress fence before the Sandbox CR and creates +the CR in a suspended state. It then observes the namespace, Sandbox, +Deployment, and fence identities; creates one immutable bootstrap Secret with +separately mounted boundary and control records; and starts control before +releasing the workload. The boundary/control TCP protocol uses a per-sandbox CA +and server certificate in addition to its bootstrap-token authentication. + +The workload `NetworkPolicy` denies every workload-initiated egress connection +and permits ingress only from the uniquely paired control pod to the configured +boundary port. The workload receives no gateway endpoint, gateway TLS secret, +projected service-account token, provider identity socket, or upstream-proxy +credential. Its boundary resolves process and socket-owner identity locally and +streams attributed connections to control. Control performs destination and L7 +policy evaluation and makes the actual upstream connection. + +Stop suspends the workload before scaling control to zero. Start restores +control before resuming the workload. Delete retains the unowned egress fence +until workload-pod deletion is confirmed; Kubernetes garbage collection removes +the Sandbox-owned Secret, Service, Deployment, and control policy. +The control Deployment becomes Available only after its TCP readiness probe +observes that control has attached, confirmed enforcement, started or resumed +the boundary process, and installed the access plane. The driver combines that +availability with the boundary Service and workload-fence presence when +publishing sandbox readiness. Periodic reconciliation repairs control replica +drift. It suspends the workload if an immutable-identity companion is deleted +and cannot be reconstructed without changing the boundary's trusted claims. +The control container runs as the namespace-resolved non-root UID/GID with no Linux +capabilities, a read-only root filesystem, and the runtime-default seccomp +profile. Orphan cleanup retains newly created workload fences for at least five +minutes so it cannot race the fence-first Sandbox creation sequence. +If gateway failure interrupts initial companion creation, reconciliation keeps +the Sandbox suspended and rolls back the unchanged partial CR after five +minutes. Bootstrap Secrets remain create-only to avoid granting the gateway +read access to arbitrary Secrets in operator namespaces. A deleted Secret does +not affect an already-mounted control; any replacement pod fails its semantic +readiness probe and the driver reports the Sandbox not ready. + +This posture requires more than accepting a `NetworkPolicy` object. Policies +are additive, and Kubernetes does not expose portable CNI enforcement +attestation. The sandbox namespace must therefore be controlled so untrusted +principals cannot create permissive policies, create pods, read bootstrap +Secrets, or spoof the pair labels. The explicit acknowledgement is an operator +contract, not runtime CNI detection. + ## Sandbox Resource The driver works with the `agents.x-k8s.io` `Sandbox` custom resource. It diff --git a/crates/openshell-driver-kubernetes/src/config.rs b/crates/openshell-driver-kubernetes/src/config.rs index 805c0314b0..593a908015 100644 --- a/crates/openshell-driver-kubernetes/src/config.rs +++ b/crates/openshell-driver-kubernetes/src/config.rs @@ -72,6 +72,9 @@ pub enum SupervisorTopology { /// Run network supervision in a privileged sidecar and process supervision /// as a low-capability wrapper in the agent container. Sidecar, + /// Run the shared RFC 0012 control role in a separate pod and the boundary + /// role beside the workload. + ProxyPod, } impl std::fmt::Display for SupervisorTopology { @@ -79,6 +82,7 @@ impl std::fmt::Display for SupervisorTopology { match self { Self::Combined => f.write_str("combined"), Self::Sidecar => f.write_str("sidecar"), + Self::ProxyPod => f.write_str("proxy-pod"), } } } @@ -90,11 +94,47 @@ impl FromStr for SupervisorTopology { match s { "combined" => Ok(Self::Combined), "sidecar" => Ok(Self::Sidecar), + "proxy-pod" => Ok(Self::ProxyPod), other => Err(format!("unknown topology '{other}'")), } } } +/// Driver-owned requirements for the cross-pod RFC 0012 topology. +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(default, deny_unknown_fields)] +pub struct KubernetesProxyPodConfig { + /// Explicit operator assertion that the cluster CNI enforces + /// `networking.k8s.io/v1` `NetworkPolicy` for the sandbox namespaces. + pub network_policy_enforced: bool, + /// TCP port exposed by the workload boundary to its paired control pod. + pub boundary_port: u16, +} + +impl Default for KubernetesProxyPodConfig { + fn default() -> Self { + Self { + network_policy_enforced: false, + boundary_port: 5500, + } + } +} + +impl KubernetesProxyPodConfig { + pub fn validate(&self) -> Result<(), String> { + if !self.network_policy_enforced { + return Err( + "proxy-pod topology requires proxy_pod.network_policy_enforced = true after the operator has verified CNI NetworkPolicy enforcement" + .to_string(), + ); + } + if self.boundary_port == 0 { + return Err("proxy_pod.boundary_port must be nonzero".to_string()); + } + Ok(()) + } +} + /// How workspaces map to Kubernetes namespaces. #[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)] #[serde(rename_all = "kebab-case")] @@ -326,6 +366,8 @@ pub struct KubernetesComputeConfig { pub topology: SupervisorTopology, /// Sidecar-only settings used when `topology = "sidecar"`. pub sidecar: KubernetesSidecarConfig, + /// Cross-pod boundary settings used when `topology = "proxy-pod"`. + pub proxy_pod: KubernetesProxyPodConfig, /// Corporate HTTP forward proxy used by the network supervisor for /// policy-approved TLS CONNECT egress. pub https_proxy: Option, @@ -451,6 +493,7 @@ impl Default for KubernetesComputeConfig { supervisor_sideload_method: SupervisorSideloadMethod::default(), topology: SupervisorTopology::default(), sidecar: KubernetesSidecarConfig::default(), + proxy_pod: KubernetesProxyPodConfig::default(), https_proxy: None, no_proxy: None, proxy_auth_secret_name: None, @@ -503,7 +546,11 @@ impl KubernetesComputeConfig { } pub fn validate_proxy_uid(&self) -> Result<(), String> { - self.sidecar.validate_proxy_uid() + self.sidecar.validate_proxy_uid()?; + if self.topology == SupervisorTopology::ProxyPod { + self.proxy_pod.validate()?; + } + Ok(()) } /// Validate the operator-owned corporate upstream proxy configuration. @@ -580,7 +627,7 @@ impl KubernetesComputeConfig { } if self.topology == SupervisorTopology::Combined { return Err( - "proxy credential Secrets require topology = \"sidecar\"; combined topology shares the credential mount with the workload and fsGroup can make it readable by the sandbox user" + "proxy credential Secrets require topology = \"sidecar\" or \"proxy-pod\"; combined topology shares the credential mount with the workload and fsGroup can make it readable by the sandbox user" .to_string(), ); } @@ -946,6 +993,22 @@ mod tests { assert_eq!(cfg.topology, SupervisorTopology::Combined); } + #[test] + fn proxy_pod_requires_network_policy_enforcement_acknowledgement() { + let mut cfg = KubernetesComputeConfig { + topology: SupervisorTopology::ProxyPod, + ..KubernetesComputeConfig::default() + }; + assert!( + cfg.validate_proxy_uid() + .unwrap_err() + .contains("network_policy_enforced") + ); + + cfg.proxy_pod.network_policy_enforced = true; + cfg.validate_proxy_uid().unwrap(); + } + #[test] fn serde_rejects_sidecar_binary_identity_field() { let json = serde_json::json!({ diff --git a/crates/openshell-driver-kubernetes/src/driver.rs b/crates/openshell-driver-kubernetes/src/driver.rs index afe3f579e0..0ecdf5a7e2 100644 --- a/crates/openshell-driver-kubernetes/src/driver.rs +++ b/crates/openshell-driver-kubernetes/src/driver.rs @@ -10,13 +10,22 @@ use crate::config::{ SupervisorSideloadMethod, SupervisorTopology, WorkspaceMode, is_dns_1123_label, managed_namespace, managed_namespace_prefix, validate_managed_namespace_name, }; +use crate::isolation::{BOUNDARY_PAIR_LABEL, BOUNDARY_ROLE_LABEL, KubernetesProxyPodBoundarySpec}; +use crate::proxy_pod::{ + BOUNDARY_CERTIFICATE_KEY, BOUNDARY_CERTIFICATE_PATH, BOUNDARY_CONFIG_KEY, BOUNDARY_CONFIG_PATH, + BOUNDARY_PRIVATE_KEY, BOUNDARY_PRIVATE_KEY_PATH, BOUNDARY_TLS_SERVER_NAME, ProxyPodNames, + bootstrap_secret, boundary_service, control_deployment, control_egress_policy, + generate_boundary_tls_material, generate_proxy_ca_material, + sandbox_owner_reference as proxy_pod_sandbox_owner_reference, workload_fence, +}; use futures::{Stream, StreamExt, TryStreamExt}; +use k8s_openapi::api::apps::v1::{Deployment, ReplicaSet}; use k8s_openapi::api::authentication::v1::{ TokenReview, TokenReviewSpec, TokenReviewStatus, UserInfo, }; use k8s_openapi::api::core::v1::{ Event as KubeEventObj, Namespace, Node, PersistentVolumeClaimVolumeSource, Pod, Secret, - ServiceAccount, Volume, VolumeMount, + Service, ServiceAccount, Volume, VolumeMount, }; use k8s_openapi::api::networking::v1::{ NetworkPolicy, NetworkPolicyIngressRule, NetworkPolicyPeer, NetworkPolicyPort, @@ -30,6 +39,7 @@ use kube::api::{ use kube::core::gvk::GroupVersionKind; use kube::core::{DynamicObject, ObjectMeta}; use kube::runtime::WatchStreamExt; +use kube::runtime::wait::await_condition; use kube::runtime::watcher::{self, Event}; use kube::{Client, Error as KubeError}; use openshell_core::driver_mounts; @@ -52,6 +62,7 @@ use openshell_core::proto::compute::v1::{ watch_sandboxes_event, }; use openshell_core::proto_struct::{struct_to_json_object, value_to_json}; +use rand::RngCore as _; use serde::Deserialize; use std::collections::{BTreeMap, HashSet}; use std::path::{Path, PathBuf}; @@ -67,6 +78,9 @@ pub type WatchStream = const MANAGED_SSH_NETWORK_POLICY_NAME: &str = "openshell-sandbox-ssh"; const AGENT_SANDBOX_TRACE_CONTEXT_ANNOTATION: &str = "opentelemetry.io/trace-context"; +const ANNOTATION_SUPERVISOR_TOPOLOGY: &str = "openshell.ai/supervisor-topology"; +const ANNOTATION_PROXY_POD_BOOTSTRAPPING: &str = "openshell.ai/proxy-pod-bootstrapping"; +const ANNOTATION_PROXY_POD_READINESS: &str = "openshell.ai/proxy-pod-readiness"; #[derive(Debug, thiserror::Error)] pub enum KubernetesDriverError { @@ -107,6 +121,13 @@ impl From for openshell_core::ComputeDriverError { /// This prevents gRPC handlers from blocking indefinitely when the k8s /// API server is unreachable or slow. const KUBE_API_TIMEOUT: Duration = Duration::from_secs(30); +const PROXY_POD_RECONCILE_INTERVAL: Duration = Duration::from_secs(30); +const PROXY_POD_FENCE_QUIESCE_WINDOW: Duration = Duration::from_secs(6); +const PROXY_POD_FENCE_QUIESCE_INTERVAL: Duration = Duration::from_secs(2); +/// A pre-created fence may legitimately have no Sandbox CR while create is in flight. +const PROXY_POD_ORPHAN_FENCE_MIN_AGE: Duration = Duration::from_secs(5 * 60); +/// Bound how long a crash-interrupted, fail-closed bootstrap may remain stranded. +const PROXY_POD_BOOTSTRAP_GRACE: Duration = Duration::from_secs(5 * 60); /// Kubernetes defaults pod termination to 30 seconds when the pod template /// omits `terminationGracePeriodSeconds`. @@ -621,7 +642,9 @@ impl KubernetesComputeDriver { })?; validate_pod_uid(&pod, &identity.pod_uid)?; let sandbox_id = pod_sandbox_id(&pod)?; - let owner = sandbox_owner_reference(&pod)?; + let (owner, via_proxy_control) = self + .resolve_sandbox_owner(&pod, &identity.namespace, &sandbox_id) + .await?; let sandboxes = self .supported_agent_sandbox_api(self.client.clone(), &identity.namespace) .await @@ -632,10 +655,85 @@ impl KubernetesComputeDriver { warn!(sandbox = %owner.name, %error, "failed to read authenticated Sandbox resource"); tonic::Status::internal("failed to read authenticated Sandbox resource") })?.ok_or_else(|| tonic::Status::permission_denied("sandbox owner not found"))?; - validate_sandbox_owner_identity(owner, &sandbox_id, &sandbox)?; + validate_sandbox_owner_identity(&owner, &sandbox_id, &sandbox)?; + if via_proxy_control && object_topology(&sandbox) != SupervisorTopology::ProxyPod { + return Err(tonic::Status::permission_denied( + "control pod owner is not a proxy-pod Sandbox", + )); + } Ok(sandbox_id) } + #[allow(clippy::result_large_err)] + async fn resolve_sandbox_owner( + &self, + pod: &Pod, + namespace: &str, + sandbox_id: &str, + ) -> Result<(OwnerReference, bool), tonic::Status> { + if let Ok(owner) = sandbox_owner_reference(pod) { + return Ok((owner.clone(), false)); + } + validate_proxy_control_labels(pod, sandbox_id)?; + let names = ProxyPodNames::new(sandbox_id); + let replica_set_owner = controller_owner_reference( + pod.metadata.owner_references.as_deref().unwrap_or_default(), + ) + .filter(|owner| owner.api_version == "apps/v1" && owner.kind == "ReplicaSet") + .ok_or_else(|| tonic::Status::permission_denied("pod is not controlled by a Sandbox"))?; + let replica_set = Api::::namespaced(self.client.clone(), namespace) + .get(&replica_set_owner.name) + .await + .map_err(|error| { + tonic::Status::internal(format!("ReplicaSet lookup failed: {error}")) + })?; + validate_owner_uid(&replica_set.metadata.uid, replica_set_owner)?; + let deployment_owner = controller_owner_reference( + replica_set + .metadata + .owner_references + .as_deref() + .unwrap_or_default(), + ) + .filter(|owner| owner.api_version == "apps/v1" && owner.kind == "Deployment") + .ok_or_else(|| { + tonic::Status::permission_denied("ReplicaSet is not controlled by a Deployment") + })?; + if deployment_owner.name != names.control_deployment { + return Err(tonic::Status::permission_denied( + "control pod is not owned by the expected proxy-pod Deployment", + )); + } + let deployment = Api::::namespaced(self.client.clone(), namespace) + .get(&deployment_owner.name) + .await + .map_err(|error| { + tonic::Status::internal(format!("Deployment lookup failed: {error}")) + })?; + validate_owner_uid(&deployment.metadata.uid, deployment_owner)?; + validate_proxy_control_labels_from_metadata(&deployment.metadata, sandbox_id)?; + let owner = deployment + .metadata + .owner_references + .as_deref() + .and_then(|owners| { + owners + .iter() + .find(|owner| owner.controller == Some(true) && owner.kind == SANDBOX_KIND) + }) + .filter(|owner| { + matches!( + owner.api_version.as_str(), + "agents.x-k8s.io/v1beta1" | "agents.x-k8s.io/v1alpha1" + ) + }) + .cloned() + .ok_or_else(|| { + tonic::Status::permission_denied("Deployment is not controlled by a Sandbox") + })?; + Ok((owner, true)) + } + fn accepts_auth_namespace(&self, namespace: &str) -> bool { accepts_auth_namespace(&self.config, self.operator_allowlist.as_ref(), namespace) } @@ -1325,20 +1423,23 @@ impl KubernetesComputeDriver { let selector = self.sandbox_lookup_selector(sandbox_id); let lp = ListParams::default().labels(&selector); match tokio::time::timeout(KUBE_API_TIMEOUT, agent_sandbox_api.api.list(&lp)).await { - Ok(Ok(list)) => list.items.into_iter().next().map_or_else( - || { + Ok(Ok(list)) => { + let Some(obj) = list.items.into_iter().next() else { debug!(sandbox_id = %sandbox_id, "Sandbox not found in Kubernetes"); - Ok(None) - }, - |obj| { - let ns = obj - .metadata - .namespace - .clone() - .unwrap_or_else(|| self.config.namespace.clone()); - Ok(sandbox_from_object(&ns, obj).ok().map(|(_, s)| s)) - }, - ), + return Ok(None); + }; + let ns = obj + .metadata + .namespace + .clone() + .unwrap_or_else(|| self.config.namespace.clone()); + Ok( + sandbox_from_object_with_proxy_pod_readiness(&self.client, &ns, obj) + .await + .ok() + .map(|(_, sandbox)| sandbox), + ) + } Ok(Err(err)) => { warn!( sandbox_id = %sandbox_id, @@ -1380,25 +1481,22 @@ impl KubernetesComputeDriver { .await { Ok(Ok(list)) => { - let mut sandboxes: Vec = list - .items - .into_iter() - .filter_map(|obj| { - let name = obj.metadata.name.clone().unwrap_or_default(); - let ns = obj - .metadata - .namespace - .clone() - .unwrap_or_else(|| self.config.namespace.clone()); - match sandbox_from_object(&ns, obj) { - Ok((_, s)) => Some(s), - Err(err) => { - warn!(object_name = %name, error = %err, "skipping unrecognized Sandbox in list"); - None - } + let mut sandboxes = Vec::new(); + for obj in list.items { + let name = obj.metadata.name.clone().unwrap_or_default(); + let ns = obj + .metadata + .namespace + .clone() + .unwrap_or_else(|| self.config.namespace.clone()); + match sandbox_from_object_with_proxy_pod_readiness(&self.client, &ns, obj).await + { + Ok((_, sandbox)) => sandboxes.push(sandbox), + Err(err) => { + warn!(object_name = %name, error = %err, "skipping unrecognized Sandbox in list"); } - }) - .collect(); + } + } sandboxes.sort_by(|left, right| { left.name .cmp(&right.name) @@ -1544,12 +1642,35 @@ impl KubernetesComputeDriver { }; validate_sidecar_proxy_identity(¶ms)?; - let data = sandbox_to_k8s_spec(sandbox.spec.as_ref(), ¶ms) - .map_err(KubernetesDriverError::InvalidArgument)?; let kube_name = self.config.kube_resource_name(workspace, name); + let mut data = sandbox_to_k8s_spec(sandbox.spec.as_ref(), ¶ms) + .map_err(KubernetesDriverError::InvalidArgument)?; + let proxy_names = ProxyPodNames::new(&sandbox.id); + if self.config.topology == SupervisorTopology::ProxyPod { + self.create_proxy_pod_fence(&target_namespace, &proxy_names, sandbox, &kube_name) + .await?; + // A missing bootstrap Secret keeps both pods inert as defense in + // depth, but the CR is also created suspended so the controller + // never races an unfenced workload into execution. + if agent_sandbox_api.resource.version == SANDBOX_VERSION_V1ALPHA1 { + data["spec"]["replicas"] = serde_json::json!(0); + } else { + data["spec"]["operatingMode"] = serde_json::json!("Suspended"); + } + } let mut obj = DynamicObject::new(&kube_name, &agent_sandbox_api.resource); let mut annotations = sandbox_annotations(sandbox); add_trace_context_annotation(&mut annotations); + annotations.insert( + ANNOTATION_SUPERVISOR_TOPOLOGY.to_string(), + self.config.topology.to_string(), + ); + if self.config.topology == SupervisorTopology::ProxyPod { + annotations.insert( + ANNOTATION_PROXY_POD_BOOTSTRAPPING.to_string(), + "true".to_string(), + ); + } for key in [ crate::config::ANNOTATION_SCC_UID_RANGE, crate::config::ANNOTATION_SCC_SUPPLEMENTAL_GROUPS, @@ -1559,27 +1680,27 @@ impl KubernetesComputeDriver { } } obj.metadata = ObjectMeta { - name: Some(kube_name), - namespace: Some(target_namespace), + name: Some(kube_name.clone()), + namespace: Some(target_namespace.clone()), labels: Some(sandbox_labels(sandbox, Some(&self.config.gateway_id))), annotations: Some(annotations), ..Default::default() }; obj.data = data; - match tokio::time::timeout( + let created = match tokio::time::timeout( KUBE_API_TIMEOUT, agent_sandbox_api.api.create(&PostParams::default(), &obj), ) .await { - Ok(Ok(_result)) => { + Ok(Ok(result)) => { info!( sandbox_id = %sandbox.id, sandbox_name = %name, "Sandbox created in Kubernetes successfully" ); - Ok(()) + result } Ok(Err(err)) => { warn!( @@ -1588,7 +1709,7 @@ impl KubernetesComputeDriver { error = %err, "Failed to create sandbox in Kubernetes" ); - Err(KubernetesDriverError::from_kube(err)) + return Err(KubernetesDriverError::from_kube(err)); } Err(_elapsed) => { warn!( @@ -1597,12 +1718,328 @@ impl KubernetesComputeDriver { timeout_secs = KUBE_API_TIMEOUT.as_secs(), "Timed out creating sandbox in Kubernetes" ); - Err(KubernetesDriverError::Message(format!( + return Err(KubernetesDriverError::Message(format!( "timed out after {}s waiting for Kubernetes API", KUBE_API_TIMEOUT.as_secs() - ))) + ))); } + }; + if self.config.topology == SupervisorTopology::ProxyPod + && let Err(error) = self + .create_proxy_pod_companions( + sandbox, + &target_namespace, + &kube_name, + &agent_sandbox_api, + &created, + &proxy_names, + resolved_user_id, + resolved_group_id, + ) + .await + { + warn!(sandbox_id = %sandbox.id, %error, "proxy-pod provisioning failed; rolling back Sandbox CR"); + let _ = agent_sandbox_api + .api + .delete(&kube_name, &DeleteParams::default()) + .await; + return Err(error); + } + Ok(()) + } + + async fn create_proxy_pod_fence( + &self, + namespace: &str, + names: &ProxyPodNames, + sandbox: &Sandbox, + workload_pod_name: &str, + ) -> Result<(), KubernetesDriverError> { + let mut policy = workload_fence( + namespace, + names, + &sandbox.id, + self.config.proxy_pod.boundary_port, + ) + .workload_policy; + let labels = policy.metadata.labels.get_or_insert_default(); + labels.insert( + LABEL_MANAGED_BY.to_string(), + LABEL_MANAGED_BY_VALUE.to_string(), + ); + labels.insert(LABEL_GATEWAY_ID.to_string(), self.config.gateway_id.clone()); + labels.insert(LABEL_SANDBOX_ID.to_string(), sandbox.id.clone()); + labels.insert( + "openshell.ai/component".to_string(), + "boundary-workload-fence".to_string(), + ); + policy.metadata.annotations.get_or_insert_default().insert( + "openshell.ai/workload-pod-name".to_string(), + workload_pod_name.to_string(), + ); + let policies: Api = Api::namespaced(self.client.clone(), namespace); + create_or_validate_proxy_pod_fence(&policies, &policy).await + } + + #[allow(clippy::too_many_arguments, clippy::similar_names)] + async fn create_proxy_pod_companions( + &self, + sandbox: &Sandbox, + namespace: &str, + cr_name: &str, + sandbox_api: &AgentSandboxApi, + sandbox_cr: &DynamicObject, + names: &ProxyPodNames, + agent_uid: u32, + agent_gid: u32, + ) -> Result<(), KubernetesDriverError> { + let cr_uid = sandbox_cr.metadata.uid.as_deref().ok_or_else(|| { + KubernetesDriverError::Message("created Sandbox CR has no UID".to_string()) + })?; + let namespace_uid = Api::::all(self.client.clone()) + .get(namespace) + .await + .map_err(KubernetesDriverError::from_kube)? + .metadata + .uid + .ok_or_else(|| { + KubernetesDriverError::Message("sandbox namespace has no UID".to_string()) + })?; + let dependent_owner = proxy_pod_sandbox_owner_reference( + cr_name, + cr_uid, + &sandbox_api.resource.api_version, + false, + ); + let controller_owner = proxy_pod_sandbox_owner_reference( + cr_name, + cr_uid, + &sandbox_api.resource.api_version, + true, + ); + + let services: Api = Api::namespaced(self.client.clone(), namespace); + let service = services + .create( + &PostParams::default(), + &boundary_service( + namespace, + names, + &sandbox.id, + self.config.proxy_pod.boundary_port, + dependent_owner.clone(), + ), + ) + .await + .map_err(KubernetesDriverError::from_kube)?; + let service_ip: std::net::IpAddr = service + .spec + .and_then(|spec| spec.cluster_ip) + .filter(|ip| ip != "None") + .ok_or_else(|| { + KubernetesDriverError::Message("boundary Service has no ClusterIP".to_string()) + })? + .parse() + .map_err(|error| { + KubernetesDriverError::Message(format!( + "invalid boundary Service ClusterIP: {error}" + )) + })?; + + let main_process_spec = openshell_core::sandbox_env::MainProcessConfig::encode_driver_spec( + sandbox.spec.as_ref(), + ) + .map_err(|error| { + KubernetesDriverError::InvalidArgument(format!("encode main process spec: {error}")) + })?; + let log_level = openshell_core::driver_utils::sandbox_log_level(sandbox, "info"); + let deployments: Api = Api::namespaced(self.client.clone(), namespace); + let deployment = deployments + .create( + &PostParams::default(), + &control_deployment( + namespace, + names, + &sandbox.id, + &sandbox.name, + &self.config.gateway_id, + &self.config.supervisor_image, + &self.config.supervisor_image_pull_policy, + &self.config.service_account_name, + agent_uid, + agent_gid, + &self.config.image_pull_secrets, + &self.config.grpc_endpoint, + &self.config.client_tls_secret_name, + &main_process_spec, + &log_level, + self.config.effective_sa_token_ttl_secs(), + self.config.https_proxy.as_deref(), + self.config.no_proxy.as_deref(), + self.config + .proxy_auth_secret_name + .as_deref() + .zip(self.config.proxy_auth_secret_key.as_deref()), + self.config.proxy_auth_allow_insecure == Some(true), + self.config.proxy_connect_by_hostname == Some(true), + self.config.provider_spiffe_enabled().then_some( + self.config + .provider_spiffe_workload_api_socket_path + .as_str(), + ), + controller_owner, + ), + ) + .await + .map_err(KubernetesDriverError::from_kube)?; + let deployment_uid = deployment.metadata.uid.ok_or_else(|| { + KubernetesDriverError::Message("control Deployment has no UID".to_string()) + })?; + + let policies: Api = Api::namespaced(self.client.clone(), namespace); + policies + .create( + &PostParams::default(), + &control_egress_policy(namespace, names, &sandbox.id, dependent_owner.clone()), + ) + .await + .map_err(KubernetesDriverError::from_kube)?; + let fence = policies + .get(&names.workload_policy) + .await + .map_err(KubernetesDriverError::from_kube)?; + let fence_uid = fence.metadata.uid.ok_or_else(|| { + KubernetesDriverError::Message("workload NetworkPolicy has no UID".to_string()) + })?; + let fence_resource_version = fence.metadata.resource_version.ok_or_else(|| { + KubernetesDriverError::Message( + "workload NetworkPolicy has no resourceVersion".to_string(), + ) + })?; + let mut token_bytes = [0_u8; 32]; + rand::rng().fill_bytes(&mut token_bytes); + let (token_left, token_right) = token_bytes.split_at(16); + let bootstrap_token = format!( + "{:032x}{:032x}", + u128::from_be_bytes(token_left.try_into().expect("token half has fixed size")), + u128::from_be_bytes(token_right.try_into().expect("token half has fixed size")) + ); + let mut child_env = sandbox + .spec + .as_ref() + .and_then(|spec| spec.template.as_ref()) + .map_or_else(std::collections::HashMap::new, |template| { + template.environment.clone() + }); + if let Some(spec) = sandbox.spec.as_ref() { + child_env.extend(spec.environment.clone()); + } + let host_gateway_ip = self.config.host_gateway_ip.parse().ok(); + let tls = generate_boundary_tls_material().map_err(KubernetesDriverError::Message)?; + let proxy_ca = generate_proxy_ca_material().map_err(KubernetesDriverError::Message)?; + let provisioned = KubernetesProxyPodBoundarySpec { + boundary_id: sandbox.id.clone(), + bootstrap_token, + namespace_uid, + sandbox_resource_uid: cr_uid.to_string(), + control_deployment_uid: deployment_uid, + egress_policy_uid: fence_uid, + egress_policy_resource_version: fence_resource_version, + boundary_listener: std::net::SocketAddr::new( + if service_ip.is_ipv6() { + std::net::IpAddr::V6(std::net::Ipv6Addr::UNSPECIFIED) + } else { + std::net::IpAddr::V4(std::net::Ipv4Addr::UNSPECIFIED) + }, + self.config.proxy_pod.boundary_port, + ), + control_address: std::net::SocketAddr::new( + service_ip, + self.config.proxy_pod.boundary_port, + ), + boundary_server_name: BOUNDARY_TLS_SERVER_NAME.to_string(), + boundary_ca_certificate_pem: tls.ca_certificate_pem, + boundary_certificate_chain_path: PathBuf::from(BOUNDARY_CERTIFICATE_PATH), + boundary_private_key_path: PathBuf::from(BOUNDARY_PRIVATE_KEY_PATH), + host_gateway_ip, + agent_uid, + agent_gid, + trusted_runtime_root: PathBuf::from(format!( + "{SUPERVISOR_MOUNT_PATH}/openshell-runtime" + )), + child_env, } + .provision(); + let descriptor = provisioned + .topology + .descriptor(crate::isolation::BACKEND_NAME) + .map_err(|error| KubernetesDriverError::Message(error.to_string()))?; + let secret = bootstrap_secret( + namespace, + names, + &sandbox.id, + provisioned + .boundary_config + .encode() + .map_err(|error| KubernetesDriverError::Message(error.to_string()))?, + descriptor.payload, + tls.certificate_pem.into_bytes(), + tls.private_key_pem.into_bytes(), + proxy_ca.certificate_pem.into_bytes(), + proxy_ca.private_key_pem.into_bytes(), + dependent_owner, + ); + Api::::namespaced(self.client.clone(), namespace) + .create(&PostParams::default(), &secret) + .await + .map_err(KubernetesDriverError::from_kube)?; + + deployments + .patch( + &names.control_deployment, + &PatchParams::default(), + &Patch::Merge(&serde_json::json!({"spec": {"replicas": 1}})), + ) + .await + .map_err(KubernetesDriverError::from_kube)?; + // The Agent Sandbox controller may update status while companions are + // being created. Re-read the CR so the bootstrap release patch does + // not use the stale resourceVersion returned by CREATE. + let current_sandbox = sandbox_api + .api + .get(cr_name) + .await + .map_err(KubernetesDriverError::from_kube)?; + let resource_version = current_sandbox + .metadata + .resource_version + .as_deref() + .unwrap_or_default(); + let running_patch = + sandbox_operating_state_patch(&sandbox_api.resource.version, resource_version, true); + sandbox_api + .api + .patch( + cr_name, + &PatchParams::default(), + &Patch::Merge(&running_patch), + ) + .await + .map_err(KubernetesDriverError::from_kube)?; + spawn_proxy_pod_bootstrap_completion( + deployments.clone(), + sandbox_api.api.clone(), + names.control_deployment.clone(), + cr_name.to_string(), + current_sandbox.metadata.uid.clone(), + ); + // Return while the CR remains explicitly bootstrapping. The gateway + // can now commit the sandbox configuration required by a policy-less + // control process without deadlocking behind this driver call. Only + // boundary PID 1 is running at this point; the agent process cannot + // start until control attaches and confirms enforcement. Reconcile + // removes the marker after the control Deployment becomes Available. + Ok(()) } #[tracing::instrument( @@ -1661,6 +2098,16 @@ impl KubernetesComputeDriver { pod_is_gone, ); if stop_is_complete { + if object_topology(&object) == SupervisorTopology::ProxyPod + && let Err(error) = self + .scale_proxy_pod_control(sandbox_id, &namespace, 0) + .await + { + // The workload is already stopped. Treat a stranded + // control replica as resource drift and let periodic + // reconciliation retry rather than failing the stop. + warn!(sandbox_id, %error, "failed to scale stopped proxy-pod control Deployment to zero"); + } return Ok(()); } let now = tokio::time::Instant::now(); @@ -1733,7 +2180,7 @@ impl KubernetesComputeDriver { &namespace, ); let stop_timeout = kubernetes_sandbox_stop_timeout(&object); - let kube_name = object.metadata.name.ok_or_else(|| { + let kube_name = object.metadata.name.clone().ok_or_else(|| { KubernetesDriverError::Message("sandbox resource has no name".to_string()) })?; let pod_name = object @@ -1743,7 +2190,17 @@ impl KubernetesComputeDriver { .and_then(|annotations| annotations.get(SANDBOX_POD_NAME_ANNOTATION)) .cloned() .unwrap_or_else(|| kube_name.clone()); - let resource_version = object.metadata.resource_version.unwrap_or_default(); + let resource_version = object.metadata.resource_version.clone().unwrap_or_default(); + if running && object_topology(&object) == SupervisorTopology::ProxyPod { + if proxy_pod_bootstrap_in_progress(&object) { + return Err(KubernetesDriverError::Precondition( + "proxy-pod bootstrap has not completed; wait for reconciliation or recreate the sandbox" + .to_string(), + )); + } + self.scale_proxy_pod_control(sandbox_id, &namespace, 1) + .await?; + } let desired = sandbox_operating_state_patch( &agent_sandbox_api.resource.version, &resource_version, @@ -1781,6 +2238,24 @@ impl KubernetesComputeDriver { )) } + async fn scale_proxy_pod_control( + &self, + sandbox_id: &str, + namespace: &str, + replicas: i32, + ) -> Result<(), KubernetesDriverError> { + let names = ProxyPodNames::new(sandbox_id); + Api::::namespaced(self.client.clone(), namespace) + .patch( + &names.control_deployment, + &PatchParams::default(), + &Patch::Merge(&serde_json::json!({"spec": {"replicas": replicas}})), + ) + .await + .map_err(KubernetesDriverError::from_kube)?; + Ok(()) + } + #[tracing::instrument( name = "kubernetes.delete_sandbox", skip(self), @@ -1808,60 +2283,66 @@ impl KubernetesComputeDriver { .await?; let selector = self.sandbox_lookup_selector(sandbox_id); let lp = ListParams::default().labels(&selector); - let (kube_name, obj_namespace, _workspace, preconditions) = match tokio::time::timeout( - KUBE_API_TIMEOUT, - lookup_api.api.list(&lp), - ) - .await - { - Ok(Ok(list)) => { - if let Some(obj) = list.items.into_iter().next() { - match obj.metadata.name { - Some(name) => { - let ns = obj - .metadata - .namespace - .clone() - .unwrap_or_else(|| self.config.namespace.clone()); - let ws = obj - .metadata - .labels - .as_ref() - .and_then(|l| l.get(LABEL_SANDBOX_WORKSPACE).cloned()) - .unwrap_or_default(); - let pc = Preconditions { - uid: obj.metadata.uid, - resource_version: obj.metadata.resource_version, - }; - (name, ns, ws, pc) + let (kube_name, obj_namespace, _workspace, preconditions, pod_name, topology) = + match tokio::time::timeout(KUBE_API_TIMEOUT, lookup_api.api.list(&lp)).await { + Ok(Ok(list)) => { + if let Some(obj) = list.items.into_iter().next() { + match obj.metadata.name.clone() { + Some(name) => { + let ns = obj + .metadata + .namespace + .clone() + .unwrap_or_else(|| self.config.namespace.clone()); + let ws = obj + .metadata + .labels + .as_ref() + .and_then(|l| l.get(LABEL_SANDBOX_WORKSPACE).cloned()) + .unwrap_or_default(); + let topology = object_topology(&obj); + let pc = Preconditions { + uid: obj.metadata.uid, + resource_version: obj.metadata.resource_version, + }; + let pod_name = obj + .metadata + .annotations + .as_ref() + .and_then(|annotations| { + annotations.get(SANDBOX_POD_NAME_ANNOTATION) + }) + .cloned() + .unwrap_or_else(|| name.clone()); + (name, ns, ws, pc, pod_name, topology) + } + None => return Ok(false), } - None => return Ok(false), + } else { + debug!(sandbox_id = %sandbox_id, "Sandbox not found in Kubernetes (already deleted)"); + return Ok(false); } - } else { - debug!(sandbox_id = %sandbox_id, "Sandbox not found in Kubernetes (already deleted)"); - return Ok(false); } - } - Ok(Err(err)) => { - warn!( - sandbox_id = %sandbox_id, - error = %err, - "Failed to list sandbox for deletion from Kubernetes" - ); - return Err(err.to_string()); - } - Err(_elapsed) => { - warn!( - sandbox_id = %sandbox_id, - timeout_secs = KUBE_API_TIMEOUT.as_secs(), - "Timed out listing sandbox for deletion from Kubernetes" - ); - return Err(format!( - "timed out after {}s waiting for Kubernetes API", - KUBE_API_TIMEOUT.as_secs() - )); - } - }; + Ok(Err(err)) => { + warn!( + sandbox_id = %sandbox_id, + error = %err, + "Failed to list sandbox for deletion from Kubernetes" + ); + return Err(err.to_string()); + } + Err(_elapsed) => { + warn!( + sandbox_id = %sandbox_id, + timeout_secs = KUBE_API_TIMEOUT.as_secs(), + "Timed out listing sandbox for deletion from Kubernetes" + ); + return Err(format!( + "timed out after {}s waiting for Kubernetes API", + KUBE_API_TIMEOUT.as_secs() + )); + } + }; let delete_api = self .supported_agent_sandbox_api(self.client.clone(), &obj_namespace) @@ -1870,6 +2351,84 @@ impl KubernetesComputeDriver { match tokio::time::timeout(KUBE_API_TIMEOUT, delete_api.api.delete(&kube_name, &dp)).await { Ok(Ok(_response)) => { info!(sandbox_id = %sandbox_id, namespace = %obj_namespace, "Sandbox deleted from Kubernetes"); + if topology == SupervisorTopology::ProxyPod { + let pod_api = Api::::namespaced(self.client.clone(), &obj_namespace); + let deadline = tokio::time::Instant::now() + + DEFAULT_POD_TERMINATION_GRACE_PERIOD + + KUBE_API_TIMEOUT; + loop { + match kubernetes_sandbox_pod_is_gone(&pod_api, &pod_name, deadline).await { + Ok(true) => break, + Ok(false) if tokio::time::Instant::now() < deadline => { + tokio::time::sleep(STOP_INITIAL_POLL_INTERVAL).await; + } + Ok(false) | Err(_) => { + warn!( + sandbox_id, + "retaining proxy-pod workload fence because workload Pod deletion was not confirmed" + ); + return Ok(true); + } + } + } + // API acceptance of DELETE does not mean the CR is gone. + // Keep the unowned fence while finalizers can still leave + // the controller able to reconcile a workload Pod. + if !matches!( + tokio::time::timeout(KUBE_API_TIMEOUT, delete_api.api.get(&kube_name)) + .await, + Ok(Err(KubeError::Api(kube::core::ErrorResponse { + code: 404, + .. + }))) + ) { + warn!( + sandbox_id, + "retaining proxy-pod workload fence because Sandbox CR deletion was not confirmed" + ); + return Ok(true); + } + let quiesce_deadline = + tokio::time::Instant::now() + PROXY_POD_FENCE_QUIESCE_WINDOW; + while tokio::time::Instant::now() < quiesce_deadline { + tokio::time::sleep(PROXY_POD_FENCE_QUIESCE_INTERVAL).await; + let cr_is_gone = matches!( + tokio::time::timeout(KUBE_API_TIMEOUT, delete_api.api.get(&kube_name)) + .await, + Ok(Err(KubeError::Api(kube::core::ErrorResponse { + code: 404, + .. + }))) + ); + let pod_is_gone = kubernetes_sandbox_pod_is_gone( + &pod_api, + &pod_name, + tokio::time::Instant::now() + KUBE_API_TIMEOUT, + ) + .await + .unwrap_or(false); + if !cr_is_gone || !pod_is_gone { + warn!( + sandbox_id, + "retaining proxy-pod workload fence because deletion did not remain quiescent" + ); + return Ok(true); + } + } + let names = ProxyPodNames::new(sandbox_id); + let policies: Api = + Api::namespaced(self.client.clone(), &obj_namespace); + match policies + .delete(&names.workload_policy, &DeleteParams::default()) + .await + { + Ok(_) + | Err(KubeError::Api(kube::core::ErrorResponse { code: 404, .. })) => {} + Err(error) => { + warn!(sandbox_id, %error, "failed to delete proxy-pod workload fence; reconciliation will retry"); + } + } + } Ok(true) } Ok(Err(KubeError::Api(err))) if err.code == 404 || err.code == 409 => { @@ -1914,27 +2473,536 @@ impl KubernetesComputeDriver { } } - // Kept `async` to match the gRPC handler signature in `grpc.rs`, which awaits this method. - #[allow(clippy::unused_async)] - pub async fn watch_sandboxes(&self) -> Result { - if self.config.is_multi_namespace() { - self.watch_sandboxes_cluster_wide().await - } else { - self.watch_sandboxes_single_namespace().await - } - } - - async fn watch_sandboxes_single_namespace(&self) -> Result { - let namespace = self.config.namespace.clone(); - let agent_sandbox_api = self - .supported_agent_sandbox_api(self.watch_client.clone(), &self.config.namespace) - .await?; - let event_api: Api = Api::namespaced(self.watch_client.clone(), &namespace); - let watcher_config = watcher::Config::default().labels(&openshell_sandbox_label_selector()); - let mut sandbox_stream = recovering_watcher_stream( - watcher::watcher(agent_sandbox_api.api, watcher_config), - "sandbox-resource", - ) + /// Repair lifecycle drift and reap unowned workload fences while the + /// gateway's sandbox watch is alive. Bootstrap material is immutable and + /// intentionally not read by the gateway, so this pass only repairs state + /// that can be proven from the Sandbox CR and named companion objects. + async fn reconcile_proxy_pod_resources(&self) { + let lookup_api = match self + .supported_sandbox_api_for_lookup(self.client.clone()) + .await + { + Ok(api) => api, + Err(error) => { + warn!(%error, "skipping proxy-pod reconciliation: Sandbox API unavailable"); + return; + } + }; + let list = match tokio::time::timeout( + KUBE_API_TIMEOUT, + lookup_api + .api + .list(&ListParams::default().labels(&self.openshell_sandbox_selector())), + ) + .await + { + Ok(Ok(list)) => list, + Ok(Err(error)) => { + warn!(%error, "skipping proxy-pod reconciliation: Sandbox list failed"); + return; + } + Err(_) => { + warn!("skipping proxy-pod reconciliation: Sandbox list timed out"); + return; + } + }; + + let mut live_ids = HashSet::new(); + for object in list.items { + let Ok(sandbox_id) = sandbox_id_from_object(&object) else { + continue; + }; + live_ids.insert(sandbox_id.clone()); + if object_topology(&object) != SupervisorTopology::ProxyPod { + continue; + } + let namespace = object + .metadata + .namespace + .as_deref() + .unwrap_or(&self.config.namespace); + let cr_name = object.metadata.name.as_deref().unwrap_or_default(); + let names = ProxyPodNames::new(&sandbox_id); + let policies = Api::::namespaced(self.client.clone(), namespace); + let fence_was_missing = matches!( + tokio::time::timeout(KUBE_API_TIMEOUT, policies.get_opt(&names.workload_policy)) + .await, + Ok(Ok(None)) + ); + let sandbox = Sandbox { + id: sandbox_id.clone(), + ..Default::default() + }; + match self + .create_proxy_pod_fence(namespace, &names, &sandbox, cr_name) + .await + { + Ok(()) => {} + Err(KubernetesDriverError::Precondition(error)) => { + warn!(sandbox_id, %error, "proxy-pod workload fence is altered; suspending workload"); + self.suspend_proxy_pod_after_dependency_failure(&lookup_api, &object) + .await; + continue; + } + Err(error) => { + warn!(sandbox_id, %error, "could not verify proxy-pod workload fence; reconciliation will retry"); + continue; + } + } + if fence_was_missing { + // Recreating the deny-all fence closes direct egress first; + // suspend next because the immutable boundary descriptor names + // the prior policy UID/resourceVersion and can no longer be + // re-confirmed after a control restart. + warn!( + sandbox_id, + "recreated missing proxy-pod workload fence; suspending stale boundary" + ); + self.suspend_proxy_pod_after_dependency_failure(&lookup_api, &object) + .await; + continue; + } + if proxy_pod_bootstrap_in_progress(&object) { + if proxy_pod_control_availability(&self.client, namespace, &sandbox_id).await + == ProxyPodControlAvailability::Available + { + self.complete_proxy_pod_bootstrap(&lookup_api, &object) + .await; + } else { + self.reap_stale_proxy_pod_bootstrap(&lookup_api, &object) + .await; + } + continue; + } + let desired = desired_proxy_pod_control_replicas(&object); + match self + .reconcile_proxy_pod_control_replicas(&sandbox_id, namespace, desired) + .await + { + Ok(()) => {} + Err(KubernetesDriverError::Precondition(error)) => { + warn!(sandbox_id, %error, "proxy-pod companion is missing; suspending workload"); + self.suspend_proxy_pod_after_dependency_failure(&lookup_api, &object) + .await; + } + Err(error) => { + warn!(sandbox_id, %error, "failed to reconcile proxy-pod control Deployment"); + } + } + let availability = + proxy_pod_control_availability(&self.client, namespace, &sandbox_id).await; + self.publish_proxy_pod_readiness_transition(&lookup_api, &object, availability) + .await; + } + + self.reap_orphaned_proxy_pod_fences(&live_ids).await; + } + + async fn publish_proxy_pod_readiness_transition( + &self, + lookup_api: &AgentSandboxApi, + object: &DynamicObject, + availability: ProxyPodControlAvailability, + ) { + let state = match availability { + ProxyPodControlAvailability::Available => "ready", + ProxyPodControlAvailability::Unavailable => "unavailable", + ProxyPodControlAvailability::Unknown => return, + }; + if object + .metadata + .annotations + .as_ref() + .and_then(|annotations| annotations.get(ANNOTATION_PROXY_POD_READINESS)) + .is_some_and(|current| current == state) + { + return; + } + let (Some(name), Some(resource_version)) = ( + object.metadata.name.as_deref(), + object.metadata.resource_version.as_deref(), + ) else { + return; + }; + let namespace = object + .metadata + .namespace + .as_deref() + .unwrap_or(&self.config.namespace); + let api = + Self::agent_sandbox_api(self.client.clone(), &lookup_api.resource.version, namespace); + let patch = proxy_pod_readiness_transition_patch(resource_version, state); + match tokio::time::timeout( + KUBE_API_TIMEOUT, + api.api + .patch(name, &PatchParams::default(), &Patch::Merge(&patch)), + ) + .await + { + Ok(Ok(_)) => {} + Ok(Err(error)) => { + debug!(sandbox = name, %error, "proxy-pod readiness transition publication raced; reconciliation will retry"); + } + Err(_) => { + warn!( + sandbox = name, + "timed out publishing proxy-pod readiness transition" + ); + } + } + } + + async fn suspend_proxy_pod_after_dependency_failure( + &self, + lookup_api: &AgentSandboxApi, + object: &DynamicObject, + ) { + let Some(name) = object.metadata.name.as_deref() else { + return; + }; + let namespace = object + .metadata + .namespace + .as_deref() + .unwrap_or(&self.config.namespace); + let api = + Self::agent_sandbox_api(self.client.clone(), &lookup_api.resource.version, namespace); + let patch = sandbox_operating_state_patch( + &lookup_api.resource.version, + object + .metadata + .resource_version + .as_deref() + .unwrap_or_default(), + false, + ); + match tokio::time::timeout( + KUBE_API_TIMEOUT, + api.api + .patch(name, &PatchParams::default(), &Patch::Merge(&patch)), + ) + .await + { + Ok(Ok(_)) => {} + Ok(Err(error)) => { + warn!(sandbox = name, %error, "failed to suspend proxy-pod after fence failure"); + } + Err(error) => { + warn!(sandbox = name, %error, "timed out suspending proxy-pod after fence failure"); + } + } + } + + async fn reap_stale_proxy_pod_bootstrap( + &self, + lookup_api: &AgentSandboxApi, + object: &DynamicObject, + ) { + if !object_is_older_than(object, SystemTime::now(), PROXY_POD_BOOTSTRAP_GRACE) { + return; + } + let (Some(name), Some(uid), Some(resource_version)) = ( + object.metadata.name.as_deref(), + object.metadata.uid.clone(), + object.metadata.resource_version.clone(), + ) else { + return; + }; + let namespace = object + .metadata + .namespace + .as_deref() + .unwrap_or(&self.config.namespace); + let api = + Self::agent_sandbox_api(self.client.clone(), &lookup_api.resource.version, namespace); + let params = DeleteParams::default().preconditions(Preconditions { + uid: Some(uid), + resource_version: Some(resource_version), + }); + match tokio::time::timeout(KUBE_API_TIMEOUT, api.api.delete(name, ¶ms)).await { + Ok(Ok(_)) => warn!( + sandbox = name, + "rolled back stale fail-closed proxy-pod bootstrap" + ), + Ok(Err(KubeError::Api(error))) if error.code == 404 || error.code == 409 => {} + Ok(Err(error)) => { + warn!(sandbox = name, %error, "failed to roll back stale proxy-pod bootstrap"); + } + Err(_) => { + warn!( + sandbox = name, + "timed out rolling back stale proxy-pod bootstrap" + ); + } + } + } + + async fn complete_proxy_pod_bootstrap( + &self, + lookup_api: &AgentSandboxApi, + object: &DynamicObject, + ) { + let (Some(name), Some(resource_version)) = ( + object.metadata.name.as_deref(), + object.metadata.resource_version.as_deref(), + ) else { + return; + }; + let namespace = object + .metadata + .namespace + .as_deref() + .unwrap_or(&self.config.namespace); + let api = + Self::agent_sandbox_api(self.client.clone(), &lookup_api.resource.version, namespace); + let patch = proxy_pod_bootstrap_completion_patch(resource_version); + match tokio::time::timeout( + KUBE_API_TIMEOUT, + api.api + .patch(name, &PatchParams::default(), &Patch::Merge(&patch)), + ) + .await + { + Ok(Ok(_)) => {} + Ok(Err(error)) => { + debug!(sandbox = name, %error, "proxy-pod bootstrap completion raced; reconciliation will retry"); + } + Err(_) => warn!(sandbox = name, "timed out completing proxy-pod bootstrap"), + } + } + + async fn reconcile_proxy_pod_control_replicas( + &self, + sandbox_id: &str, + namespace: &str, + desired: i32, + ) -> Result<(), KubernetesDriverError> { + let names = ProxyPodNames::new(sandbox_id); + let services = Api::::namespaced(self.client.clone(), namespace); + let service_exists = + tokio::time::timeout(KUBE_API_TIMEOUT, services.get_opt(&names.boundary_service)) + .await + .map_err(|_| { + KubernetesDriverError::Message( + "timed out reading proxy-pod boundary Service".to_string(), + ) + })? + .map_err(KubernetesDriverError::from_kube)? + .is_some(); + if !service_exists { + return Err(KubernetesDriverError::Precondition(format!( + "proxy-pod boundary Service {} is missing and its allocated address cannot be safely reconstructed", + names.boundary_service + ))); + } + let deployments = Api::::namespaced(self.client.clone(), namespace); + let deployment = tokio::time::timeout( + KUBE_API_TIMEOUT, + deployments.get_opt(&names.control_deployment), + ) + .await + .map_err(|_| { + KubernetesDriverError::Message( + "timed out reading proxy-pod control Deployment".to_string(), + ) + })? + .map_err(KubernetesDriverError::from_kube)? + .ok_or_else(|| { + KubernetesDriverError::Precondition(format!( + "proxy-pod control Deployment {} is missing and cannot be safely reconstructed from the Sandbox CR", + names.control_deployment + )) + })?; + if deployment.spec.as_ref().and_then(|spec| spec.replicas) == Some(desired) { + return Ok(()); + } + self.scale_proxy_pod_control(sandbox_id, namespace, desired) + .await + } + + async fn reap_orphaned_proxy_pod_fences(&self, live_ids: &HashSet) { + let policies: Api = if self.config.is_multi_namespace() { + Api::all(self.client.clone()) + } else { + Api::namespaced(self.client.clone(), &self.config.namespace) + }; + let selector = format!( + "{LABEL_MANAGED_BY}={LABEL_MANAGED_BY_VALUE},{LABEL_GATEWAY_ID}={},openshell.ai/component=boundary-workload-fence", + self.config.gateway_id + ); + let list = match tokio::time::timeout( + KUBE_API_TIMEOUT, + policies.list(&ListParams::default().labels(&selector)), + ) + .await + { + Ok(Ok(list)) => list, + Ok(Err(error)) => { + warn!(%error, "failed to list proxy-pod workload fences for orphan cleanup"); + return; + } + Err(_) => { + warn!("timed out listing proxy-pod workload fences for orphan cleanup"); + return; + } + }; + for policy in list.items { + let sandbox_id = policy + .metadata + .labels + .as_ref() + .and_then(|labels| labels.get(LABEL_SANDBOX_ID)) + .cloned() + .unwrap_or_default(); + if sandbox_id.is_empty() || live_ids.contains(&sandbox_id) { + continue; + } + if !proxy_pod_fence_is_old_enough(&policy, SystemTime::now()) { + debug!( + sandbox_id, + "retaining young proxy-pod fence while Sandbox creation may be in flight" + ); + continue; + } + // Reconfirm against the API after the two list snapshots. A sandbox + // created between them must never lose the fence that precedes its CR. + if self.sandbox_exists(&sandbox_id).await != Ok(false) { + continue; + } + let Some(namespace) = policy.metadata.namespace.as_deref() else { + continue; + }; + let Some(pod_name) = policy + .metadata + .annotations + .as_ref() + .and_then(|annotations| annotations.get("openshell.ai/workload-pod-name")) + else { + warn!( + sandbox_id, + "retaining orphaned proxy-pod fence without workload Pod annotation" + ); + continue; + }; + let pods = Api::::namespaced(self.client.clone(), namespace); + if !matches!( + tokio::time::timeout(KUBE_API_TIMEOUT, pods.get_opt(pod_name)).await, + Ok(Ok(None)) + ) { + debug!( + sandbox_id, + pod_name, "retaining proxy-pod fence until workload Pod absence is confirmed" + ); + continue; + } + let quiesce_deadline = tokio::time::Instant::now() + PROXY_POD_FENCE_QUIESCE_WINDOW; + let mut quiescent = true; + while tokio::time::Instant::now() < quiesce_deadline { + tokio::time::sleep(PROXY_POD_FENCE_QUIESCE_INTERVAL).await; + if self.sandbox_exists(&sandbox_id).await != Ok(false) + || !matches!( + tokio::time::timeout(KUBE_API_TIMEOUT, pods.get_opt(pod_name)).await, + Ok(Ok(None)) + ) + { + quiescent = false; + break; + } + } + if !quiescent { + debug!( + sandbox_id, + pod_name, "retaining orphaned proxy-pod fence after quiescence recheck" + ); + continue; + } + let Some(name) = policy.metadata.name.as_deref() else { + continue; + }; + let scoped = Api::::namespaced(self.client.clone(), namespace); + match tokio::time::timeout( + KUBE_API_TIMEOUT, + scoped.delete(name, &DeleteParams::default()), + ) + .await + { + Ok(Ok(_) | Err(KubeError::Api(kube::core::ErrorResponse { code: 404, .. }))) => { + info!( + sandbox_id, + policy = name, + "reaped orphaned proxy-pod workload fence" + ); + } + Ok(Err(error)) => { + warn!(sandbox_id, policy = name, %error, "failed to reap orphaned proxy-pod workload fence"); + } + Err(_) => warn!( + sandbox_id, + policy = name, + "timed out reaping orphaned proxy-pod workload fence" + ), + } + } + } + + fn spawn_proxy_pod_periodic_reconcile( + &self, + tx: mpsc::Sender>, + ) { + if self.config.topology != SupervisorTopology::ProxyPod { + return; + } + let driver = self.clone(); + tokio::spawn(async move { + let mut interval = tokio::time::interval(PROXY_POD_RECONCILE_INTERVAL); + interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + interval.tick().await; + loop { + tokio::select! { + _ = interval.tick() => { + driver.reconcile_proxy_pod_resources().await; + if let Ok(sandboxes) = driver.list_sandboxes().await { + for sandbox in sandboxes { + if tx.send(Ok(WatchSandboxesEvent { + payload: Some(watch_sandboxes_event::Payload::Sandbox( + WatchSandboxesSandboxEvent { sandbox: Some(sandbox) }, + )), + })).await.is_err() { + return; + } + } + } + } + () = tx.closed() => return, + } + } + }); + } + + // Kept `async` to match the gRPC handler signature in `grpc.rs`, which awaits this method. + #[allow(clippy::unused_async)] + pub async fn watch_sandboxes(&self) -> Result { + if self.config.topology == SupervisorTopology::ProxyPod { + self.reconcile_proxy_pod_resources().await; + } + if self.config.is_multi_namespace() { + self.watch_sandboxes_cluster_wide().await + } else { + self.watch_sandboxes_single_namespace().await + } + } + + async fn watch_sandboxes_single_namespace(&self) -> Result { + let namespace = self.config.namespace.clone(); + let agent_sandbox_api = self + .supported_agent_sandbox_api(self.watch_client.clone(), &self.config.namespace) + .await?; + let event_api: Api = Api::namespaced(self.watch_client.clone(), &namespace); + let watcher_config = watcher::Config::default().labels(&openshell_sandbox_label_selector()); + let mut sandbox_stream = recovering_watcher_stream( + watcher::watcher(agent_sandbox_api.api, watcher_config), + "sandbox-resource", + ) .boxed(); let mut event_stream = recovering_watcher_stream( watcher::watcher(event_api, watcher::Config::default()), @@ -1942,6 +3010,8 @@ impl KubernetesComputeDriver { ) .boxed(); let (tx, rx) = mpsc::channel(256); + self.spawn_proxy_pod_periodic_reconcile(tx.clone()); + let readiness_client = self.watch_client.clone(); tokio::spawn(async move { let mut sandbox_name_to_id = std::collections::HashMap::::new(); @@ -1951,7 +3021,7 @@ impl KubernetesComputeDriver { tokio::select! { event = sandbox_stream.next() => match event { Some(Event::Applied(obj)) => { - if let Ok((kube_name, sandbox)) = sandbox_from_object(&namespace, obj) { + if let Ok((kube_name, sandbox)) = sandbox_from_object_with_proxy_pod_readiness(&readiness_client, &namespace, obj).await { update_indexes(&mut sandbox_name_to_id, &mut agent_pod_to_id, &kube_name, &sandbox); let event = WatchSandboxesEvent { payload: Some(watch_sandboxes_event::Payload::Sandbox( @@ -1980,7 +3050,7 @@ impl KubernetesComputeDriver { } Some(Event::Restarted(objs)) => { for obj in objs { - if let Ok((kube_name, sandbox)) = sandbox_from_object(&namespace, obj) { + if let Ok((kube_name, sandbox)) = sandbox_from_object_with_proxy_pod_readiness(&readiness_client, &namespace, obj).await { update_indexes(&mut sandbox_name_to_id, &mut agent_pod_to_id, &kube_name, &sandbox); let event = WatchSandboxesEvent { payload: Some(watch_sandboxes_event::Payload::Sandbox( @@ -2053,15 +3123,124 @@ impl KubernetesComputeDriver { Ok(cluster_wide_watch_stream( sandbox_stream, self.config.namespace.clone(), + self.watch_client.clone(), + self.clone(), )) } } -fn cluster_wide_watch_stream(mut sandbox_stream: S, default_namespace: String) -> WatchStream +fn proxy_pod_bootstrap_completion_patch(resource_version: &str) -> serde_json::Value { + serde_json::json!({ + "metadata": { + "resourceVersion": resource_version, + "annotations": { + ANNOTATION_PROXY_POD_BOOTSTRAPPING: serde_json::Value::Null, + ANNOTATION_PROXY_POD_READINESS: "ready", + } + } + }) +} + +fn spawn_proxy_pod_bootstrap_completion( + deployments: Api, + sandboxes: Api, + deployment_name: String, + sandbox_name: String, + expected_sandbox_uid: Option, +) { + tokio::spawn(async move { + let available = |deployment: Option<&Deployment>| { + deployment.is_some_and(|deployment| { + proxy_pod_control_availability_from_deployment(deployment) + == ProxyPodControlAvailability::Available + }) + }; + match tokio::time::timeout( + PROXY_POD_BOOTSTRAP_GRACE, + await_condition(deployments, &deployment_name, available), + ) + .await + { + Ok(Ok(Some(_))) => {} + Ok(Ok(None)) | Err(_) => return, + Ok(Err(error)) => { + debug!(%error, sandbox = sandbox_name, "proxy-pod bootstrap availability watch failed; reconciliation will retry"); + return; + } + } + + let Ok(object) = sandboxes.get(&sandbox_name).await else { + return; + }; + if object.metadata.uid != expected_sandbox_uid || !proxy_pod_bootstrap_in_progress(&object) + { + return; + } + let Some(resource_version) = object.metadata.resource_version.as_deref() else { + return; + }; + let patch = proxy_pod_bootstrap_completion_patch(resource_version); + if let Err(error) = sandboxes + .patch( + &sandbox_name, + &PatchParams::default(), + &Patch::Merge(&patch), + ) + .await + { + debug!(%error, sandbox = sandbox_name, "proxy-pod bootstrap completion raced; reconciliation will retry"); + } + }); +} + +fn proxy_pod_fence_is_old_enough(policy: &NetworkPolicy, now: SystemTime) -> bool { + metadata_is_older_than(&policy.metadata, now, PROXY_POD_ORPHAN_FENCE_MIN_AGE) +} + +fn proxy_pod_bootstrap_in_progress(object: &DynamicObject) -> bool { + object + .metadata + .annotations + .as_ref() + .and_then(|annotations| annotations.get(ANNOTATION_PROXY_POD_BOOTSTRAPPING)) + .is_some_and(|value| value == "true") +} + +fn proxy_pod_readiness_transition_patch(resource_version: &str, state: &str) -> serde_json::Value { + serde_json::json!({ + "metadata": { + "resourceVersion": resource_version, + "annotations": { ANNOTATION_PROXY_POD_READINESS: state }, + } + }) +} + +fn object_is_older_than(object: &DynamicObject, now: SystemTime, minimum_age: Duration) -> bool { + metadata_is_older_than(&object.metadata, now, minimum_age) +} + +fn metadata_is_older_than(metadata: &ObjectMeta, now: SystemTime, minimum_age: Duration) -> bool { + let Some(created) = metadata.creation_timestamp.as_ref() else { + return false; + }; + let Ok(created_seconds) = u64::try_from(created.0.timestamp()) else { + return false; + }; + now.duration_since(SystemTime::UNIX_EPOCH + Duration::from_secs(created_seconds)) + .is_ok_and(|age| age >= minimum_age) +} + +fn cluster_wide_watch_stream( + mut sandbox_stream: S, + default_namespace: String, + readiness_client: Client, + driver: KubernetesComputeDriver, +) -> WatchStream where S: Stream> + Send + Unpin + 'static, { let (tx, rx) = mpsc::channel(256); + driver.spawn_proxy_pod_periodic_reconcile(tx.clone()); tokio::spawn(async move { loop { @@ -2070,7 +3249,7 @@ where Some(Event::Applied(obj)) => { let ns = obj.metadata.namespace.clone() .unwrap_or_else(|| default_namespace.clone()); - if let Ok((_kube_name, sandbox)) = sandbox_from_object(&ns, obj) { + if let Ok((_kube_name, sandbox)) = sandbox_from_object_with_proxy_pod_readiness(&readiness_client, &ns, obj).await { let event = WatchSandboxesEvent { payload: Some(watch_sandboxes_event::Payload::Sandbox( WatchSandboxesSandboxEvent { sandbox: Some(sandbox) } @@ -2099,7 +3278,7 @@ where for obj in objs { let ns = obj.metadata.namespace.clone() .unwrap_or_else(|| default_namespace.clone()); - if let Ok((_kube_name, sandbox)) = sandbox_from_object(&ns, obj) { + if let Ok((_kube_name, sandbox)) = sandbox_from_object_with_proxy_pod_readiness(&readiness_client, &ns, obj).await { let event = WatchSandboxesEvent { payload: Some(watch_sandboxes_event::Payload::Sandbox( WatchSandboxesSandboxEvent { sandbox: Some(sandbox) } @@ -2437,6 +3616,38 @@ fn validate_pod_uid(pod: &Pod, expected_uid: &str) -> Result<(), tonic::Status> )) } +#[allow(clippy::result_large_err)] +fn validate_proxy_control_labels(pod: &Pod, sandbox_id: &str) -> Result<(), tonic::Status> { + validate_proxy_control_labels_from_metadata(&pod.metadata, sandbox_id) +} + +#[allow(clippy::result_large_err)] +fn validate_proxy_control_labels_from_metadata( + metadata: &ObjectMeta, + sandbox_id: &str, +) -> Result<(), tonic::Status> { + let labels = metadata.labels.as_ref().ok_or_else(|| { + tonic::Status::permission_denied("control workload has no proxy-pod labels") + })?; + let expected_pair = crate::proxy_pod::pair_label_value(sandbox_id); + let matches = labels + .get(BOUNDARY_ROLE_LABEL) + .is_some_and(|role| role == "control") + && labels + .get(BOUNDARY_PAIR_LABEL) + .is_some_and(|pair| pair == &expected_pair) + && labels + .get(LABEL_SANDBOX_ID) + .is_some_and(|actual| actual == sandbox_id); + if matches { + Ok(()) + } else { + Err(tonic::Status::permission_denied( + "control workload proxy-pod labels do not match the sandbox identity", + )) + } +} + #[allow(clippy::result_large_err)] fn sandbox_owner_reference(pod: &Pod) -> Result<&OwnerReference, tonic::Status> { let mut owners = pod @@ -2467,6 +3678,24 @@ fn sandbox_owner_reference(pod: &Pod) -> Result<&OwnerReference, tonic::Status> Ok(owner) } +fn controller_owner_reference(owners: &[OwnerReference]) -> Option<&OwnerReference> { + owners.iter().find(|owner| owner.controller == Some(true)) +} + +#[allow(clippy::result_large_err)] +fn validate_owner_uid( + actual_uid: &Option, + expected_owner: &OwnerReference, +) -> Result<(), tonic::Status> { + if actual_uid.as_deref() == Some(expected_owner.uid.as_str()) { + Ok(()) + } else { + Err(tonic::Status::permission_denied( + "Kubernetes controller owner UID mismatch", + )) + } +} + #[allow(clippy::result_large_err)] fn validate_sandbox_owner_identity( owner: &OwnerReference, @@ -2563,6 +3792,186 @@ fn sandbox_from_object(namespace: &str, obj: DynamicObject) -> Result<(String, S )) } +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum ProxyPodControlAvailability { + Available, + Unavailable, + Unknown, +} + +fn proxy_pod_control_availability_from_deployment( + deployment: &Deployment, +) -> ProxyPodControlAvailability { + if deployment + .status + .as_ref() + .and_then(|status| status.available_replicas) + .unwrap_or_default() + > 0 + { + ProxyPodControlAvailability::Available + } else { + ProxyPodControlAvailability::Unavailable + } +} + +async fn proxy_pod_control_availability( + client: &Client, + namespace: &str, + sandbox_id: &str, +) -> ProxyPodControlAvailability { + let names = ProxyPodNames::new(sandbox_id); + let deployments = Api::::namespaced(client.clone(), namespace); + let services = Api::::namespaced(client.clone(), namespace); + let policies = Api::::namespaced(client.clone(), namespace); + let deployment = Box::pin(tokio::time::timeout( + KUBE_API_TIMEOUT, + deployments.get_opt(&names.control_deployment), + )); + let service = Box::pin(tokio::time::timeout( + KUBE_API_TIMEOUT, + services.get_opt(&names.boundary_service), + )); + let fence = Box::pin(tokio::time::timeout( + KUBE_API_TIMEOUT, + policies.get_opt(&names.workload_policy), + )); + let (deployment, service, fence) = tokio::join!(deployment, service, fence); + let control = match deployment { + Ok(Ok(Some(deployment))) => proxy_pod_control_availability_from_deployment(&deployment), + Ok(Ok(None)) => ProxyPodControlAvailability::Unavailable, + Ok(Err(error)) => { + warn!(sandbox_id, %error, "could not determine proxy-pod control availability"); + ProxyPodControlAvailability::Unknown + } + Err(_) => { + warn!( + sandbox_id, + "timed out checking proxy-pod control availability" + ); + ProxyPodControlAvailability::Unknown + } + }; + let service = match service { + Ok(Ok(Some(_))) => ProxyPodControlAvailability::Available, + Ok(Ok(None)) => ProxyPodControlAvailability::Unavailable, + Ok(Err(error)) => { + warn!(sandbox_id, %error, "could not determine proxy-pod boundary Service availability"); + ProxyPodControlAvailability::Unknown + } + Err(_) => { + warn!( + sandbox_id, + "timed out checking proxy-pod boundary Service availability" + ); + ProxyPodControlAvailability::Unknown + } + }; + + // Deployment availability alone is not sufficient: deletion of the + // unowned fence would otherwise leave a live boundary with direct pod + // egress while the driver continued to publish Ready. + let fence = match fence { + Ok(Ok(Some(_))) => ProxyPodControlAvailability::Available, + Ok(Ok(None)) => ProxyPodControlAvailability::Unavailable, + Ok(Err(error)) => { + warn!(sandbox_id, %error, "could not determine proxy-pod workload fence availability"); + ProxyPodControlAvailability::Unknown + } + Err(_) => { + warn!( + sandbox_id, + "timed out checking proxy-pod workload fence availability" + ); + ProxyPodControlAvailability::Unknown + } + }; + if [control, service, fence].contains(&ProxyPodControlAvailability::Unavailable) { + ProxyPodControlAvailability::Unavailable + } else if [control, service, fence].contains(&ProxyPodControlAvailability::Unknown) { + ProxyPodControlAvailability::Unknown + } else { + ProxyPodControlAvailability::Available + } +} + +async fn sandbox_from_object_with_proxy_pod_readiness( + client: &Client, + namespace: &str, + obj: DynamicObject, +) -> Result<(String, Sandbox), String> { + let topology = object_topology(&obj); + let bootstrapping = proxy_pod_bootstrap_in_progress(&obj); + let sandbox_id = sandbox_id_from_object(&obj).unwrap_or_default(); + let object_namespace = obj + .metadata + .namespace + .clone() + .unwrap_or_else(|| namespace.to_string()); + let (name, mut sandbox) = sandbox_from_object(namespace, obj)?; + if topology == SupervisorTopology::ProxyPod && bootstrapping { + mark_proxy_pod_bootstrapping(&mut sandbox); + } + if topology == SupervisorTopology::ProxyPod + && !sandbox_id.is_empty() + && proxy_pod_control_availability(client, &object_namespace, &sandbox_id).await + != ProxyPodControlAvailability::Available + { + mark_proxy_pod_control_unavailable(&mut sandbox); + } + Ok((name, sandbox)) +} + +fn mark_proxy_pod_bootstrapping(sandbox: &mut Sandbox) { + if let Some(status) = sandbox.status.as_mut() { + status + .conditions + .retain(|condition| condition.r#type != SANDBOX_SUSPENDED_CONDITION); + } + mark_proxy_pod_control_unavailable(sandbox); +} + +fn mark_proxy_pod_control_unavailable(sandbox: &mut Sandbox) { + const REASON: &str = "DependenciesNotReady"; + const MESSAGE: &str = "proxy-pod enforcement dependencies are not ready"; + let Some(status) = sandbox.status.as_mut() else { + return; + }; + if let Some(ready) = status + .conditions + .iter_mut() + .find(|condition| condition.r#type == "Ready") + { + ready.status = "False".to_string(); + ready.reason = REASON.to_string(); + ready.message = MESSAGE.to_string(); + } else { + status.conditions.push(SandboxCondition { + r#type: "Ready".to_string(), + status: "False".to_string(), + reason: REASON.to_string(), + message: MESSAGE.to_string(), + last_transition_time: String::new(), + }); + } +} + +fn desired_proxy_pod_control_replicas(obj: &DynamicObject) -> i32 { + if let Some(mode) = obj + .data + .get("spec") + .and_then(|spec| spec.get("operatingMode")) + .and_then(serde_json::Value::as_str) + { + return i32::from(!mode.eq_ignore_ascii_case("Suspended")); + } + obj.data + .get("spec") + .and_then(|spec| spec.get("replicas")) + .and_then(serde_json::Value::as_i64) + .map_or(1, |replicas| i32::from(replicas > 0)) +} + fn update_indexes( sandbox_name_to_id: &mut std::collections::HashMap, agent_pod_to_id: &mut std::collections::HashMap, @@ -2865,7 +4274,6 @@ fn apply_supervisor_sideload_with_params( params.supervisor_image_pull_policy, params.supervisor_sideload_method, ); - // Find the agent container and add volume mount + command override let Some(containers) = spec.get_mut("containers").and_then(|v| v.as_array_mut()) else { return; @@ -3393,6 +4801,189 @@ fn apply_supervisor_sidecar_topology( )); } +/// Render the workload pod half of the RFC 0012 proxy-pod topology. +/// +/// The pod receives no gateway credential or endpoint. Its root boundary owns +/// the workload process and the nested network namespace; only the paired +/// control Deployment can reach its boundary listener through `NetworkPolicy`. +fn apply_supervisor_proxy_pod_boundary( + pod_template: &mut serde_json::Value, + params: &SandboxPodParams<'_>, +) { + let names = ProxyPodNames::new(params.sandbox_id); + let metadata = pod_template + .as_object_mut() + .expect("pod template must be an object") + .entry("metadata") + .or_insert_with(|| serde_json::json!({})); + let labels = metadata + .as_object_mut() + .expect("pod metadata must be an object") + .entry("labels") + .or_insert_with(|| serde_json::json!({})); + let labels = labels + .as_object_mut() + .expect("pod labels must be an object"); + labels.insert( + BOUNDARY_PAIR_LABEL.to_string(), + serde_json::json!(crate::proxy_pod::pair_label_value(params.sandbox_id)), + ); + labels.insert( + BOUNDARY_ROLE_LABEL.to_string(), + serde_json::json!("workload"), + ); + + let Some(spec) = pod_template + .get_mut("spec") + .and_then(serde_json::Value::as_object_mut) + else { + return; + }; + apply_supervisor_binary_source( + spec, + params.supervisor_image, + params.supervisor_image_pull_policy, + params.supervisor_sideload_method, + ); + if params.supervisor_sideload_method == SupervisorSideloadMethod::InitContainer { + let init_containers = spec + .entry("initContainers") + .or_insert_with(|| serde_json::json!([])) + .as_array_mut() + .expect("pod init containers must be an array"); + let mut runtime_installer = serde_json::json!({ + "name": "openshell-boundary-runtime-install", + "image": params.supervisor_image, + "command": ["/bin/sh", "-ec", "cp -a /openshell-runtime /opt/openshell/bin/openshell-runtime"], + "securityContext": {"runAsUser": 0, "allowPrivilegeEscalation": false, "capabilities": {"drop": ["ALL"]}}, + "volumeMounts": [{"name": SUPERVISOR_VOLUME_NAME, "mountPath": SUPERVISOR_MOUNT_PATH, "readOnly": false}] + }); + if !params.supervisor_image_pull_policy.is_empty() { + runtime_installer["imagePullPolicy"] = + serde_json::json!(params.supervisor_image_pull_policy); + } + init_containers.push(runtime_installer); + } + let volumes = spec + .entry("volumes") + .or_insert_with(|| serde_json::json!([])) + .as_array_mut() + .expect("pod volumes must be an array"); + volumes.retain(|volume| { + !matches!( + volume.get("name").and_then(serde_json::Value::as_str), + Some("openshell-sa-token" | "openshell-client-tls" | "spiffe-workload-api") + ) + }); + volumes.push(serde_json::json!({ + "name": "openshell-boundary-bootstrap", + "secret": {"secretName": names.bootstrap_secret, "defaultMode": 0o400} + })); + + let containers = spec + .get_mut("containers") + .and_then(serde_json::Value::as_array_mut) + .expect("pod containers must be an array"); + let index = containers + .iter() + .position(|container| { + container.get("name").and_then(serde_json::Value::as_str) == Some("agent") + }) + .unwrap_or(0); + let container = containers[index] + .as_object_mut() + .expect("agent container must be an object"); + container.insert( + "command".to_string(), + serde_json::json!([ + format!("{SUPERVISOR_MOUNT_PATH}/openshell-sandbox"), + "--mode=boundary", + "--boundary-config", + BOUNDARY_CONFIG_PATH, + ]), + ); + container.insert( + "securityContext".to_string(), + serde_json::json!({ + "runAsUser": 0, + "allowPrivilegeEscalation": false, + "capabilities": { + "drop": ["ALL"], + "add": [ + "SYS_ADMIN", + "NET_ADMIN", + "SYS_PTRACE", + "SYSLOG", + "CHOWN", + "SETUID", + "SETGID", + "SETPCAP", + "DAC_READ_SEARCH" + ] + } + }), + ); + let mounts = container + .entry("volumeMounts") + .or_insert_with(|| serde_json::json!([])) + .as_array_mut() + .expect("agent volume mounts must be an array"); + mounts.retain(|mount| { + !matches!( + mount.get("name").and_then(serde_json::Value::as_str), + Some("openshell-sa-token" | "openshell-client-tls" | "spiffe-workload-api") + ) + }); + if !mounts.iter().any(|mount| { + mount.get("name").and_then(serde_json::Value::as_str) == Some(SUPERVISOR_VOLUME_NAME) + }) { + mounts.push(supervisor_volume_mount()); + } + mounts.push(serde_json::json!({ + "name": "openshell-boundary-bootstrap", + "mountPath": BOUNDARY_CONFIG_PATH, + "subPath": BOUNDARY_CONFIG_KEY, + "readOnly": true + })); + mounts.push(serde_json::json!({ + "name": "openshell-boundary-bootstrap", + "mountPath": BOUNDARY_CERTIFICATE_PATH, + "subPath": BOUNDARY_CERTIFICATE_KEY, + "readOnly": true + })); + mounts.push(serde_json::json!({ + "name": "openshell-boundary-bootstrap", + "mountPath": BOUNDARY_PRIVATE_KEY_PATH, + "subPath": BOUNDARY_PRIVATE_KEY, + "readOnly": true + })); + let env = container + .entry("env") + .or_insert_with(|| serde_json::json!([])) + .as_array_mut() + .expect("agent environment must be an array"); + for key in [ + openshell_core::sandbox_env::ENDPOINT, + openshell_core::sandbox_env::GATEWAY_TLS_SERVER_NAME, + openshell_core::sandbox_env::TLS_CA, + openshell_core::sandbox_env::TLS_CERT, + openshell_core::sandbox_env::TLS_KEY, + openshell_core::sandbox_env::SANDBOX_TOKEN, + openshell_core::sandbox_env::SANDBOX_TOKEN_FILE, + openshell_core::sandbox_env::K8S_SA_TOKEN_FILE, + openshell_core::sandbox_env::PROVIDER_SPIFFE_WORKLOAD_API_SOCKET, + openshell_core::sandbox_env::MAIN_PROCESS_SPEC, + ] { + remove_env(env, key); + } + upsert_env( + env, + openshell_core::sandbox_env::SUPERVISOR_TOPOLOGY, + "proxy-pod", + ); + apply_resolved_identity_env(env, params.sandbox_uid, params.sandbox_gid); +} + /// Apply workspace persistence transforms to an already-built pod template. /// /// This injects: @@ -3413,7 +5004,8 @@ fn apply_workspace_persistence( pod_template: &mut serde_json::Value, image: &str, image_pull_policy: &str, - sandbox_gid: u32, + sandbox_gid: Option, + workspace_owner: Option<(u32, u32)>, ) { let Some(spec) = pod_template.get_mut("spec").and_then(|v| v.as_object_mut()) else { return; @@ -3421,11 +5013,13 @@ fn apply_workspace_persistence( // fsGroup is a pod-level field — it instructs kubelet to chown mounted // volumes to this GID. It is invalid at the container securityContext level. - let pod_sc = spec - .entry("securityContext") - .or_insert_with(|| serde_json::json!({})); - if let Some(pod_sc_obj) = pod_sc.as_object_mut() { - pod_sc_obj.insert("fsGroup".to_string(), serde_json::json!(sandbox_gid)); + if let Some(sandbox_gid) = sandbox_gid { + let pod_sc = spec + .entry("securityContext") + .or_insert_with(|| serde_json::json!({})); + if let Some(pod_sc_obj) = pod_sc.as_object_mut() { + pod_sc_obj.insert("fsGroup".to_string(), serde_json::json!(sandbox_gid)); + } } // 1. Add workspace volume mount to the agent container @@ -3476,6 +5070,11 @@ fn apply_workspace_persistence( // The inner `[ -d ... ]` guard handles custom images that don't have // a /sandbox directory — the copy is skipped but the sentinel is // still written so subsequent starts are instant. + let ownership_cmd = workspace_owner.map_or_else(String::new, |(uid, gid)| { + format!( + " && chown -R {uid}:{gid} {WORKSPACE_INIT_MOUNT_PATH} && chmod u+rwx {WORKSPACE_INIT_MOUNT_PATH}" + ) + }); let copy_cmd = format!( "if [ ! -f {WORKSPACE_INIT_MOUNT_PATH}/{WORKSPACE_SENTINEL} ]; then \ if [ -d {WORKSPACE_MOUNT_PATH} ]; then \ @@ -3487,7 +5086,7 @@ fn apply_workspace_persistence( fi; \ fi && \ touch {WORKSPACE_INIT_MOUNT_PATH}/{WORKSPACE_SENTINEL}; \ - fi" + fi{ownership_cmd}" ); let mut init_spec = serde_json::json!({ @@ -4013,7 +5612,7 @@ fn sandbox_template_to_k8s_with_validated_config( let mut volumes: Vec = Vec::new(); if !params.client_tls_secret_name.is_empty() { let client_tls_default_mode = match params.topology { - SupervisorTopology::Combined => 0o400, + SupervisorTopology::Combined | SupervisorTopology::ProxyPod => 0o400, SupervisorTopology::Sidecar => 0o440, }; volumes.push(serde_json::json!({ @@ -4035,7 +5634,7 @@ fn sandbox_template_to_k8s_with_validated_config( // network supervision. Sidecar mode uses the pod fsGroup already // required for its non-root network supervisor. let default_mode = match params.topology { - SupervisorTopology::Combined => 0o400, + SupervisorTopology::Combined | SupervisorTopology::ProxyPod => 0o400, SupervisorTopology::Sidecar => 0o440, }; volumes.push(serde_json::json!({ @@ -4065,7 +5664,7 @@ fn sandbox_template_to_k8s_with_validated_config( // JWT via `IssueSandboxToken` once at startup. In sidecar topology both // supervisor containers run with the sandbox GID and need group-read access. let sa_token_default_mode = match params.topology { - SupervisorTopology::Combined => 0o400, + SupervisorTopology::Combined | SupervisorTopology::ProxyPod => 0o400, SupervisorTopology::Sidecar => 0o440, }; volumes.push(serde_json::json!({ @@ -4120,17 +5719,23 @@ fn sandbox_template_to_k8s_with_validated_config( params, ); } + SupervisorTopology::ProxyPod => { + apply_supervisor_proxy_pod_boundary(&mut result, params); + } } // Inject workspace persistence (init container + PVC volume mount) so // that /sandbox data survives pod rescheduling. Skipped when the user // provides custom storage through driver_config. if inject_workspace { + let proxy_workspace_owner = (params.topology == SupervisorTopology::ProxyPod) + .then_some((params.sandbox_uid, params.sandbox_gid)); apply_workspace_persistence( &mut result, image, params.image_pull_policy, - params.sandbox_gid, + (params.topology != SupervisorTopology::ProxyPod).then_some(params.sandbox_gid), + proxy_workspace_owner, ); } @@ -4549,6 +6154,93 @@ fn status_from_object(obj: &DynamicObject) -> Option { }) } +fn object_topology(obj: &DynamicObject) -> SupervisorTopology { + match obj + .metadata + .annotations + .as_ref() + .and_then(|annotations| annotations.get(ANNOTATION_SUPERVISOR_TOPOLOGY)) + .map(String::as_str) + { + Some("sidecar") => SupervisorTopology::Sidecar, + Some("proxy-pod") => SupervisorTopology::ProxyPod, + _ => SupervisorTopology::Combined, + } +} + +async fn create_or_validate_proxy_pod_fence( + policies: &Api, + expected: &NetworkPolicy, +) -> Result<(), KubernetesDriverError> { + match tokio::time::timeout( + KUBE_API_TIMEOUT, + policies.create(&PostParams::default(), expected), + ) + .await + { + Ok(Ok(_)) => Ok(()), + Ok(Err(KubeError::Api(error))) if error.code == 409 => { + let name = expected.metadata.name.as_deref().unwrap_or_default(); + let existing = tokio::time::timeout(KUBE_API_TIMEOUT, policies.get(name)) + .await + .map_err(|_| { + KubernetesDriverError::Message( + "timed out validating existing proxy-pod workload fence".to_string(), + ) + })? + .map_err(KubernetesDriverError::from_kube)?; + if proxy_pod_fence_matches(&existing, expected) { + Ok(()) + } else { + Err(KubernetesDriverError::Precondition(format!( + "proxy-pod workload fence {name} exists but does not match the intended enforcement" + ))) + } + } + Ok(Err(error)) => Err(KubernetesDriverError::from_kube(error)), + Err(_) => Err(KubernetesDriverError::Message( + "timed out creating proxy-pod workload fence".to_string(), + )), + } +} + +fn proxy_pod_fence_matches(existing: &NetworkPolicy, expected: &NetworkPolicy) -> bool { + fn normalized_spec(mut spec: Option) -> Option { + if let Some(spec) = spec.as_mut() { + // The Kubernetes API server omits explicitly empty rule arrays when it + // persists a NetworkPolicy. For a policy type named in `policyTypes`, + // an omitted rule array and an empty rule array both deny all traffic. + if spec.egress.as_ref().is_some_and(Vec::is_empty) { + spec.egress = None; + } + if spec.ingress.as_ref().is_some_and(Vec::is_empty) { + spec.ingress = None; + } + } + spec + } + + fn contains_required_metadata( + actual: &Option>, + required: &Option>, + ) -> bool { + required.as_ref().is_none_or(|required| { + actual.as_ref().is_some_and(|actual| { + required + .iter() + .all(|(key, value)| actual.get(key) == Some(value)) + }) + }) + } + + normalized_spec(existing.spec.clone()) == normalized_spec(expected.spec.clone()) + && contains_required_metadata(&existing.metadata.labels, &expected.metadata.labels) + && contains_required_metadata( + &existing.metadata.annotations, + &expected.metadata.annotations, + ) +} + fn kubernetes_sandbox_has_stopped_condition(obj: &DynamicObject) -> bool { obj.data .get("status") @@ -4655,15 +6347,39 @@ fn sandbox_operating_state_patch( running: bool, ) -> serde_json::Value { if api_version == SANDBOX_VERSION_V1BETA1 { - serde_json::json!({ - "metadata": {"resourceVersion": resource_version}, - "spec": {"operatingMode": if running { "Running" } else { "Suspended" }} - }) + if running { + serde_json::json!({ + "metadata": {"resourceVersion": resource_version}, + "spec": {"operatingMode": "Running"} + }) + } else { + serde_json::json!({ + "metadata": { + "resourceVersion": resource_version, + "annotations": { + ANNOTATION_PROXY_POD_BOOTSTRAPPING: serde_json::Value::Null, + }, + }, + "spec": {"operatingMode": "Suspended"} + }) + } } else { - serde_json::json!({ - "metadata": {"resourceVersion": resource_version}, - "spec": {"replicas": i32::from(running)} - }) + if running { + serde_json::json!({ + "metadata": {"resourceVersion": resource_version}, + "spec": {"replicas": 1} + }) + } else { + serde_json::json!({ + "metadata": { + "resourceVersion": resource_version, + "annotations": { + ANNOTATION_PROXY_POD_BOOTSTRAPPING: serde_json::Value::Null, + }, + }, + "spec": {"replicas": 0} + }) + } } } @@ -5104,7 +6820,13 @@ mod tests { ]) .chain(futures::stream::pending()); let sandbox_stream = recovering_watcher_stream(source, "sandbox-resource").boxed(); - let mut outward = cluster_wide_watch_stream(sandbox_stream, "default".to_string()); + let driver = KubernetesComputeDriver::new_for_test(KubernetesComputeConfig::default()); + let mut outward = cluster_wide_watch_stream( + sandbox_stream, + "default".to_string(), + driver.watch_client.clone(), + driver, + ); let event = outward .next() @@ -5304,6 +7026,123 @@ mod tests { assert_eq!(error.code(), tonic::Code::PermissionDenied); } + #[test] + fn proxy_control_identity_requires_exact_pair_and_role_labels() { + let sandbox_id = "sandbox-id-a"; + let mut pod = Pod { + metadata: ObjectMeta { + labels: Some(BTreeMap::from([ + (LABEL_SANDBOX_ID.to_string(), sandbox_id.to_string()), + ( + BOUNDARY_PAIR_LABEL.to_string(), + crate::proxy_pod::pair_label_value(sandbox_id), + ), + (BOUNDARY_ROLE_LABEL.to_string(), "control".to_string()), + ])), + ..Default::default() + }, + ..Default::default() + }; + validate_proxy_control_labels(&pod, sandbox_id).unwrap(); + + pod.metadata + .labels + .as_mut() + .unwrap() + .insert(BOUNDARY_ROLE_LABEL.to_string(), "workload".to_string()); + assert_eq!( + validate_proxy_control_labels(&pod, sandbox_id) + .unwrap_err() + .code(), + tonic::Code::PermissionDenied + ); + } + + #[test] + fn orphan_fence_reaping_requires_a_creation_age_floor() { + let policy: NetworkPolicy = serde_json::from_value(serde_json::json!({ + "apiVersion": "networking.k8s.io/v1", + "kind": "NetworkPolicy", + "metadata": {"creationTimestamp": "2026-01-01T00:00:00Z"}, + "spec": {"podSelector": {}} + })) + .unwrap(); + let created = SystemTime::UNIX_EPOCH + Duration::from_secs(1_767_225_600); + assert!(!proxy_pod_fence_is_old_enough( + &policy, + created + PROXY_POD_ORPHAN_FENCE_MIN_AGE - Duration::from_secs(1) + )); + assert!(proxy_pod_fence_is_old_enough( + &policy, + created + PROXY_POD_ORPHAN_FENCE_MIN_AGE + )); + assert!(!proxy_pod_fence_is_old_enough( + &NetworkPolicy::default(), + created + PROXY_POD_ORPHAN_FENCE_MIN_AGE + )); + } + + #[test] + fn proxy_pod_fence_validation_accepts_api_normalization_and_injected_metadata() { + let names = ProxyPodNames::new("sandbox-id-a"); + let mut expected = + workload_fence("namespace-a", &names, "sandbox-id-a", 5000).workload_policy; + expected.metadata.labels = Some(BTreeMap::from([( + LABEL_MANAGED_BY.to_string(), + LABEL_MANAGED_BY_VALUE.to_string(), + )])); + expected.metadata.annotations = Some(BTreeMap::from([( + "openshell.ai/workload-pod-name".to_string(), + "sandbox-a".to_string(), + )])); + + let mut persisted = expected.clone(); + persisted.spec.as_mut().unwrap().egress = None; + persisted + .metadata + .labels + .as_mut() + .unwrap() + .insert("admission.example/injected".to_string(), "true".to_string()); + assert!(proxy_pod_fence_matches(&persisted, &expected)); + + persisted.spec.as_mut().unwrap().policy_types = Some(vec!["Ingress".to_string()]); + assert!(!proxy_pod_fence_matches(&persisted, &expected)); + } + + #[test] + fn proxy_pod_bootstrap_marker_and_age_gate_rollback() { + let mut object: DynamicObject = serde_json::from_value(serde_json::json!({ + "apiVersion": "agents.x-k8s.io/v1beta1", + "kind": "Sandbox", + "metadata": { + "name": "sandbox-a", + "creationTimestamp": "2026-01-01T00:00:00Z", + "annotations": {ANNOTATION_PROXY_POD_BOOTSTRAPPING: "true"} + } + })) + .unwrap(); + assert!(proxy_pod_bootstrap_in_progress(&object)); + let created = SystemTime::UNIX_EPOCH + Duration::from_secs(1_767_225_600); + assert!(!object_is_older_than( + &object, + created + PROXY_POD_BOOTSTRAP_GRACE - Duration::from_secs(1), + PROXY_POD_BOOTSTRAP_GRACE + )); + assert!(object_is_older_than( + &object, + created + PROXY_POD_BOOTSTRAP_GRACE, + PROXY_POD_BOOTSTRAP_GRACE + )); + object + .metadata + .annotations + .as_mut() + .unwrap() + .remove(ANNOTATION_PROXY_POD_BOOTSTRAPPING); + assert!(!proxy_pod_bootstrap_in_progress(&object)); + } + #[test] fn sandbox_owner_identity_requires_matching_uid_and_sandbox_id() { let owner = sandbox_owner_for_test("sandbox-a", "sandbox-uid-a"); @@ -5335,11 +7174,16 @@ mod tests { fn lifecycle_patch_uses_version_specific_operating_state() { let beta_stop = sandbox_operating_state_patch(SANDBOX_VERSION_V1BETA1, "42", false); assert_eq!(beta_stop["metadata"]["resourceVersion"], "42"); + assert_eq!( + beta_stop["metadata"]["annotations"][ANNOTATION_PROXY_POD_BOOTSTRAPPING], + serde_json::Value::Null + ); assert_eq!(beta_stop["spec"]["operatingMode"], "Suspended"); assert!(beta_stop["spec"].get("replicas").is_none()); let alpha_start = sandbox_operating_state_patch(SANDBOX_VERSION_V1ALPHA1, "43", true); assert_eq!(alpha_start["metadata"]["resourceVersion"], "43"); + assert!(alpha_start["metadata"].get("annotations").is_none()); assert_eq!(alpha_start["spec"]["replicas"], 1); assert!(alpha_start["spec"].get("operatingMode").is_none()); } @@ -6726,6 +8570,98 @@ mod tests { })); } + #[test] + fn proxy_pod_topology_renders_credential_free_boundary_workload() { + let params = SandboxPodParams { + topology: SupervisorTopology::ProxyPod, + supervisor_sideload_method: SupervisorSideloadMethod::InitContainer, + supervisor_image: "supervisor-image:latest", + grpc_endpoint: "https://openshell.openshell.svc:8080", + client_tls_secret_name: "openshell-client-tls", + sandbox_id: "sandbox-123", + sandbox_uid: 1500, + sandbox_gid: 1500, + ..SandboxPodParams::default() + }; + let pod_template = sandbox_template_to_k8s( + &SandboxTemplate::default(), + false, + &std::collections::HashMap::new(), + true, + ¶ms, + ); + let agent = &pod_template["spec"]["containers"][0]; + + assert_eq!( + agent["command"], + serde_json::json!([ + format!("{SUPERVISOR_MOUNT_PATH}/openshell-sandbox"), + "--mode=boundary", + "--boundary-config", + BOUNDARY_CONFIG_PATH + ]) + ); + assert_eq!(agent["securityContext"]["runAsUser"], 0); + assert!(pod_template["spec"]["securityContext"]["fsGroup"].is_null()); + let workspace_init = pod_template["spec"]["initContainers"] + .as_array() + .unwrap() + .iter() + .find(|container| container["name"] == WORKSPACE_INIT_CONTAINER_NAME) + .unwrap(); + assert!( + workspace_init["command"][2] + .as_str() + .unwrap() + .contains("chown -R 1500:1500") + ); + let capabilities = agent["securityContext"]["capabilities"]["add"] + .as_array() + .expect("boundary capabilities"); + for required in [ + "SYS_ADMIN", + "NET_ADMIN", + "CHOWN", + "SETUID", + "SETGID", + "SETPCAP", + ] { + assert!( + capabilities.iter().any(|capability| capability == required), + "boundary must retain {required}" + ); + } + assert_eq!( + rendered_env(agent, openshell_core::sandbox_env::ENDPOINT), + None + ); + assert_eq!( + rendered_env(agent, openshell_core::sandbox_env::K8S_SA_TOKEN_FILE), + None + ); + assert_eq!( + pod_template["metadata"]["labels"][BOUNDARY_ROLE_LABEL], + "workload" + ); + + let mounts = agent["volumeMounts"].as_array().unwrap(); + assert!( + mounts + .iter() + .any(|mount| mount["mountPath"] == BOUNDARY_CONFIG_PATH) + ); + assert!( + !mounts + .iter() + .any(|mount| mount["name"] == CLIENT_TLS_VOLUME_NAME) + ); + assert!( + !mounts + .iter() + .any(|mount| mount["name"] == SERVICE_ACCOUNT_TOKEN_VOLUME_NAME) + ); + } + #[test] fn sidecar_topology_can_relax_process_binary_aware_network_policy() { let params = SandboxPodParams { @@ -7353,7 +9289,8 @@ mod tests { &mut pod_template, "openshell/sandbox:latest", "IfNotPresent", - 1000, // sandbox_gid + Some(1000), // sandbox_gid + None, ); // Init container @@ -7412,7 +9349,8 @@ mod tests { &mut pod_template, "my-custom-image:v2", "IfNotPresent", - 1000, + Some(1000), + None, ); let init_image = pod_template["spec"]["initContainers"][0]["image"] @@ -7435,7 +9373,7 @@ mod tests { } }); - apply_workspace_persistence(&mut pod_template, "img:latest", "Always", 1000); + apply_workspace_persistence(&mut pod_template, "img:latest", "Always", Some(1000), None); let cmd = pod_template["spec"]["initContainers"][0]["command"] .as_array() @@ -8716,4 +10654,143 @@ mod tests { namespace_watcher_retry_delay(3, 2) ); } + + #[test] + fn proxy_pod_control_availability_requires_an_available_replica() { + let mut deployment = Deployment::default(); + assert_eq!( + proxy_pod_control_availability_from_deployment(&deployment), + ProxyPodControlAvailability::Unavailable + ); + deployment.status = Some(k8s_openapi::api::apps::v1::DeploymentStatus { + available_replicas: Some(1), + ..Default::default() + }); + assert_eq!( + proxy_pod_control_availability_from_deployment(&deployment), + ProxyPodControlAvailability::Available + ); + } + + #[test] + fn proxy_pod_readiness_transitions_bump_the_watched_cr() { + let unavailable = proxy_pod_readiness_transition_patch("42", "unavailable"); + let ready = proxy_pod_readiness_transition_patch("42", "ready"); + + assert_eq!(unavailable["metadata"]["resourceVersion"], "42"); + assert_eq!(ready["metadata"]["resourceVersion"], "42"); + assert_eq!( + unavailable["metadata"]["annotations"][ANNOTATION_PROXY_POD_READINESS], + "unavailable" + ); + assert_eq!( + ready["metadata"]["annotations"][ANNOTATION_PROXY_POD_READINESS], + "ready" + ); + assert_ne!(unavailable, ready); + } + + #[test] + fn proxy_pod_bootstrap_completion_is_resource_version_guarded_and_publishes_ready() { + let patch = proxy_pod_bootstrap_completion_patch("42"); + + assert_eq!(patch["metadata"]["resourceVersion"], "42"); + assert_eq!( + patch["metadata"]["annotations"][ANNOTATION_PROXY_POD_BOOTSTRAPPING], + serde_json::Value::Null + ); + assert_eq!( + patch["metadata"]["annotations"][ANNOTATION_PROXY_POD_READINESS], + "ready" + ); + } + + #[test] + fn proxy_pod_readiness_is_downgraded_with_a_transient_reason() { + let mut sandbox = Sandbox { + status: Some(SandboxStatus { + conditions: vec![SandboxCondition { + r#type: "Ready".to_string(), + status: "True".to_string(), + ..Default::default() + }], + ..Default::default() + }), + ..Default::default() + }; + mark_proxy_pod_control_unavailable(&mut sandbox); + let ready = &sandbox.status.unwrap().conditions[0]; + assert_eq!(ready.status, "False"); + assert_eq!(ready.reason, "DependenciesNotReady"); + } + + #[test] + fn proxy_pod_bootstrap_does_not_publish_a_terminal_suspension() { + let mut sandbox = Sandbox { + status: Some(SandboxStatus { + conditions: vec![SandboxCondition { + r#type: SANDBOX_SUSPENDED_CONDITION.to_string(), + status: "True".to_string(), + reason: "PodTerminated".to_string(), + ..Default::default() + }], + ..Default::default() + }), + ..Default::default() + }; + + mark_proxy_pod_bootstrapping(&mut sandbox); + + let conditions = &sandbox.status.unwrap().conditions; + assert!( + !conditions + .iter() + .any(|condition| condition.r#type == SANDBOX_SUSPENDED_CONDITION) + ); + assert!(conditions.iter().any(|condition| { + condition.r#type == "Ready" + && condition.status == "False" + && condition.reason == "DependenciesNotReady" + })); + } + + #[test] + fn completed_proxy_pod_bootstrap_preserves_real_suspension() { + let sandbox = Sandbox { + status: Some(SandboxStatus { + conditions: vec![SandboxCondition { + r#type: SANDBOX_SUSPENDED_CONDITION.to_string(), + status: "True".to_string(), + reason: "PodTerminated".to_string(), + ..Default::default() + }], + ..Default::default() + }), + ..Default::default() + }; + + assert!(sandbox.status.unwrap().conditions.iter().any(|condition| { + condition.r#type == SANDBOX_SUSPENDED_CONDITION && condition.status == "True" + })); + } + + #[test] + fn desired_proxy_pod_control_replicas_tracks_both_sandbox_apis() { + let resource = ApiResource::from_gvk(&GroupVersionKind::gvk( + SANDBOX_GROUP, + SANDBOX_VERSION_V1BETA1, + SANDBOX_KIND, + )); + let mut beta = DynamicObject::new("beta", &resource); + beta.data = serde_json::json!({"spec": {"operatingMode": "Suspended"}}); + assert_eq!(desired_proxy_pod_control_replicas(&beta), 0); + beta.data = serde_json::json!({"spec": {"operatingMode": "Running"}}); + assert_eq!(desired_proxy_pod_control_replicas(&beta), 1); + + let mut alpha = DynamicObject::new("alpha", &resource); + alpha.data = serde_json::json!({"spec": {"replicas": 0}}); + assert_eq!(desired_proxy_pod_control_replicas(&alpha), 0); + alpha.data = serde_json::json!({"spec": {"replicas": 1}}); + assert_eq!(desired_proxy_pod_control_replicas(&alpha), 1); + } } diff --git a/crates/openshell-driver-kubernetes/src/grpc.rs b/crates/openshell-driver-kubernetes/src/grpc.rs index 027c350a26..4f57a48b90 100644 --- a/crates/openshell-driver-kubernetes/src/grpc.rs +++ b/crates/openshell-driver-kubernetes/src/grpc.rs @@ -224,17 +224,19 @@ impl ComputeDriver for ComputeDriverService { &self, request: Request, ) -> Result, Status> { - self.trace_rpc("driver.create_sandbox", "create_sandbox", async { - let sandbox = request - .into_inner() - .sandbox - .ok_or_else(|| Status::invalid_argument("sandbox is required"))?; - self.driver - .create_sandbox(&sandbox) - .await - .map_err(|e| Status::from(openshell_core::ComputeDriverError::from(e)))?; - Ok(Response::new(CreateSandboxResponse {})) - }) + Box::pin( + self.trace_rpc("driver.create_sandbox", "create_sandbox", async { + let sandbox = request + .into_inner() + .sandbox + .ok_or_else(|| Status::invalid_argument("sandbox is required"))?; + self.driver + .create_sandbox(&sandbox) + .await + .map_err(|e| Status::from(openshell_core::ComputeDriverError::from(e)))?; + Ok(Response::new(CreateSandboxResponse {})) + }), + ) .await } diff --git a/crates/openshell-driver-kubernetes/src/isolation.rs b/crates/openshell-driver-kubernetes/src/isolation.rs new file mode 100644 index 0000000000..1033f8e932 --- /dev/null +++ b/crates/openshell-driver-kubernetes/src/isolation.rs @@ -0,0 +1,346 @@ +// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Kubernetes provisioning for the shared authenticated boundary protocol. +//! +//! This module deliberately contains no lifecycle, process, network, identity, +//! or wire implementation. The driver chooses the proxy-pod placement, binds +//! immutable Kubernetes resource identities, and provisions TCP coordinates; +//! `openshell-isolation-interface` and `openshell-sandbox` provide the common +//! control and boundary behavior. + +use std::collections::{BTreeMap, HashMap}; +use std::net::{IpAddr, SocketAddr}; +use std::path::PathBuf; + +use k8s_openapi::api::networking::v1::{ + NetworkPolicy, NetworkPolicyIngressRule, NetworkPolicyPeer, NetworkPolicyPort, + NetworkPolicySpec, +}; +use k8s_openapi::apimachinery::pkg::apis::meta::v1::LabelSelector; +use k8s_openapi::apimachinery::pkg::util::intstr::IntOrString; +use kube::core::ObjectMeta; +use openshell_isolation_interface::boundary_protocol::{ + BOUNDARY_PROTOCOL_VERSION, BoundaryAgentIdentity, BoundaryConfig, BoundaryListener, + BoundaryTopology, BoundaryTransport, +}; + +/// Registered RFC 0012 backend name for the proxy-pod topology. +pub const BACKEND_NAME: &str = "kubernetes-proxy-pod"; + +/// Label that binds the workload and control pods in one unique pair. +pub const BOUNDARY_PAIR_LABEL: &str = "openshell.ai/boundary-pair"; + +/// Label distinguishing the two pods in a boundary pair. +pub const BOUNDARY_ROLE_LABEL: &str = "openshell.ai/boundary-role"; + +const WORKLOAD_ROLE: &str = "workload"; +const CONTROL_ROLE: &str = "control"; + +/// Driver-owned inputs for the workload pod's Kubernetes network fence. +/// +/// This is the first phase of proxy-pod provisioning. The driver applies the +/// returned labels to the respective pods and creates the returned policy. It +/// then observes the policy UID and resourceVersion and supplies both to +/// `KubernetesProxyPodBoundarySpec`. +pub struct KubernetesProxyPodNetworkFenceSpec { + pub namespace: String, + pub policy_name: String, + /// A unique, Kubernetes-label-safe value generated for this pod pair. + pub pair_label_value: String, + pub boundary_port: u16, +} + +/// Labels and policy needed to remove direct workload-pod egress. +pub struct KubernetesProxyPodNetworkFence { + pub workload_labels: BTreeMap, + pub control_labels: BTreeMap, + pub workload_policy: NetworkPolicy, +} + +impl KubernetesProxyPodNetworkFenceSpec { + /// Render a default-deny workload fence with one control-to-boundary path. + /// + /// Kubernetes `NetworkPolicy` is connection-aware: traffic returning over + /// the control-initiated boundary connection is allowed even though the + /// workload pod has no egress rules. The control pod remains responsible + /// for opening policy-approved upstream connections. + #[must_use] + pub fn provision(self) -> KubernetesProxyPodNetworkFence { + let workload_labels = boundary_pair_labels(&self.pair_label_value, WORKLOAD_ROLE); + let control_labels = boundary_pair_labels(&self.pair_label_value, CONTROL_ROLE); + + let workload_policy = NetworkPolicy { + metadata: ObjectMeta { + name: Some(self.policy_name), + namespace: Some(self.namespace), + ..Default::default() + }, + spec: Some(NetworkPolicySpec { + pod_selector: LabelSelector { + match_labels: Some(workload_labels.clone()), + ..Default::default() + }, + policy_types: Some(vec!["Ingress".to_string(), "Egress".to_string()]), + // Only the exactly paired control pod in this namespace may + // establish the authenticated boundary connection. + ingress: Some(vec![NetworkPolicyIngressRule { + from: Some(vec![NetworkPolicyPeer { + pod_selector: Some(LabelSelector { + match_labels: Some(control_labels.clone()), + ..Default::default() + }), + ..Default::default() + }]), + ports: Some(vec![NetworkPolicyPort { + port: Some(IntOrString::Int(i32::from(self.boundary_port))), + protocol: Some("TCP".to_string()), + ..Default::default() + }]), + }]), + // An explicit empty list selects the pod for egress and allows + // no new workload-initiated connections, including DNS and the + // Kubernetes API. Reply traffic for allowed ingress remains + // permitted by conforming NetworkPolicy implementations. + egress: Some(Vec::new()), + }), + status: None, + }; + + KubernetesProxyPodNetworkFence { + workload_labels, + control_labels, + workload_policy, + } + } +} + +fn boundary_pair_labels(pair: &str, role: &str) -> BTreeMap { + BTreeMap::from([ + (BOUNDARY_PAIR_LABEL.to_string(), pair.to_string()), + (BOUNDARY_ROLE_LABEL.to_string(), role.to_string()), + ]) +} + +/// Driver-owned inputs that bind one workload/proxy pair to one boundary. +/// +/// The driver constructs this only after Kubernetes has assigned every UID and +/// after it has observed the exact egress policy resource version. The workload +/// stays held until the matching boundary config and control topology have been +/// installed. +pub struct KubernetesProxyPodBoundarySpec { + pub boundary_id: String, + pub bootstrap_token: String, + pub namespace_uid: String, + pub sandbox_resource_uid: String, + pub control_deployment_uid: String, + pub egress_policy_uid: String, + pub egress_policy_resource_version: String, + pub boundary_listener: SocketAddr, + pub control_address: SocketAddr, + pub boundary_server_name: String, + pub boundary_ca_certificate_pem: String, + pub boundary_certificate_chain_path: PathBuf, + pub boundary_private_key_path: PathBuf, + pub host_gateway_ip: Option, + pub agent_uid: u32, + pub agent_gid: u32, + pub trusted_runtime_root: PathBuf, + pub child_env: HashMap, +} + +/// Protected workload-pod config and matching proxy-pod descriptor. +pub struct KubernetesProxyPodBoundaryProvisioning { + pub boundary_config: BoundaryConfig, + pub topology: BoundaryTopology, +} + +impl KubernetesProxyPodBoundarySpec { + /// Produce both sides of the common protocol from one observed Kubernetes + /// resource set so a stale or recreated object cannot be attached. + #[must_use] + pub fn provision(self) -> KubernetesProxyPodBoundaryProvisioning { + let resource_claims = BTreeMap::from([ + ("kubernetes.namespace_uid".to_string(), self.namespace_uid), + ( + "kubernetes.sandbox_resource_uid".to_string(), + self.sandbox_resource_uid, + ), + ( + "kubernetes.control_deployment_uid".to_string(), + self.control_deployment_uid, + ), + ( + "kubernetes.egress_policy_uid".to_string(), + self.egress_policy_uid, + ), + ( + "kubernetes.egress_policy_resource_version".to_string(), + self.egress_policy_resource_version, + ), + ]); + KubernetesProxyPodBoundaryProvisioning { + boundary_config: BoundaryConfig { + protocol_version: BOUNDARY_PROTOCOL_VERSION, + boundary_id: self.boundary_id.clone(), + bootstrap_token: self.bootstrap_token.clone(), + listener: BoundaryListener::TlsTcp { + address: self.boundary_listener, + certificate_chain_path: self.boundary_certificate_chain_path, + private_key_path: self.boundary_private_key_path, + }, + resource_claims: resource_claims.clone(), + agent_identity: BoundaryAgentIdentity::Resolved { + uid: self.agent_uid, + gid: self.agent_gid, + }, + // Kubernetes projects this immutable Secret root-owned and + // mode 0400. Projected volumes are read-only, so attempting a + // defensive chown/chmod here would fail. + protect_config_file: false, + trusted_runtime_root: self.trusted_runtime_root, + child_env: self.child_env, + }, + topology: BoundaryTopology { + protocol_version: BOUNDARY_PROTOCOL_VERSION, + boundary_id: self.boundary_id, + transport: BoundaryTransport::TlsTcp { + address: self.control_address, + server_name: self.boundary_server_name, + ca_certificate_pem: self.boundary_ca_certificate_pem, + }, + host_gateway_ip: self.host_gateway_ip, + resource_claims, + bootstrap_token: self.bootstrap_token, + }, + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn spec() -> KubernetesProxyPodBoundarySpec { + KubernetesProxyPodBoundarySpec { + boundary_id: "sandbox-1".to_string(), + bootstrap_token: "a".repeat(64), + namespace_uid: "namespace-uid".to_string(), + sandbox_resource_uid: "sandbox-resource-uid".to_string(), + control_deployment_uid: "control-deployment-uid".to_string(), + egress_policy_uid: "network-policy-uid".to_string(), + egress_policy_resource_version: "1945".to_string(), + boundary_listener: "0.0.0.0:5500".parse().expect("valid listener"), + control_address: "10.42.0.7:5500".parse().expect("valid target"), + boundary_server_name: "boundary.sandbox.openshell".to_string(), + boundary_ca_certificate_pem: "test-ca".to_string(), + boundary_certificate_chain_path: PathBuf::from("/run/boundary/tls.crt"), + boundary_private_key_path: PathBuf::from("/run/boundary/tls.key"), + host_gateway_ip: Some("10.42.0.1".parse().expect("valid gateway IP")), + agent_uid: 1000, + agent_gid: 1000, + trusted_runtime_root: PathBuf::from("/opt/openshell/bin/openshell-runtime"), + child_env: HashMap::new(), + } + } + + #[test] + fn provisioning_binds_identical_kubernetes_resource_claims() { + let provisioned = spec().provision(); + + assert_eq!( + provisioned.boundary_config.resource_claims, + provisioned.topology.resource_claims + ); + assert_eq!( + provisioned.topology.resource_claims["kubernetes.sandbox_resource_uid"], + "sandbox-resource-uid" + ); + assert_eq!( + provisioned.topology.resource_claims["kubernetes.egress_policy_resource_version"], + "1945" + ); + } + + #[test] + fn provisioning_uses_one_shared_tcp_protocol_across_pods() { + let provisioned = spec().provision(); + + assert_eq!( + provisioned.boundary_config.listener, + BoundaryListener::TlsTcp { + address: "0.0.0.0:5500".parse().expect("valid listener"), + certificate_chain_path: PathBuf::from("/run/boundary/tls.crt"), + private_key_path: PathBuf::from("/run/boundary/tls.key"), + } + ); + assert_eq!( + provisioned.topology.transport, + BoundaryTransport::TlsTcp { + address: "10.42.0.7:5500".parse().expect("valid target"), + server_name: "boundary.sandbox.openshell".to_string(), + ca_certificate_pem: "test-ca".to_string(), + } + ); + } + + #[test] + fn network_fence_denies_all_workload_initiated_egress() { + let fence = KubernetesProxyPodNetworkFenceSpec { + namespace: "sandbox-ns".to_string(), + policy_name: "openshell-boundary-sandbox-1".to_string(), + pair_label_value: "pair-1".to_string(), + boundary_port: 5500, + } + .provision(); + + let policy_spec = fence.workload_policy.spec.expect("policy has a spec"); + assert_eq!( + policy_spec.policy_types, + Some(vec!["Ingress".to_string(), "Egress".to_string()]) + ); + assert_eq!(policy_spec.egress, Some(Vec::new())); + assert_eq!( + policy_spec.pod_selector.match_labels, + Some(fence.workload_labels) + ); + } + + #[test] + fn network_fence_allows_only_paired_control_to_boundary_port() { + let fence = KubernetesProxyPodNetworkFenceSpec { + namespace: "sandbox-ns".to_string(), + policy_name: "openshell-boundary-sandbox-1".to_string(), + pair_label_value: "pair-1".to_string(), + boundary_port: 5500, + } + .provision(); + + let policy_spec = fence.workload_policy.spec.expect("policy has a spec"); + let ingress = policy_spec + .ingress + .expect("policy has ingress rules") + .pop() + .expect("policy has one ingress rule"); + let peer = ingress + .from + .expect("rule has peers") + .pop() + .expect("rule has one peer"); + assert_eq!( + peer.pod_selector + .expect("peer has a pod selector") + .match_labels, + Some(fence.control_labels) + ); + assert!(peer.namespace_selector.is_none()); + + let port = ingress + .ports + .expect("rule has ports") + .pop() + .expect("rule has one port"); + assert_eq!(port.protocol.as_deref(), Some("TCP")); + assert_eq!(port.port, Some(IntOrString::Int(5500))); + } +} diff --git a/crates/openshell-driver-kubernetes/src/lib.rs b/crates/openshell-driver-kubernetes/src/lib.rs index 28d3c77a7d..c960e633f1 100644 --- a/crates/openshell-driver-kubernetes/src/lib.rs +++ b/crates/openshell-driver-kubernetes/src/lib.rs @@ -4,13 +4,15 @@ pub mod config; pub mod driver; pub mod grpc; +pub mod isolation; pub mod otel_tracing; +mod proxy_pod; pub use config::{ AppArmorProfile, DEFAULT_GATEWAY_ID, DEFAULT_PROXY_UID, DEFAULT_SANDBOX_SERVICE_ACCOUNT_NAME, - DEFAULT_WORKSPACE_STORAGE_SIZE, KubernetesComputeConfig, KubernetesSidecarConfig, - ManagedSshIngressConfig, SupervisorSideloadMethod, SupervisorTopology, WorkspaceMode, - managed_namespace_prefix, + DEFAULT_WORKSPACE_STORAGE_SIZE, KubernetesComputeConfig, KubernetesProxyPodConfig, + KubernetesSidecarConfig, ManagedSshIngressConfig, SupervisorSideloadMethod, SupervisorTopology, + WorkspaceMode, managed_namespace_prefix, }; pub use driver::{KubernetesComputeDriver, KubernetesDriverError}; pub use grpc::ComputeDriverService; diff --git a/crates/openshell-driver-kubernetes/src/main.rs b/crates/openshell-driver-kubernetes/src/main.rs index 7690ccee85..99b102eccb 100644 --- a/crates/openshell-driver-kubernetes/src/main.rs +++ b/crates/openshell-driver-kubernetes/src/main.rs @@ -16,8 +16,8 @@ use openshell_driver_kubernetes::otel_tracing::compute_driver_rpc_layer; use openshell_driver_kubernetes::{ AppArmorProfile, ComputeDriverService, DEFAULT_GATEWAY_ID, DEFAULT_PROXY_UID, DEFAULT_SANDBOX_SERVICE_ACCOUNT_NAME, KubernetesComputeConfig, KubernetesComputeDriver, - KubernetesSidecarConfig, ManagedSshIngressConfig, SupervisorSideloadMethod, SupervisorTopology, - WorkspaceMode, + KubernetesProxyPodConfig, KubernetesSidecarConfig, ManagedSshIngressConfig, + SupervisorSideloadMethod, SupervisorTopology, WorkspaceMode, }; #[derive(Parser, Debug)] @@ -145,6 +145,20 @@ struct Args { )] sidecar_process_binary_aware_network_policy: bool, + #[arg( + long, + env = "OPENSHELL_K8S_PROXY_POD_NETWORK_POLICY_ENFORCED", + default_value_t = false + )] + proxy_pod_network_policy_enforced: bool, + + #[arg( + long, + env = "OPENSHELL_K8S_PROXY_POD_BOUNDARY_PORT", + default_value_t = 5500 + )] + proxy_pod_boundary_port: u16, + /// Corporate HTTP forward proxy for policy-approved TLS CONNECT egress. #[arg(long, env = "OPENSHELL_UPSTREAM_PROXY")] https_proxy: Option, @@ -278,6 +292,10 @@ async fn main() -> Result<()> { process_binary_aware_network_policy: args .sidecar_process_binary_aware_network_policy, }, + proxy_pod: KubernetesProxyPodConfig { + network_policy_enforced: args.proxy_pod_network_policy_enforced, + boundary_port: args.proxy_pod_boundary_port, + }, https_proxy: args.https_proxy, no_proxy: args.no_proxy, proxy_auth_secret_name: args.proxy_auth_secret_name, diff --git a/crates/openshell-driver-kubernetes/src/proxy_pod.rs b/crates/openshell-driver-kubernetes/src/proxy_pod.rs new file mode 100644 index 0000000000..bad199c4a3 --- /dev/null +++ b/crates/openshell-driver-kubernetes/src/proxy_pod.rs @@ -0,0 +1,721 @@ +// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Kubernetes resources owned by the RFC 0012 proxy-pod topology. + +use std::collections::BTreeMap; +use std::path::Path; + +use k8s_openapi::ByteString; +use k8s_openapi::api::apps::v1::Deployment; +use k8s_openapi::api::core::v1::{Secret, Service}; +use k8s_openapi::api::networking::v1::{NetworkPolicy, NetworkPolicyEgressRule, NetworkPolicySpec}; +use k8s_openapi::apimachinery::pkg::apis::meta::v1::{LabelSelector, OwnerReference}; +use kube::core::ObjectMeta; +use rcgen::{CertificateParams, DnType, IsCa, KeyPair, KeyUsagePurpose}; + +use crate::isolation::{ + BOUNDARY_PAIR_LABEL, BOUNDARY_ROLE_LABEL, KubernetesProxyPodNetworkFence, + KubernetesProxyPodNetworkFenceSpec, +}; + +pub const BOOTSTRAP_SECRET_COMPONENT: &str = "boundary-bootstrap"; +pub const BOUNDARY_CONFIG_KEY: &str = "boundary.json"; +pub const TOPOLOGY_PAYLOAD_KEY: &str = "topology.json"; +pub const BOUNDARY_CERTIFICATE_KEY: &str = "tls.crt"; +pub const BOUNDARY_PRIVATE_KEY: &str = "tls.key"; +pub const PROXY_CA_CERTIFICATE_KEY: &str = "proxy-ca.crt"; +pub const PROXY_CA_PRIVATE_KEY: &str = "proxy-ca.key"; +pub const BOUNDARY_CONFIG_PATH: &str = "/run/openshell-boundary/boundary.json"; +pub const TOPOLOGY_PAYLOAD_PATH: &str = "/run/openshell-boundary/topology.json"; +pub const BOUNDARY_CERTIFICATE_PATH: &str = "/run/openshell-boundary/tls.crt"; +pub const BOUNDARY_PRIVATE_KEY_PATH: &str = "/run/openshell-boundary/tls.key"; +pub const PROXY_CA_CERTIFICATE_PATH: &str = "/run/openshell-boundary/proxy-ca.crt"; +pub const PROXY_CA_PRIVATE_KEY_PATH: &str = "/run/openshell-boundary/proxy-ca.key"; +pub const BOUNDARY_TLS_SERVER_NAME: &str = "boundary.openshell.internal"; +pub const CONTROL_HEALTH_PORT: u16 = 8080; + +#[allow(clippy::struct_field_names)] +pub struct BoundaryTlsMaterial { + pub ca_certificate_pem: String, + pub certificate_pem: String, + pub private_key_pem: String, +} + +pub struct ProxyCaMaterial { + pub certificate_pem: String, + pub private_key_pem: String, +} + +pub fn generate_proxy_ca_material() -> Result { + let key = KeyPair::generate().map_err(|error| format!("generate proxy CA key: {error}"))?; + let mut params = CertificateParams::default(); + params.is_ca = IsCa::Ca(rcgen::BasicConstraints::Unconstrained); + params + .distinguished_name + .push(DnType::CommonName, "OpenShell Sandbox CA"); + params + .distinguished_name + .push(DnType::OrganizationName, "OpenShell"); + params.key_usages = vec![KeyUsagePurpose::KeyCertSign, KeyUsagePurpose::CrlSign]; + let certificate = params + .self_signed(&key) + .map_err(|error| format!("generate proxy CA certificate: {error}"))?; + Ok(ProxyCaMaterial { + certificate_pem: certificate.pem(), + private_key_pem: key.serialize_pem(), + }) +} + +pub fn generate_boundary_tls_material() -> Result { + let ca_key = + KeyPair::generate().map_err(|error| format!("generate boundary CA key: {error}"))?; + let mut ca_params = CertificateParams::default(); + ca_params.is_ca = IsCa::Ca(rcgen::BasicConstraints::Unconstrained); + ca_params + .distinguished_name + .push(DnType::CommonName, "OpenShell boundary CA"); + ca_params.key_usages = vec![KeyUsagePurpose::KeyCertSign, KeyUsagePurpose::CrlSign]; + let ca = ca_params + .self_signed(&ca_key) + .map_err(|error| format!("generate boundary CA certificate: {error}"))?; + let server_key = + KeyPair::generate().map_err(|error| format!("generate boundary server key: {error}"))?; + let server_params = CertificateParams::new(vec![BOUNDARY_TLS_SERVER_NAME.to_string()]) + .map_err(|error| format!("build boundary server certificate: {error}"))?; + let server = server_params + .signed_by(&server_key, &ca, &ca_key) + .map_err(|error| format!("sign boundary server certificate: {error}"))?; + Ok(BoundaryTlsMaterial { + ca_certificate_pem: ca.pem(), + certificate_pem: server.pem(), + private_key_pem: server_key.serialize_pem(), + }) +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct ProxyPodNames { + pub bootstrap_secret: String, + pub boundary_service: String, + pub control_deployment: String, + pub workload_policy: String, + pub control_policy: String, +} + +impl ProxyPodNames { + #[must_use] + pub fn new(sandbox_id: &str) -> Self { + let suffix = sandbox_id.to_ascii_lowercase(); + Self { + bootstrap_secret: format!("os-boundary-{suffix}"), + boundary_service: format!("os-boundary-{suffix}"), + control_deployment: format!("os-control-{suffix}"), + workload_policy: format!("os-boundary-{suffix}"), + control_policy: format!("os-control-{suffix}"), + } + } +} + +#[must_use] +pub fn pair_label_value(sandbox_id: &str) -> String { + sandbox_id.to_ascii_lowercase() +} + +#[must_use] +pub fn workload_fence( + namespace: &str, + names: &ProxyPodNames, + sandbox_id: &str, + boundary_port: u16, +) -> KubernetesProxyPodNetworkFence { + KubernetesProxyPodNetworkFenceSpec { + namespace: namespace.to_string(), + policy_name: names.workload_policy.clone(), + pair_label_value: pair_label_value(sandbox_id), + boundary_port, + } + .provision() +} + +#[must_use] +pub fn boundary_service( + namespace: &str, + names: &ProxyPodNames, + sandbox_id: &str, + boundary_port: u16, + owner: OwnerReference, +) -> Service { + serde_json::from_value(serde_json::json!({ + "apiVersion": "v1", + "kind": "Service", + "metadata": { + "name": names.boundary_service, + "namespace": namespace, + "ownerReferences": [owner], + "labels": common_labels(sandbox_id, "boundary-service"), + }, + "spec": { + "selector": pair_labels(sandbox_id, "workload"), + "ports": [{"name": "boundary", "protocol": "TCP", "port": boundary_port, "targetPort": boundary_port}], + } + })) + .expect("boundary Service renderer must produce a valid object") +} + +#[allow(clippy::too_many_arguments, clippy::similar_names)] +#[must_use] +pub fn control_deployment( + namespace: &str, + names: &ProxyPodNames, + sandbox_id: &str, + sandbox_name: &str, + gateway_id: &str, + supervisor_image: &str, + supervisor_pull_policy: &str, + service_account_name: &str, + control_uid: u32, + control_gid: u32, + image_pull_secrets: &[String], + grpc_endpoint: &str, + client_tls_secret_name: &str, + main_process_spec: &str, + log_level: &str, + sa_token_ttl_secs: i64, + https_proxy: Option<&str>, + no_proxy: Option<&str>, + proxy_auth_secret: Option<(&str, &str)>, + proxy_auth_allow_insecure: bool, + proxy_connect_by_hostname: bool, + provider_spiffe_socket_path: Option<&str>, + owner: OwnerReference, +) -> Deployment { + let labels = control_labels(sandbox_id, gateway_id); + let mut environment = vec![ + env_var( + "OPENSHELL_ADMITTED_ISOLATION_BACKEND", + crate::isolation::BACKEND_NAME, + ), + env_var("OPENSHELL_ENDPOINT", grpc_endpoint), + env_var("OPENSHELL_SANDBOX_ID", sandbox_id), + env_var("OPENSHELL_SANDBOX", sandbox_name), + env_var("OPENSHELL_MAIN_PROCESS_SPEC", main_process_spec), + env_var( + "OPENSHELL_K8S_SA_TOKEN_FILE", + "/var/run/secrets/openshell/token", + ), + env_var("OPENSHELL_SSH_SOCKET_PATH", "/run/openshell/ssh.sock"), + env_var(openshell_core::sandbox_env::SSH_SOCKET_SHARED, "true"), + env_var("OPENSHELL_PROXY_TLS_DIR", "/run/openshell/proxy-tls"), + env_var( + openshell_core::sandbox_env::PROXY_CA_CERT, + PROXY_CA_CERTIFICATE_PATH, + ), + env_var( + openshell_core::sandbox_env::PROXY_CA_KEY, + PROXY_CA_PRIVATE_KEY_PATH, + ), + env_var("OPENSHELL_LOG_LEVEL", log_level), + env_var( + openshell_core::sandbox_env::TELEMETRY_ENABLED, + openshell_core::telemetry::enabled_env_value(), + ), + env_var( + openshell_core::sandbox_env::NETWORK_RUNTIME_CAPABILITIES, + "", + ), + serde_json::json!({ + "name": "OPENSHELL_HEALTH_BIND_IP", + "valueFrom": {"fieldRef": {"fieldPath": "status.podIP"}} + }), + ]; + let mut volume_mounts = vec![ + serde_json::json!({"name": "bootstrap", "mountPath": TOPOLOGY_PAYLOAD_PATH, "subPath": TOPOLOGY_PAYLOAD_KEY, "readOnly": true}), + serde_json::json!({"name": "bootstrap", "mountPath": PROXY_CA_CERTIFICATE_PATH, "subPath": PROXY_CA_CERTIFICATE_KEY, "readOnly": true}), + serde_json::json!({"name": "bootstrap", "mountPath": PROXY_CA_PRIVATE_KEY_PATH, "subPath": PROXY_CA_PRIVATE_KEY, "readOnly": true}), + serde_json::json!({"name": "sa-token", "mountPath": "/var/run/secrets/openshell", "readOnly": true}), + serde_json::json!({"name": "run", "mountPath": "/run/openshell"}), + serde_json::json!({"name": "logs", "mountPath": "/var/log"}), + ]; + let mut volumes = vec![ + serde_json::json!({"name": "bootstrap", "secret": {"secretName": names.bootstrap_secret, "defaultMode": 0o440}}), + serde_json::json!({"name": "sa-token", "projected": {"sources": [{"serviceAccountToken": {"audience": "openshell-gateway", "expirationSeconds": sa_token_ttl_secs, "path": "token"}}], "defaultMode": 0o440}}), + serde_json::json!({"name": "run", "emptyDir": {}}), + serde_json::json!({"name": "logs", "emptyDir": {}}), + ]; + if !client_tls_secret_name.is_empty() { + environment.extend([ + env_var("OPENSHELL_TLS_CA", "/var/run/secrets/openshell-tls/ca.crt"), + env_var( + "OPENSHELL_TLS_CERT", + "/var/run/secrets/openshell-tls/tls.crt", + ), + env_var( + "OPENSHELL_TLS_KEY", + "/var/run/secrets/openshell-tls/tls.key", + ), + ]); + volume_mounts.push(serde_json::json!({"name": "client-tls", "mountPath": "/var/run/secrets/openshell-tls", "readOnly": true})); + volumes.push(serde_json::json!({"name": "client-tls", "secret": {"secretName": client_tls_secret_name, "defaultMode": 0o440}})); + } + let mut command = vec![ + "/openshell-sandbox".to_string(), + "--mode=control".to_string(), + "--topology-backend-name".to_string(), + crate::isolation::BACKEND_NAME.to_string(), + "--topology-version".to_string(), + openshell_isolation_interface::contract::INTERFACE_VERSION.to_string(), + "--topology-payload-file".to_string(), + TOPOLOGY_PAYLOAD_PATH.to_string(), + "--workdir".to_string(), + "/sandbox".to_string(), + "--health-check".to_string(), + "--health-port".to_string(), + CONTROL_HEALTH_PORT.to_string(), + ]; + if let Some(url) = https_proxy { + command.extend(["--upstream-proxy".to_string(), url.to_string()]); + } + if let Some(hosts) = no_proxy { + command.extend(["--upstream-no-proxy".to_string(), hosts.to_string()]); + } + if proxy_auth_secret.is_some() { + command.extend([ + "--upstream-proxy-auth-file".to_string(), + openshell_core::container_paths::UPSTREAM_PROXY_AUTH_MOUNT_PATH.to_string(), + ]); + } + if proxy_auth_allow_insecure { + command.push("--upstream-proxy-auth-allow-insecure".to_string()); + } + if proxy_connect_by_hostname { + command.push("--upstream-proxy-connect-by-hostname".to_string()); + } + if let Some((secret_name, secret_key)) = proxy_auth_secret { + let auth_path = Path::new(openshell_core::container_paths::UPSTREAM_PROXY_AUTH_MOUNT_PATH); + volume_mounts.push(serde_json::json!({ + "name": "upstream-proxy-auth", + "mountPath": auth_path.parent().and_then(Path::to_str).expect("auth path has parent"), + "readOnly": true + })); + volumes.push(serde_json::json!({ + "name": "upstream-proxy-auth", + "secret": {"secretName": secret_name, "defaultMode": 0o440, "items": [{ + "key": secret_key, + "path": auth_path.file_name().and_then(|name| name.to_str()).expect("auth path has file name") + }]} + })); + } + if let Some(socket_path) = provider_spiffe_socket_path { + environment.push(env_var( + openshell_core::sandbox_env::PROVIDER_SPIFFE_WORKLOAD_API_SOCKET, + socket_path, + )); + volume_mounts.push(serde_json::json!({ + "name": "spiffe-workload-api", + "mountPath": Path::new(socket_path).parent().and_then(Path::to_str).expect("SPIFFE socket has parent"), + "readOnly": true + })); + volumes.push(serde_json::json!({ + "name": "spiffe-workload-api", + "csi": {"driver": "csi.spiffe.io", "readOnly": true} + })); + } + let mut container = serde_json::json!({ + "name": "control", + "image": supervisor_image, + "command": command, + "terminationMessagePolicy": "FallbackToLogsOnError", + "env": environment, + "ports": [{"name": "health", "containerPort": CONTROL_HEALTH_PORT, "protocol": "TCP"}], + "readinessProbe": { + "tcpSocket": {"port": "health"}, + "periodSeconds": 1, + "failureThreshold": 1 + }, + "securityContext": { + "runAsUser": control_uid, + "runAsGroup": control_gid, + "runAsNonRoot": true, + "readOnlyRootFilesystem": true, + "allowPrivilegeEscalation": false, + "capabilities": {"drop": ["ALL"]} + }, + "volumeMounts": volume_mounts, + }); + if !supervisor_pull_policy.is_empty() { + container["imagePullPolicy"] = serde_json::json!(supervisor_pull_policy); + } + serde_json::from_value(serde_json::json!({ + "apiVersion": "apps/v1", + "kind": "Deployment", + "metadata": {"name": names.control_deployment, "namespace": namespace, "ownerReferences": [owner], "labels": labels}, + "spec": { + "replicas": 0, + "strategy": {"type": "Recreate"}, + "selector": {"matchLabels": pair_labels(sandbox_id, "control")}, + "template": { + "metadata": {"labels": control_labels(sandbox_id, gateway_id), "annotations": {"openshell.ai/sandbox-id": sandbox_id}}, + "spec": { + "serviceAccountName": service_account_name, + "imagePullSecrets": image_pull_secrets.iter().map(|name| serde_json::json!({"name": name})).collect::>(), + "automountServiceAccountToken": false, + "securityContext": { + "fsGroup": control_gid, + "fsGroupChangePolicy": "OnRootMismatch", + "seccompProfile": {"type": "RuntimeDefault"} + }, + "restartPolicy": "Always", + "containers": [container], + "volumes": volumes + } + } + } + })) + .expect("control Deployment renderer must produce a valid object") +} + +#[must_use] +pub fn control_egress_policy( + namespace: &str, + names: &ProxyPodNames, + sandbox_id: &str, + owner: OwnerReference, +) -> NetworkPolicy { + NetworkPolicy { + metadata: ObjectMeta { + name: Some(names.control_policy.clone()), + namespace: Some(namespace.to_string()), + owner_references: Some(vec![owner]), + labels: Some(common_labels(sandbox_id, "control-egress")), + ..Default::default() + }, + spec: Some(NetworkPolicySpec { + pod_selector: LabelSelector { + match_labels: Some(pair_labels(sandbox_id, "control")), + ..Default::default() + }, + policy_types: Some(vec!["Egress".to_string()]), + // Control is the policy-enforcing egress principal. Namespace-wide + // default-deny policies must not prevent its gateway/DNS/upstream dials. + egress: Some(vec![NetworkPolicyEgressRule::default()]), + ..Default::default() + }), + status: None, + } +} + +#[allow(clippy::too_many_arguments)] +#[must_use] +pub fn bootstrap_secret( + namespace: &str, + names: &ProxyPodNames, + sandbox_id: &str, + boundary_config: Vec, + topology_payload: Vec, + boundary_certificate: Vec, + boundary_private_key: Vec, + proxy_ca_certificate: Vec, + proxy_ca_private_key: Vec, + owner: OwnerReference, +) -> Secret { + Secret { + metadata: ObjectMeta { + name: Some(names.bootstrap_secret.clone()), + namespace: Some(namespace.to_string()), + owner_references: Some(vec![owner]), + labels: Some(common_labels(sandbox_id, BOOTSTRAP_SECRET_COMPONENT)), + ..Default::default() + }, + data: Some(BTreeMap::from([ + (BOUNDARY_CONFIG_KEY.to_string(), ByteString(boundary_config)), + ( + TOPOLOGY_PAYLOAD_KEY.to_string(), + ByteString(topology_payload), + ), + ( + BOUNDARY_CERTIFICATE_KEY.to_string(), + ByteString(boundary_certificate), + ), + ( + BOUNDARY_PRIVATE_KEY.to_string(), + ByteString(boundary_private_key), + ), + ( + PROXY_CA_CERTIFICATE_KEY.to_string(), + ByteString(proxy_ca_certificate), + ), + ( + PROXY_CA_PRIVATE_KEY.to_string(), + ByteString(proxy_ca_private_key), + ), + ])), + immutable: Some(true), + type_: Some("Opaque".to_string()), + ..Default::default() + } +} + +#[must_use] +pub fn sandbox_owner_reference( + name: &str, + uid: &str, + api_version: &str, + controller: bool, +) -> OwnerReference { + OwnerReference { + api_version: api_version.to_string(), + kind: "Sandbox".to_string(), + name: name.to_string(), + uid: uid.to_string(), + controller: controller.then_some(true), + // The driver's RBAC intentionally does not permit mutating Sandbox + // finalizers. Kubernetes garbage collection does not require this bit, + // and setting it would make admission fail under + // OwnerReferencesPermissionEnforcement. + block_owner_deletion: Some(false), + } +} + +fn pair_labels(sandbox_id: &str, role: &str) -> BTreeMap { + BTreeMap::from([ + ( + BOUNDARY_PAIR_LABEL.to_string(), + pair_label_value(sandbox_id), + ), + (BOUNDARY_ROLE_LABEL.to_string(), role.to_string()), + ]) +} + +fn common_labels(sandbox_id: &str, component: &str) -> BTreeMap { + BTreeMap::from([ + ( + "openshell.ai/managed-by".to_string(), + "openshell".to_string(), + ), + ( + "openshell.ai/sandbox-id".to_string(), + sandbox_id.to_string(), + ), + ("openshell.ai/component".to_string(), component.to_string()), + ]) +} + +fn control_labels(sandbox_id: &str, gateway_id: &str) -> BTreeMap { + let mut labels = common_labels(sandbox_id, "boundary-control"); + labels.extend(pair_labels(sandbox_id, "control")); + labels.insert( + "openshell.ai/gateway-id".to_string(), + gateway_id.to_string(), + ); + labels +} + +fn env_var(name: &str, value: &str) -> serde_json::Value { + serde_json::json!({"name": name, "value": value}) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn owner() -> OwnerReference { + sandbox_owner_reference("demo", "uid-1", "agents.x-k8s.io/v1beta1", true) + } + + #[test] + fn service_selects_only_the_workload_boundary() { + let names = ProxyPodNames::new("4b67c0d0-1111-2222-3333-444444444444"); + let service = boundary_service("sandbox", &names, "pair", 5500, owner()); + assert_eq!( + service.spec.unwrap().selector.unwrap()[BOUNDARY_ROLE_LABEL], + "workload" + ); + } + + #[test] + fn owner_reference_does_not_require_finalizer_mutation_permission() { + assert_eq!(owner().block_owner_deletion, Some(false)); + } + + #[test] + fn control_policy_explicitly_allows_control_egress() { + let names = ProxyPodNames::new("pair"); + let policy = control_egress_policy("sandbox", &names, "pair", owner()); + let spec = policy.spec.unwrap(); + assert_eq!(spec.policy_types.unwrap(), ["Egress"]); + assert_eq!(spec.egress.unwrap().len(), 1); + } + + #[test] + fn control_deployment_is_singleton_ready_and_unprivileged() { + let names = ProxyPodNames::new("pair"); + let deployment = control_deployment( + "sandbox", + &names, + "pair", + "demo", + "gateway", + "supervisor:latest", + "IfNotPresent", + "sandbox-sa", + 1000, + 1000, + &["registry-credentials".to_string()], + "https://gateway:8080", + "client-tls", + "{}", + "info", + 600, + None, + None, + None, + false, + false, + None, + owner(), + ); + let spec = deployment.spec.as_ref().unwrap(); + assert_eq!( + spec.strategy + .as_ref() + .and_then(|strategy| strategy.type_.as_deref()), + Some("Recreate") + ); + let container = &spec.template.spec.as_ref().unwrap().containers[0]; + let pod_spec = spec.template.spec.as_ref().unwrap(); + assert_eq!(pod_spec.automount_service_account_token, Some(false)); + assert_eq!( + pod_spec.image_pull_secrets.as_ref().unwrap()[0] + .name + .as_deref(), + Some("registry-credentials") + ); + let pod_security = + serde_json::to_value(pod_spec.security_context.as_ref().unwrap()).unwrap(); + assert_eq!(pod_security["fsGroup"], 1000); + assert_eq!(pod_security["seccompProfile"]["type"], "RuntimeDefault"); + let container_security = + serde_json::to_value(container.security_context.as_ref().unwrap()).unwrap(); + assert_eq!(container_security["runAsUser"], 1000); + assert_eq!(container_security["runAsNonRoot"], true); + assert_eq!(container_security["readOnlyRootFilesystem"], true); + assert_eq!( + container_security["capabilities"]["drop"], + serde_json::json!(["ALL"]) + ); + assert_eq!( + container + .readiness_probe + .as_ref() + .and_then(|probe| probe.tcp_socket.as_ref()) + .map(|socket| &socket.port), + Some( + &k8s_openapi::apimachinery::pkg::util::intstr::IntOrString::String( + "health".to_string() + ) + ) + ); + let command = container.command.as_ref().unwrap(); + let health_port = CONTROL_HEALTH_PORT.to_string(); + assert!( + command + .windows(2) + .any(|args| args == ["--health-port", health_port.as_str()]) + ); + let health_bind = container + .env + .as_ref() + .unwrap() + .iter() + .find(|env| env.name == "OPENSHELL_HEALTH_BIND_IP") + .unwrap(); + assert_eq!( + health_bind + .value_from + .as_ref() + .and_then(|source| source.field_ref.as_ref()) + .map(|field| field.field_path.as_str()), + Some("status.podIP") + ); + let env = container.env.as_ref().unwrap(); + let env_value = |name: &str| { + env.iter() + .find(|variable| variable.name == name) + .and_then(|variable| variable.value.as_deref()) + }; + assert_eq!( + env_value(openshell_core::sandbox_env::SSH_SOCKET_SHARED), + Some("true") + ); + assert_eq!( + env_value(openshell_core::sandbox_env::PROXY_CA_CERT), + Some(PROXY_CA_CERTIFICATE_PATH) + ); + assert_eq!( + env_value(openshell_core::sandbox_env::PROXY_CA_KEY), + Some(PROXY_CA_PRIVATE_KEY_PATH) + ); + let mounts = container.volume_mounts.as_ref().unwrap(); + for (path, key) in [ + (PROXY_CA_CERTIFICATE_PATH, PROXY_CA_CERTIFICATE_KEY), + (PROXY_CA_PRIVATE_KEY_PATH, PROXY_CA_PRIVATE_KEY), + ] { + let mount = mounts + .iter() + .find(|mount| mount.mount_path == path) + .expect("durable proxy CA material is mounted into control"); + assert_eq!(mount.name, "bootstrap"); + assert_eq!(mount.sub_path.as_deref(), Some(key)); + assert_eq!(mount.read_only, Some(true)); + } + } + + #[test] + fn bootstrap_secret_uses_valid_component_label() { + let names = ProxyPodNames::new("pair"); + let secret = bootstrap_secret( + "sandbox", + &names, + "pair", + Vec::new(), + Vec::new(), + Vec::new(), + Vec::new(), + Vec::new(), + Vec::new(), + owner(), + ); + assert_eq!( + secret.metadata.labels.unwrap()["openshell.ai/component"], + BOOTSTRAP_SECRET_COMPONENT + ); + assert_eq!(secret.immutable, Some(true)); + let keys = secret + .data + .unwrap() + .into_keys() + .collect::>(); + assert_eq!( + keys, + std::collections::BTreeSet::from([ + BOUNDARY_CERTIFICATE_KEY.to_string(), + BOUNDARY_CONFIG_KEY.to_string(), + BOUNDARY_PRIVATE_KEY.to_string(), + PROXY_CA_CERTIFICATE_KEY.to_string(), + PROXY_CA_PRIVATE_KEY.to_string(), + TOPOLOGY_PAYLOAD_KEY.to_string(), + ]) + ); + } + + #[test] + fn generated_proxy_ca_material_is_pem_encoded() { + let material = generate_proxy_ca_material().unwrap(); + assert!( + material + .certificate_pem + .starts_with("-----BEGIN CERTIFICATE-----") + ); + assert!(material.private_key_pem.contains("PRIVATE KEY")); + } +} diff --git a/deploy/helm/openshell/README.md b/deploy/helm/openshell/README.md index 7dbd7964d0..dc02475ec8 100644 --- a/deploy/helm/openshell/README.md +++ b/deploy/helm/openshell/README.md @@ -300,10 +300,12 @@ discovery endpoint or its TLS CA. | supervisor.image.pullPolicy | string | `""` | Supervisor image pull policy. Defaults to the gateway image pull policy when empty. | | supervisor.image.repository | string | `"ghcr.io/nvidia/openshell/supervisor"` | Supervisor image repository. Changing it uses the effective gateway image tag unless tag is also set. | | supervisor.image.tag | string | `""` | Supervisor image tag override. Empty uses the version pinned into the gateway unless repository is changed. | +| supervisor.proxyPod.boundaryPort | int | `5500` | Workload boundary TLS listener port. | +| supervisor.proxyPod.networkPolicyEnforced | bool | `false` | Required operator acknowledgement that the cluster CNI enforces NetworkPolicy. | | supervisor.sidecar.processBinaryAwareNetworkPolicy | bool | `true` | Keep process/binary-aware network policy enabled in sidecar topology. When false, the network sidecar runs as proxyUid, drops the extra /proc inspection capabilities, and enforces endpoint/L7 policy without matching policy.binaries. | | supervisor.sidecar.proxyUid | int | `1337` | UID for relaxed long-running network sidecars in sidecar topology. Strict process/binary-aware sidecars run as UID 0 so Kubernetes grants the required /proc inspection capabilities into the effective set. The network init container installs nftables rules that exempt the effective sidecar UID. | | supervisor.sideloadMethod | string | `""` | How the supervisor binary is delivered into sandbox pods. Empty (default) = auto-detect from cluster version: K8s >= v1.35 -> "image-volume" (ImageVolume enabled by default; GA in v1.36) K8s < v1.35 -> "init-container" (copies via init container + emptyDir) On K8s v1.33-v1.34 with the ImageVolume feature gate manually enabled, set this to "image-volume" explicitly. | -| supervisor.topology | string | `"combined"` | Supervisor pod topology for Kubernetes sandboxes. "combined" runs the current single supervisor container in the agent pod. "sidecar" runs network enforcement in a dedicated sidecar and the process supervisor as a low-capability wrapper in the agent container. | +| supervisor.topology | string | `"combined"` | Supervisor pod topology for Kubernetes sandboxes. "combined" runs the current single supervisor container in the agent pod. "sidecar" runs network enforcement in a dedicated sidecar and the process supervisor as a low-capability wrapper in the agent container. "proxy-pod" runs RFC 0012 control in a separate pod and boundary beside the workload. | | tolerations | list | `[]` | Tolerations for the gateway pod. | | upstreamProxy | object | `{"authAllowInsecure":false,"authSecret":{"key":"","name":""},"connectByHostname":false,"noProxy":"","url":""}` | Operator-owned corporate forward proxy for policy-approved TLS egress from Kubernetes sandboxes. The workload cannot select or override it. | | upstreamProxy.authAllowInsecure | bool | `false` | Required when authSecret is configured because Basic auth to an HTTP proxy is cleartext. | diff --git a/deploy/helm/openshell/ci/values-proxy-pod.yaml b/deploy/helm/openshell/ci/values-proxy-pod.yaml new file mode 100644 index 0000000000..1824a65d50 --- /dev/null +++ b/deploy/helm/openshell/ci/values-proxy-pod.yaml @@ -0,0 +1,7 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +supervisor: + topology: proxy-pod + proxyPod: + networkPolicyEnforced: true diff --git a/deploy/helm/openshell/templates/clusterrole.yaml b/deploy/helm/openshell/templates/clusterrole.yaml index eb1ed8e1d0..3fe3775d31 100644 --- a/deploy/helm/openshell/templates/clusterrole.yaml +++ b/deploy/helm/openshell/templates/clusterrole.yaml @@ -84,6 +84,23 @@ rules: {{- end }} {{- end }} {{- end }} + {{- if and (ne $workspaceMode "shared") (eq (.Values.supervisor.topology | default "combined") "proxy-pod") }} + - apiGroups: ["apps"] + resources: ["deployments"] + verbs: ["create", "get", "patch"] + - apiGroups: ["apps"] + resources: ["replicasets"] + verbs: ["get"] + - apiGroups: [""] + resources: ["services"] + verbs: ["create", "get"] + - apiGroups: [""] + resources: ["secrets"] + verbs: ["create"] + - apiGroups: ["networking.k8s.io"] + resources: ["networkpolicies"] + verbs: ["create", "delete", "get", "list"] + {{- end }} {{- $copiedSecretNames = uniq $copiedSecretNames }} {{- if $copiedSecretNames }} # Copy only explicitly configured TLS and image-pull Secrets into workspace diff --git a/deploy/helm/openshell/templates/gateway-config.yaml b/deploy/helm/openshell/templates/gateway-config.yaml index 083748aee3..43d678ec2a 100644 --- a/deploy/helm/openshell/templates/gateway-config.yaml +++ b/deploy/helm/openshell/templates/gateway-config.yaml @@ -219,6 +219,12 @@ data: proxy_uid = {{ .Values.supervisor.sidecar.proxyUid | default 1337 }} process_binary_aware_network_policy = {{ .Values.supervisor.sidecar.processBinaryAwareNetworkPolicy }} + {{- if eq (.Values.supervisor.topology | default "combined") "proxy-pod" }} + [openshell.drivers.kubernetes.proxy_pod] + network_policy_enforced = {{ .Values.supervisor.proxyPod.networkPolicyEnforced }} + boundary_port = {{ .Values.supervisor.proxyPod.boundaryPort | default 5500 }} + {{- end }} + {{- if not $credentialDrivers }} [openshell.gateway.credential_storage] diff --git a/deploy/helm/openshell/templates/role.yaml b/deploy/helm/openshell/templates/role.yaml index dfd6423615..eeec0fecc8 100644 --- a/deploy/helm/openshell/templates/role.yaml +++ b/deploy/helm/openshell/templates/role.yaml @@ -43,4 +43,21 @@ rules: - pods verbs: - get + {{- if eq (.Values.supervisor.topology | default "combined") "proxy-pod" }} + - apiGroups: ["apps"] + resources: ["deployments"] + verbs: ["create", "get", "patch"] + - apiGroups: ["apps"] + resources: ["replicasets"] + verbs: ["get"] + - apiGroups: [""] + resources: ["services"] + verbs: ["create", "get"] + - apiGroups: [""] + resources: ["secrets"] + verbs: ["create"] + - apiGroups: ["networking.k8s.io"] + resources: ["networkpolicies"] + verbs: ["create", "delete", "get", "list"] + {{- end }} {{- end }} diff --git a/deploy/helm/openshell/tests/clusterrole_test.yaml b/deploy/helm/openshell/tests/clusterrole_test.yaml index afecada9b6..f7ebafda99 100644 --- a/deploy/helm/openshell/tests/clusterrole_test.yaml +++ b/deploy/helm/openshell/tests/clusterrole_test.yaml @@ -122,3 +122,39 @@ tests: apiGroups: [""] resources: ["secrets"] any: true + + - it: grants managed proxy-pod companion permissions + set: + server.drivers.kubernetes.workspaceMode: managed + supervisor.topology: proxy-pod + asserts: + - contains: + path: rules + content: + apiGroups: ["apps"] + resources: ["deployments"] + verbs: ["create", "get", "patch"] + - contains: + path: rules + content: + apiGroups: [""] + resources: ["secrets"] + verbs: ["create"] + + - it: grants operator proxy-pod companion permissions + set: + server.drivers.kubernetes.workspaceMode: operator + supervisor.topology: proxy-pod + asserts: + - contains: + path: rules + content: + apiGroups: ["apps"] + resources: ["deployments"] + verbs: ["create", "get", "patch"] + - contains: + path: rules + content: + apiGroups: [""] + resources: ["secrets"] + verbs: ["create"] diff --git a/deploy/helm/openshell/values.yaml b/deploy/helm/openshell/values.yaml index 8f5b6fa51a..0e75d0a8f9 100644 --- a/deploy/helm/openshell/values.yaml +++ b/deploy/helm/openshell/values.yaml @@ -48,6 +48,7 @@ supervisor: # "combined" runs the current single supervisor container in the agent pod. # "sidecar" runs network enforcement in a dedicated sidecar and the process # supervisor as a low-capability wrapper in the agent container. + # "proxy-pod" runs RFC 0012 control in a separate pod and boundary beside the workload. topology: "combined" sidecar: # -- UID for relaxed long-running network sidecars in sidecar topology. @@ -61,6 +62,11 @@ supervisor: # inspection capabilities, and enforces endpoint/L7 policy without matching # policy.binaries. processBinaryAwareNetworkPolicy: true + proxyPod: + # -- Required operator acknowledgement that the cluster CNI enforces NetworkPolicy. + networkPolicyEnforced: false + # -- Workload boundary TLS listener port. + boundaryPort: 5500 # -- Operator-owned corporate forward proxy for policy-approved TLS egress # from Kubernetes sandboxes. The workload cannot select or override it. diff --git a/docs/reference/gateway-config.mdx b/docs/reference/gateway-config.mdx index 7553e7deb4..b99e13496f 100644 --- a/docs/reference/gateway-config.mdx +++ b/docs/reference/gateway-config.mdx @@ -493,7 +493,9 @@ supervisor_image_pull_policy = "IfNotPresent" supervisor_sideload_method = "image-volume" # "combined" runs the existing single supervisor container with full process, # filesystem, and network enforcement in the agent container. "sidecar" moves -# pod-level network enforcement and gateway session handling into a network sidecar. +# network enforcement into the workload pod's sidecar. "proxy-pod" places the +# credential-holding control supervisor in a separate pod and leaves only the +# process-owning boundary beside the workload. topology = "combined" # Optional corporate HTTP forward proxy for policy-approved TLS egress. The # sandbox workload cannot select or override these settings. Only http:// proxy @@ -512,9 +514,9 @@ topology = "combined" # Secret prevents that Pod from starting; unreadable or malformed `user:pass` # content is validated fail-closed by the supervisor at startup and never # falls back to direct egress. -# Proxy credential Secrets require `topology = "sidecar"`. Combined topology -# shares its credential mount with the workload and can make it readable by the -# sandbox group through Kubernetes `fsGroup` volume permission handling. +# Proxy credential Secrets require `topology = "sidecar"` or `"proxy-pod"`. +# Combined topology shares its credential mount with the workload and can make +# it readable by the sandbox group through Kubernetes `fsGroup` handling. # Required with a credential Secret: Basic authentication to an http:// proxy # is cleartext on the connection to that proxy. # proxy_auth_allow_insecure = true @@ -574,6 +576,13 @@ proxy_uid = 1337 # inspection capabilities, and enforce endpoint/L7 policy without matching # policy.binaries. process_binary_aware_network_policy = true + +[openshell.drivers.kubernetes.proxy_pod] +# Required acknowledgement that the cluster CNI enforces NetworkPolicy and the +# sandbox namespaces prevent untrusted policy, pod, label, and Secret changes. +network_policy_enforced = true +# TLS-protected boundary listener reached only by the paired control pod. +boundary_port = 5500 ``` In managed workspace mode, the Kubernetes driver copies each explicitly named diff --git a/skills/debug-openshell-cluster/SKILL.md b/skills/debug-openshell-cluster/SKILL.md index 8e86bc0643..46fbd1635f 100644 --- a/skills/debug-openshell-cluster/SKILL.md +++ b/skills/debug-openshell-cluster/SKILL.md @@ -554,6 +554,46 @@ kubectl -n logs -c openshell-supervisor-networ kubectl -n logs -c agent --tail=200 ``` +If `topology = "proxy-pod"` is rendered, Helm must also render +`network_policy_enforced = true`. This is an explicit operator assertion that +the cluster CNI enforces Kubernetes NetworkPolicy; the Kubernetes API cannot +attest enforcement. Run proxy-pod topology only in a trusted sandbox namespace +where tenants cannot create pods, copy the OpenShell pair labels, or read the +bootstrap Secret. + +The workload pod runs only `openshell-sandbox --mode=boundary`. It has no +gateway credentials and no direct egress. A deny-all workload NetworkPolicy is +created before the suspended Sandbox resource, then a per-sandbox Service, +immutable bootstrap Secret, control Deployment, control-egress policy, and +boundary-ingress policy are provisioned before the workload is released. The +control pod runs `openshell-sandbox --mode=control` with all Linux capabilities +dropped. It reaches the boundary over per-sandbox TLS with server-certificate +verification plus bootstrap-token client authentication, and owns gateway +policy, provider credentials, DNS, and mediated upstream connections. + +Inspect the complete companion set when a proxy-pod sandbox remains Starting +or loses readiness: + +```bash +kubectl -n get sandbox,pod,deployment,replicaset,service,secret,networkpolicy -l openshell.ai/sandbox-id= +kubectl -n describe deployment -l openshell.ai/sandbox-id= +kubectl -n logs deployment/ --tail=200 +kubectl -n get pod -l openshell.ai/sandbox-id=,openshell.ai/boundary-role=workload -o yaml +kubectl -n get networkpolicy -l openshell.ai/sandbox-id= -o yaml +``` + +Creation and recovery fail closed. A missing Secret leaves both pods inert; a +missing or unobserved workload fence must prevent the driver from releasing the +Sandbox; and readiness requires both Agent Sandbox readiness and an Available +control Deployment. The Deployment's TCP readiness probe opens only after +control has attached, confirmed enforcement, started or resumed the workload, +and installed the access plane. Use the control container termination message +and logs for boundary bootstrap errors. `EPERM` while preparing the workspace commonly +means the boundary lost a required bootstrap capability; the boundary needs +the narrow bootstrap set rendered by the driver, while the untrusted child has +its capability bounding set cleared before exec. Do not add gateway egress or +credentials to the workload pod as a workaround. + #### Corporate upstream proxy When the deployment routes sandbox egress through a corporate HTTP forward diff --git a/tasks/test.toml b/tasks/test.toml index 4a5cda0890..81d0efba73 100644 --- a/tasks/test.toml +++ b/tasks/test.toml @@ -186,6 +186,11 @@ env = { OPENSHELL_E2E_KUBE_EXTRA_VALUES = "deploy/helm/openshell/ci/values-sidec depends = ["e2e:conformance:build"] run = "OPENSHELL_CONFORMANCE_BIN=\"${OPENSHELL_CONFORMANCE_BIN:-$PWD/target/debug/openshell-conformance}\" e2e/rust/e2e-kubernetes.sh" +["e2e:kubernetes:proxy-pod"] +description = "Run Kubernetes e2e with the RFC 0012 control/boundary proxy-pod topology" +env = { OPENSHELL_E2E_KUBE_EXTRA_VALUES = "deploy/helm/openshell/ci/values-proxy-pod.yaml" } +run = "e2e/rust/e2e-kubernetes.sh" + ["e2e:kubernetes:db"] description = "Run Kubernetes e2e with all database backend scenarios (SQLite and external PostgreSQL with existingSecret)" env = { OPENSHELL_E2E_KUBE_DB_SCENARIOS = "1" }