diff --git a/Cargo.toml b/Cargo.toml index 9fd97931..257a5ab1 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -4,7 +4,7 @@ # SPDX-License-Identifier: CC0-1.0 [workspace] -members = ["attestation-key-register", "compute-pcrs", "lib", "operator", "register-server", "test_utils", "tests"] +members = ["attestation-key-register", "compute-pcrs", "kbs-event-proxy", "lib", "operator", "register-server", "test_utils", "tests"] resolver = "3" [workspace.package] diff --git a/Containerfile b/Containerfile index c2971a6d..a7d8bfb0 100644 --- a/Containerfile +++ b/Containerfile @@ -28,8 +28,10 @@ COPY register-server/Cargo.toml register-server/ COPY register-server/src/lib.rs register-server/src/ COPY attestation-key-register/Cargo.toml attestation-key-register/ COPY attestation-key-register/src/lib.rs attestation-key-register/src/ +COPY kbs-event-proxy/Cargo.toml kbs-event-proxy/ +COPY kbs-event-proxy/src/main.rs kbs-event-proxy/src/ -RUN sed -i 's/members = .*/members = ["lib", "operator", "compute-pcrs", "register-server", "attestation-key-register"]/' Cargo.toml && \ +RUN sed -i 's/members = .*/members = ["lib", "operator", "compute-pcrs", "register-server", "attestation-key-register", "kbs-event-proxy"]/' Cargo.toml && \ sed -i '/\[dev-dependencies\]/,$d' operator/Cargo.toml && \ sed -i '/\[dev-dependencies\]/,$d' register-server/Cargo.toml && \ sed -i '/trusted-cluster-operator-test-utils/d' lib/Cargo.toml @@ -44,13 +46,14 @@ RUN --mount=type=cache,target=/build/target \ RUN --mount=type=cache,target=/build/target \ --mount=type=cache,target=/usr/local/cargo/registry \ if [ "$build_type" = debug ]; then \ - cargo build -p operator -p compute-pcrs -p register-server -p attestation-key-register; \ + cargo build -p operator -p compute-pcrs -p register-server -p attestation-key-register -p kbs-event-proxy; \ fi COPY operator/src operator/src COPY compute-pcrs/src compute-pcrs/src COPY register-server/src register-server/src COPY attestation-key-register/src attestation-key-register/src +COPY kbs-event-proxy/src kbs-event-proxy/src RUN --mount=type=cache,target=/build/target \ --mount=type=cache,target=/usr/local/cargo/registry \ @@ -61,6 +64,7 @@ RUN --mount=type=cache,target=/build/target \ -p compute-pcrs \ -p register-server \ -p attestation-key-register \ + -p kbs-event-proxy \ $release_flag RUN --mount=type=cache,target=/build/target \ @@ -70,7 +74,8 @@ RUN --mount=type=cache,target=/build/target \ cp /build/target/${profile_dir}/operator /output/ && \ cp /build/target/${profile_dir}/compute-pcrs /output/ && \ cp /build/target/${profile_dir}/register-server /output/ && \ - cp /build/target/${profile_dir}/attestation-key-register /output/ + cp /build/target/${profile_dir}/attestation-key-register /output/ && \ + cp /build/target/${profile_dir}/kbs-event-proxy /output/ # Distribution stages FROM ${deployment_base} AS operator @@ -87,6 +92,10 @@ COPY --from=builder /output/register-server /usr/bin EXPOSE 3030 ENTRYPOINT ["/usr/bin/register-server"] +FROM ${deployment_base} AS kbs-event-proxy +COPY --from=builder /output/kbs-event-proxy /usr/bin +EXPOSE 8080 +ENTRYPOINT ["/usr/bin/kbs-event-proxy"] FROM builder AS compute-pcrs-data RUN rv_line=$(cargo metadata --format-version=1 | jq -r '.packages[] | select(.name == "reference-values") | .source') && \ diff --git a/Makefile b/Makefile index 5d962066..b9b1ac1b 100644 --- a/Makefile +++ b/Makefile @@ -5,8 +5,8 @@ .PHONY: all build build-tools crds-rs generate manifests cluster-up cluster-down \ install-trustee install clean fmt-check clippy lint test test-release release-tarball prepare-release \ - operator-image compute-pcrs-image reg-server-image attestation-key-register-image image \ - push-operator push-compute-pcrs push-reg-server push-attestation-key-register push \ + operator-image compute-pcrs-image reg-server-image attestation-key-register-image kbs-event-proxy-image image \ + push-operator push-compute-pcrs push-reg-server push-attestation-key-register push-kbs-event-proxy push \ SHELL := /bin/bash @@ -53,6 +53,7 @@ OPERATOR_IMAGE ?= $(REGISTRY)/trusted-cluster-operator:$(TAG) COMPUTE_PCRS_IMAGE=$(REGISTRY)/compute-pcrs:$(TAG) REG_SERVER_IMAGE=$(REGISTRY)/registration-server:$(TAG) ATTESTATION_KEY_REGISTER_IMAGE=$(REGISTRY)/attestation-key-register:$(TAG) +KBS_EVENT_PROXY_IMAGE=$(REGISTRY)/kbs-event-proxy:$(TAG) TRUSTEE_IMAGE ?= quay.io/trusted-execution-clusters/key-broker-service:v0.20.0 TEST_IMAGE ?= quay.io/trusted-execution-clusters/fedora-coreos-kubevirt:20260831 @@ -109,6 +110,7 @@ manifests: trusted-cluster-gen generate -pcrs-compute-image $(COMPUTE_PCRS_IMAGE) \ -register-server-image $(REG_SERVER_IMAGE) \ -attestation-key-register-image $(ATTESTATION_KEY_REGISTER_IMAGE) \ + -kbs-event-proxy-image $(KBS_EVENT_PROXY_IMAGE) \ -approved-image coreos,$(APPROVED_IMAGE) cluster-up: @@ -134,8 +136,10 @@ reg-server-image: $(CONTAINER_CLI) build $(IMAGE_BUILD_OPTIONS) --target register-server -t $(REG_SERVER_IMAGE) -f Containerfile . attestation-key-register-image: $(CONTAINER_CLI) build $(IMAGE_BUILD_OPTIONS) --target attestation-key-register -t $(ATTESTATION_KEY_REGISTER_IMAGE) -f Containerfile . +kbs-event-proxy-image: + $(CONTAINER_CLI) build $(IMAGE_BUILD_OPTIONS) --target kbs-event-proxy -t $(KBS_EVENT_PROXY_IMAGE) -f Containerfile . -image: operator-image compute-pcrs-image reg-server-image attestation-key-register-image +image: operator-image compute-pcrs-image reg-server-image attestation-key-register-image kbs-event-proxy-image define push-image $(CONTAINER_CLI) push $(1) $(PUSH_FLAGS) @@ -150,8 +154,10 @@ push-reg-server: reg-server-image $(call push-image,$(REG_SERVER_IMAGE)) push-attestation-key-register: attestation-key-register-image $(call push-image,$(ATTESTATION_KEY_REGISTER_IMAGE)) +push-kbs-event-proxy: kbs-event-proxy-image + $(call push-image,$(KBS_EVENT_PROXY_IMAGE)) -push: push-operator push-compute-pcrs push-reg-server push-attestation-key-register +push: push-operator push-compute-pcrs push-reg-server push-attestation-key-register push-kbs-event-proxy release-tarball: manifests tar -cf trusted-execution-operator-$(TAG).tar config diff --git a/README.md b/README.md index 6013f2bc..af876b7e 100644 --- a/README.md +++ b/README.md @@ -18,6 +18,7 @@ The operator relies on Rust crates for its functionality. - `/register-server`: A server that provides Clevis PINs for key retrieval with random UUIDs. - `/attestation-key-register`: A server that accepts attestation key registrations from VMs and creates AttestationKey resources. - `/compute-pcrs`: A program to compute PCR reference values using the [compute-pcrs library](https://github.com/trusted-execution-clusters/compute-pcrs) and insert them into a ConfigMap, run as a Job. +- `/kbs-event-proxy`: A reverse proxy sidecar for KBS that emits Kubernetes events for attestation activity. See [design doc](docs/design/kbs-event-proxy.md). ### Other crates diff --git a/api/trusted-cluster-gen.go b/api/trusted-cluster-gen.go index 914d3885..7325f488 100644 --- a/api/trusted-cluster-gen.go +++ b/api/trusted-cluster-gen.go @@ -66,6 +66,7 @@ type Args struct { pcrsComputeImage string registerServerImage string attestationKeyRegisterImage string + kbsEventProxyImage string approvedImages approvedImageSlice } @@ -78,6 +79,7 @@ func main() { flag.StringVar(&args.pcrsComputeImage, "pcrs-compute-image", "quay.io/trusted-execution-clusters/compute-pcrs:latest", "Container image with the Trusted Execution Clusters compute-pcrs binary") flag.StringVar(&args.registerServerImage, "register-server-image", "quay.io/trusted-execution-clusters/register-server:latest", "Register server image to use in the deployment") flag.StringVar(&args.attestationKeyRegisterImage, "attestation-key-register-image", "quay.io/trusted-execution-clusters/attestation-key-register:latest", "Attestation key register image to use in the deployment") + flag.StringVar(&args.kbsEventProxyImage, "kbs-event-proxy-image", "quay.io/trusted-execution-clusters/kbs-event-proxy:latest", "KBS event proxy sidecar image") flag.Var(&args.approvedImages, "approved-image", "When set, defines an initial approved image. It must be a comma-separated name,image-ref pair. Must be a bootable container image with SHA reference. Can be set multiple times.") flag.Parse() @@ -146,6 +148,10 @@ func generateOperator(args *Args) error { Name: "RELATED_IMAGE_ATTESTATION_KEY_REGISTER", Value: args.attestationKeyRegisterImage, }, + { + Name: "RELATED_IMAGE_KBS_EVENT_PROXY", + Value: args.kbsEventProxyImage, + }, }, }, }, diff --git a/docs/design/kbs-event-proxy.md b/docs/design/kbs-event-proxy.md new file mode 100644 index 00000000..2b3861d0 --- /dev/null +++ b/docs/design/kbs-event-proxy.md @@ -0,0 +1,128 @@ +# KBS Event Proxy + +## Overview + +The KBS event proxy is a reverse proxy sidecar that runs alongside the KBS (Key Broker Service) container in the Trustee pod. It intercepts RCAR attestation HTTP traffic between nodes and KBS and emits Kubernetes events for attestation activity. + +Without the proxy, attestation outcomes are only visible in KBS pod logs. The proxy surfaces these outcomes as first-class Kubernetes events on Machine and TrustedExecutionCluster resources. + +## Problem + +Trustee has no webhook, callback, or audit log for attestation outcomes. An administrator cannot answer "Did machine X attest successfully?" without reading KBS container logs. Kubernetes events provide a standard, queryable interface for this information. + +## Architecture + +The proxy runs as a sidecar container in the same pod as KBS. The Kubernetes Service routes external traffic to the proxy on port 8080. The proxy forwards all requests to KBS on localhost port 8081. + +``` +Nodes --> Service:8080 --> Proxy:8080 --(TLS)--> KBS:8081 (localhost) + | + inspects request/response + | + emits K8s events +``` + +Both hops use TLS. The proxy terminates external TLS from nodes, then connects to KBS via HTTPS on localhost. Both containers mount the same TLS secret volume. + +### Why a reverse proxy + +A metrics-based sidecar polls Prometheus counters and sees counter deltas, not individual events. The reverse proxy provides: + +- Real-time event emission per attestation attempt +- Distinction between attestation failure (401) and resource policy denial (403) +- Session correlation across the three RCAR protocol steps +- Per-machine event attribution for resource requests + +## RCAR Protocol + +The RCAR (Remote CoCo Attestation and Retrieval) protocol has three HTTP steps. The proxy tracks sessions via the `kbs-session-id` cookie. + +| Step | Endpoint | What the proxy observes | +|---|---|---| +| Auth | `POST /kbs/v0/auth` | TEE type from request body. Session cookie in response. | +| Attest | `POST /kbs/v0/attest` | 200 = attestation passed. Non-200 = failure. Session cookie identifies the session. | +| Resource | `GET /kbs/v0/resource/default/{id}/root` | Machine ID from URL path. 200 = key released. 403 = policy denied. 401 = rejected. | + +### Session tracking + +The proxy maintains an in-memory HashMap that maps session IDs (from the `kbs-session-id` cookie) to session metadata: + +``` +session_id -> SessionInfo { tee_type, created } +``` + +Sessions expire after 5 minutes (matching the KBS session timeout). The proxy cleans up expired sessions after each request. + +## Events emitted + +| Reason | Event type | Target resource | Trigger | +|---|---|---|---| +| `AttestationSucceeded` | Normal | Machine | Resource endpoint returns 200 for `default/{machine-id}/root` | +| `AttestationFailed` | Warning | TrustedExecutionCluster | Attest endpoint returns non-200 | +| `AttestationFailed` | Warning | Machine | Resource endpoint returns 401 for `default/{machine-id}/root` | +| `ResourcePolicyDenied` | Warning | Machine | Resource endpoint returns 403 for `default/{machine-id}/root` | + +The proxy emits `AttestationFailed` on the TrustedExecutionCluster (not on a Machine) at the attest step because the RCAR protocol does not carry a machine identifier at that point. The session carries only the TEE type. + +The resource step does carry the machine ID in the URL path. The proxy resolves machine IDs to Machine custom resources via the Kubernetes API. + +## Deployment + +### Pod spec changes + +The operator modifies the Trustee pod spec in `operator/src/trustee.rs`: + +1. KBS container listens on `127.0.0.1:8081` (internal only) +2. Proxy container listens on `0.0.0.0:8080` (exposed via Service) +3. Both containers mount the TLS secret volume +4. The pod uses the `trusted-cluster-operator` ServiceAccount for RBAC +5. The proxy receives `CONTROLLER_POD_NAME` via the downward API for event reporting + +### Image resolution + +The operator resolves the proxy image from the `RELATED_IMAGE_KBS_EVENT_PROXY` environment variable. If unset, it falls back to `{TEC_REGISTRY}/kbs-event-proxy:{COMPONENT_VERSION}`. + +### RBAC + +The proxy reuses the `trusted-cluster-operator` ServiceAccount. The ClusterRole includes: + +- `events.k8s.io` API group: `create`, `patch` (for emitting events via the `events.k8s.io/v1` API) + +The proxy also reads Machine and TrustedExecutionCluster resources to resolve object references for event targets. These permissions are already present in the operator's ClusterRole. + +## TLS + +The proxy accepts invalid TLS certificates when connecting to KBS on localhost. This is safe because the connection stays within the same pod on the loopback interface. The KBS TLS certificate contains the external hostname, not `127.0.0.1`, so strict validation would reject the connection. + +The external-facing TLS termination uses the same certificate and key that KBS previously used directly. Nodes see no change in TLS behavior. + +## Code structure + +The proxy source is in `kbs-event-proxy/src/main.rs`, organized into six sections: + +1. **Types and state**: CLI arguments, session info, proxy state with HTTP client, Kubernetes client, event recorder, and session map +2. **Request/response parsing**: Extract session IDs from Cookie and Set-Cookie headers, extract machine IDs from URL paths +3. **Kubernetes object lookups**: Resolve TrustedExecutionCluster and Machine custom resources to ObjectReferences for event targets +4. **RCAR attestation event handlers**: One handler per RCAR step (auth, attest, resource) that inspects the forwarded response and emits events +5. **Reverse proxy core**: Request forwarding, error responses, and the main handler that dispatches to event handlers based on URL path +6. **Entry point**: Client initialization, TLS configuration, and server startup + +## Dependencies + +The proxy reuses workspace dependencies: + +- `axum` and `axum-server`: HTTP server and TLS termination (also used by register-server and attestation-key-register) +- `reqwest`: HTTP client for forwarding requests to KBS +- `kube` and `k8s-openapi`: Kubernetes API access and event recording +- `trusted-cluster-operator-lib`: Shared types (`Machine`, `record_event`, `get_trusted_execution_cluster`) + +## Verification + +After deployment, verify events with: + +```bash +kubectl get events.events.k8s.io -n +kubectl describe machine +``` + +A successful attestation produces an `AttestationSucceeded` event on the Machine resource. A failed attestation produces an `AttestationFailed` warning on the TrustedExecutionCluster resource. diff --git a/docs/design/operator-architecture.md b/docs/design/operator-architecture.md index 12375c5e..28e378c0 100644 --- a/docs/design/operator-architecture.md +++ b/docs/design/operator-architecture.md @@ -16,7 +16,8 @@ The operator consists of several interconnected components: 4. **Machine Controller**: Reconciles Machine custom resources representing individual nodes. Part of the *operator* pod 5. **Secret Management**: Generates and manages LUKS. encryption keys and attestation key secrets. Part of the *operator* pod. 6. **Attestation Server and KBS**: [Trustee](https://github.com/confidential-containers/trustee) deployment handle the attestation request, the reference values and secrets. -7. **Reference Values calculation**: calculate the reference values provided by the approved images. +7. **KBS Event Proxy**: Reverse proxy sidecar in the Trustee pod that intercepts attestation traffic and emits Kubernetes events. See [KBS Event Proxy Design](kbs-event-proxy.md). +8. **Reference Values calculation**: calculate the reference values provided by the approved images. ## Architecture Components diff --git a/kbs-event-proxy/Cargo.toml b/kbs-event-proxy/Cargo.toml new file mode 100644 index 00000000..447e53db --- /dev/null +++ b/kbs-event-proxy/Cargo.toml @@ -0,0 +1,25 @@ +# SPDX-FileCopyrightText: Yair Podemsky +# +# SPDX-License-Identifier: CC0-1.0 + +[package] +name = "kbs-event-proxy" +version = "0.1.0" +edition.workspace = true +rust-version.workspace = true + +[dependencies] +anyhow.workspace = true +axum.workspace = true +axum-server.workspace = true +clap.workspace = true +env_logger.workspace = true +http.workspace = true +k8s-openapi.workspace = true +kube.workspace = true +log.workspace = true +reqwest = { version = "0.12", default-features = false, features = ["native-tls"] } +serde.workspace = true +serde_json.workspace = true +tokio.workspace = true +trusted-cluster-operator-lib = { path = "../lib" } diff --git a/kbs-event-proxy/src/main.rs b/kbs-event-proxy/src/main.rs new file mode 100644 index 00000000..bd70aa0f --- /dev/null +++ b/kbs-event-proxy/src/main.rs @@ -0,0 +1,378 @@ +// SPDX-FileCopyrightText: Yair Podemsky +// +// SPDX-License-Identifier: MIT + +use std::collections::HashMap; +use std::net::SocketAddr; +use std::sync::Arc; +use std::time::{Duration, Instant}; + +use anyhow::{Context, Result}; +use axum::Router; +use axum::body::{Body, Bytes}; +use axum::extract::State; +use axum::http::{Request, Response, StatusCode}; +use axum_server::tls_openssl::OpenSSLConfig; +use clap::Parser; +use env_logger::Env; +use k8s_openapi::api::core::v1::ObjectReference; +use kube::runtime::events::{EventType, Recorder}; +use kube::runtime::reflector::{self, Store}; +use kube::{Client, Resource}; +use log::{error, info, warn}; +use tokio::sync::Mutex; + +use trusted_cluster_operator_lib::{ + Machine, get_trusted_execution_cluster, new_recorder, record_event, spawn_reflector, sync_cache, +}; + +// -- Types and state -- + +const SESSION_TIMEOUT: Duration = Duration::from_secs(300); +const MAX_BODY_SIZE: usize = 16 * 1024 * 1024; + +const KBS_AUTH_PATH: &str = "/kbs/v0/auth"; +const KBS_ATTEST_PATH: &str = "/kbs/v0/attest"; +const KBS_RESOURCE_PATH_PREFIX: &str = "/kbs/v0/resource/default/"; + +#[derive(Parser)] +#[command(name = "kbs-event-proxy")] +#[command(about = "Reverse proxy for KBS that emits Kubernetes events for attestation activity")] +struct Args { + #[arg(long, default_value = "8080")] + listen_port: u16, + + #[arg(long, default_value = "https://127.0.0.1:8081")] + backend_url: String, + + #[arg(long)] + cert_path: Option, + + #[arg(long)] + key_path: Option, +} + +struct SessionInfo { + tee_type: String, + created: Instant, +} + +struct ProxyState { + http_client: reqwest::Client, + kube_client: Client, + machine_store: Store, + recorder: Recorder, + backend_url: String, + sessions: Mutex>, +} + +impl ProxyState { + async fn cleanup_expired_sessions(&self) { + let mut sessions = self.sessions.lock().await; + sessions.retain(|_, s| s.created.elapsed() < SESSION_TIMEOUT); + } +} + +// -- Request/response parsing -- + +fn session_id_from_request(headers: &http::HeaderMap) -> Option { + headers + .get_all(http::header::COOKIE) + .iter() + .filter_map(|v| v.to_str().ok()) + .flat_map(|s| s.split(';')) + .map(str::trim) + .find_map(|pair| pair.strip_prefix("kbs-session-id=").map(|v| v.to_string())) +} + +fn session_id_from_response(headers: &http::HeaderMap) -> Option { + headers + .get_all(http::header::SET_COOKIE) + .iter() + .filter_map(|v| v.to_str().ok()) + .find_map(|s| { + s.split(';') + .next() + .and_then(|pair| pair.strip_prefix("kbs-session-id=")) + .map(|v| v.to_string()) + }) +} + +fn machine_id_from_path(path: &str) -> Option<&str> { + let stripped = path.strip_prefix(KBS_RESOURCE_PATH_PREFIX)?; + stripped.strip_suffix("/root") +} + +// -- Kubernetes object lookups -- + +async fn lookup_cluster_ref(client: &Client) -> Option { + match get_trusted_execution_cluster(client.clone()).await { + Ok(tec) => Some(tec.object_ref(&())), + Err(e) => { + warn!("Failed to look up TrustedExecutionCluster for event: {e}"); + None + } + } +} + +fn lookup_machine_ref(store: &Store, machine_id: &str) -> Option { + let machine_name = format!("machine-{machine_id}"); + store + .state() + .iter() + .find(|m| m.meta().name.as_deref() == Some(machine_name.as_str())) + .map(|m| m.object_ref(&())) +} + +// -- RCAR attestation event handlers (auth -> attest -> resource) -- + +async fn handle_auth_response(state: &ProxyState, req_body: &[u8], resp_headers: &http::HeaderMap) { + let session_id = match session_id_from_response(resp_headers) { + Some(id) => id, + None => return, + }; + + let tee_type = serde_json::from_slice::(req_body) + .ok() + .and_then(|v| v.get("tee")?.as_str().map(String::from)) + .unwrap_or_else(|| "unknown".to_string()); + + state.sessions.lock().await.insert( + session_id, + SessionInfo { + tee_type, + created: Instant::now(), + }, + ); +} + +async fn handle_attest_response( + state: &ProxyState, + req_headers: &http::HeaderMap, + resp_status: StatusCode, +) { + let session_id = match session_id_from_request(req_headers) { + Some(id) => id, + None => return, + }; + + let succeeded = resp_status == StatusCode::OK; + + let tee_type = { + let sessions = state.sessions.lock().await; + sessions + .get(&session_id) + .map(|s| s.tee_type.clone()) + .unwrap_or_else(|| "unknown".to_string()) + }; + + if !succeeded && let Some(tec_ref) = lookup_cluster_ref(&state.kube_client).await { + record_event( + &state.recorder, + &tec_ref, + EventType::Warning, + "AttestationFailed", + format!("Attestation failed for TEE type: {tee_type}"), + "Attesting", + None, + ) + .await; + } +} + +async fn handle_resource_response(state: &ProxyState, path: &str, resp_status: StatusCode) { + let machine_id = match machine_id_from_path(path) { + Some(id) => id, + None => return, + }; + + let machine_ref = match lookup_machine_ref(&state.machine_store, machine_id) { + Some(r) => r, + None => return, + }; + + let (event_type, reason, note) = match resp_status { + StatusCode::OK => ( + EventType::Normal, + "AttestationSucceeded", + format!("KBS released secret for machine {machine_id}"), + ), + StatusCode::FORBIDDEN => ( + EventType::Warning, + "ResourcePolicyDenied", + format!("Resource policy denied access for machine {machine_id}"), + ), + StatusCode::UNAUTHORIZED => ( + EventType::Warning, + "AttestationFailed", + format!("Attestation rejected for machine {machine_id}"), + ), + _ => return, + }; + record_event( + &state.recorder, + &machine_ref, + event_type, + reason, + note, + "Attesting", + None, + ) + .await; +} + +// -- Reverse proxy core -- + +fn bad_gateway(msg: &str) -> Response { + Response::builder() + .status(StatusCode::BAD_GATEWAY) + .body(Body::from(msg.to_string())) + .unwrap() +} + +struct BackendResponse { + status: StatusCode, + headers: http::HeaderMap, + body: Bytes, +} + +async fn forward_request( + state: &ProxyState, + method: http::Method, + backend_uri: &str, + headers: &http::HeaderMap, + body: Bytes, +) -> Result> { + let mut forwarded = state + .http_client + .request(method, backend_uri) + .headers(headers.clone()) + .body(body) + .build() + .map_err(|e| { + error!("Failed to build backend request: {e}"); + bad_gateway("proxy error") + })?; + forwarded.headers_mut().remove(http::header::HOST); + + let resp = state.http_client.execute(forwarded).await.map_err(|e| { + error!("Backend request failed: {e}"); + bad_gateway("backend unavailable") + })?; + + let status = resp.status(); + let headers = resp.headers().clone(); + let body = resp.bytes().await.map_err(|e| { + error!("Failed to read backend response: {e}"); + bad_gateway("proxy error") + })?; + + Ok(BackendResponse { + status, + headers, + body, + }) +} + +async fn proxy_handler(State(state): State>, req: Request) -> Response { + let method = req.method().clone(); + let uri = req.uri().clone(); + let path = uri.path().to_string(); + let req_headers = req.headers().clone(); + + let body_bytes = match axum::body::to_bytes(req.into_body(), MAX_BODY_SIZE).await { + Ok(b) => b, + Err(e) => { + error!("Failed to read request body: {e}"); + return bad_gateway("proxy error"); + } + }; + + let backend_uri = format!( + "{}{}", + state.backend_url, + uri.path_and_query().map(|pq| pq.as_str()).unwrap_or("/") + ); + + let backend = match forward_request( + &state, + method, + &backend_uri, + &req_headers, + body_bytes.clone(), + ) + .await + { + Ok(r) => r, + Err(resp) => return resp, + }; + + if path == KBS_AUTH_PATH && backend.status == StatusCode::OK { + handle_auth_response(&state, &body_bytes, &backend.headers).await; + } else if path == KBS_ATTEST_PATH { + handle_attest_response(&state, &req_headers, backend.status).await; + } else if path.starts_with(KBS_RESOURCE_PATH_PREFIX) { + handle_resource_response(&state, &path, backend.status).await; + } + + state.cleanup_expired_sessions().await; + + let mut response = Response::builder().status(backend.status); + for (key, value) in &backend.headers { + response = response.header(key, value); + } + response.body(Body::from(backend.body)).unwrap() +} + +// -- Entry point -- + +#[tokio::main] +async fn main() -> Result<()> { + env_logger::Builder::from_env(Env::default().default_filter_or("info")).init(); + + let args = Args::parse(); + + let http_client = reqwest::ClientBuilder::new() + .danger_accept_invalid_certs(true) + .build() + .context("Failed to create HTTP client")?; + + let kube_client = Client::try_default() + .await + .context("Failed to create Kubernetes client")?; + + let (machine_store, machine_writer) = reflector::store::(); + spawn_reflector::(machine_writer, kube_client.clone(), "Machine"); + sync_cache(&machine_store, "Machine", Duration::from_secs(30)).await?; + + let state = Arc::new(ProxyState { + http_client, + kube_client: kube_client.clone(), + machine_store, + recorder: new_recorder(kube_client, "kbs-event-proxy"), + backend_url: args.backend_url, + sessions: Mutex::new(HashMap::new()), + }); + + let app = Router::new().fallback(proxy_handler).with_state(state); + let addr = SocketAddr::from(([0, 0, 0, 0], args.listen_port)); + let service = app.into_make_service(); + + if let (Some(cert_path), Some(key_path)) = (args.cert_path, args.key_path) { + let config = OpenSSLConfig::from_pem_file(cert_path, key_path) + .context("Invalid PEM files for TLS")?; + info!("Proxy listening on https://{addr}"); + axum_server::bind_openssl(addr, config) + .serve(service) + .await + .context("Server failed")?; + } else { + info!("Proxy listening on http://{addr}"); + axum_server::bind(addr) + .serve(service) + .await + .context("Server failed")?; + } + + Ok(()) +} diff --git a/lib/Cargo.toml b/lib/Cargo.toml index 051dd55d..eca12b56 100644 --- a/lib/Cargo.toml +++ b/lib/Cargo.toml @@ -15,8 +15,10 @@ anyhow.workspace = true compute-pcrs-lib.workspace = true k8s-openapi.workspace = true kube.workspace = true +futures-util = "0.3.32" log.workspace = true serde.workspace = true +tokio.workspace = true serde_json.workspace = true [dev-dependencies] diff --git a/lib/src/endpoints.rs b/lib/src/endpoints.rs index c95449f9..fefe8ce4 100644 --- a/lib/src/endpoints.rs +++ b/lib/src/endpoints.rs @@ -5,6 +5,7 @@ pub const TRUSTEE_SERVICE: &str = "kbs-service"; pub const TRUSTEE_DEPLOYMENT: &str = "trustee-deployment"; pub const TRUSTEE_PORT: i32 = 8080; +pub const KBS_INTERNAL_PORT: i32 = 8081; pub const TRUSTEE_APP_LABEL: &str = "kbs"; pub const REGISTER_SERVER_SERVICE: &str = "register-server"; pub const REGISTER_SERVER_DEPLOYMENT: &str = "register-server"; diff --git a/lib/src/images.rs b/lib/src/images.rs index 2ab7fe3c..5c5c4693 100644 --- a/lib/src/images.rs +++ b/lib/src/images.rs @@ -6,3 +6,4 @@ pub const RELATED_IMAGE_COMPUTE_PCRS: &str = "RELATED_IMAGE_COMPUTE_PCRS"; pub const RELATED_IMAGE_TRUSTEE: &str = "RELATED_IMAGE_TRUSTEE"; pub const RELATED_IMAGE_REGISTRATION_SERVER: &str = "RELATED_IMAGE_REGISTRATION_SERVER"; pub const RELATED_IMAGE_ATTESTATION_KEY_REGISTER: &str = "RELATED_IMAGE_ATTESTATION_KEY_REGISTER"; +pub const RELATED_IMAGE_KBS_EVENT_PROXY: &str = "RELATED_IMAGE_KBS_EVENT_PROXY"; diff --git a/lib/src/lib.rs b/lib/src/lib.rs index 37560bcd..6c42133e 100644 --- a/lib/src/lib.rs +++ b/lib/src/lib.rs @@ -25,10 +25,14 @@ pub use vendor_kopium::virtualmachines; use anyhow::{Context, Result, anyhow}; use conditions::*; +use futures_util::StreamExt; use k8s_openapi::api::core::v1::ObjectReference; use k8s_openapi::apimachinery::pkg::apis::meta::v1::{Condition, OwnerReference, Time}; -use kube::runtime::events::{Event as K8sEvent, EventType, Recorder}; +use kube::runtime::events::{Event as K8sEvent, EventType, Recorder, Reporter}; +use kube::runtime::reflector::{self, Store}; +use kube::runtime::watcher::watcher; use kube::{Api, Client, Resource}; +use std::time::Duration; #[macro_export] macro_rules! update_status { @@ -152,6 +156,45 @@ pub fn generate_owner_reference>( }) } +pub fn new_recorder(client: Client, controller_name: &str) -> Recorder { + let reporter = Reporter { + controller: controller_name.into(), + instance: std::env::var("CONTROLLER_POD_NAME").ok(), + }; + Recorder::new(client, reporter) +} + +pub fn spawn_reflector(writer: reflector::store::Writer, client: Client, name: &'static str) +where + K: Resource, + K: Clone + serde::de::DeserializeOwned + std::fmt::Debug + Send + Sync + 'static, + K::DynamicType: Default + Eq + std::hash::Hash + Clone, +{ + let watcher = watcher(Api::::default_namespaced(client), Default::default()); + let reflector = reflector::reflector(writer, watcher).for_each(move |res| async move { + if let Err(e) = res { + log::warn!("{name} reflector error: {e}"); + } + }); + tokio::spawn(reflector); +} + +pub async fn sync_cache(store: &Store, name: &str, sync_timeout: Duration) -> Result<()> +where + K: 'static + Clone + reflector::Lookup, + K::DynamicType: Eq + std::hash::Hash + Clone, +{ + let err = anyhow!( + "Timed out after {sync_timeout:?} waiting for {name} cache to sync. \ + Ensure the CRD is installed and the API server is reachable." + ); + tokio::time::timeout(sync_timeout, store.wait_until_ready()) + .await + .map_err(|_| err)? + .map_err(|e| anyhow!("Cache writer for {name} was dropped: {e}"))?; + Ok(()) +} + pub async fn get_opt_trusted_execution_cluster( client: Client, ) -> Result> { diff --git a/operator/src/lib.rs b/operator/src/lib.rs index 31ae7052..2b4dd71e 100644 --- a/operator/src/lib.rs +++ b/operator/src/lib.rs @@ -9,27 +9,26 @@ // Use in other crates is not an intended purpose. use anyhow::{Result, anyhow}; -use futures_util::StreamExt; use k8s_openapi::api::core::v1::{ConfigMap, Secret, SecretVolumeSource, Volume, VolumeMount}; use k8s_openapi::apimachinery::pkg::apis::meta::v1::{Condition, Time}; use k8s_openapi::jiff::Timestamp; use kube::Resource; -use kube::runtime::events::{Recorder, Reporter}; +use kube::runtime::events::Recorder; use kube::runtime::reflector::{self, Store}; -use kube::runtime::watcher::watcher; use kube::{Api, Client, runtime::controller::Action}; use log::{info, warn}; use std::fmt::{Debug, Display}; use std::{sync::Arc, time::Duration}; -use tokio::time::timeout; // Re-export common functions from the lib use kube::api::{Patch, PatchParams}; use trusted_cluster_operator_lib::Conditions; -pub use trusted_cluster_operator_lib::generate_owner_reference; use trusted_cluster_operator_lib::{ ApprovedImage, AttestationKey, Machine, TrustedExecutionCluster, }; +pub use trusted_cluster_operator_lib::{ + generate_owner_reference, new_recorder, spawn_reflector, sync_cache, +}; /// Unified context shared across all controllers. /// Stores give local cache access to avoid repeated API-server reads. @@ -92,14 +91,6 @@ pub async fn controller_info(res: Result) { } } -pub fn new_recorder(client: Client, controller_name: &str) -> Recorder { - let reporter = Reporter { - controller: controller_name.into(), - instance: std::env::var("CONTROLLER_POD_NAME").ok(), - }; - Recorder::new(client, reporter) -} - #[macro_export] macro_rules! create_or_info_if_exists { ($client:expr, $type:ident, $resource:ident) => { @@ -156,37 +147,6 @@ pub async fn read_certificate( Ok(Some((volume, volume_mount))) } -pub fn spawn_reflector(writer: reflector::store::Writer, client: Client, name: &'static str) -where - K: Resource, - K: Clone + serde::de::DeserializeOwned + std::fmt::Debug + Send + Sync + 'static, - K::DynamicType: Default + Eq + std::hash::Hash + Clone, -{ - let watcher = watcher(Api::::default_namespaced(client), Default::default()); - let reflector = reflector::reflector(writer, watcher).for_each(move |res| async move { - if let Err(e) = res { - warn!("{name} reflector error: {e}"); - } - }); - tokio::spawn(reflector); -} - -pub async fn sync_cache(store: &Store, name: &str, sync_timeout: Duration) -> Result<()> -where - K: 'static + Clone + reflector::Lookup, - K::DynamicType: Eq + std::hash::Hash + Clone, -{ - let err = anyhow::anyhow!( - "Timed out after {sync_timeout:?} waiting for {name} cache to sync. \ - Ensure the CRD is installed and the API server is reachable." - ); - timeout(sync_timeout, store.wait_until_ready()) - .await - .map_err(|_| err)? - .map_err(|e| anyhow::anyhow!("Cache writer for {name} was dropped: {e}"))?; - Ok(()) -} - // TODO: Port this functionality to kube-rs API. // Update condition if already present, otherwise append(insert) it into the conditions vector. // Inspired by k8s.io/apimachinery/pkg/api/meta.SetStatusCondition diff --git a/operator/src/main.rs b/operator/src/main.rs index e4223796..6d164acd 100644 --- a/operator/src/main.rs +++ b/operator/src/main.rs @@ -171,9 +171,19 @@ async fn install_trustee_configuration( let default = format!("{TEC_REGISTRY}/key-broker-service:{TRUSTEE_VERSION}"); let trustee_image = env::var(RELATED_IMAGE_TRUSTEE).ok().unwrap_or(default); - trustee::generate_kbs_deployment(client, owner_reference, &trustee_image, trustee_secret) - .await - .context("Failed to create the KBS deployment")?; + let default_proxy = format!("{TEC_REGISTRY}/kbs-event-proxy:{COMPONENT_VERSION}"); + let proxy_image = env::var(RELATED_IMAGE_KBS_EVENT_PROXY) + .ok() + .unwrap_or(default_proxy); + trustee::generate_kbs_deployment( + client, + owner_reference, + &trustee_image, + &proxy_image, + trustee_secret, + ) + .await + .context("Failed to create the KBS deployment")?; info!("Generated the KBS deployment"); Ok(()) diff --git a/operator/src/trustee.rs b/operator/src/trustee.rs index 52b2619e..245fd205 100644 --- a/operator/src/trustee.rs +++ b/operator/src/trustee.rs @@ -14,8 +14,8 @@ use futures_util::StreamExt; use k8s_openapi::api::apps::v1::{Deployment, DeploymentSpec}; use k8s_openapi::api::core::v1::{ ConfigMap, ConfigMapVolumeSource, Container, ContainerPort, EmptyDirVolumeSource, EnvVar, - KeyToPath, PodSpec, PodTemplateSpec, Secret, SecretVolumeSource, Service, ServicePort, - ServiceSpec, Volume, VolumeMount, + EnvVarSource, KeyToPath, ObjectFieldSelector, PodSpec, PodTemplateSpec, Secret, + SecretVolumeSource, Service, ServicePort, ServiceSpec, Volume, VolumeMount, }; use k8s_openapi::apimachinery::pkg::{ apis::meta::v1::{LabelSelector, OwnerReference}, @@ -500,6 +500,14 @@ fn generate_kbs_config(has_certificate: bool) -> Result { let server_err = "http_server is not a table"; let http_server = http_section.as_table_mut().context(server_err)?; + // KBS listens on localhost only — the event proxy sidecar handles + // external traffic and TLS termination. + let internal_socket = format!("127.0.0.1:{KBS_INTERNAL_PORT}"); + http_server.insert( + "sockets".to_string(), + toml::Value::Array(vec![toml::Value::String(internal_socket)]), + ); + if has_certificate { let tls_key = toml::Value::String(format!("{TLS_DIR}/tls.key")); http_server.insert("private_key".to_string(), tls_key); @@ -623,7 +631,11 @@ fn generate_kbs_volume_templates() -> [(&'static str, &'static str, Volume); 3] ] } -fn generate_kbs_pod_spec(image: &str, tls_volumes: Option<(Volume, VolumeMount)>) -> PodSpec { +fn generate_kbs_pod_spec( + image: &str, + proxy_image: &str, + tls_volumes: Option<(Volume, VolumeMount)>, +) -> PodSpec { let volume_templates = generate_kbs_volume_templates(); let mut volumes: Vec = volume_templates .iter() @@ -633,7 +645,7 @@ fn generate_kbs_pod_spec(image: &str, tls_volumes: Option<(Volume, VolumeMount)> volume }) .collect(); - let mut volume_mounts: Vec = volume_templates + let mut kbs_volume_mounts: Vec = volume_templates .iter() .map(|(name, mount_path, _)| VolumeMount { name: name.to_string(), @@ -642,32 +654,91 @@ fn generate_kbs_pod_spec(image: &str, tls_volumes: Option<(Volume, VolumeMount)> }) .collect(); + let has_tls = tls_volumes.is_some(); + let mut proxy_volume_mounts = Vec::new(); + if let Some((volume, volume_mount)) = tls_volumes { volumes.push(volume); - volume_mounts.push(volume_mount); + proxy_volume_mounts.push(volume_mount.clone()); + kbs_volume_mounts.push(volume_mount); } - PodSpec { - containers: vec![Container { - command: Some(vec![ - "/usr/local/bin/kbs".to_string(), - "--config-file".to_string(), - format!("{TRUSTEE_DATA_DIR}/{KBS_CONFIG_FILE}"), - ]), - env: Some(vec![EnvVar { - name: "RUST_LOG".to_string(), - value: Some("debug".to_string()), - ..Default::default() - }]), - image: Some(image.to_string()), - name: "kbs".to_string(), - ports: Some(vec![ContainerPort { - container_port: TRUSTEE_PORT, + let backend_scheme = if has_tls { "https" } else { "http" }; + let mut proxy_command = vec![ + "/usr/bin/kbs-event-proxy".to_string(), + "--backend-url".to_string(), + format!("{backend_scheme}://127.0.0.1:{KBS_INTERNAL_PORT}"), + ]; + if has_tls { + proxy_command.extend([ + "--cert-path".to_string(), + format!("{TLS_DIR}/tls.crt"), + "--key-path".to_string(), + format!("{TLS_DIR}/tls.key"), + ]); + } + + let proxy_env = vec![ + EnvVar { + name: "RUST_LOG".to_string(), + value: Some("info".to_string()), + ..Default::default() + }, + EnvVar { + name: "CONTROLLER_POD_NAME".to_string(), + value_from: Some(EnvVarSource { + field_ref: Some(ObjectFieldSelector { + field_path: "metadata.name".to_string(), + ..Default::default() + }), ..Default::default() - }]), - volume_mounts: Some(volume_mounts), + }), ..Default::default() - }], + }, + ]; + + let proxy_container = Container { + command: Some(proxy_command), + env: Some(proxy_env), + image: Some(proxy_image.to_string()), + name: "kbs-event-proxy".to_string(), + ports: Some(vec![ContainerPort { + container_port: TRUSTEE_PORT, + ..Default::default() + }]), + volume_mounts: if proxy_volume_mounts.is_empty() { + None + } else { + Some(proxy_volume_mounts) + }, + ..Default::default() + }; + + PodSpec { + service_account_name: Some("trusted-cluster-operator".to_string()), + containers: vec![ + Container { + command: Some(vec![ + "/usr/local/bin/kbs".to_string(), + "--config-file".to_string(), + format!("{TRUSTEE_DATA_DIR}/{KBS_CONFIG_FILE}"), + ]), + env: Some(vec![EnvVar { + name: "RUST_LOG".to_string(), + value: Some("debug".to_string()), + ..Default::default() + }]), + image: Some(image.to_string()), + name: "kbs".to_string(), + ports: Some(vec![ContainerPort { + container_port: KBS_INTERNAL_PORT, + ..Default::default() + }]), + volume_mounts: Some(kbs_volume_mounts), + ..Default::default() + }, + proxy_container, + ], volumes: Some(volumes), ..Default::default() } @@ -677,6 +748,7 @@ pub async fn generate_kbs_deployment( client: Client, owner_reference: OwnerReference, image: &str, + proxy_image: &str, secret: &Option, ) -> Result<()> { let selector = Some(BTreeMap::from([( @@ -684,7 +756,7 @@ pub async fn generate_kbs_deployment( TRUSTEE_APP_LABEL.to_string(), )])); let tls_volumes = read_certificate(client.clone(), secret).await?; - let pod_spec = generate_kbs_pod_spec(image, tls_volumes); + let pod_spec = generate_kbs_pod_spec(image, proxy_image, tls_volumes); // Inspired by trustee-operator let deployment = Deployment { @@ -891,13 +963,17 @@ mod tests { #[tokio::test] async fn test_generate_kbs_depl_success() { - let clos = |client| generate_kbs_deployment(client, Default::default(), "image", &None); + let clos = |client| { + generate_kbs_deployment(client, Default::default(), "image", "proxy-image", &None) + }; test_create_success::<_, _, Deployment>(clos).await; } #[tokio::test] async fn test_generate_kbs_depl_error() { - let clos = |client| generate_kbs_deployment(client, Default::default(), "image", &None); + let clos = |client| { + generate_kbs_deployment(client, Default::default(), "image", "proxy-image", &None) + }; test_error_method!(clos, Method::POST); } diff --git a/test_utils/src/lib.rs b/test_utils/src/lib.rs index 118177cf..b1c3aef0 100644 --- a/test_utils/src/lib.rs +++ b/test_utils/src/lib.rs @@ -827,11 +827,14 @@ impl TestContext { .unwrap_or_else(|_| format!("{repo}/registration-server:{tag}")); let att_reg_img = env::var(RELATED_IMAGE_ATTESTATION_KEY_REGISTER) .unwrap_or_else(|_| format!("{repo}/attestation-key-register:{tag}")); + let proxy_img = env::var(RELATED_IMAGE_KBS_EVENT_PROXY) + .unwrap_or_else(|_| format!("{repo}/kbs-event-proxy:{tag}")); args.extend(&["-image", &operator_img]); args.extend(&["-pcrs-compute-image", &compute_pcrs_img]); args.extend(&["-trustee-image", &trustee_image]); args.extend(&["-register-server-image", ®_srv_img]); args.extend(&["-attestation-key-register-image", &att_reg_img]); + args.extend(&["-kbs-event-proxy-image", &proxy_img]); let primary_approved_arg = format!("{},{approved_image}", constants::APPROVED_IMAGE_NAME); args.extend(&["-approved-image", &primary_approved_arg]); let approved_args: Vec = approved_images diff --git a/tests/attestation.rs b/tests/attestation.rs index a60c7bec..e6709f96 100644 --- a/tests/attestation.rs +++ b/tests/attestation.rs @@ -346,3 +346,36 @@ async fn test_attestation_events() -> anyhow::Result<()> { Ok(()) } } + +virt_test! { +async fn test_kbs_proxy_attestation_events() -> anyhow::Result<()> { + let test_ctx = setup!().await?; + let client = test_ctx.client(); + let namespace = test_ctx.namespace(); + + let vm_name = "test-coreos-proxy-events"; + let att_ctx = SingleAttestationContext::new(vm_name, &test_ctx).await?; + + test_ctx.info("Verifying encrypted root device"); + let has_encrypted_root = att_ctx.verify_encrypted_root().await?; + assert!(has_encrypted_root, "VM {ENCRYPTED_ROOT_ASSERT}"); + test_ctx.info("Attestation successful, verifying KBS proxy events"); + + let machines: Api = Api::namespaced(client.clone(), namespace); + let machine_list = machines.list(&Default::default()).await?; + assert_eq!(machine_list.items.len(), 1, "Expected exactly one Machine in namespace"); + let machine_name = machine_list.items.first() + .expect("No Machine found in namespace") + .metadata + .name + .as_ref() + .expect("Machine should have a name"); + + wait_for_event(client, namespace, machine_name, "AttestationSucceeded", scaled_timeout(60)).await?; + test_ctx.info("Event AttestationSucceeded verified on Machine"); + + att_ctx.cleanup().await?; + test_ctx.cleanup().await?; + Ok(()) +} +}