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

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/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.
@@ -197,38 +148,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,9 +55,8 @@ 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)
-            .await
-            .expect("Failed to listen on free port");
+        let (relay_port, relay_handle) =
+            init_ohttp_relay(root_store).await.expect("Failed to listen on free port");
         let relay_task = tokio::spawn(async move {
             if let Err(e) = relay_handle.await {
                 eprintln!("Relay failed: {}", e);
@@ -91,60 +91,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,9 +326,8 @@ 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)
-                .await
-                .expect("Failed to listen on free port");
+            let (relay_port, relay_handle) =
+                init_ohttp_relay(root_store).await.expect("Failed to listen on free port");
             let relay_task = tokio::spawn(async move {
                 if let Err(e) = relay_handle.await {
                     eprintln!("Relay failed: {}", e);

payjoin-test-utils/src/lib.rs

@@ -138,7 +138,7 @@ pub async fn init_directory(
     Ok((port, handle))
 }
 
-async fn init_ohttp_relay(
+pub async fn init_ohttp_relay(
     root_store: RootCertStore,
 ) -> std::result::Result<
     (u16, tokio::task::JoinHandle<std::result::Result<(), BoxSendSyncError>>),