rws_k8s/k8s/
webhook.rs

1//! Admission and conversion webhooks as an rws application: validating
2//! (allow or deny a create/update/delete), mutating (change the object; the
3//! JSON patch is computed for you) and CRD conversion between versions —
4//! plus the TLS certificate the API server needs to trust them.
5//!
6//! ```rust,no_run
7//! use rws_k8s::k8s::webhook::{Webhooks, WebhookCertificate};
8//! # #[derive(rws_k8s::k8s::kube::CustomResource, serde::Deserialize, serde::Serialize, Clone, Debug, rws_k8s::k8s::schemars::JsonSchema)]
9//! # #[kube(group = "example.com", version = "v1", kind = "Backup", namespaced)]
10//! # pub struct BackupSpec { pub schedule: String }
11//!
12//! let webhooks = Webhooks::new()
13//!     .validate::<Backup, _>("/validate-backups", |request| {
14//!         match &request.object {
15//!             Some(b) if b.spec.schedule.is_empty() => Err("spec.schedule is required".into()),
16//!             _ => Ok(()),
17//!         }
18//!     })
19//!     .mutate::<Backup, _>("/mutate-backups", |mut backup| {
20//!         backup.labels_mut().entry("backups.example.com/managed".into()).or_insert("true".into());
21//!         Ok(backup)
22//!     });
23//!
24//! // A certificate for <service>.<namespace>.svc, served by run_tls.
25//! let cert = WebhookCertificate::generate("backup-operator", "backups").unwrap();
26//! cert.install("/tmp/webhook-tls").unwrap();   // sets RWS_CONFIG_TLS_CERT_FILE / KEY_FILE
27//! # use rws_k8s::k8s::kube::ResourceExt;
28//! ```
29//!
30//! Requests to other paths fall through to the wrapped app (`.wrap(app)`),
31//! so one server can answer webhooks, probes and metrics. A request that
32//! isn't an `AdmissionReview` gets `400`; a handler's `Err` denies with
33//! that message (`kubectl` shows it).
34
35use std::sync::Arc;
36
37use kube::core::admission::{AdmissionRequest, AdmissionResponse, AdmissionReview, Operation};
38use kube::core::conversion::{ConversionRequest, ConversionResponse, ConversionReview};
39use kube::core::Status;
40use kube::Resource;
41use serde::de::DeserializeOwned;
42use serde::Serialize;
43
44use rws_server::application::Application;
45use rws_server::request::Request;
46use rws_server::response::{Response, STATUS_CODE_REASON_PHRASE};
47use rws_server::server::ConnectionInfo;
48
49pub use kube::core::admission::Operation as AdmissionOperation;
50
51type Handler = Arc<dyn Fn(&[u8]) -> Result<Vec<u8>, String> + Send + Sync>;
52
53/// Webhook routes, an [`Application`](rws_server::application::Application).
54#[derive(Clone, Default)]
55pub struct Webhooks {
56    routes: Vec<(String, Handler)>,
57    fallback: Option<Arc<dyn Application + Send + Sync>>,
58}
59
60fn json_response(body: Vec<u8>) -> Response {
61    Response::json(STATUS_CODE_REASON_PHRASE.n200_ok, body)
62}
63
64/// The AdmissionReview answer to `body` from `decide`, which gets the
65/// request and answers `Ok(None)` (allow), `Ok(Some(patch))` (allow with a
66/// patch) or `Err(message)` (deny).
67pub fn admission<K, F>(body: &[u8], decide: F) -> Result<Vec<u8>, String>
68where
69    K: Resource + DeserializeOwned + Serialize + Clone,
70    F: Fn(&AdmissionRequest<K>) -> Result<Option<json_patch::Patch>, String>,
71{
72    let review: AdmissionReview<K> = serde_json::from_slice(body).map_err(|e| format!("not an AdmissionReview: {e}"))?;
73    let request: AdmissionRequest<K> = review.try_into().map_err(|_| "the AdmissionReview has no request".to_string())?;
74    let response = match decide(&request) {
75        Ok(None) => AdmissionResponse::from(&request),
76        Ok(Some(patch)) => AdmissionResponse::from(&request).with_patch(patch).map_err(|e| e.to_string())?,
77        Err(message) => AdmissionResponse::from(&request).deny(message),
78    };
79    // kube serializes the patch's bytes as a JSON array of numbers; the
80    // API's wire form for bytes is base64, as Go's encoding/json writes it.
81    let mut review = serde_json::to_value(response.into_review()).map_err(|e| e.to_string())?;
82    if let Some(bytes) = review["response"]["patch"].as_array().map(|a| a.iter().filter_map(|b| b.as_u64().map(|b| b as u8)).collect::<Vec<u8>>()) {
83        review["response"]["patch"] = serde_json::Value::String(super::manifests::base64(&bytes));
84    }
85    serde_json::to_vec(&review).map_err(|e| e.to_string())
86}
87
88/// The ConversionReview answer to `body`: each object converted to the
89/// desired `apiVersion` by `convert(object, desired_api_version)`.
90pub fn conversion<F>(body: &[u8], convert: F) -> Result<Vec<u8>, String>
91where
92    F: Fn(serde_json::Value, &str) -> Result<serde_json::Value, String>,
93{
94    let review: ConversionReview = serde_json::from_slice(body).map_err(|e| format!("not a ConversionReview: {e}"))?;
95    let request = ConversionRequest::from_review(review).map_err(|_| "the ConversionReview has no request".to_string())?;
96    let desired = request.desired_api_version.clone();
97    let objects = request.objects.clone();
98    let response = ConversionResponse::for_request(request);
99    let mut converted = Vec::with_capacity(objects.len());
100    let mut failure = None;
101    for object in objects {
102        match convert(object, &desired) {
103            Ok(mut o) => {
104                o["apiVersion"] = serde_json::Value::String(desired.clone());
105                converted.push(o);
106            }
107            Err(e) => {
108                failure = Some(e);
109                break;
110            }
111        }
112    }
113    let response = match failure {
114        None => response.success(converted),
115        Some(e) => response.failure(Status::failure(&e, "ConversionFailed")),
116    };
117    serde_json::to_vec(&response.into_review()).map_err(|e| e.to_string())
118}
119
120impl Webhooks {
121    pub fn new() -> Self {
122        Webhooks::default()
123    }
124
125    /// A validating webhook at `path`: `check` gets the request (its
126    /// `operation`, `object`, `old_object`, `user_info`) and answers `Ok`
127    /// to allow or `Err(message)` to deny.
128    pub fn validate<K, F>(mut self, path: &str, check: F) -> Self
129    where
130        K: Resource + DeserializeOwned + Serialize + Clone + 'static,
131        F: Fn(&AdmissionRequest<K>) -> Result<(), String> + Send + Sync + 'static,
132    {
133        self.routes.push((path.to_string(), Arc::new(move |body| admission::<K, _>(body, |r| check(r).map(|_| None)))));
134        self
135    }
136
137    /// A mutating webhook at `path`: `change` gets the object (on create and
138    /// update) and returns it changed; the JSON patch from the old to the
139    /// new is what the API server applies. `Err` denies.
140    pub fn mutate<K, F>(mut self, path: &str, change: F) -> Self
141    where
142        K: Resource + DeserializeOwned + Serialize + Clone + 'static,
143        F: Fn(K) -> Result<K, String> + Send + Sync + 'static,
144    {
145        self.routes.push((
146            path.to_string(),
147            Arc::new(move |body| {
148                admission::<K, _>(body, |request| {
149                    let Some(object) = request.object.clone() else { return Ok(None) };
150                    if !matches!(request.operation, Operation::Create | Operation::Update) {
151                        return Ok(None);
152                    }
153                    let before = serde_json::to_value(&object).map_err(|e| e.to_string())?;
154                    let after = serde_json::to_value(change(object)?).map_err(|e| e.to_string())?;
155                    let patch = json_patch::diff(&before, &after);
156                    Ok((!patch.0.is_empty()).then_some(patch))
157                })
158            }),
159        ));
160        self
161    }
162
163    /// A CRD conversion webhook at `path` (see
164    /// [`manifests::with_conversion_webhook`](super::manifests::with_conversion_webhook)):
165    /// `convert(object, desired_api_version)` returns the object in that
166    /// version (its `apiVersion` is set for you).
167    pub fn convert<F>(mut self, path: &str, convert: F) -> Self
168    where
169        F: Fn(serde_json::Value, &str) -> Result<serde_json::Value, String> + Send + Sync + 'static,
170    {
171        self.routes.push((path.to_string(), Arc::new(move |body| conversion(body, &convert))));
172        self
173    }
174
175    /// Requests to other paths go to `app`.
176    pub fn wrap(mut self, app: impl Application + Send + Sync + 'static) -> Self {
177        self.fallback = Some(Arc::new(app));
178        self
179    }
180}
181
182impl Application for Webhooks {
183    fn execute(&self, request: &Request, connection: &ConnectionInfo) -> Result<Response, String> {
184        let path = request.request_uri.split('?').next().unwrap_or("");
185        if let Some((_, handler)) = self.routes.iter().find(|(p, _)| p == path) {
186            if request.method != "POST" {
187                return Ok(Response::text(STATUS_CODE_REASON_PHRASE.n405_method_not_allowed, "POST an AdmissionReview or a ConversionReview"));
188            }
189            return Ok(match handler(&request.body) {
190                Ok(body) => json_response(body),
191                Err(message) => Response::text(STATUS_CODE_REASON_PHRASE.n400_bad_request, &message),
192            });
193        }
194        match &self.fallback {
195            Some(app) => app.execute(request, connection),
196            None => {
197                use rws_server::core::New;
198                rws_server::app::App::new().execute(request, connection)
199            }
200        }
201    }
202}
203
204/// A CA and a serving certificate for `<service>.<namespace>.svc` (and the
205/// short names): the API server checks the webhook's certificate against
206/// the CA in the webhook configuration's `caBundle`.
207#[derive(Clone)]
208pub struct WebhookCertificate {
209    pub ca_pem: String,
210    pub cert_pem: String,
211    pub key_pem: String,
212}
213
214impl WebhookCertificate {
215    /// A new CA and a certificate signed by it for the Service `service`
216    /// in `namespace`, valid for ten years.
217    pub fn generate(service: &str, namespace: &str) -> Result<Self, String> {
218        let names = vec![
219            service.to_string(),
220            format!("{service}.{namespace}"),
221            format!("{service}.{namespace}.svc"),
222            format!("{service}.{namespace}.svc.cluster.local"),
223        ];
224        Self::generate_for(&format!("{service}.{namespace}.svc"), names)
225    }
226
227    /// The same for other DNS names (a webhook the API server reaches by
228    /// URL rather than through a Service).
229    pub fn generate_for(common_name: &str, names: Vec<String>) -> Result<Self, String> {
230        use rcgen::{BasicConstraints, CertificateParams, DnType, IsCa, KeyPair, KeyUsagePurpose};
231        let ca_key = KeyPair::generate().map_err(|e| e.to_string())?;
232        let mut ca_params = CertificateParams::new(Vec::<String>::new()).map_err(|e| e.to_string())?;
233        ca_params.is_ca = IsCa::Ca(BasicConstraints::Unconstrained);
234        ca_params.distinguished_name.push(DnType::CommonName, format!("{common_name} webhook CA"));
235        ca_params.key_usages = vec![KeyUsagePurpose::KeyCertSign, KeyUsagePurpose::CrlSign];
236        ten_years(&mut ca_params);
237        let ca = ca_params.self_signed(&ca_key).map_err(|e| e.to_string())?;
238        let issuer = rcgen::Issuer::new(ca_params, ca_key);
239        let key = KeyPair::generate().map_err(|e| e.to_string())?;
240        let mut params = CertificateParams::new(names).map_err(|e| e.to_string())?;
241        params.distinguished_name.push(DnType::CommonName, common_name.to_string());
242        ten_years(&mut params);
243        let cert = params.signed_by(&key, &issuer).map_err(|e| e.to_string())?;
244        Ok(WebhookCertificate { ca_pem: ca.pem(), cert_pem: cert.pem(), key_pem: key.serialize_pem() })
245    }
246
247    /// The certificate every replica of an operator uses: read from the
248    /// TLS Secret `secret` in `namespace`, or made and stored there by the
249    /// first replica to ask (a create race is settled by the API server:
250    /// the loser reads the winner's). Replicas that each made their own
251    /// would overwrite each other's `caBundle`. The pod needs `get` and
252    /// `create` on `secrets` in the namespace.
253    pub async fn shared(client: &kube::Client, namespace: &str, secret: &str, service: &str) -> super::Result<Self> {
254        use k8s_openapi::api::core::v1::Secret;
255        use kube::api::{Api, PostParams};
256        let api: Api<Secret> = Api::namespaced(client.clone(), namespace);
257        if let Some(existing) = api.get_opt(secret).await? {
258            return Self::from_secret(&existing).ok_or_else(|| super::Error::Other(format!("secret {namespace}/{secret} has no tls.crt, tls.key and ca.crt")));
259        }
260        let cert = Self::generate(service, namespace)?;
261        match api.create(&PostParams::default(), &cert.to_secret(secret, namespace)).await {
262            Ok(_) => Ok(cert),
263            Err(kube::Error::Api(status)) if status.is_already_exists() => {
264                let existing = api.get(secret).await?;
265                Self::from_secret(&existing).ok_or_else(|| super::Error::Other(format!("secret {namespace}/{secret} has no tls.crt, tls.key and ca.crt")))
266            }
267            Err(e) => Err(e.into()),
268        }
269    }
270
271    /// A `kubernetes.io/tls` Secret holding it (`tls.crt`, `tls.key`, `ca.crt`).
272    pub fn to_secret(&self, name: &str, namespace: &str) -> k8s_openapi::api::core::v1::Secret {
273        use k8s_openapi::ByteString;
274        k8s_openapi::api::core::v1::Secret {
275            metadata: k8s_openapi::apimachinery::pkg::apis::meta::v1::ObjectMeta { name: Some(name.to_string()), namespace: Some(namespace.to_string()), ..Default::default() },
276            type_: Some("kubernetes.io/tls".to_string()),
277            data: Some(std::collections::BTreeMap::from([
278                ("tls.crt".to_string(), ByteString(self.cert_pem.clone().into_bytes())),
279                ("tls.key".to_string(), ByteString(self.key_pem.clone().into_bytes())),
280                ("ca.crt".to_string(), ByteString(self.ca_pem.clone().into_bytes())),
281            ])),
282            ..Default::default()
283        }
284    }
285
286    /// Read one from a Secret written by [`to_secret`](Self::to_secret)
287    /// (or cert-manager, whose Secrets have the same keys).
288    pub fn from_secret(secret: &k8s_openapi::api::core::v1::Secret) -> Option<Self> {
289        let data = secret.data.as_ref()?;
290        let text = |key: &str| data.get(key).and_then(|v| String::from_utf8(v.0.clone()).ok()).filter(|t| !t.is_empty());
291        Some(WebhookCertificate { cert_pem: text("tls.crt")?, key_pem: text("tls.key")?, ca_pem: text("ca.crt")? })
292    }
293
294    /// Write `tls.crt`, `tls.key` and `ca.crt` into `dir` and point rws's
295    /// TLS settings at them (`RWS_CONFIG_TLS_CERT_FILE`/`KEY_FILE`), so
296    /// `Server::run_tls` serves them.
297    pub fn install(&self, dir: &str) -> Result<(), String> {
298        std::fs::create_dir_all(dir).map_err(|e| format!("{dir}: {e}"))?;
299        let path = |f: &str| format!("{}/{f}", dir.trim_end_matches('/'));
300        std::fs::write(path("tls.crt"), &self.cert_pem).map_err(|e| e.to_string())?;
301        std::fs::write(path("tls.key"), &self.key_pem).map_err(|e| e.to_string())?;
302        std::fs::write(path("ca.crt"), &self.ca_pem).map_err(|e| e.to_string())?;
303        // SAFETY of set_var: settings are read through ServerConfig snapshots.
304        unsafe {
305            std::env::set_var("RWS_CONFIG_TLS_CERT_FILE", path("tls.crt"));
306            std::env::set_var("RWS_CONFIG_TLS_KEY_FILE", path("tls.key"));
307        }
308        rws_server::server_config::invalidate();
309        Ok(())
310    }
311
312    /// Put the CA into the `caBundle` of every webhook in the
313    /// Validating/MutatingWebhookConfiguration `name` (and a CRD's
314    /// conversion webhook, when `crd` is given) — for an operator that
315    /// makes its own certificate at startup.
316    pub async fn patch_ca_bundle(&self, client: &kube::Client, validating: Option<&str>, mutating: Option<&str>, crd: Option<&str>) -> super::Result<()> {
317        use k8s_openapi::api::admissionregistration::v1::{MutatingWebhookConfiguration, ValidatingWebhookConfiguration};
318        use k8s_openapi::apiextensions_apiserver::pkg::apis::apiextensions::v1::CustomResourceDefinition;
319        use kube::api::{Api, Patch, PatchParams};
320        let bundle = super::manifests::base64(self.ca_pem.as_bytes());
321        let params = PatchParams::default();
322        if let Some(name) = validating {
323            let api: Api<ValidatingWebhookConfiguration> = Api::all(client.clone());
324            let current = api.get(name).await?;
325            let hooks: Vec<serde_json::Value> = current.webhooks.unwrap_or_default().iter().map(|w| serde_json::json!({ "name": w.name, "clientConfig": { "caBundle": bundle } })).collect();
326            api.patch(name, &params, &Patch::Strategic(serde_json::json!({ "webhooks": hooks }))).await?;
327        }
328        if let Some(name) = mutating {
329            let api: Api<MutatingWebhookConfiguration> = Api::all(client.clone());
330            let current = api.get(name).await?;
331            let hooks: Vec<serde_json::Value> = current.webhooks.unwrap_or_default().iter().map(|w| serde_json::json!({ "name": w.name, "clientConfig": { "caBundle": bundle } })).collect();
332            api.patch(name, &params, &Patch::Strategic(serde_json::json!({ "webhooks": hooks }))).await?;
333        }
334        if let Some(name) = crd {
335            let api: Api<CustomResourceDefinition> = Api::all(client.clone());
336            let patch = serde_json::json!({ "spec": { "conversion": { "webhook": { "clientConfig": { "caBundle": bundle } } } } });
337            api.patch(name, &params, &Patch::Merge(&patch)).await?;
338        }
339        Ok(())
340    }
341}
342
343fn ten_years(params: &mut rcgen::CertificateParams) {
344    let now = std::time::SystemTime::now();
345    params.not_before = now.into();
346    params.not_after = (now + std::time::Duration::from_secs(10 * 365 * 86_400)).into();
347}
348
349#[cfg(test)]
350mod tests;
rws_k8s/k8s/webhook.rs

rws_k8s/k8s/
webhook.rs
