lenso_api/
lib.rs

1use anyhow::Context as _;
2use axum::Router;
3use axum::body::Body;
4use axum::http::{HeaderName, HeaderValue, Method, header};
5use axum::middleware;
6use axum::response::Html;
7use hyper::body::Incoming;
8use hyper::service::service_fn;
9use hyper_util::{
10    rt::{TokioExecutor, TokioIo},
11    server::conn::auto::Builder as HyperConnectionBuilder,
12};
13use platform_core::{
14    AppConfig, AppContext, LoggingEventPublisher, PostgresRuntimeConfigProvider,
15    RuntimeConfigRegistry, Shutdown, connect_pool, connect_redis, telemetry,
16};
17use platform_http::request_context_middleware;
18use spiffe_rustls::{LocalOnly, authorizer, mtls_server};
19use spiffe_rustls_tokio::TlsAcceptor;
20use std::convert::Infallible;
21use std::future::Future;
22use std::net::SocketAddr;
23use std::sync::Arc;
24use tokio::sync::watch;
25use tower::ServiceExt as _;
26use tower_http::cors::CorsLayer;
27use tracing::info;
28
29pub mod openapi;
30
31pub use openapi::openapi_document;
32
33pub async fn run_from_env() -> anyhow::Result<()> {
34    run_from_env_with_composition(lenso_bootstrap::HostComposition::default()).await
35}
36
37pub async fn run_from_env_with_composition(
38    composition: lenso_bootstrap::HostComposition,
39) -> anyhow::Result<()> {
40    let config = AppConfig::try_from_env().context("invalid application configuration")?;
41    telemetry::init(&config.telemetry)?;
42
43    let db = connect_pool(&config.database).await?;
44    let redis = connect_redis(&config.redis).await?;
45    let mut ctx = AppContext::new(config, db, Arc::new(LoggingEventPublisher)).with_redis(redis);
46
47    let descriptors =
48        lenso_bootstrap::runtime_config_descriptors_with_composition(&ctx, &composition)
49            .context("failed to collect runtime-config descriptors")?;
50    let groups =
51        lenso_bootstrap::runtime_config_group_descriptors_with_composition(&ctx, &composition)
52            .context("failed to collect runtime-config groups")?;
53    let registry = RuntimeConfigRegistry::try_new_with_groups(descriptors, groups)
54        .context("duplicate runtime-config descriptor registered")?;
55    let runtime_config =
56        PostgresRuntimeConfigProvider::connect(ctx.db.clone(), Arc::new(registry), "api")
57            .await
58            .context("failed to load runtime-config snapshot")?;
59    runtime_config.spawn_listener();
60    ctx = ctx.with_runtime_config_provider(runtime_config);
61
62    let provider_plan = lenso_bootstrap::provider_runtime_plan_from_workspace(".")
63        .context("failed to compile Provider Runtime Plan")?;
64    if let Some(provider_runtime) = lenso_bootstrap::load_provider_runtime_with_composition(
65        &ctx,
66        &composition,
67        provider_plan.as_ref(),
68    )
69    .await
70    .context("failed to load locked Provider Runtime")?
71    {
72        platform_provider::install_provider_http_proxy_registry(provider_runtime.proxy_registry());
73    }
74
75    let app = try_build_router_with_composition(ctx.clone(), &composition)
76        .context("failed to build API router")?;
77    let address: SocketAddr = format!("{}:{}", ctx.config.http.host, ctx.config.http.port)
78        .parse()
79        .context("invalid HTTP bind address")?;
80
81    info!(%address, "starting API server");
82    let listener = tokio::net::TcpListener::bind(address).await?;
83    let shutdown = ctx.shutdown.clone();
84    axum::serve(
85        listener,
86        app.into_make_service_with_connect_info::<SocketAddr>(),
87    )
88    .with_graceful_shutdown(async move {
89        let mut shutdown_rx = shutdown.subscribe();
90        tokio::select! {
91            () = Shutdown::wait_for_signal() => {},
92            changed = shutdown_rx.changed() => {
93                let _ = changed;
94            },
95        }
96    })
97    .await?;
98
99    Ok(())
100}
101
102pub fn build_router(ctx: AppContext) -> Router {
103    try_build_router(ctx).expect("Runtime API router should build with a valid composition profile")
104}
105
106pub fn try_build_router(ctx: AppContext) -> platform_core::AppResult<Router> {
107    try_build_router_with_composition(ctx, &lenso_bootstrap::HostComposition::default())
108}
109
110pub fn try_build_router_with_composition(
111    mut ctx: AppContext,
112    composition: &lenso_bootstrap::HostComposition,
113) -> platform_core::AppResult<Router> {
114    if let Some(actor_resolver) =
115        lenso_bootstrap::auth_actor_resolver_for_context_with_composition(&ctx, composition)?
116    {
117        ctx = ctx.with_actor_resolver(actor_resolver);
118    }
119    let host_wiring = lenso_bootstrap::host_wiring_for_context_with_composition(&ctx, composition)?;
120    let (router, mut document) =
121        openapi::api_router_for_context_with_composition(&ctx, composition)?.split_for_parts();
122    openapi::normalize_error_response_content_types(&mut document);
123    let document = Arc::new(document);
124
125    Ok(router
126        .route("/docs", axum::routing::get(scalar_docs))
127        .route("/openapi.json", axum::routing::get(serve_openapi))
128        .layer(axum::Extension(document))
129        .layer(axum::Extension(host_wiring.auth_session_policy()))
130        .layer(middleware::from_fn_with_state(
131            ctx.clone(),
132            request_context_middleware,
133        ))
134        .layer(cors_layer(&ctx))
135        .with_state(ctx))
136}
137
138/// Builds the independent System Plane Router. Callers must serve it on a
139/// listener that injects a transport binding authenticated by mTLS.
140pub fn try_build_router_with_composition_and_system_plane(
141    _ctx: AppContext,
142    _composition: &lenso_bootstrap::HostComposition,
143    runtime: &lenso_bootstrap::HostSystemPlaneRuntime,
144) -> platform_core::AppResult<Router> {
145    let core = Some(Arc::clone(&runtime.core));
146    let installations = Some(Arc::clone(&runtime.service_installations));
147    let observability = runtime.runtime_observability.clone();
148    let operations = runtime.runtime_operations.clone();
149    let (router, _document) = platform_system_plane::router(core.clone())
150        .merge(platform_module_management::system_plane_router(
151            installations,
152        ))
153        .merge(platform_runtime_observability::router(observability))
154        .merge(platform_runtime_operations::router(operations))
155        .layer(axum::Extension(core))
156        .split_for_parts();
157    Ok(router)
158}
159
160/// Serves a System Plane-only Router over rotating SPIFFE X.509-SVID mTLS.
161/// The verified peer SPIFFE ID is injected as the transport binding; request
162/// headers cannot manufacture or replace it.
163pub async fn run_production_system_plane<F>(
164    listener: tokio::net::TcpListener,
165    router: Router,
166    identity: Arc<lenso_service::SpiffeWorkloadIdentityProvider>,
167    allowed_peer_spiffe_ids: impl IntoIterator<Item = String>,
168    shutdown: F,
169) -> anyhow::Result<()>
170where
171    F: Future<Output = ()> + Send,
172{
173    let allowed_peer_spiffe_ids = allowed_peer_spiffe_ids.into_iter().collect::<Vec<_>>();
174    if allowed_peer_spiffe_ids.is_empty() {
175        anyhow::bail!("production System Plane requires at least one allowed peer SPIFFE ID");
176    }
177    let tls = mtls_server(identity.x509_source())
178        .authorize(
179            authorizer::exact(allowed_peer_spiffe_ids)
180                .context("invalid System Plane peer SPIFFE allow list")?,
181        )
182        .trust_domain_policy(LocalOnly(identity.config().trust_domain().clone()))
183        .with_alpn_protocols([b"http/1.1"])
184        .build()
185        .context("failed to build System Plane mTLS configuration")?;
186    let acceptor = TlsAcceptor::new(Arc::new(tls));
187    let (shutdown_tx, shutdown_rx) = watch::channel(false);
188    let mut connections = tokio::task::JoinSet::new();
189    tokio::pin!(shutdown);
190
191    loop {
192        tokio::select! {
193            () = &mut shutdown => break,
194            accepted = listener.accept() => {
195                let (stream, peer_address) = accepted.context("System Plane listener failed")?;
196                let acceptor = acceptor.clone();
197                let router = router.clone();
198                let mut connection_shutdown = shutdown_rx.clone();
199                connections.spawn(async move {
200                    let (tls, peer) = match acceptor.accept(stream).await {
201                        Ok(result) => result,
202                        Err(error) => {
203                            tracing::warn!(%peer_address, %error, "rejected System Plane mTLS connection");
204                            return;
205                        }
206                    };
207                    let Some(peer_spiffe_id) = peer.spiffe_id() else {
208                        tracing::warn!(%peer_address, "rejected System Plane peer without a SPIFFE ID");
209                        return;
210                    };
211                    let binding =
212                        lenso_service::SpiffeWorkloadIdentityProvider::authenticated_transport_binding(
213                            peer_spiffe_id,
214                        );
215                    let service = service_fn(move |request: hyper::Request<Incoming>| {
216                        let router = router.clone();
217                        let binding = binding.clone();
218                        async move {
219                            let (mut parts, incoming) = request.into_parts();
220                            parts.extensions.insert(binding);
221                            let request = hyper::Request::from_parts(parts, Body::new(incoming));
222                            let response = router.oneshot(request).await?;
223                            Ok::<_, Infallible>(response)
224                        }
225                    });
226                    let connection_builder = HyperConnectionBuilder::new(TokioExecutor::new());
227                    let connection =
228                        connection_builder.serve_connection(TokioIo::new(tls), service);
229                    tokio::pin!(connection);
230                    tokio::select! {
231                        result = &mut connection => {
232                            if let Err(error) = result {
233                                tracing::warn!(%peer_address, %error, "System Plane connection failed");
234                            }
235                        }
236                        changed = connection_shutdown.changed() => {
237                            if changed.is_ok() {
238                                connection.as_mut().graceful_shutdown();
239                                if let Err(error) = connection.await {
240                                    tracing::warn!(%peer_address, %error, "System Plane graceful shutdown failed");
241                                }
242                            }
243                        }
244                    }
245                });
246            }
247        }
248    }
249
250    let _ = shutdown_tx.send(true);
251    while let Some(result) = connections.join_next().await {
252        if let Err(error) = result {
253            tracing::warn!(%error, "System Plane connection task failed");
254        }
255    }
256    identity.shutdown().await;
257    Ok(())
258}
259
260async fn scalar_docs() -> ([(HeaderName, HeaderValue); 3], Html<&'static str>) {
261    (
262        [
263            (
264                HeaderName::from_static("content-security-policy"),
265                HeaderValue::from_static(SCALAR_DOCS_CSP),
266            ),
267            (
268                HeaderName::from_static("referrer-policy"),
269                HeaderValue::from_static("no-referrer"),
270            ),
271            (
272                HeaderName::from_static("x-content-type-options"),
273                HeaderValue::from_static("nosniff"),
274            ),
275        ],
276        Html(SCALAR_DOCS_HTML),
277    )
278}
279
280async fn serve_openapi(
281    axum::Extension(document): axum::Extension<Arc<utoipa::openapi::OpenApi>>,
282) -> axum::Json<utoipa::openapi::OpenApi> {
283    axum::Json((*document).clone())
284}
285
286fn cors_layer(ctx: &AppContext) -> CorsLayer {
287    let origins: Vec<HeaderValue> = ctx
288        .config
289        .http
290        .cors_allowed_origins
291        .iter()
292        .filter_map(|origin| origin.parse().ok())
293        .collect();
294
295    CorsLayer::new()
296        .allow_origin(origins)
297        .allow_methods([
298            Method::GET,
299            Method::POST,
300            Method::PUT,
301            Method::PATCH,
302            Method::DELETE,
303            Method::OPTIONS,
304        ])
305        .allow_headers([header::ACCEPT, header::AUTHORIZATION, header::CONTENT_TYPE])
306}
307
308const SCALAR_DOCS_CSP: &str = "default-src 'none'; script-src https://cdn.jsdelivr.net 'sha256-wT12sSim/cr/4i3SfCUXmSC76WSRp+uWevWj0uNZ/vU='; style-src 'unsafe-inline'; connect-src 'self'; img-src 'self' data: https:; font-src 'self' data: https:; base-uri 'none'; form-action 'none'; frame-ancestors 'none'";
309
310const SCALAR_DOCS_HTML: &str = r##"<!doctype html>
311<html lang="en">
312  <head>
313    <meta charset="utf-8" />
314    <meta name="viewport" content="width=device-width, initial-scale=1" />
315    <title>Lenso API Docs</title>
316    <script src="https://cdn.jsdelivr.net/npm/@scalar/api-reference@1.62.5" integrity="sha384-jVBCKhcCfx34USN27x4iQK1SBNdL/HxKq3KuBAxTS4WPaP5w80K4fjpwB+DezJL5" crossorigin="anonymous"></script>
317    <style>
318      body {
319        margin: 0;
320      }
321    </style>
322  </head>
323  <body>
324    <div id="app"></div>
325    <script>Scalar.createApiReference("#app",{url:"/openapi.json",theme:"default"});</script>
326  </body>
327</html>
328"##;
lenso_api/lib.rs

lenso_api/
lib.rs
