Delete standalone ohttp-relay listeners

These were previously necessary for the integration test, but we can now
use an actual mailroom instance instead of a custom ohttp_relay
listener.

The unix socket test wasn't adding much value AFAICT so I deleted it.
Unix sockets are supported by the mailroom `listener` parameter but we'd
just be testing an external dependency, not internal logic.

Author: spacebear21

Cargo-minimal.lock

@@ -2871,7 +2871,6 @@ dependencies = [
  "tokio-rustls-acme",
  "tokio-stream",
  "tokio-tungstenite",
- "tokio-util",
  "tower",
  "tracing",
  "tracing-subscriber",

Cargo-recent.lock

@@ -2871,7 +2871,6 @@ dependencies = [
  "tokio-rustls-acme",
  "tokio-stream",
  "tokio-tungstenite",
- "tokio-util",
  "tower",
  "tracing",
  "tracing-subscriber",

payjoin-mailroom/Cargo.toml

@@ -72,7 +72,6 @@ tokio-listener = { version = "0.5", features = ["axum08", "serde"] }
 tokio-rustls-acme = { version = "0.9.0", features = ["axum"], optional = true }
 tokio-stream = { version = "0.1.17", features = ["net"] }
 tokio-tungstenite = { version = "0.27.0", optional = true }
-tokio-util = { version = "0.7.16", features = ["net", "codec"] }
 tower = "0.5"
 tracing = "0.1"
 tracing-subscriber = { version = "0.3", features = ["env-filter", "json"] }

payjoin-mailroom/src/lib.rs

@@ -79,6 +79,7 @@ pub async fn serve_manual_tls(
     config: Config,
     tls_config: Option<axum_server::tls_rustls::RustlsConfig>,
     root_store: rustls::RootCertStore,
+    default_gateway: Option<crate::ohttp_relay::GatewayUri>,
 ) -> anyhow::Result<(u16, tokio::task::JoinHandle<anyhow::Result<()>>)> {
     use std::net::SocketAddr;
 
@@ -91,7 +92,12 @@ pub async fn serve_manual_tls(
 
     let services = Services {
         directory,
-        relay: crate::ohttp_relay::Service::new_with_roots(root_store, sentinel_tag).await,
+        relay: crate::ohttp_relay::Service::new_with_roots(
+            sentinel_tag,
+            root_store,
+            default_gateway,
+        )
+        .await,
         metrics: MetricsService::new(None),
         #[cfg(feature = "access-control")]
         geoip,
@@ -439,7 +445,8 @@ mod tests {
         root_store.add(CertificateDer::from(cert_der.clone())).unwrap();
         let tls_config = RustlsConfig::from_der(vec![cert_der], key_der).await.unwrap();
 
-        let (port, handle) = serve_manual_tls(config, Some(tls_config), root_store).await.unwrap();
+        let (port, handle) =
+            serve_manual_tls(config, Some(tls_config), root_store, None).await.unwrap();
         (port, handle, tempdir)
     }
 

payjoin-mailroom/src/ohttp_relay/mod.rs

@@ -14,19 +14,13 @@ use hyper::header::{
     HeaderValue, ACCESS_CONTROL_ALLOW_HEADERS, ACCESS_CONTROL_ALLOW_METHODS,
     ACCESS_CONTROL_ALLOW_ORIGIN, CONTENT_LENGTH, CONTENT_TYPE,
 };
-use hyper::server::conn::http1;
 use hyper::{Method, Request, Response};
 use hyper_rustls::builderstates::WantsSchemes;
 use hyper_rustls::{HttpsConnector, HttpsConnectorBuilder};
 use hyper_util::client::legacy::connect::HttpConnector;
 use hyper_util::client::legacy::Client;
-use hyper_util::rt::{TokioExecutor, TokioIo};
-use hyper_util::service::TowerToHyperService;
-use tokio::io::{AsyncRead, AsyncWrite};
-#[cfg(unix)]
-use tokio::net::UnixListener;
-use tokio_util::net::Listener;
-use tracing::{error, instrument};
+use hyper_util::rt::TokioExecutor;
+use tracing::instrument;
 
 pub mod error;
 #[cfg(not(feature = "_test-util"))]
@@ -45,47 +39,6 @@ pub mod bootstrap;
 pub const EXPECTED_MEDIA_TYPE: HeaderValue = HeaderValue::from_static("message/ohttp-req");
 pub const DEFAULT_GATEWAY: &str = "https://payjo.in";
 
-#[instrument]
-#[cfg(unix)]
-pub async fn listen_socket(
-    socket_path: &str,
-    gateway_origin: GatewayUri,
-) -> Result<tokio::task::JoinHandle<Result<(), BoxError>>, BoxError> {
-    let listener = UnixListener::bind(socket_path)?;
-    tracing::info!("OHTTP relay listening on socket: {}", socket_path);
-    let sentinel_tag = SentinelTag::new([0u8; 32]);
-    ohttp_relay(listener, RelayConfig::new_with_default_client(gateway_origin, sentinel_tag)).await
-}
-
-#[instrument]
-#[cfg(not(unix))]
-pub async fn listen_socket(
-    socket_path: &str,
-    gateway_origin: GatewayUri,
-) -> Result<tokio::task::JoinHandle<Result<(), BoxError>>, BoxError> {
-    // Keep API parity across targets while making the limitation explicit at runtime.
-    let _ = (socket_path, gateway_origin);
-    Err(std::io::Error::new(
-        std::io::ErrorKind::Unsupported,
-        "UNIX_SOCKET is only supported on unix targets; use PORT instead",
-    )
-    .into())
-}
-
-#[cfg(feature = "_test-util")]
-pub async fn listen_tcp_on_free_port(
-    default_gateway: GatewayUri,
-    root_store: rustls::RootCertStore,
-) -> Result<(u16, tokio::task::JoinHandle<Result<(), BoxError>>), BoxError> {
-    let listener = tokio::net::TcpListener::bind("[::]:0").await?;
-    let port = listener.local_addr()?.port();
-    println!("OHTTP relay binding to port {}", listener.local_addr()?);
-    let sentinel_tag = SentinelTag::new([0u8; 32]);
-    let config = RelayConfig::new(default_gateway, root_store, sentinel_tag);
-    let handle = ohttp_relay(listener, config).await?;
-    Ok((port, handle))
-}
-
 #[derive(Debug)]
 struct RelayConfig {
     default_gateway: GatewayUri,
@@ -116,8 +69,6 @@ pub struct Service {
 }
 
 impl Service {
-    fn from_config(config: Arc<RelayConfig>) -> Self { Self { config } }
-
     pub async fn new(sentinel_tag: SentinelTag) -> Self {
         // The default gateway is hardcoded because it is obsolete and required only for backwards
         // compatibility.
@@ -131,10 +82,12 @@ impl Service {
 
     #[cfg(feature = "_test-util")]
     pub async fn new_with_roots(
-        root_store: rustls::RootCertStore,
         sentinel_tag: SentinelTag,
+        root_store: rustls::RootCertStore,
+        default_gateway: Option<GatewayUri>,
     ) -> Self {
-        let gateway_origin = GatewayUri::from_str(DEFAULT_GATEWAY).expect("valid gateway uri");
+        let gateway_origin = default_gateway
+            .unwrap_or_else(|| GatewayUri::from_str(DEFAULT_GATEWAY).expect("valid gateway uri"));
         let config = RelayConfig::new(gateway_origin, root_store, sentinel_tag);
         config.prober.assert_opt_in(&config.default_gateway).await;
         Self { config: Arc::new(config) }
@@ -197,38 +150,6 @@ impl From<rustls::RootCertStore> for HttpClient {
     }
 }
 
-#[instrument(skip(listener))]
-async fn ohttp_relay<L>(
-    mut listener: L,
-    config: RelayConfig,
-) -> Result<tokio::task::JoinHandle<Result<(), BoxError>>, BoxError>
-where
-    L: Listener + Unpin + Send + 'static,
-    L::Io: AsyncRead + AsyncWrite + Unpin + Send + 'static,
-{
-    config.prober.assert_opt_in(&config.default_gateway).await;
-
-    let config = Arc::new(config);
-
-    let handle = tokio::spawn(async move {
-        while let Ok((stream, _)) = listener.accept().await {
-            let service = Service::from_config(config.clone());
-            let io = TokioIo::new(stream);
-            tokio::spawn(async move {
-                let hyper_service = TowerToHyperService::new(service);
-                if let Err(err) =
-                    http1::Builder::new().serve_connection(io, hyper_service).with_upgrades().await
-                {
-                    error!("Error serving connection: {:?}", err);
-                }
-            });
-        }
-        Ok(())
-    });
-
-    Ok(handle)
-}
-
 #[instrument]
 async fn serve_ohttp_relay<B>(
     req: Request<B>,

payjoin-mailroom/tests/integration.rs

@@ -22,6 +22,7 @@ mod integration {
         ALLOWED_PURPOSES_CONTENT_TYPE, MAGIC_BIP77_PURPOSE,
     };
     use payjoin_mailroom::ohttp_relay::*;
+    use payjoin_test_utils::init_ohttp_relay;
     use rcgen::Certificate;
     use rustls::pki_types::{CertificateDer, PrivateKeyDer, PrivatePkcs8KeyDer};
     use tempfile::NamedTempFile;
@@ -54,7 +55,7 @@ mod integration {
         let mut root_store = rustls::RootCertStore::empty();
         root_store.add(nginx_cert_der.clone()).unwrap();
 
-        let (relay_port, relay_handle) = listen_tcp_on_free_port(gateway.clone(), root_store)
+        let (relay_port, relay_handle) = init_ohttp_relay(root_store, Some(gateway.clone()))
             .await
             .expect("Failed to listen on free port");
         let relay_task = tokio::spawn(async move {
@@ -91,60 +92,6 @@ mod integration {
         }
     }
 
-    // UNIX domain socket transport is only available on unix targets.
-    #[tokio::test]
-    #[cfg(unix)]
-    async fn test_request_response_socket() -> Result<(), Box<dyn std::error::Error>> {
-        init_crypto_provider();
-        let temp_dir = std::env::temp_dir();
-        let socket_path = temp_dir.as_path().join("test.socket");
-
-        if socket_path.exists() {
-            std::fs::remove_file(&socket_path).expect("Failed to remove existing socket file");
-        }
-
-        let gateway_port = find_free_port();
-        let gateway = GatewayUri::from_str(&format!("http://0.0.0.0:{}", gateway_port)).unwrap();
-        let nginx_cert = gen_localhost_cert();
-        let nginx_cert_der = cert_to_cert_der(&nginx_cert);
-        let socket_path_str = socket_path.to_str().unwrap();
-        let relay_handle = listen_socket(socket_path_str, gateway.clone())
-            .await
-            .expect("Failed to listen on socket");
-        let relay_task = tokio::spawn(async move {
-            if let Err(e) = relay_handle.await {
-                eprintln!("Relay failed: {}", e);
-            }
-        });
-        let n_http_port = find_free_port();
-        let n_https_port = find_free_port();
-        let _nginx = match start_nginx(
-            n_http_port,
-            n_https_port,
-            format!("unix:{}", socket_path_str),
-            nginx_cert,
-        )
-        .await
-        {
-            Ok(nginx) => nginx,
-            Err(NginxInitError::NginxNotAvailable) => {
-                eprintln!("Skipping test: NGINX_EXE environment variable not set");
-                return Ok(());
-            }
-            Err(e) => return Err(Box::new(e) as Box<dyn std::error::Error>),
-        };
-        tokio::select! {
-            _ = example_gateway_http(gateway_port) => {
-                panic!("Gateway is long running");
-            }
-            _ = relay_task => {
-                panic!("Relay is long running");
-            }
-            _ = ohttp_req(n_https_port, nginx_cert_der, gateway) => {}
-        }
-        Ok(())
-    }
-
     async fn example_gateway_http(port: u16) -> Result<(), Box<dyn std::error::Error>> {
         example_gateway(port, |stream| {
             tokio::spawn(async move {
@@ -380,7 +327,7 @@ mod integration {
             let mut root_store = rustls::RootCertStore::empty();
             root_store.add(gateway_cert_der.clone()).unwrap();
             root_store.add(nginx_cert_der).unwrap();
-            let (relay_port, relay_handle) = listen_tcp_on_free_port(gateway.clone(), root_store)
+            let (relay_port, relay_handle) = init_ohttp_relay(root_store, Some(gateway.clone()))
                 .await
                 .expect("Failed to listen on free port");
             let relay_task = tokio::spawn(async move {

payjoin-test-utils/src/lib.rs

@@ -61,7 +61,7 @@ impl TestServices {
         root_store.add(CertificateDer::from(cert.cert.der().to_vec())).unwrap();
 
         let directory = init_directory(cert_key, root_store.clone()).await?;
-        let ohttp_relay = init_ohttp_relay(root_store).await?;
+        let ohttp_relay = init_ohttp_relay(root_store, None).await?;
 
         let http_agent: Arc<Client> = Arc::new(http_agent(cert_der)?);
 
@@ -126,9 +126,10 @@ pub async fn init_directory(
 
     let tls_config = RustlsConfig::from_der(vec![local_cert_key.0], local_cert_key.1).await?;
 
-    let (port, handle) = payjoin_mailroom::serve_manual_tls(config, Some(tls_config), root_store)
-        .await
-        .map_err(|e| e.to_string())?;
+    let (port, handle) =
+        payjoin_mailroom::serve_manual_tls(config, Some(tls_config), root_store, None)
+            .await
+            .map_err(|e| e.to_string())?;
 
     let handle = tokio::spawn(async move {
         let _tempdir = tempdir; // keep the tempdir until the directory shuts down
@@ -138,8 +139,9 @@ pub async fn init_directory(
     Ok((port, handle))
 }
 
-async fn init_ohttp_relay(
+pub async fn init_ohttp_relay(
     root_store: RootCertStore,
+    default_gateway: Option<payjoin_mailroom::ohttp_relay::GatewayUri>,
 ) -> std::result::Result<
     (u16, tokio::task::JoinHandle<std::result::Result<(), BoxSendSyncError>>),
     BoxSendSyncError,
@@ -152,9 +154,10 @@ async fn init_ohttp_relay(
         None,
     );
 
-    let (port, handle) = payjoin_mailroom::serve_manual_tls(config, None, root_store)
-        .await
-        .map_err(|e| e.to_string())?;
+    let (port, handle) =
+        payjoin_mailroom::serve_manual_tls(config, None, root_store, default_gateway)
+            .await
+            .map_err(|e| e.to_string())?;
 
     let handle = tokio::spawn(async move {
         let _tempdir = tempdir; // keep the tempdir until the relay shuts down