TLS configuration and building TlsAcceptors

mirrord/agent/env/Cargo.toml

@@ -18,7 +18,9 @@ default = ["k8s-openapi"]
 k8s-openapi = ["dep:k8s-openapi"]
 
 [dependencies]
+base64.workspace = true
 k8s-openapi = { workspace = true, optional = true }
 serde.workspace = true
+serde_json.workspace = true
 thiserror.workspace = true
 tracing.workspace = true

mirrord/agent/env/src/checked_env.rs

@@ -8,10 +8,13 @@ use std::{
     str::{FromStr, Utf8Error},
 };
 
+use base64::{engine::general_purpose, Engine};
 #[cfg(feature = "k8s-openapi")]
 use k8s_openapi::api::core::v1::EnvVar;
 use thiserror::Error;
 
+use crate::steal_tls::StealTlsConfig;
+
 /// Type of an environment variable value.
 pub trait EnvValue: Sized {
     /// Error that can occur when producing the value representation.
@@ -194,3 +197,30 @@ impl EnvValue for Vec<IpAddr> {
             .map_err(ParseEnvError::ParseError)
     }
 }
+
+#[derive(Error, Debug)]
+pub enum ParseStealTlsConfigError {
+    #[error("failed to decode as base64: {0}")]
+    DecodeBase64Error(#[from] base64::DecodeError),
+    #[error("failed to deserialize as JSON: {0}")]
+    DeserializeError(#[from] serde_json::Error),
+}
+
+impl EnvValue for StealTlsConfig {
+    type IntoReprError = Infallible;
+    type FromReprError = ParseStealTlsConfigError;
+
+    fn as_repr(&self) -> Result<String, Self::IntoReprError> {
+        let as_bytes = serde_json::to_vec(self).expect("serializing to memory should not fail");
+        let encoded = general_purpose::STANDARD_NO_PAD.encode(as_bytes);
+
+        Ok(encoded)
+    }
+
+    fn from_repr(repr: &[u8]) -> Result<Self, Self::FromReprError> {
+        let decoded = general_purpose::STANDARD_NO_PAD.decode(repr)?;
+        let deserialized = serde_json::from_slice(&decoded)?;
+
+        Ok(deserialized)
+    }
+}

mirrord/agent/env/src/envs.rs

@@ -4,7 +4,7 @@
 
 use std::net::{IpAddr, SocketAddr};
 
-use crate::checked_env::CheckedEnv;
+use crate::{checked_env::CheckedEnv, steal_tls::StealTlsConfig};
 
 /// Used to pass operator's x509 certificate to the agent.
 ///
@@ -51,3 +51,7 @@ pub const POD_IPS: CheckedEnv<Vec<IpAddr>> = CheckedEnv::new("MIRRORD_AGENT_POD_
 ///
 /// Should follow `tracing` format, e.g `mirrord=trace`.
 pub const LOG_LEVEL: CheckedEnv<String> = CheckedEnv::new("RUST_LOG");
+
+/// Provides the agent with a steal TLS configuration.
+pub const STEAL_TLS_CONFIG: CheckedEnv<StealTlsConfig> =
+    CheckedEnv::new("MIRRORD_AGENT_STEAL_TLS_CONFIG");

mirrord/agent/env/src/steal_tls.rs

@@ -4,7 +4,7 @@ use serde::{Deserialize, Serialize};
 
 pub type StealTlsConfig = HashMap<u16, StealPortTlsConfig>;
 
-#[derive(Deserialize, Serialize)]
+#[derive(Deserialize, Serialize, Clone, Debug)]
 pub struct StealPortTlsConfig {
     pub server_cert_pem: PathBuf,
     pub server_key_pem: PathBuf,

mirrord/agent/src/steal/tls.rs

@@ -1,3 +1,5 @@
+#![allow(unused)]
+
 use std::{
     collections::HashMap,
     io,

mirrord/cli/src/connection.rs

@@ -173,7 +173,7 @@ where
 
     let agent_connect_info = tokio::time::timeout(
         Duration::from_secs(config.agent.startup_timeout),
-        k8s_api.create_agent(progress, &config.target, Some(config), None, None),
+        k8s_api.create_agent(progress, &config.target, Some(config), None, None, None),
     )
     .await
     .unwrap_or(Err(KubeApiError::AgentReadyTimeout))

mirrord/kube/src/api/container.rs

@@ -1,7 +1,7 @@
 use std::{collections::HashSet, net::IpAddr, sync::LazyLock};
 
 use k8s_openapi::api::core::v1::{ContainerStatus, Pod};
-use mirrord_agent_env::mesh::MeshVendor;
+use mirrord_agent_env::{mesh::MeshVendor, steal_tls::StealTlsConfig};
 use mirrord_config::agent::AgentConfig;
 use mirrord_progress::Progress;
 use rand::{
@@ -46,6 +46,7 @@ pub struct ContainerParams {
     pub pod_ips: Option<Vec<IpAddr>>,
     /// Support IPv6-only clusters
     pub support_ipv6: bool,
+    pub steal_tls_config: Option<StealTlsConfig>,
 }
 
 impl ContainerParams {
@@ -54,6 +55,7 @@ impl ContainerParams {
         pod_ips: Option<Vec<IpAddr>>,
         support_ipv6: bool,
         port: Option<u16>,
+        steal_tls_config: Option<StealTlsConfig>,
     ) -> ContainerParams {
         let port = port.unwrap_or_else(|| rand::thread_rng().gen_range(30000..=65535));
         let gid: u16 = rand::thread_rng().gen_range(3000..u16::MAX);
@@ -72,6 +74,7 @@ impl ContainerParams {
             tls_cert,
             pod_ips,
             support_ipv6,
+            steal_tls_config,
         }
     }
 }

mirrord/kube/src/api/container/job.rs

@@ -250,6 +250,7 @@ mod test {
             tls_cert: None,
             pod_ips: None,
             support_ipv6,
+            steal_tls_config: None,
         };
 
         let update = JobVariant::new(&agent, &params).as_update();
@@ -338,6 +339,7 @@ mod test {
             tls_cert: None,
             pod_ips: None,
             support_ipv6,
+            steal_tls_config: None,
         };
 
         let update = JobTargetedVariant::new(

mirrord/kube/src/api/container/util.rs

@@ -33,7 +33,7 @@ pub(super) fn get_capabilities(agent: &AgentConfig) -> Vec<LinuxCapability> {
 }
 
 /// Builds mirrord agent environment variables.
-pub(super) fn agent_env(agent: &AgentConfig, params: &&ContainerParams) -> Vec<EnvVar> {
+pub(super) fn agent_env(agent: &AgentConfig, params: &ContainerParams) -> Vec<EnvVar> {
     let mut env = vec![
         envs::LOG_LEVEL.as_k8s_spec(&agent.log_level),
         envs::STEALER_FLUSH_CONNECTIONS.as_k8s_spec(&agent.flush_connections),
@@ -66,6 +66,10 @@ pub(super) fn agent_env(agent: &AgentConfig, params: &&ContainerParams) -> Vec<E
         env.push(envs::OPERATOR_CERT.as_k8s_spec(cert));
     }
 
+    if let Some(config) = &params.steal_tls_config {
+        env.push(envs::STEAL_TLS_CONFIG.as_k8s_spec(config));
+    }
+
     env
 }
 

mirrord/kube/src/api/kubernetes.rs

@@ -6,7 +6,7 @@ use kube::{
     config::{KubeConfigOptions, Kubeconfig},
     Api, Client, Config, Discovery,
 };
-use mirrord_agent_env::mesh::MeshVendor;
+use mirrord_agent_env::{mesh::MeshVendor, steal_tls::StealTlsConfig};
 use mirrord_config::{
     agent::AgentConfig,
     target::{Target, TargetConfig},
@@ -176,13 +176,15 @@ impl KubernetesAPI {
     ///   creating an agent from the operator. In usage from this repo this is always `None`.
     /// * `agent_port` - port number on which the agent will listen for client connections. If
     ///   [`None`] is given, a random high port will be user.
+    /// * `steal_tls_config` - optional config to enable filtered stealing of TLS traffic.
     #[tracing::instrument(level = "trace", skip(self), ret, err)]
     pub async fn create_agent_params(
         &self,
         target: &TargetConfig,
         tls_cert: Option<String>,
         support_ipv6: bool,
         agent_port: Option<u16>,
+        steal_tls_config: Option<StealTlsConfig>,
     ) -> Result<(ContainerParams, Option<RuntimeData>), KubeApiError> {
         let runtime_data = match target.path.as_ref().unwrap_or(&Target::Targetless) {
             Target::Targetless => None,
@@ -197,7 +199,13 @@ impl KubernetesAPI {
             .map(|runtime_data| runtime_data.pod_ips.clone())
             .filter(|pod_ips| !pod_ips.is_empty());
 
-        let params = ContainerParams::new(tls_cert, pod_ips, support_ipv6, agent_port);
+        let params = ContainerParams::new(
+            tls_cert,
+            pod_ips,
+            support_ipv6,
+            agent_port,
+            steal_tls_config,
+        );
 
         Ok((params, runtime_data))
     }
@@ -209,6 +217,7 @@ impl KubernetesAPI {
     ///   creating an agent from the operator. In usage from this repo this is always `None`.
     /// * `agent_port` - port number on which the agent will listen for client connections. If
     ///   [`None`] is given, a random high port will be used.
+    /// * `steal_tls_config` - optional config to enable filtered stealing of TLS traffic.
     #[tracing::instrument(level = "trace", skip(self, progress))]
     pub async fn create_agent<P>(
         &self,
@@ -217,6 +226,7 @@ impl KubernetesAPI {
         config: Option<&LayerConfig>,
         tls_cert: Option<String>,
         agent_port: Option<u16>,
+        steal_tls_config: Option<StealTlsConfig>,
     ) -> Result<AgentKubernetesConnectInfo, KubeApiError>
     where
         P: Progress + Send + Sync,
@@ -225,7 +235,7 @@ impl KubernetesAPI {
             .map(|layer_conf| layer_conf.feature.network.ipv6)
             .unwrap_or_default();
         let (params, runtime_data) = self
-            .create_agent_params(target, tls_cert, support_ipv6, agent_port)
+            .create_agent_params(target, tls_cert, support_ipv6, agent_port, steal_tls_config)
             .await?;
         if let Some(RuntimeData {
             guessed_container: true,