diff --git a/examples/timing_attack_check.rs b/examples/timing_attack_check.rs index 90e3308..5929f9f 100644 --- a/examples/timing_attack_check.rs +++ b/examples/timing_attack_check.rs @@ -1,7 +1,7 @@ use dudect_bencher::{BenchRng, Class, CtRunner, ctbench_main}; use commit_bridge::domain::NonEmptyString; -use commit_bridge::verify_api_key; +use commit_bridge::http::router::verify_api_key; use dudect_bencher::rand::RngExt; use rand::distr::{Alphanumeric, SampleString}; diff --git a/src/http.rs b/src/http.rs new file mode 100644 index 0000000..404d2af --- /dev/null +++ b/src/http.rs @@ -0,0 +1,20 @@ +//! HTTP networking: router wiring, server runtime, and the outbound HTTP client. + +pub mod handler; +pub mod router; +pub(crate) mod server; +pub mod state; + +use reqwest::Client; + +use crate::{config::Config, error::ClientCreationError}; + +/// Creates a new HTTP client. +pub(crate) fn build_http_client(config: &Config) -> Result { + let client = Client::builder() + .user_agent(config.server.user_agent.to_string()) + .timeout(config.server.out_request_timeout) + .build()?; + + Ok(client) +} diff --git a/src/handler.rs b/src/http/handler.rs similarity index 99% rename from src/handler.rs rename to src/http/handler.rs index 8859e2d..8843e66 100644 --- a/src/handler.rs +++ b/src/http/handler.rs @@ -56,8 +56,8 @@ mod tests { use super::update::update_subscription_inner; use crate::domain::{BranchName, EventType, RepoUrl, TargetRepo}; use crate::error::HandlerError; + use crate::http::state::AppState; use crate::model::{CreateSubscription, UpdateSubscription}; - use crate::state::AppState; use crate::test_utils::create_test_db; use axum::Json; use axum::extract::{Path, Query, State}; diff --git a/src/handler/create.rs b/src/http/handler/create.rs similarity index 98% rename from src/handler/create.rs rename to src/http/handler/create.rs index 7a743dd..9eef579 100644 --- a/src/handler/create.rs +++ b/src/http/handler/create.rs @@ -2,9 +2,9 @@ use super::map_to_hal; use crate::error::HandlerError; +use crate::http::state::AppState; use crate::model::{CreateSubscription, SubscriptionHal}; use crate::repository::subscription::SubscriptionRepository; -use crate::state::AppState; use axum::{Json, extract::State}; use rovo::rovo; use tracing::{info, instrument}; diff --git a/src/handler/delete.rs b/src/http/handler/delete.rs similarity index 97% rename from src/handler/delete.rs rename to src/http/handler/delete.rs index 3bdb85c..78f6da0 100644 --- a/src/handler/delete.rs +++ b/src/http/handler/delete.rs @@ -1,8 +1,8 @@ //! Delete a subscription handler. use crate::error::HandlerError; +use crate::http::state::AppState; use crate::repository::subscription::SubscriptionRepository; -use crate::state::AppState; use axum::extract::{Path, State}; use rovo::rovo; use tracing::instrument; diff --git a/src/handler/get.rs b/src/http/handler/get.rs similarity index 97% rename from src/handler/get.rs rename to src/http/handler/get.rs index bd6ea8c..9872d14 100644 --- a/src/handler/get.rs +++ b/src/http/handler/get.rs @@ -2,9 +2,9 @@ use super::map_to_hal; use crate::error::HandlerError; +use crate::http::state::AppState; use crate::model::SubscriptionHal; use crate::repository::subscription::SubscriptionRepository; -use crate::state::AppState; use axum::{ Json, extract::{Path, State}, diff --git a/src/handler/list.rs b/src/http/handler/list.rs similarity index 98% rename from src/handler/list.rs rename to src/http/handler/list.rs index 7f9df7d..b217854 100644 --- a/src/handler/list.rs +++ b/src/http/handler/list.rs @@ -2,9 +2,9 @@ use super::map_to_hal; use crate::error::HandlerError; +use crate::http::state::AppState; use crate::model::{HalLink, SubscriptionHal, SubscriptionPage, SubscriptionPageLinks}; use crate::repository::subscription::SubscriptionRepository; -use crate::state::AppState; use axum::{ Json, extract::{Query, State}, diff --git a/src/handler/update.rs b/src/http/handler/update.rs similarity index 98% rename from src/handler/update.rs rename to src/http/handler/update.rs index cf3011c..8a5ab67 100644 --- a/src/handler/update.rs +++ b/src/http/handler/update.rs @@ -2,9 +2,9 @@ use super::map_to_hal; use crate::error::HandlerError; +use crate::http::state::AppState; use crate::model::{SubscriptionHal, UpdateSubscription}; use crate::repository::subscription::SubscriptionRepository; -use crate::state::AppState; use axum::{ Json, extract::{Path, State}, diff --git a/src/http/router.rs b/src/http/router.rs new file mode 100644 index 0000000..cd701df --- /dev/null +++ b/src/http/router.rs @@ -0,0 +1,223 @@ +//! Axum router wiring and middlewares. + +use std::time::Duration; + +use axum::{ + Router, + body::Body, + extract::{MatchedPath, State}, + http::{HeaderValue, Request, Response, StatusCode, header}, + middleware::{self, Next}, + response::IntoResponse, +}; +use rovo::Router as RovoRouter; +use rovo::aide::openapi::OpenApi; +use rovo::rovo; +use subtle::ConstantTimeEq; +use tower_http::timeout::TimeoutLayer; +use tower_http::trace::{MakeSpan, OnResponse, TraceLayer}; +use tracing::Span; + +use crate::{ + config::Config, + domain::NonEmptyString, + http::handler::{ + create_subscription, delete_subscription, get_subscription, list_subscriptions, + update_subscription, + }, + http::state::AppState, +}; + +/// Middleware to authorize requests with an API key. +/// +/// Records the `authenticated` attribute onto the `http.request` span +/// (the span is current while this middleware runs, +/// since it is layered below the `TraceLayer`). +async fn auth_middleware( + State(state): State, + req: Request, + next: Next, +) -> Response { + let needs_authentication = + req.uri().path().starts_with("/subscriptions") && !state.config.auth.allow_unauthenticated; + + if needs_authentication { + let auth_header = req.headers().get("X-API-KEY").and_then(|v| v.to_str().ok()); + + if !verify_api_key(state.config.auth.api_key.as_ref(), auth_header) { + tracing::Span::current().record("authenticated", false); + return StatusCode::UNAUTHORIZED.into_response(); + } + + tracing::Span::current().record("authenticated", true); + } + + next.run(req).await +} + +/// Uses constant-time verification to check API key correspondence. +/// +/// If the API keys correspond, returns `true`, +/// otherwise `false`. +/// +/// ### Cryptographic security +/// +/// The function short-circuits if lengths of `expected` and `provided` are unequal. +/// While this allows an attacker to extract the key length, +/// it is order of magnitudes safer than using a simple string equality test, +/// which would allow the attacker to gradually know the exact key +/// over many requests. +pub fn verify_api_key(expected: Option<&NonEmptyString>, provided: Option<&str>) -> bool { + expected.zip(provided).is_some_and(|(key, header)| { + let key_bytes = key.as_bytes(); + let header_bytes = header.as_bytes(); + key_bytes.ct_eq(header_bytes).into() + }) +} + +/// Middleware to set Cache-Control header. +async fn set_no_cache_header(req: Request, next: Next) -> Response { + let path = req.uri().path().to_string(); + let mut response = next.run(req).await; + if path.starts_with("/subscriptions") { + response.headers_mut().insert( + header::CACHE_CONTROL, + HeaderValue::from_static("private, no-cache, no-store, must-revalidate, max-age=0"), + ); + } + response +} + +/// Records the matched route path (`http.route`) +/// onto the current HTTP request span. +/// +/// Runs after routing, so the [`MatchedPath`] extension is available. +async fn record_http_route(req: Request, next: Next) -> Response { + if let Some(matched) = req.extensions().get::() { + tracing::Span::current().record("http.route", matched.as_str()); + } + next.run(req).await +} + +#[allow(missing_docs, clippy::missing_docs_in_private_items)] +mod health_handler { + use super::*; + #[rovo] + #[tracing::instrument(skip_all, fields(otel.kind = "internal"))] + pub async fn health_check(State(_state): State) -> &'static str { + "CommitBridge is alive" + } +} + +/// Span factory for incoming HTTP requests, +/// following OpenTelemetry semantic conventions. +/// +/// The span is created within this crate +/// (instead of using the default `tower_http` span factory) +/// so that its attributes follow OpenTelemetry conventions. +#[derive(Clone, Copy)] +struct HttpRequestSpan; + +impl MakeSpan for HttpRequestSpan { + fn make_span(&mut self, request: &Request) -> Span { + tracing::info_span!( + "http.request", + otel.kind = "server", + http.request.method = %request.method(), + url.path = %request.uri().path(), + http.route = tracing::field::Empty, + http.response.status_code = tracing::field::Empty, + otel.status_code = tracing::field::Empty, + error.type = tracing::field::Empty, + authenticated = tracing::field::Empty, + ) + } +} + +/// Records HTTP response metadata onto the request span, +/// following OpenTelemetry semantic conventions. +/// +/// Must be used with [`HttpRequestSpan`], +/// which declares the fields recorded here. +#[derive(Clone, Copy)] +pub(crate) struct HttpRequestOnResponse { + /// Whether client error responses (4xx) + /// should be marked as errors in exported traces. + mark_client_errors: bool, +} + +impl HttpRequestOnResponse { + /// Creates a new [`HttpRequestOnResponse`]. + pub(crate) const fn new(mark_client_errors: bool) -> Self { + Self { mark_client_errors } + } + + /// Returns `true` if the response status should be marked as an error + /// in exported traces. + /// + /// Server errors (5xx) are always marked; + /// client errors (4xx) are only marked when `mark_client_errors` is set. + pub(crate) fn should_mark_error(&self, status: StatusCode) -> bool { + status.is_server_error() || (self.mark_client_errors && status.is_client_error()) + } +} + +impl OnResponse for HttpRequestOnResponse { + fn on_response(self, response: &Response, _latency: Duration, span: &Span) { + span.record("http.response.status_code", response.status().as_u16()); + if self.should_mark_error(response.status()) { + span.record("otel.status_code", "ERROR"); + span.record("error.type", response.status().as_u16().to_string()); + } + } +} + +/// Builds the application router. +pub fn build_router( + repository: std::sync::Arc, + config: &Config, +) -> Router { + let state = AppState { + config: std::sync::Arc::new(config.clone()), + repository, + }; + + let mut api = OpenApi::default(); + api.info.title = "CommitBridge API".to_string(); + api.info.description = + Some("API for managing repository subscriptions and triggering workflows".to_string()); + + let subscriptions = RovoRouter::::new() + .route( + "/", + rovo::routing::post(create_subscription).get(list_subscriptions), + ) + .route( + "/{id}", + rovo::routing::get(get_subscription) + .patch(update_subscription) + .delete(delete_subscription), + ); + + RovoRouter::::new() + .route("/health", rovo::routing::get(health_handler::health_check)) + .nest("/subscriptions", subscriptions) + .with_oas(api) + .with_scalar("/scalar") + .with_state(state.clone()) + .finish() + .layer(middleware::from_fn_with_state(state, auth_middleware)) + .layer(middleware::from_fn(set_no_cache_header)) + .layer(middleware::from_fn(record_http_route)) + .layer(TimeoutLayer::with_status_code( + StatusCode::REQUEST_TIMEOUT, + config.server.in_request_timeout, + )) + .layer( + TraceLayer::new_for_http() + .make_span_with(HttpRequestSpan) + .on_response(HttpRequestOnResponse::new( + config.telemetry.mark_client_errors_as_error, + )), + ) +} diff --git a/src/http/server.rs b/src/http/server.rs new file mode 100644 index 0000000..4332980 --- /dev/null +++ b/src/http/server.rs @@ -0,0 +1,55 @@ +//! HTTP server runtime and lifecycle. + +use axum::Router; +use tokio::signal; +use tokio_util::sync::CancellationToken; +use tracing::info; + +use crate::{config::Config, error::FatalError}; + +/// Runs the server. +pub(crate) async fn run_server( + app: Router, + config: &Config, + token: CancellationToken, +) -> Result<(), FatalError> { + let listener = tokio::net::TcpListener::bind(config.server.address) + .await + .map_err(FatalError::TcpBinding)?; + println!("Server listening on http://{}", config.server.address); + println!( + "Scalar UI available at http://{}/scalar", + config.server.address + ); + + axum::serve(listener, app) + .with_graceful_shutdown(shutdown_signal(token)) + .await + .map_err(FatalError::Serve)?; + + Ok(()) +} + +/// Creates a future that resolves when a termination signal is received. +async fn shutdown_signal(token: CancellationToken) { + let ctrl_c = signal::ctrl_c(); + + #[cfg(unix)] + let terminate = async { + if let Ok(mut signal) = signal::unix::signal(signal::unix::SignalKind::terminate()) { + signal.recv().await; + } else { + std::future::pending::<()>().await; + } + }; + + #[cfg(not(unix))] + let terminate = std::future::pending::<()>(); + + tokio::select! { + _ = ctrl_c => {}, + _ = terminate => {}, + _ = token.cancelled() => {}, + } + info!("Shutdown signal received, initiating graceful shutdown..."); +} diff --git a/src/state.rs b/src/http/state.rs similarity index 91% rename from src/state.rs rename to src/http/state.rs index 7c6d821..7d9a03b 100644 --- a/src/state.rs +++ b/src/http/state.rs @@ -6,7 +6,7 @@ use std::sync::Arc; /// Holds data accessible from each [handler]. /// /// -/// [handler]: crate::handler +/// [handler]: crate::http::handler #[derive(Debug, Clone)] pub struct AppState { /// Application configuration. diff --git a/src/lib.rs b/src/lib.rs index 642c590..cd25ac3 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -10,41 +10,18 @@ )] use std::fs; -use std::time::Duration; -use axum::{ - Router, - body::Body, - extract::{MatchedPath, State}, - http::{HeaderValue, Request, Response, StatusCode, header}, - middleware::{self, Next}, - response::IntoResponse, -}; use jsonwebtoken::EncodingKey; use reqwest::Client; -use rovo::Router as RovoRouter; -use rovo::aide::openapi::OpenApi; -use rovo::rovo; -use subtle::ConstantTimeEq; -use tokio::signal; use tokio_util::sync::CancellationToken; use tokio_util::task::TaskTracker; -use tower_http::timeout::TimeoutLayer; -use tower_http::trace::{MakeSpan, OnResponse, TraceLayer}; -use tracing::{Span, info}; use crate::{ config::Config, context::SharedContext, - domain::NonEmptyString, engine::AsyncEngine, - error::{ClientCreationError, FatalError}, - handler::{ - create_subscription, delete_subscription, get_subscription, list_subscriptions, - update_subscription, - }, + error::FatalError, polling::PollingEngine, - state::AppState, trigger::{GitHubAuthenticator, TriggerEngine}, }; @@ -54,11 +31,10 @@ pub mod context; pub mod domain; pub mod engine; pub mod error; -pub mod handler; +pub mod http; pub mod model; pub mod polling; pub mod repository; -pub mod state; pub mod telemetry; #[cfg(test)] mod test_utils; @@ -74,7 +50,7 @@ pub async fn run_app(tracker: &TaskTracker, token: &CancellationToken) -> Result let config = Config::load()?; let repository = std::sync::Arc::new(crate::repository::SqliteRepository::connect(&config.database).await?); - let http_client = build_http_client(&config)?; + let http_client = crate::http::build_http_client(&config)?; let ctx = init_context(repository.clone(), config.clone(), token.clone())?; @@ -87,9 +63,9 @@ pub async fn run_app(tracker: &TaskTracker, token: &CancellationToken) -> Result crate::engine::start_engine(engine, message, tracker); } - let app = build_router(repository, &config); + let app = crate::http::router::build_router(repository, &config); - run_server(app, &ctx.config, token.clone()).await?; + crate::http::server::run_server(app, &ctx.config, token.clone()).await?; Ok(()) } @@ -159,254 +135,3 @@ fn init_engines(ctx: &SharedContext, http_client: Client) -> Result, - req: Request, - next: Next, -) -> Response { - let needs_authentication = - req.uri().path().starts_with("/subscriptions") && !state.config.auth.allow_unauthenticated; - - if needs_authentication { - let auth_header = req.headers().get("X-API-KEY").and_then(|v| v.to_str().ok()); - - if !verify_api_key(state.config.auth.api_key.as_ref(), auth_header) { - tracing::Span::current().record("authenticated", false); - return StatusCode::UNAUTHORIZED.into_response(); - } - - tracing::Span::current().record("authenticated", true); - } - - next.run(req).await -} - -/// Uses constant-time verification to check API key correspondence. -/// -/// If the API keys correspond, returns `true`, -/// otherwise `false`. -/// -/// ### Cryptographic security -/// -/// The function short-circuits if lengths of `expected` and `provided` are unequal. -/// While this allows an attacker to extract the key length, -/// it is order of magnitudes safer than using a simple string equality test, -/// which would allow the attacker to gradually know the exact key -/// over many requests. -pub fn verify_api_key(expected: Option<&NonEmptyString>, provided: Option<&str>) -> bool { - expected.zip(provided).is_some_and(|(key, header)| { - let key_bytes = key.as_bytes(); - let header_bytes = header.as_bytes(); - key_bytes.ct_eq(header_bytes).into() - }) -} - -/// Middleware to set Cache-Control header. -async fn set_no_cache_header(req: Request, next: Next) -> Response { - let path = req.uri().path().to_string(); - let mut response = next.run(req).await; - if path.starts_with("/subscriptions") { - response.headers_mut().insert( - header::CACHE_CONTROL, - HeaderValue::from_static("private, no-cache, no-store, must-revalidate, max-age=0"), - ); - } - response -} - -/// Records the matched route path (`http.route`) -/// onto the current HTTP request span. -/// -/// Runs after routing, so the [`MatchedPath`] extension is available. -async fn record_http_route(req: Request, next: Next) -> Response { - if let Some(matched) = req.extensions().get::() { - tracing::Span::current().record("http.route", matched.as_str()); - } - next.run(req).await -} - -#[allow(missing_docs, clippy::missing_docs_in_private_items)] -mod health_handler { - use super::*; - #[rovo] - #[tracing::instrument(skip_all, fields(otel.kind = "internal"))] - pub async fn health_check(State(_state): State) -> &'static str { - "CommitBridge is alive" - } -} - -/// Span factory for incoming HTTP requests, -/// following OpenTelemetry semantic conventions. -/// -/// The span is created within this crate -/// (instead of using the default `tower_http` span factory) -/// so that its attributes follow OpenTelemetry conventions. -#[derive(Clone, Copy)] -struct HttpRequestSpan; - -impl MakeSpan for HttpRequestSpan { - fn make_span(&mut self, request: &Request) -> Span { - tracing::info_span!( - "http.request", - otel.kind = "server", - http.request.method = %request.method(), - url.path = %request.uri().path(), - http.route = tracing::field::Empty, - http.response.status_code = tracing::field::Empty, - otel.status_code = tracing::field::Empty, - error.type = tracing::field::Empty, - authenticated = tracing::field::Empty, - ) - } -} - -/// Records HTTP response metadata onto the request span, -/// following OpenTelemetry semantic conventions. -/// -/// Must be used with [`HttpRequestSpan`], -/// which declares the fields recorded here. -#[derive(Clone, Copy)] -pub(crate) struct HttpRequestOnResponse { - /// Whether client error responses (4xx) - /// should be marked as errors in exported traces. - mark_client_errors: bool, -} - -impl HttpRequestOnResponse { - /// Creates a new [`HttpRequestOnResponse`]. - pub(crate) const fn new(mark_client_errors: bool) -> Self { - Self { mark_client_errors } - } - - /// Returns `true` if the response status should be marked as an error - /// in exported traces. - /// - /// Server errors (5xx) are always marked; - /// client errors (4xx) are only marked when `mark_client_errors` is set. - pub(crate) fn should_mark_error(&self, status: StatusCode) -> bool { - status.is_server_error() || (self.mark_client_errors && status.is_client_error()) - } -} - -impl OnResponse for HttpRequestOnResponse { - fn on_response(self, response: &Response, _latency: Duration, span: &Span) { - span.record("http.response.status_code", response.status().as_u16()); - if self.should_mark_error(response.status()) { - span.record("otel.status_code", "ERROR"); - span.record("error.type", response.status().as_u16().to_string()); - } - } -} - -/// Builds the application router. -pub fn build_router( - repository: std::sync::Arc, - config: &Config, -) -> Router { - let state = AppState { - config: std::sync::Arc::new(config.clone()), - repository, - }; - - let mut api = OpenApi::default(); - api.info.title = "CommitBridge API".to_string(); - api.info.description = - Some("API for managing repository subscriptions and triggering workflows".to_string()); - - let subscriptions = RovoRouter::::new() - .route( - "/", - rovo::routing::post(create_subscription).get(list_subscriptions), - ) - .route( - "/{id}", - rovo::routing::get(get_subscription) - .patch(update_subscription) - .delete(delete_subscription), - ); - - RovoRouter::::new() - .route("/health", rovo::routing::get(health_handler::health_check)) - .nest("/subscriptions", subscriptions) - .with_oas(api) - .with_scalar("/scalar") - .with_state(state.clone()) - .finish() - .layer(middleware::from_fn_with_state(state, auth_middleware)) - .layer(middleware::from_fn(set_no_cache_header)) - .layer(middleware::from_fn(record_http_route)) - .layer(TimeoutLayer::with_status_code( - StatusCode::REQUEST_TIMEOUT, - config.server.in_request_timeout, - )) - .layer( - TraceLayer::new_for_http() - .make_span_with(HttpRequestSpan) - .on_response(HttpRequestOnResponse::new( - config.telemetry.mark_client_errors_as_error, - )), - ) -} - -/// Runs the server. -async fn run_server( - app: Router, - config: &Config, - token: CancellationToken, -) -> Result<(), FatalError> { - let listener = tokio::net::TcpListener::bind(config.server.address) - .await - .map_err(FatalError::TcpBinding)?; - println!("Server listening on http://{}", config.server.address); - println!( - "Scalar UI available at http://{}/scalar", - config.server.address - ); - - axum::serve(listener, app) - .with_graceful_shutdown(shutdown_signal(token)) - .await - .map_err(FatalError::Serve)?; - - Ok(()) -} - -/// Creates a future that resolves when a termination signal is received. -async fn shutdown_signal(token: CancellationToken) { - let ctrl_c = signal::ctrl_c(); - - #[cfg(unix)] - let terminate = async { - if let Ok(mut signal) = signal::unix::signal(signal::unix::SignalKind::terminate()) { - signal.recv().await; - } else { - std::future::pending::<()>().await; - } - }; - - #[cfg(not(unix))] - let terminate = std::future::pending::<()>(); - - tokio::select! { - _ = ctrl_c => {}, - _ = terminate => {}, - _ = token.cancelled() => {}, - } - info!("Shutdown signal received, initiating graceful shutdown..."); -} - -/// Creates a new HTTP client. -pub fn build_http_client(config: &Config) -> Result { - let client = Client::builder() - .user_agent(config.server.user_agent.to_string()) - .timeout(config.server.out_request_timeout) - .build()?; - - Ok(client) -} diff --git a/src/tests/api_routes.rs b/src/tests/api_routes.rs index 6b7feb5..d87083d 100644 --- a/src/tests/api_routes.rs +++ b/src/tests/api_routes.rs @@ -1,4 +1,4 @@ -use crate::{build_router, repository::SqliteRepository, test_utils::create_test_db}; +use crate::{http::router::build_router, repository::SqliteRepository, test_utils::create_test_db}; use axum::{ body::Body, http::{Request, StatusCode}, diff --git a/src/tests/auth_tests.rs b/src/tests/auth_tests.rs index fa3fea9..8492306 100644 --- a/src/tests/auth_tests.rs +++ b/src/tests/auth_tests.rs @@ -1,5 +1,6 @@ use crate::{ - build_router, domain::NonEmptyString, repository::SqliteRepository, test_utils::create_test_db, + domain::NonEmptyString, http::router::build_router, repository::SqliteRepository, + test_utils::create_test_db, }; use axum::{ body::Body, diff --git a/src/tests/mark_error_tests.rs b/src/tests/mark_error_tests.rs index d56309b..8350f5a 100644 --- a/src/tests/mark_error_tests.rs +++ b/src/tests/mark_error_tests.rs @@ -1,4 +1,4 @@ -use crate::HttpRequestOnResponse; +use crate::http::router::HttpRequestOnResponse; use axum::http::StatusCode; #[test]