rws_gateway_api/gateway_api/mesh/
inject.rs

1//! Sidecar injection: a mutating admission webhook the controller serves
2//! (over TLS, the certificate in `RWS_CONFIG_GATEWAY_API_MESH_WEBHOOK_CERT_FILE`
3//! / `_KEY_FILE`). Pods created in the namespaces the
4//! MutatingWebhookConfiguration selects get two init containers, first:
5//!
6//! - `rws-mesh-init` ([`InjectSettings::init_image`]): sets the pod's
7//!   iptables rules — outbound TCP to the sidecar's port, except the
8//!   sidecar's own (UID [`SIDECAR_UID`](super::sidecar::SIDECAR_UID)) and
9//!   loopback. Runs as root with `NET_ADMIN` and `NET_RAW`, then exits.
10//! - `rws-mesh`: rws in `mesh-sidecar` mode, a native sidecar
11//!   (`restartPolicy: Always`, Kubernetes ≥ 1.29) — started before the
12//!   app's containers and kept running beside them.
13//!
14//! A pod opts out with the annotation `mesh.rws8.tech/inject: "false"`;
15//! `hostNetwork` pods and pods that already have the sidecar are left alone.
16//! The webhook allows every request: injection never blocks a pod.
17
18use serde_json::{json, Value};
19
20use crate::application::Application;
21use crate::request::Request;
22use crate::response::{Response, STATUS_CODE_REASON_PHRASE};
23use crate::server::ConnectionInfo;
24
25use super::sidecar::{OUTBOUND_PORT, SIDECAR_UID};
26
27/// Where the webhook answers.
28pub const PATH: &str = "/mesh/v1/inject";
29/// The default port of the webhook.
30pub const DEFAULT_PORT: u16 = 15017;
31/// The opt-out annotation (`"false"`).
32pub const INJECT_ANNOTATION: &str = "mesh.rws8.tech/inject";
33/// Set on injected pods.
34pub const INJECTED_ANNOTATION: &str = "mesh.rws8.tech/injected";
35/// The sidecar container's name.
36pub const SIDECAR_NAME: &str = "rws-mesh";
37/// The init container's name.
38pub const INIT_NAME: &str = "rws-mesh-init";
39/// The sidecar's health port (its rws server, for the probes).
40pub const SIDECAR_HEALTH_PORT: u16 = 15000;
41
42/// What injected containers run.
43#[derive(Clone, Debug)]
44pub struct InjectSettings {
45    /// The sidecar's image (the rws image).
46    pub sidecar_image: String,
47    /// The image setting the iptables rules.
48    pub init_image: String,
49    /// The controller's configuration endpoint, for the sidecar.
50    pub config_url: String,
51}
52
53/// The JSON patch injecting the sidecar into `pod`, or `None` when the pod
54/// is left as it is.
55pub fn patch(pod: &Value, settings: &InjectSettings) -> Option<Vec<Value>> {
56    let annotations = &pod["metadata"]["annotations"];
57    if annotations[INJECT_ANNOTATION].as_str().is_some_and(|v| v.eq_ignore_ascii_case("false")) {
58        return None;
59    }
60    if pod["spec"]["hostNetwork"].as_bool() == Some(true) {
61        return None;
62    }
63    let named = |list: &Value, name: &str| list.as_array().is_some_and(|l| l.iter().any(|c| c["name"] == name));
64    if named(&pod["spec"]["initContainers"], SIDECAR_NAME) || named(&pod["spec"]["containers"], SIDECAR_NAME) {
65        return None;
66    }
67    let (init, sidecar) = (init_container(settings), sidecar_container(settings));
68    let mut ops = Vec::new();
69    if pod["spec"]["initContainers"].as_array().is_some_and(|l| !l.is_empty()) {
70        // Insert at the front, sidecar first so the init container ends up before it.
71        ops.push(json!({"op": "add", "path": "/spec/initContainers/0", "value": sidecar}));
72        ops.push(json!({"op": "add", "path": "/spec/initContainers/0", "value": init}));
73    } else {
74        ops.push(json!({"op": "add", "path": "/spec/initContainers", "value": [init, sidecar]}));
75    }
76    if annotations.is_object() {
77        ops.push(json!({"op": "add", "path": format!("/metadata/annotations/{}", pointer_escape(INJECTED_ANNOTATION)), "value": "true"}));
78    } else {
79        ops.push(json!({"op": "add", "path": "/metadata/annotations", "value": {INJECTED_ANNOTATION: "true"}}));
80    }
81    Some(ops)
82}
83
84/// A JSON Pointer token (RFC 6901): `~` → `~0`, `/` → `~1`.
85fn pointer_escape(token: &str) -> String {
86    token.replace('~', "~0").replace('/', "~1")
87}
88
89fn init_container(settings: &InjectSettings) -> Value {
90    json!({
91        "name": INIT_NAME,
92        "image": settings.init_image,
93        "env": [
94            {"name": "RWS_MESH_OUTBOUND_PORT", "value": OUTBOUND_PORT.to_string()},
95            {"name": "RWS_MESH_UID", "value": SIDECAR_UID.to_string()}
96        ],
97        "securityContext": {
98            "runAsUser": 0,
99            "runAsNonRoot": false,
100            "allowPrivilegeEscalation": false,
101            "capabilities": {"add": ["NET_ADMIN", "NET_RAW"], "drop": ["ALL"]}
102        },
103        "resources": {"requests": {"cpu": "10m", "memory": "16Mi"}, "limits": {"memory": "64Mi"}}
104    })
105}
106
107fn sidecar_container(settings: &InjectSettings) -> Value {
108    let probe = json!({"httpGet": {"path": "/healthz", "port": SIDECAR_HEALTH_PORT}, "periodSeconds": 1, "failureThreshold": 60});
109    json!({
110        "name": SIDECAR_NAME,
111        "image": settings.sidecar_image,
112        "restartPolicy": "Always",
113        "env": [
114            {"name": "RWS_CONFIG_GATEWAY_API_MODE", "value": "mesh-sidecar"},
115            {"name": "RWS_CONFIG_GATEWAY_API_MESH_CONFIG_URL", "value": settings.config_url},
116            {"name": "RWS_CONFIG_PORT", "value": SIDECAR_HEALTH_PORT.to_string()},
117            {"name": "RWS_CONFIG_IP", "value": "0.0.0.0"},
118            {"name": "RWS_CONFIG_LOG_FORMAT", "value": "off"},
119            {"name": "POD_NAMESPACE", "valueFrom": {"fieldRef": {"fieldPath": "metadata.namespace"}}}
120        ],
121        "startupProbe": probe,
122        "securityContext": {
123            "runAsUser": SIDECAR_UID,
124            "runAsGroup": SIDECAR_UID,
125            "runAsNonRoot": true,
126            "allowPrivilegeEscalation": false,
127            "readOnlyRootFilesystem": true,
128            "capabilities": {"drop": ["ALL"]}
129        },
130        "resources": {"requests": {"cpu": "10m", "memory": "32Mi"}, "limits": {"memory": "256Mi"}}
131    })
132}
133
134/// Answer an `AdmissionReview` (`admission.k8s.io/v1`): allowed, with the
135/// patch when the pod gets the sidecar. `Err` for a body that isn't one.
136pub fn review(body: &[u8], settings: &InjectSettings) -> Result<Vec<u8>, String> {
137    let review: Value = serde_json::from_slice(body).map_err(|e| format!("not an AdmissionReview: {e}"))?;
138    let request = &review["request"];
139    let uid = request["uid"].as_str().ok_or("an AdmissionReview without request.uid")?;
140    let mut response = json!({"uid": uid, "allowed": true});
141    let is_pod_create = request["kind"]["kind"] == "Pod" && request["operation"] == "CREATE";
142    if is_pod_create {
143        if let Some(ops) = patch(&request["object"], settings) {
144            let patch = serde_json::to_vec(&ops).map_err(|e| e.to_string())?;
145            response["patchType"] = json!("JSONPatch");
146            response["patch"] = json!(crate::core::base64::Base64::encode(&patch).map_err(|e| e.to_string())?);
147        }
148    }
149    let api_version = review["apiVersion"].as_str().unwrap_or("admission.k8s.io/v1");
150    serde_json::to_vec(&json!({"apiVersion": api_version, "kind": "AdmissionReview", "response": response})).map_err(|e| e.to_string())
151}
152
153/// The webhook as an [`Application`].
154#[derive(Clone)]
155pub struct Injector {
156    pub settings: InjectSettings,
157}
158
159impl Application for Injector {
160    fn execute(&self, request: &Request, _connection: &ConnectionInfo) -> Result<Response, String> {
161        if request.method != "POST" || request.request_uri.split('?').next() != Some(PATH) {
162            return Ok(Response::text(STATUS_CODE_REASON_PHRASE.n404_not_found, "not found"));
163        }
164        Ok(match review(&request.body, &self.settings) {
165            Ok(body) => {
166                let mut response = Response::text(STATUS_CODE_REASON_PHRASE.n200_ok, &String::from_utf8_lossy(&body));
167                response.headers.retain(|h| !h.name.eq_ignore_ascii_case("content-type"));
168                response.headers.push(crate::header::Header::new("Content-Type", "application/json"));
169                response
170            }
171            Err(e) => Response::text(STATUS_CODE_REASON_PHRASE.n400_bad_request, &e),
172        })
173    }
174}
175
176/// Serve the webhook over TLS on `port` until the runtime stops.
177pub async fn serve(injector: Injector, port: u16, cert_file: String, key_file: String) {
178    let acceptor = match crate::tls::create_tls_acceptor(&cert_file, &key_file) {
179        Ok(acceptor) => acceptor,
180        Err(e) => {
181            eprintln!("[gateway-api] mesh: can't serve the injection webhook: {e}");
182            return;
183        }
184    };
185    let listener = match tokio::net::TcpListener::bind(("0.0.0.0", port)).await {
186        Ok(listener) => listener,
187        Err(e) => {
188            eprintln!("[gateway-api] mesh: can't serve the injection webhook on port {port}: {e}");
189            return;
190        }
191    };
192    loop {
193        let Ok((stream, peer)) = listener.accept().await else {
194            tokio::time::sleep(std::time::Duration::from_millis(50)).await;
195            continue;
196        };
197        tokio::spawn(crate::server::serve_tls_connection(stream, peer, acceptor.clone(), injector.clone()));
198    }
199}
rws_gateway_api/gateway_api/mesh/inject.rs

rws_gateway_api/gateway_api/mesh/
inject.rs
