diff --git a/Cargo.lock b/Cargo.lock index cccce54..ac3695e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1677,6 +1677,7 @@ dependencies = [ "dotenvy", "hyper", "hyper-util", + "reqwest", "signal-gateway", "syslog_rfc5424", "tokio", diff --git a/signal-gateway-bin/Cargo.toml b/signal-gateway-bin/Cargo.toml index ed9aac0..8e10fa5 100644 --- a/signal-gateway-bin/Cargo.toml +++ b/signal-gateway-bin/Cargo.toml @@ -23,6 +23,7 @@ conf-extra = { workspace = true } dotenvy = { workspace = true } hyper = { workspace = true } hyper-util = { workspace = true } +reqwest = { workspace = true } syslog_rfc5424 = { workspace = true } tokio = { workspace = true, features = ["rt-multi-thread", "signal"] } tokio-util = { workspace = true } diff --git a/signal-gateway-bin/src/admin_http/mod.rs b/signal-gateway-bin/src/admin_http/mod.rs new file mode 100644 index 0000000..0bbb1ed --- /dev/null +++ b/signal-gateway-bin/src/admin_http/mod.rs @@ -0,0 +1,64 @@ +//! Admin HTTP client for forwarding messages to an HTTP server +//! +//! This module handles admin messages not handled by the gateway by making an HTTP POST request +//! with the message as the body, and returning the response body as the reply. + +use conf::Conf; +use signal_gateway::{AdminMessageResponse, MessageHandler, MessageHandlerResult}; +use std::time::Duration; + +/// Configuration for the admin HTTP client +#[derive(Clone, Conf, Debug)] +pub struct AdminHttpConfig { + /// URL to POST admin commands to + #[conf(long, env)] + pub url: String, + /// Timeout for the HTTP request + #[conf(long, env, default_value = "5s", value_parser = conf_extra::parse_duration)] + pub timeout: Duration, +} + +impl AdminHttpConfig { + /// Create a message handler function from this config. + /// + /// The returned handler makes an HTTP POST request to the configured URL + /// with the message as the body, and returns the response body. + pub fn into_handler(self) -> MessageHandler { + let client = reqwest::Client::builder() + .timeout(self.timeout) + .build() + .expect("Failed to build HTTP client"); + + Box::new(move |message: String| { + let client = client.clone(); + let url = self.url.clone(); + Box::pin(async move { handle_message(&client, &url, message).await }) + }) + } +} + +/// Handle a message by POSTing it to the configured HTTP server +async fn handle_message( + client: &reqwest::Client, + url: &str, + message: String, +) -> MessageHandlerResult { + let response = client + .post(url) + .body(message) + .send() + .await + .map_err(|err| (502u16, format!("HTTP request failed: {err}").into()))?; + + let status = response.status(); + let body = response + .text() + .await + .map_err(|err| (502u16, format!("Failed to read response body: {err}").into()))?; + + if !status.is_success() { + return Err((status.as_u16(), body.into())); + } + + Ok(AdminMessageResponse::new(body)) +} diff --git a/signal-gateway-bin/src/main.rs b/signal-gateway-bin/src/main.rs index 2308b3b..b3a1062 100644 --- a/signal-gateway-bin/src/main.rs +++ b/signal-gateway-bin/src/main.rs @@ -1,20 +1,44 @@ -use conf::Conf; +use conf::{Conf, Subcommands}; use hyper::service::service_fn; use hyper_util::rt::TokioIo; use hyper_util::server::conn::auto; -use signal_gateway::{Gateway, GatewayConfig}; +use signal_gateway::{Gateway, GatewayConfig, MessageHandler}; use std::{net::SocketAddr, sync::Arc, time::Duration}; use tokio::net::TcpListener; use tokio_util::sync::CancellationToken; use tracing::{error, info, warn}; use tracing_subscriber::EnvFilter; +mod admin_http; +use admin_http::AdminHttpConfig; + mod admin_netcat; use admin_netcat::AdminNetcatConfig; mod syslog; use syslog::SyslogUdpConfig; +/// Admin message handler configuration - select how non-command messages are handled +#[derive(Clone, Debug, Subcommands)] +enum AdminHandlerCommand { + /// Forward (unhandled) admin messages to a TCP endpoint (netcat-style) + /// Useful if an http server would be heavy in the target process + #[conf(name = "admin-netcat")] + Netcat(AdminNetcatConfig), + /// Forward (unhandled) admin messages to an HTTP endpoint via POST + #[conf(name = "admin-http")] + Http(AdminHttpConfig), +} + +impl AdminHandlerConfig { + fn into_handler(self) -> MessageHandler { + match self { + AdminHandlerConfig::Netcat(config) => config.into_handler(), + AdminHandlerConfig::Http(config) => config.into_handler(), + } + } +} + #[derive(Conf, Debug)] struct Config { /// If true, just validate config and don't start @@ -25,8 +49,9 @@ struct Config { http_listen_addr: SocketAddr, #[conf(flatten, prefix)] syslog_udp: Option, - #[conf(flatten, prefix)] - admin_netcat: Option, + /// Optional admin message handler (netcat or http) + #[conf(subcommands)] + admin_handler: Option, #[conf(flatten)] gateway: GatewayConfig, } @@ -73,7 +98,7 @@ async fn main() { let token = CancellationToken::new(); - let message_handler = config.admin_netcat.map(|c| c.into_handler()); + let message_handler = config.admin_handler.map(|c| c.into_handler()); let gateway = Arc::new(Gateway::new(config.gateway, token.clone(), message_handler).await); let listener = TcpListener::bind(config.http_listen_addr).await.unwrap();