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

lenso_api/
lib.rs
