diff --git a/Cargo.lock b/Cargo.lock index e82a6264..372d3217 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -32,7 +32,7 @@ dependencies = [ "actix-utils", "base64 0.22.1", "bitflags 2.13.1", - "brotli 8.0.4", + "brotli", "bytes", "bytestring", "derive_more", @@ -675,18 +675,28 @@ dependencies = [ "pkg-config", ] +[[package]] +name = "aws_lambda_events" +version = "1.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d02c123e89527e7b424f74f52d11be0d17ef2887819323a42dcae1c7630ca53d" +dependencies = [ + "base64 0.22.1", + "bytes", + "http 1.4.2", + "http-body", + "http-serde", + "query_map", + "serde", + "serde_json", +] + [[package]] name = "base16ct" version = "0.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4c7f02d4ea65f2c1853089ffd8d2787bdbc63de2f0d29dedbcf8ccdfa0ccd4cf" -[[package]] -name = "base64" -version = "0.13.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9e1b586273c5702936fe7b7d6896644d8be71e6314cfe09d3167c95f712589e8" - [[package]] name = "base64" version = "0.21.7" @@ -784,17 +794,6 @@ dependencies = [ "piper", ] -[[package]] -name = "brotli" -version = "3.5.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d640d25bc63c50fb1f0b545ffd80207d2e10a4c965530809b40ba3386825c391" -dependencies = [ - "alloc-no-stdlib", - "alloc-stdlib", - "brotli-decompressor 2.5.1", -] - [[package]] name = "brotli" version = "8.0.4" @@ -803,17 +802,7 @@ checksum = "5cc91aac060a7a1e25823bdccbfb6af1875b88f17c6daac97894eed8207166b3" dependencies = [ "alloc-no-stdlib", "alloc-stdlib", - "brotli-decompressor 5.0.3", -] - -[[package]] -name = "brotli-decompressor" -version = "2.5.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4e2e4afe60d7dd600fdd3de8d0f08c2b7ec039712e3b6137ff98b7004e82de4f" -dependencies = [ - "alloc-no-stdlib", - "alloc-stdlib", + "brotli-decompressor", ] [[package]] @@ -852,6 +841,9 @@ name = "bytes" version = "1.12.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "fc652a48c352aef3ea3aed32080501cf3ef6ed5da78602a020c991775b0aff04" +dependencies = [ + "serde", +] [[package]] name = "bytesize" @@ -2149,15 +2141,37 @@ dependencies = [ [[package]] name = "http-body" -version = "0.4.6" +version = "1.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7ceab25649e9960c0311ea418d17bee82c0dcec1bd053b5f9a66e265a693bed2" +checksum = "ca2a8f2913ee65f60facd6a5905613afaa448497a0230cc41ce022d93290bc2c" dependencies = [ "bytes", - "http 0.2.12", + "http 1.4.2", +] + +[[package]] +name = "http-body-util" +version = "0.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e9f41fd6a08e4d4ec69df65976da761afd5ad5e58a9d4acb46bd1c953a9e3ff2" +dependencies = [ + "bytes", + "futures-core", + "http 1.4.2", + "http-body", "pin-project-lite", ] +[[package]] +name = "http-serde" +version = "2.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0f056c8559e3757392c8d091e796416e4649d8e49e88b8d76df6c002f05027fd" +dependencies = [ + "http 1.4.2", + "serde", +] + [[package]] name = "httparse" version = "1.10.1" @@ -2181,25 +2195,42 @@ dependencies = [ [[package]] name = "hyper" -version = "0.14.32" +version = "1.10.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "41dfc780fdec9373c01bae43289ea34c972e40ee3c9f6b3c8801a35f35586ce7" +checksum = "55281c53a1894c864990125767da440a4e630446785086f52523b20033b74498" dependencies = [ + "atomic-waker", "bytes", "futures-channel", "futures-core", - "futures-util", - "http 0.2.12", + "http 1.4.2", "http-body", "httparse", - "httpdate", "itoa", "pin-project-lite", - "socket2 0.5.10", + "smallvec", + "tokio", + "want", +] + +[[package]] +name = "hyper-util" +version = "0.1.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "96547c2556ec9d12fb1578c4eaf448b04993e7fb79cbaad930a656880a6bdfa0" +dependencies = [ + "bytes", + "futures-channel", + "futures-util", + "http 1.4.2", + "http-body", + "hyper", + "libc", + "pin-project-lite", + "socket2 0.6.5", "tokio", "tower-service", "tracing", - "want", ] [[package]] @@ -2510,36 +2541,49 @@ dependencies = [ ] [[package]] -name = "lambda-web" -version = "0.2.1" +name = "lambda_http" +version = "1.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ea6277b60649250d681654162b7e8e875c938295ea5f883eb9a8da7e27d2c051" +checksum = "e69eb3117d123f471f7d2b5d02b5afe150f54f9f7cff3264e578c9dd03675ad9" dependencies = [ - "actix-http", - "actix-service", - "actix-web", - "base64 0.13.1", - "brotli 3.5.0", + "aws_lambda_events", + "bytes", + "encoding_rs", + "futures-util", + "http 1.4.2", + "http-body", + "http-body-util", + "hyper", "lambda_runtime", + "mime", "percent-encoding", + "pin-project-lite", "serde", "serde_json", + "serde_urlencoded", + "tokio-stream", + "url", ] [[package]] name = "lambda_runtime" -version = "0.7.3" +version = "1.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bd32d5799db2155ae4d47116bb3e169b59f531ced4d5762a10c2125bdd2bf134" +checksum = "484647f147899f866a7b5db6aaa1671e943f89f4be76e2a383c96b1b27058294" dependencies = [ "async-stream", + "base64 0.22.1", "bytes", "futures", - "http 0.2.12", + "http 1.4.2", + "http-body-util", + "http-serde", "hyper", "lambda_runtime_api_client", + "pin-project", "serde", "serde_json", + "serde_path_to_error", "tokio", "tokio-stream", "tower", @@ -2548,14 +2592,19 @@ dependencies = [ [[package]] name = "lambda_runtime_api_client" -version = "0.7.0" +version = "1.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7210012be904051520f0dc502140ba599bae3042b65b3737b87727f1aa88a7d6" +checksum = "a92e6500e47d17c1ffd3e6ad3ca224bb86382fc8b63414f6f20b1b2d98dfb7cf" dependencies = [ - "http 0.2.12", + "bytes", + "futures-channel", + "futures-util", + "http 1.4.2", + "http-body", + "http-body-util", "hyper", - "tokio", - "tower-service", + "hyper-util", + "tower", ] [[package]] @@ -3702,6 +3751,17 @@ dependencies = [ "syn", ] +[[package]] +name = "query_map" +version = "0.7.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5eab6b8b1074ef3359a863758dae650c7c0c6027927a085b7af911c8e0bf3a15" +dependencies = [ + "form_urlencoded", + "serde", + "serde_derive", +] + [[package]] name = "quote" version = "1.0.46" @@ -4519,6 +4579,7 @@ dependencies = [ "actix-http", "actix-multipart", "actix-rt", + "actix-service", "actix-web", "actix-web-httpauth", "anyhow", @@ -4538,8 +4599,9 @@ dependencies = [ "futures-util", "handlebars", "hmac 0.13.0", + "http-body-util", "include_dir", - "lambda-web", + "lambda_http", "lettre", "libflate", "log", @@ -4738,6 +4800,12 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "sync_wrapper" +version = "1.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0bf256ce5efdfa370213c1dabab5935a12e49f2c58d15e9eac2870d3b4f27263" + [[package]] name = "synstructure" version = "0.13.2" @@ -4983,17 +5051,16 @@ dependencies = [ [[package]] name = "tower" -version = "0.4.13" +version = "0.5.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b8fa9be0de6cf49e536ce1851f987bd21a43b771b09473c3549a6c853db37c1c" +checksum = "ebe5ef63511595f1344e2d5cfa636d973292adc0eec1f0ad45fae9f0851ab1d4" dependencies = [ "futures-core", "futures-util", - "pin-project", "pin-project-lite", + "sync_wrapper", "tower-layer", "tower-service", - "tracing", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index b0cf50aa..ac5adb17 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -48,7 +48,16 @@ serde_json = { version = "1.0.82", features = [ "raw_value", "arbitrary_precision", ] } -lambda-web = { version = "0.2.1", features = ["actix4"], optional = true } +lambda_http = { version = "1.3.0", default-features = false, features = [ + "alb", + "apigw_http", + "apigw_rest", + "apigw_websockets", + "vpc_lattice", +], optional = true } +actix-http = { version = "3", optional = true } +actix-service = { version = "2", optional = true } +http-body-util = { version = "0.1", optional = true } sqlparser = { version = "0.62.0", default-features = false, features = [ "std", "visitor", @@ -104,7 +113,13 @@ opentelemetry-semantic-conventions = { version = "0.32", features = ["semconv_ex [features] default = [] odbc-static = ["odbc-sys", "odbc-sys/vendored-unix-odbc"] -lambda-web = ["dep:lambda-web", "odbc-static"] +lambda-web = [ + "dep:actix-http", + "dep:actix-service", + "dep:http-body-util", + "dep:lambda_http", + "odbc-static", +] [dev-dependencies] actix-http = "3" diff --git a/src/webserver/http.rs b/src/webserver/http.rs index 0f17bb95..d03dd4bd 100644 --- a/src/webserver/http.rs +++ b/src/webserver/http.rs @@ -625,8 +625,8 @@ pub async fn run_server(config: &AppConfig, state: AppState) -> anyhow::Result<( let factory = move || create_app(web::Data::clone(&state)); #[cfg(feature = "lambda-web")] - if lambda_web::is_running_on_lambda() { - lambda_web::run_actix_on_lambda(factory) + if super::lambda_http::is_running_on_lambda() { + super::lambda_http::run(factory) .await .map_err(|e| anyhow::anyhow!("Unable to start the lambda: {e}"))?; return Ok(()); diff --git a/src/webserver/lambda_http.rs b/src/webserver/lambda_http.rs new file mode 100644 index 00000000..6d3b9d2a --- /dev/null +++ b/src/webserver/lambda_http.rs @@ -0,0 +1,220 @@ +use std::{fmt, future::poll_fn, net::SocketAddr}; + +use actix_service::{IntoServiceFactory, Service, ServiceFactory}; +use actix_web::{body::MessageBody, dev::AppConfig, web::Bytes}; +use http_body_util::Full; +use lambda_http::{Request, RequestExt, Response, request::RequestContext, service_fn}; +use tokio::sync::{mpsc, oneshot}; + +type LambdaResponse = Response>; +type LambdaResult = Result; +type LambdaRequest = (Request, oneshot::Sender); + +pub fn is_running_on_lambda() -> bool { + std::env::var_os("AWS_LAMBDA_RUNTIME_API").is_some() +} + +pub async fn run(factory: F) -> Result<(), lambda_http::Error> +where + F: Fn() -> I + Send + Clone + 'static, + I: IntoServiceFactory, + S: ServiceFactory< + actix_http::Request, + Config = AppConfig, + Response = actix_web::dev::ServiceResponse, + Error = actix_web::Error, + > + 'static, + S::InitError: fmt::Debug, + B: MessageBody + 'static, + B::Error: fmt::Display, +{ + let service = factory() + .into_factory() + .new_service(AppConfig::default()) + .await + .map_err(|error| lambda_error(format!("failed to initialize Actix service: {error:?}")))?; + let (request_sender, mut request_receiver) = mpsc::channel::(1); + + actix_web::rt::spawn(async move { + while let Some((request, response_sender)) = request_receiver.recv().await { + let result = handle_request(&service, request).await; + let _ = response_sender.send(result); + } + }); + + lambda_http::run(service_fn(move |request| { + let request_sender = request_sender.clone(); + async move { + let (response_sender, response_receiver) = oneshot::channel(); + request_sender + .send((request, response_sender)) + .await + .map_err(|_| lambda_error("Actix Lambda worker is unavailable"))?; + response_receiver + .await + .map_err(|_| lambda_error("Actix Lambda worker dropped the response"))? + } + })) + .await +} + +async fn handle_request(service: &S, request: Request) -> LambdaResult +where + S: Service< + actix_http::Request, + Response = actix_web::dev::ServiceResponse, + Error = actix_web::Error, + >, + B: MessageBody, + B::Error: fmt::Display, +{ + let request = to_actix_request(request)?; + poll_fn(|context| service.poll_ready(context)) + .await + .map_err(lambda_error)?; + let response = service.call(request).await.map_err(lambda_error)?; + to_lambda_response(response).await +} + +fn to_actix_request(request: Request) -> Result { + let peer_addr = source_ip(&request).map(|ip| SocketAddr::new(ip, 0)); + let (parts, body) = request.into_parts(); + let mut request = actix_http::Request::new(); + let head = request.head_mut(); + + head.method = actix_web::http::Method::from_bytes(parts.method.as_str().as_bytes()) + .map_err(lambda_error)?; + let path_and_query = parts + .uri + .path_and_query() + .map_or("/", lambda_http::http::uri::PathAndQuery::as_str); + head.uri = path_and_query.parse().map_err(lambda_error)?; + head.version = match parts.version { + lambda_http::http::Version::HTTP_09 => actix_web::http::Version::HTTP_09, + lambda_http::http::Version::HTTP_10 => actix_web::http::Version::HTTP_10, + lambda_http::http::Version::HTTP_2 => actix_web::http::Version::HTTP_2, + lambda_http::http::Version::HTTP_3 => actix_web::http::Version::HTTP_3, + _ => actix_web::http::Version::HTTP_11, + }; + head.peer_addr = peer_addr.or_else(|| forwarded_peer_addr(&parts.headers)); + + for (name, value) in &parts.headers { + let name = actix_web::http::header::HeaderName::from_bytes(name.as_str().as_bytes()) + .map_err(lambda_error)?; + let value = actix_web::http::header::HeaderValue::from_bytes(value.as_bytes()) + .map_err(lambda_error)?; + head.headers.append(name, value); + } + *request.payload() = actix_http::Payload::from(Bytes::copy_from_slice(body.as_ref())); + Ok(request) +} + +async fn to_lambda_response(response: actix_web::dev::ServiceResponse) -> LambdaResult +where + B: MessageBody, + B::Error: fmt::Display, +{ + let status = lambda_http::http::StatusCode::from_u16(response.status().as_u16()) + .map_err(lambda_error)?; + let mut result = Response::builder() + .status(status) + .body(Full::new(Bytes::new()))?; + + for (name, value) in response.headers() { + let name = lambda_http::http::header::HeaderName::from_bytes(name.as_str().as_bytes()) + .map_err(lambda_error)?; + let value = lambda_http::http::header::HeaderValue::from_bytes(value.as_bytes()) + .map_err(lambda_error)?; + result.headers_mut().append(name, value); + } + + let body = actix_web::body::to_bytes(response.into_body()) + .await + .map_err(lambda_error)?; + *result.body_mut() = Full::new(body); + Ok(result) +} + +fn source_ip(request: &Request) -> Option { + let source_ip = match request.request_context_ref()? { + RequestContext::ApiGatewayV1(context) => context.identity.source_ip.as_deref(), + RequestContext::ApiGatewayV2(context) => context.http.source_ip.as_deref(), + RequestContext::WebSocket(context) => context.identity.source_ip.as_deref(), + _ => None, + }?; + source_ip.parse().ok() +} + +fn forwarded_peer_addr(headers: &lambda_http::http::HeaderMap) -> Option { + let forwarded_for = headers.get(lambda_http::http::header::HeaderName::from_static( + "x-forwarded-for", + ))?; + let ip = forwarded_for + .to_str() + .ok()? + .split(',') + .next()? + .trim() + .parse() + .ok()?; + Some(SocketAddr::new(ip, 0)) +} + +fn lambda_error(error: impl fmt::Display) -> lambda_http::Error { + std::io::Error::other(error.to_string()).into() +} + +#[cfg(test)] +mod tests { + use actix_web::{HttpMessage, HttpResponse}; + use futures_util::StreamExt; + use lambda_http::http::header::{HeaderName, HeaderValue}; + + use super::*; + + #[actix_web::test] + async fn converts_lambda_request_without_losing_duplicate_headers() { + let mut request = lambda_http::http::Request::builder() + .method("POST") + .uri("/path?one=1&one=2") + .body(lambda_http::Body::Text("hello".to_owned())) + .unwrap(); + request.headers_mut().append( + HeaderName::from_static("x-value"), + HeaderValue::from_static("first"), + ); + request.headers_mut().append( + HeaderName::from_static("x-value"), + HeaderValue::from_static("second"), + ); + + let mut request = to_actix_request(request).unwrap(); + assert_eq!(request.method(), actix_web::http::Method::POST); + assert_eq!(request.uri(), "/path?one=1&one=2"); + assert_eq!( + request + .headers() + .get_all("x-value") + .map(|value| value.to_str().unwrap()) + .collect::>(), + ["first", "second"] + ); + let body = request.take_payload().next().await.unwrap().unwrap(); + assert_eq!(body, Bytes::from_static(b"hello")); + } + + #[actix_web::test] + async fn converts_actix_response_without_losing_cookies() { + let request = actix_web::test::TestRequest::default().to_http_request(); + let response = HttpResponse::Created() + .append_header(("set-cookie", "one=1")) + .append_header(("set-cookie", "two=2")) + .content_type("text/plain") + .body("hello"); + let response = actix_web::dev::ServiceResponse::new(request, response); + + let response = to_lambda_response(response).await.unwrap(); + assert_eq!(response.status(), lambda_http::http::StatusCode::CREATED); + assert_eq!(response.headers().get_all("set-cookie").iter().count(), 2); + } +} diff --git a/src/webserver/mod.rs b/src/webserver/mod.rs index 97448514..c1d00a7d 100644 --- a/src/webserver/mod.rs +++ b/src/webserver/mod.rs @@ -38,6 +38,8 @@ pub mod http_client; pub mod http_metrics; pub mod http_request_info; mod https; +#[cfg(feature = "lambda-web")] +mod lambda_http; pub mod request_variables; pub mod server_timing;