14180251374

Committed 31 Mar 2025 07:56PM UTC coverage: 81.564% (-0.2%) from 81.718%

Build # 14180251374

Build Type

Pull #616

github

Committed by

web-flow

Commit Message

Merge 999d41853 into b2aba5ea0

Pull Request Pull Request #616: Persist ohttp keys

Run Details

52 of 77 new or added lines in 4 files covered. (67.53%)

1 existing line in 1 file now uncovered.

5256 of 6444 relevant lines covered (81.56%)

717.52 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

72.93

/payjoin-directory/src/lib.rs

use std::net::{IpAddr, Ipv6Addr, SocketAddr};
use std::str::FromStr;
use std::sync::Arc;
use std::time::Duration;

use anyhow::Result;
use http_body_util::combinators::BoxBody;
use http_body_util::{BodyExt, Empty, Full};
use hyper::body::{Body, Bytes, Incoming};
use hyper::header::{HeaderValue, ACCESS_CONTROL_ALLOW_ORIGIN, CONTENT_TYPE};
use hyper::server::conn::http1;
use hyper::service::service_fn;
use hyper::{Method, Request, Response, StatusCode, Uri};
use hyper_util::rt::TokioIo;
use payjoin::directory::{ShortId, ShortIdError, ENCAPSULATED_MESSAGE_BYTES};
use tokio::net::TcpListener;
use tokio::sync::Mutex;
use tracing::{debug, error, trace};

use crate::db::DbPool;
pub mod key_config;
pub use crate::key_config::*;

pub const DEFAULT_DIR_PORT: u16 = 8080;
pub const DEFAULT_DB_HOST: &str = "localhost:6379";
pub const DEFAULT_TIMEOUT_SECS: u64 = 30;

const CHACHA20_POLY1305_NONCE_LEN: usize = 32; // chacha20poly1305 n_k
const POLY1305_TAG_SIZE: usize = 16;
pub const BHTTP_REQ_BYTES: usize =
    ENCAPSULATED_MESSAGE_BYTES - (CHACHA20_POLY1305_NONCE_LEN + POLY1305_TAG_SIZE);
const V1_MAX_BUFFER_SIZE: usize = 65536;

const V1_REJECT_RES_JSON: &str =
    r#"{{"errorCode": "original-psbt-rejected ", "message": "Body is not a string"}}"#;
const V1_UNAVAILABLE_RES_JSON: &str = r#"{{"errorCode": "unavailable", "message": "V2 receiver offline. V1 sends require synchronous communications."}}"#;

mod db;

#[cfg(feature = "_danger-local-https")]
type BoxError = Box<dyn std::error::Error + Send + Sync>;

/// Initialize the OHTTP Key Configuration in memory. For testing purposes.
pub fn init_ohttp_server() -> Result<ohttp::Server, BoxError> {
    let ohttp_config = gen_ohttp_server_config()?;
    Ok(ohttp_config.into())
}

#[cfg(feature = "_danger-local-https")]
pub async fn listen_tcp_with_tls_on_free_port(
    db_host: String,
    timeout: Duration,
    cert_key: (Vec<u8>, Vec<u8>),
    ohttp: ohttp::Server,
) -> Result<(u16, tokio::task::JoinHandle<Result<(), BoxError>>), BoxError> {
    let listener = tokio::net::TcpListener::bind("[::]:0").await?;
    let port = listener.local_addr()?.port();
    println!("Directory server binding to port {}", listener.local_addr()?);
    let handle =
        listen_tcp_with_tls_on_listener(listener, db_host, timeout, cert_key, ohttp).await?;
    Ok((port, handle))
}

// Helper function to avoid code duplication
#[cfg(feature = "_danger-local-https")]
async fn listen_tcp_with_tls_on_listener(
    listener: tokio::net::TcpListener,
    db_host: String,
    timeout: Duration,
    tls_config: (Vec<u8>, Vec<u8>),
    ohttp: ohttp::Server,
) -> Result<tokio::task::JoinHandle<Result<(), BoxError>>, BoxError> {
    let pool = DbPool::new(timeout, db_host).await?;
    let ohttp = Arc::new(Mutex::new(ohttp));
    let tls_acceptor = init_tls_acceptor(tls_config)?;
    // Spawn the connection handling loop in a separate task
    let handle = tokio::spawn(async move {
        while let Ok((stream, _)) = listener.accept().await {
            let pool = pool.clone();
            let ohttp = ohttp.clone();
            let tls_acceptor = tls_acceptor.clone();
            tokio::spawn(async move {
                let tls_stream = match tls_acceptor.accept(stream).await {
                    Ok(tls_stream) => tls_stream,
                    Err(e) => {
                        error!("TLS accept error: {}", e);
                        return;
                    }
                };
                if let Err(err) = http1::Builder::new()
                    .serve_connection(
                        TokioIo::new(tls_stream),
                        service_fn(move |req| {
                            serve_payjoin_directory(req, pool.clone(), ohttp.clone())
                        }),
                    )
                    .with_upgrades()
                    .await
                {
                    error!("Error serving connection: {:?}", err);
                }
            });
        }
        Ok(())
    });
    Ok(handle)
}

// Modify existing listen_tcp_with_tls to use the new helper
pub async fn listen_tcp(
    port: u16,
    db_host: String,
    timeout: Duration,
    ohttp: ohttp::Server,
) -> Result<(), Box<dyn std::error::Error>> {
    let pool = DbPool::new(timeout, db_host).await?;
    let ohttp = Arc::new(Mutex::new(ohttp));
    let bind_addr = SocketAddr::new(IpAddr::V6(Ipv6Addr::UNSPECIFIED), port);
    let listener = TcpListener::bind(bind_addr).await?;
    while let Ok((stream, _)) = listener.accept().await {
        let pool = pool.clone();
        let ohttp = ohttp.clone();
        let io = TokioIo::new(stream);
        tokio::spawn(async move {
            if let Err(err) = http1::Builder::new()
                .serve_connection(
                    io,
                    service_fn(move |req| {
                        serve_payjoin_directory(req, pool.clone(), ohttp.clone())
                    }),
                )
                .with_upgrades()
                .await
            {
                error!("Error serving connection: {:?}", err);
            }
        });
    }

    Ok(())
}

#[cfg(feature = "_danger-local-https")]
pub async fn listen_tcp_with_tls(
    port: u16,
    db_host: String,
    timeout: Duration,
    cert_key: (Vec<u8>, Vec<u8>),
    ohttp: ohttp::Server,
) -> Result<tokio::task::JoinHandle<Result<(), BoxError>>, BoxError> {
    let addr = format!("0.0.0.0:{}", port);
    let listener = tokio::net::TcpListener::bind(&addr).await?;
    listen_tcp_with_tls_on_listener(listener, db_host, timeout, cert_key, ohttp).await
}

#[cfg(feature = "_danger-local-https")]
fn init_tls_acceptor(cert_key: (Vec<u8>, Vec<u8>)) -> Result<tokio_rustls::TlsAcceptor> {
    use rustls::pki_types::{CertificateDer, PrivateKeyDer};
    use rustls::ServerConfig;
    use tokio_rustls::TlsAcceptor;
    let (cert, key) = cert_key;
    let cert = CertificateDer::from(cert);
    let key =
        PrivateKeyDer::try_from(key).map_err(|e| anyhow::anyhow!("Could not parse key: {}", e))?;

    let mut server_config = ServerConfig::builder()
        .with_no_client_auth()
        .with_single_cert(vec![cert], key)
        .map_err(|e| anyhow::anyhow!("TLS error: {}", e))?;
    server_config.alpn_protocols = vec![b"h2".to_vec(), b"http/1.1".to_vec(), b"http/1.0".to_vec()];
    Ok(TlsAcceptor::from(Arc::new(server_config)))
}

async fn serve_payjoin_directory(
    req: Request<Incoming>,
    pool: DbPool,
    ohttp: Arc<Mutex<ohttp::Server>>,
) -> Result<Response<BoxBody<Bytes, hyper::Error>>> {
    let path = req.uri().path().to_string();
    let query = req.uri().query().unwrap_or_default().to_string();
    let (parts, body) = req.into_parts();

    let path_segments: Vec<&str> = path.split('/').collect();
    debug!("serve_payjoin_directory: {:?}", &path_segments);
    let mut response = match (parts.method, path_segments.as_slice()) {
        (Method::POST, ["", ".well-known", "ohttp-gateway"]) =>
            handle_ohttp_gateway(body, pool, ohttp).await,
        (Method::GET, ["", ".well-known", "ohttp-gateway"]) =>
            handle_ohttp_gateway_get(&ohttp, &query).await,
        (Method::POST, ["", ""]) => handle_ohttp_gateway(body, pool, ohttp).await,
        (Method::GET, ["", "ohttp-keys"]) => get_ohttp_keys(&ohttp).await,
        (Method::POST, ["", id]) => post_fallback_v1(id, query, body, pool).await,
        (Method::GET, ["", "health"]) => health_check().await,
        _ => Ok(not_found()),
    }
    .unwrap_or_else(|e| e.to_response());

    // Allow CORS for third-party access
    response.headers_mut().insert(ACCESS_CONTROL_ALLOW_ORIGIN, HeaderValue::from_static("*"));

    Ok(response)
}

async fn handle_ohttp_gateway(
    body: Incoming,
    pool: DbPool,
    ohttp: Arc<Mutex<ohttp::Server>>,
) -> Result<Response<BoxBody<Bytes, hyper::Error>>, HandlerError> {
    // decapsulate
    let ohttp_body =
        body.collect().await.map_err(|e| HandlerError::BadRequest(e.into()))?.to_bytes();
    let ohttp_locked = ohttp.lock().await;
    let (bhttp_req, res_ctx) = ohttp_locked
        .decapsulate(&ohttp_body)
        .map_err(|e| HandlerError::OhttpKeyRejection(e.into()))?;
    drop(ohttp_locked);
    let mut cursor = std::io::Cursor::new(bhttp_req);
    let req =
        bhttp::Message::read_bhttp(&mut cursor).map_err(|e| HandlerError::BadRequest(e.into()))?;
    let uri = Uri::builder()
        .scheme(req.control().scheme().unwrap_or_default())
        .authority(req.control().authority().unwrap_or_default())
        .path_and_query(req.control().path().unwrap_or_default())
        .build()?;
    let body = req.content().to_vec();
    let mut http_req =
        Request::builder().uri(uri).method(req.control().method().unwrap_or_default());
    for header in req.header().fields() {
        http_req = http_req.header(header.name(), header.value())
    }
    let request = http_req.body(full(body))?;

    let response = handle_v2(pool, request).await?;

    let (parts, body) = response.into_parts();
    let mut bhttp_res = bhttp::Message::response(parts.status.as_u16());
    for (name, value) in parts.headers.iter() {
        bhttp_res.put_header(name.as_str(), value.to_str().unwrap_or_default());
    }
    let full_body =
        body.collect().await.map_err(|e| HandlerError::InternalServerError(e.into()))?.to_bytes();
    bhttp_res.write_content(&full_body);
    let mut bhttp_bytes = Vec::new();
    bhttp_res
        .write_bhttp(bhttp::Mode::KnownLength, &mut bhttp_bytes)
        .map_err(|e| HandlerError::InternalServerError(e.into()))?;
    bhttp_bytes.resize(BHTTP_REQ_BYTES, 0);
    let ohttp_res = res_ctx
        .encapsulate(&bhttp_bytes)
        .map_err(|e| HandlerError::InternalServerError(e.into()))?;
    assert!(ohttp_res.len() == ENCAPSULATED_MESSAGE_BYTES, "Unexpected OHTTP response size");
    Ok(Response::new(full(ohttp_res)))
}

async fn handle_v2(
    pool: DbPool,
    req: Request<BoxBody<Bytes, hyper::Error>>,
) -> Result<Response<BoxBody<Bytes, hyper::Error>>, HandlerError> {
    let path = req.uri().path().to_string();
    let (parts, body) = req.into_parts();

    let path_segments: Vec<&str> = path.split('/').collect();
    debug!("handle_v2: {:?}", &path_segments);
    match (parts.method, path_segments.as_slice()) {
        (Method::POST, &["", id]) => post_subdir(id, body, pool).await,
        (Method::GET, &["", id]) => get_subdir(id, pool).await,
        (Method::PUT, &["", id]) => put_payjoin_v1(id, body, pool).await,
        _ => Ok(not_found()),
    }
}

async fn health_check() -> Result<Response<BoxBody<Bytes, hyper::Error>>, HandlerError> {
    Ok(Response::new(empty()))
}

#[derive(Debug)]
enum HandlerError {
    PayloadTooLarge,
    InternalServerError(anyhow::Error),
    OhttpKeyRejection(anyhow::Error),
    BadRequest(anyhow::Error),
}

impl HandlerError {
    fn to_response(&self) -> Response<BoxBody<Bytes, hyper::Error>> {
        let mut res = Response::new(empty());
        match self {
            HandlerError::PayloadTooLarge => *res.status_mut() = StatusCode::PAYLOAD_TOO_LARGE,
            HandlerError::InternalServerError(e) => {
                error!("Internal server error: {}", e);
                *res.status_mut() = StatusCode::INTERNAL_SERVER_ERROR
            }
            HandlerError::OhttpKeyRejection(e) => {
                const OHTTP_KEY_REJECTION_RES_JSON: &str = r#"{"type":"https://iana.org/assignments/http-problem-types#ohttp-key", "title": "key identifier unknown"}"#;

                error!("Bad request: Key configuration rejected: {}", e);
                *res.status_mut() = StatusCode::BAD_REQUEST;
                res.headers_mut()
                    .insert(CONTENT_TYPE, HeaderValue::from_static("application/problem+json"));
                *res.body_mut() = full(OHTTP_KEY_REJECTION_RES_JSON);
            }
            HandlerError::BadRequest(e) => {
                error!("Bad request: {}", e);
                *res.status_mut() = StatusCode::BAD_REQUEST
            }
        };

        res
    }
}

impl From<hyper::http::Error> for HandlerError {
    fn from(e: hyper::http::Error) -> Self { HandlerError::InternalServerError(e.into()) }
}

impl From<ShortIdError> for HandlerError {
    fn from(_: ShortIdError) -> Self {
        HandlerError::BadRequest(anyhow::anyhow!("subdirectory ID must be 13 bech32 characters"))
    }
}

fn handle_peek(
    result: db::Result<Vec<u8>>,
    timeout_response: Response<BoxBody<Bytes, hyper::Error>>,
) -> Result<Response<BoxBody<Bytes, hyper::Error>>, HandlerError> {
    match result {
        Ok(buffered_req) => Ok(Response::new(full(buffered_req))),
        Err(e) => match e {
            db::Error::Redis(re) => {
                error!("Redis error: {}", re);
                Err(HandlerError::InternalServerError(anyhow::Error::msg("Internal server error")))
            }
            db::Error::Timeout(_) => Ok(timeout_response),
        },
    }
}

async fn post_fallback_v1(
    id: &str,
    query: String,
    body: impl Body,
    pool: DbPool,
) -> Result<Response<BoxBody<Bytes, hyper::Error>>, HandlerError> {
    trace!("Post fallback v1");
    let none_response = Response::builder()
        .status(StatusCode::SERVICE_UNAVAILABLE)
        .body(full(V1_UNAVAILABLE_RES_JSON))?;
    let bad_request_body_res =
        Response::builder().status(StatusCode::BAD_REQUEST).body(full(V1_REJECT_RES_JSON))?;

    let body_bytes = match body.collect().await {
        Ok(bytes) => bytes.to_bytes(),
        Err(_) => return Ok(bad_request_body_res),
    };

    let body_str = match String::from_utf8(body_bytes.to_vec()) {
        Ok(body_str) => body_str,
        Err(_) => return Ok(bad_request_body_res),
    };

    let v2_compat_body = format!("{}\n{}", body_str, query);
    let id = ShortId::from_str(id)?;
    pool.push_default(&id, v2_compat_body.into())
        .await
        .map_err(|e| HandlerError::BadRequest(e.into()))?;
    handle_peek(pool.peek_v1(&id).await, none_response)
}

async fn put_payjoin_v1(
    id: &str,
    body: BoxBody<Bytes, hyper::Error>,
    pool: DbPool,
) -> Result<Response<BoxBody<Bytes, hyper::Error>>, HandlerError> {
    trace!("Put_payjoin_v1");
    let ok_response = Response::builder().status(StatusCode::OK).body(empty())?;

    let id = ShortId::from_str(id)?;
    let req =
        body.collect().await.map_err(|e| HandlerError::InternalServerError(e.into()))?.to_bytes();
    if req.len() > V1_MAX_BUFFER_SIZE {
        return Err(HandlerError::PayloadTooLarge);
    }

    match pool.push_v1(&id, req.into()).await {
        Ok(_) => Ok(ok_response),
        Err(e) => Err(HandlerError::BadRequest(e.into())),
    }
}

async fn post_subdir(
    id: &str,
    body: BoxBody<Bytes, hyper::Error>,
    pool: DbPool,
) -> Result<Response<BoxBody<Bytes, hyper::Error>>, HandlerError> {
    let none_response = Response::builder().status(StatusCode::OK).body(empty())?;
    trace!("post_subdir");

    let id = ShortId::from_str(id)?;

    let req =
        body.collect().await.map_err(|e| HandlerError::InternalServerError(e.into()))?.to_bytes();
    if req.len() > V1_MAX_BUFFER_SIZE {
        return Err(HandlerError::PayloadTooLarge);
    }

    match pool.push_default(&id, req.into()).await {
        Ok(_) => Ok(none_response),
        Err(e) => Err(HandlerError::BadRequest(e.into())),
    }
}

async fn get_subdir(
    id: &str,
    pool: DbPool,
) -> Result<Response<BoxBody<Bytes, hyper::Error>>, HandlerError> {
    trace!("get_subdir");
    let id = ShortId::from_str(id)?;
    let timeout_response = Response::builder().status(StatusCode::ACCEPTED).body(empty())?;
    handle_peek(pool.peek_default(&id).await, timeout_response)
}

fn not_found() -> Response<BoxBody<Bytes, hyper::Error>> {
    let mut res = Response::default();
    *res.status_mut() = StatusCode::NOT_FOUND;
    res
}

async fn handle_ohttp_gateway_get(
    ohttp: &Arc<Mutex<ohttp::Server>>,
    query: &str,
) -> Result<Response<BoxBody<Bytes, hyper::Error>>, HandlerError> {
    match query {
        "allowed_purposes" => Ok(get_ohttp_allowed_purposes().await),
        _ => get_ohttp_keys(ohttp).await,
    }
}

async fn get_ohttp_keys(
    ohttp: &Arc<Mutex<ohttp::Server>>,
) -> Result<Response<BoxBody<Bytes, hyper::Error>>, HandlerError> {
    let ohttp_keys = ohttp
        .lock()
        .await
        .config()
        .encode()
        .map_err(|e| HandlerError::InternalServerError(e.into()))?;
    let mut res = Response::new(full(ohttp_keys));
    res.headers_mut().insert(CONTENT_TYPE, HeaderValue::from_static("application/ohttp-keys"));
    Ok(res)
}

async fn get_ohttp_allowed_purposes() -> Response<BoxBody<Bytes, hyper::Error>> {
    // Encode the magic string in the same format as a TLS ALPN protocol list (a
    // U16BE length encoded list of U8 length encoded strings).
    //
    // The string is just "BIP77" followed by a UUID, that signals to relays
    // that this OHTTP gateway will accept any requests associated with this
    // purpose.
    let mut res = Response::new(full(Bytes::from_static(
        b"\x00\x01\x2aBIP77 454403bb-9f7b-4385-b31f-acd2dae20b7e",
    )));

    res.headers_mut()
        .insert(CONTENT_TYPE, HeaderValue::from_static("application/x-ohttp-allowed-purposes"));

    res
}

fn empty() -> BoxBody<Bytes, hyper::Error> {
    Empty::<Bytes>::new().map_err(|never| match never {}).boxed()
}

fn full<T: Into<Bytes>>(chunk: T) -> BoxBody<Bytes, hyper::Error> {
    Full::new(chunk.into()).map_err(|never| match never {}).boxed()
}

1	use std::net::{IpAddr, Ipv6Addr, SocketAddr};
2	use std::str::FromStr;
3	use std::sync::Arc;
4	use std::time::Duration;
5
6	use anyhow::Result;
7	use http_body_util::combinators::BoxBody;
8	use http_body_util::{BodyExt, Empty, Full};
9	use hyper::body::{Body, Bytes, Incoming};
10	use hyper::header::{HeaderValue, ACCESS_CONTROL_ALLOW_ORIGIN, CONTENT_TYPE};
11	use hyper::server::conn::http1;
12	use hyper::service::service_fn;
13	use hyper::{Method, Request, Response, StatusCode, Uri};
14	use hyper_util::rt::TokioIo;
15	use payjoin::directory::{ShortId, ShortIdError, ENCAPSULATED_MESSAGE_BYTES};
16	use tokio::net::TcpListener;
17	use tokio::sync::Mutex;
18	use tracing::{debug, error, trace};
19
20	use crate::db::DbPool;
21	pub mod key_config;
22	pub use crate::key_config::*;
23
24	pub const DEFAULT_DIR_PORT: u16 = 8080;
25	pub const DEFAULT_DB_HOST: &str = "localhost:6379";
26	pub const DEFAULT_TIMEOUT_SECS: u64 = 30;
27
28	const CHACHA20_POLY1305_NONCE_LEN: usize = 32; // chacha20poly1305 n_k
29	const POLY1305_TAG_SIZE: usize = 16;
30	pub const BHTTP_REQ_BYTES: usize =
31	ENCAPSULATED_MESSAGE_BYTES - (CHACHA20_POLY1305_NONCE_LEN + POLY1305_TAG_SIZE);
32	const V1_MAX_BUFFER_SIZE: usize = 65536;
33
34	const V1_REJECT_RES_JSON: &str =
35	r#"{{"errorCode": "original-psbt-rejected ", "message": "Body is not a string"}}"#;
36	const V1_UNAVAILABLE_RES_JSON: &str = r#"{{"errorCode": "unavailable", "message": "V2 receiver offline. V1 sends require synchronous communications."}}"#;
37
38	mod db;
39
40	#[cfg(feature = "_danger-local-https")]
41	type BoxError = Box<dyn std::error::Error + Send + Sync>;
42
43	/// Initialize the OHTTP Key Configuration in memory. For testing purposes.
NEW 44	pub fn init_ohttp_server() -> Result<ohttp::Server, BoxError> {	×
NEW 45	let ohttp_config = gen_ohttp_server_config()?;	×
NEW 46	Ok(ohttp_config.into())	×
NEW 47	}	×
48
49	#[cfg(feature = "_danger-local-https")]
50	pub async fn listen_tcp_with_tls_on_free_port(	6✔
51	db_host: String,	6✔
52	timeout: Duration,	6✔
53	cert_key: (Vec<u8>, Vec<u8>),	6✔
54	ohttp: ohttp::Server,	6✔
55	) -> Result<(u16, tokio::task::JoinHandle<Result<(), BoxError>>), BoxError> {	6✔
56	let listener = tokio::net::TcpListener::bind("[::]:0").await?;	6✔
57	let port = listener.local_addr()?.port();	6✔
58	println!("Directory server binding to port {}", listener.local_addr()?);	6✔
59	let handle =	6✔
60	listen_tcp_with_tls_on_listener(listener, db_host, timeout, cert_key, ohttp).await?;	6✔
61	Ok((port, handle))	6✔
62	}	6✔
63
64	// Helper function to avoid code duplication
65	#[cfg(feature = "_danger-local-https")]
66	async fn listen_tcp_with_tls_on_listener(	6✔
67	listener: tokio::net::TcpListener,	6✔
68	db_host: String,	6✔
69	timeout: Duration,	6✔
70	tls_config: (Vec<u8>, Vec<u8>),	6✔
71	ohttp: ohttp::Server,	6✔
72	) -> Result<tokio::task::JoinHandle<Result<(), BoxError>>, BoxError> {	6✔
73	let pool = DbPool::new(timeout, db_host).await?;	6✔
74	let ohttp = Arc::new(Mutex::new(ohttp));	6✔
75	let tls_acceptor = init_tls_acceptor(tls_config)?;	6✔
76	// Spawn the connection handling loop in a separate task
77	let handle = tokio::spawn(async move {	6✔
78	while let Ok((stream, _)) = listener.accept().await {	22✔
79	let pool = pool.clone();	16✔
80	let ohttp = ohttp.clone();	16✔
81	let tls_acceptor = tls_acceptor.clone();	16✔
82	tokio::spawn(async move {	16✔
83	let tls_stream = match tls_acceptor.accept(stream).await {	16✔
84	Ok(tls_stream) => tls_stream,	16✔
85	Err(e) => {	×
86	error!("TLS accept error: {}", e);	×
87	return;	×
88	}
89	};
90	if let Err(err) = http1::Builder::new()	16✔
91	.serve_connection(	16✔
92	TokioIo::new(tls_stream),	16✔
93	service_fn(move \|req\| {	51✔
94	serve_payjoin_directory(req, pool.clone(), ohttp.clone())	51✔
95	}),	51✔
96	)	16✔
97	.with_upgrades()	16✔
98	.await	16✔
99	{
100	error!("Error serving connection: {:?}", err);	×
101	}	4✔
102	});	16✔
103	}
104	Ok(())	×
105	});	6✔
106	Ok(handle)	6✔
107	}	6✔
108
109	// Modify existing listen_tcp_with_tls to use the new helper
110	pub async fn listen_tcp(	×
111	port: u16,	×
112	db_host: String,	×
113	timeout: Duration,	×
NEW 114	ohttp: ohttp::Server,	×
115	) -> Result<(), Box<dyn std::error::Error>> {	×
116	let pool = DbPool::new(timeout, db_host).await?;	×
NEW 117	let ohttp = Arc::new(Mutex::new(ohttp));	×
118	let bind_addr = SocketAddr::new(IpAddr::V6(Ipv6Addr::UNSPECIFIED), port);	×
119	let listener = TcpListener::bind(bind_addr).await?;	×
120	while let Ok((stream, _)) = listener.accept().await {	×
121	let pool = pool.clone();	×
122	let ohttp = ohttp.clone();	×
123	let io = TokioIo::new(stream);	×
124	tokio::spawn(async move {	×
125	if let Err(err) = http1::Builder::new()	×
126	.serve_connection(	×
127	io,	×
128	service_fn(move \|req\| {	×
129	serve_payjoin_directory(req, pool.clone(), ohttp.clone())	×
130	}),	×
131	)	×
132	.with_upgrades()	×
133	.await	×
134	{
135	error!("Error serving connection: {:?}", err);	×
136	}	×
137	});	×
138	}
139
140	Ok(())	×
141	}	×
142
143	#[cfg(feature = "_danger-local-https")]
144	pub async fn listen_tcp_with_tls(	×
145	port: u16,	×
146	db_host: String,	×
147	timeout: Duration,	×
148	cert_key: (Vec<u8>, Vec<u8>),	×
NEW 149	ohttp: ohttp::Server,	×
150	) -> Result<tokio::task::JoinHandle<Result<(), BoxError>>, BoxError> {	×
151	let addr = format!("0.0.0.0:{}", port);	×
152	let listener = tokio::net::TcpListener::bind(&addr).await?;	×
NEW 153	listen_tcp_with_tls_on_listener(listener, db_host, timeout, cert_key, ohttp).await	×
154	}	×
155
156	#[cfg(feature = "_danger-local-https")]
157	fn init_tls_acceptor(cert_key: (Vec<u8>, Vec<u8>)) -> Result<tokio_rustls::TlsAcceptor> {	6✔
158	use rustls::pki_types::{CertificateDer, PrivateKeyDer};
159	use rustls::ServerConfig;
160	use tokio_rustls::TlsAcceptor;
161	let (cert, key) = cert_key;	6✔
162	let cert = CertificateDer::from(cert);	6✔
163	let key =	6✔
164	PrivateKeyDer::try_from(key).map_err(\|e\| anyhow::anyhow!("Could not parse key: {}", e))?;	6✔
165
166	let mut server_config = ServerConfig::builder()	6✔
167	.with_no_client_auth()	6✔
168	.with_single_cert(vec![cert], key)	6✔
169	.map_err(\|e\| anyhow::anyhow!("TLS error: {}", e))?;	6✔
170	server_config.alpn_protocols = vec![b"h2".to_vec(), b"http/1.1".to_vec(), b"http/1.0".to_vec()];	6✔
171	Ok(TlsAcceptor::from(Arc::new(server_config)))	6✔
172	}	6✔
173
174	async fn serve_payjoin_directory(	51✔
175	req: Request<Incoming>,	51✔
176	pool: DbPool,	51✔
177	ohttp: Arc<Mutex<ohttp::Server>>,	51✔
178	) -> Result<Response<BoxBody<Bytes, hyper::Error>>> {	51✔
179	let path = req.uri().path().to_string();	51✔
180	let query = req.uri().query().unwrap_or_default().to_string();	51✔
181	let (parts, body) = req.into_parts();	51✔
182		51✔
183	let path_segments: Vec<&str> = path.split('/').collect();	51✔
184	debug!("serve_payjoin_directory: {:?}", &path_segments);	51✔
185	let mut response = match (parts.method, path_segments.as_slice()) {	51✔
186	(Method::POST, ["", ".well-known", "ohttp-gateway"]) =>	38✔
187	handle_ohttp_gateway(body, pool, ohttp).await,	38✔
188	(Method::GET, ["", ".well-known", "ohttp-gateway"]) =>	5✔
189	handle_ohttp_gateway_get(&ohttp, &query).await,	5✔
190	(Method::POST, ["", ""]) => handle_ohttp_gateway(body, pool, ohttp).await,	2✔
191	(Method::GET, ["", "ohttp-keys"]) => get_ohttp_keys(&ohttp).await,	6✔
192	(Method::POST, ["", id]) => post_fallback_v1(id, query, body, pool).await,	2✔
193	(Method::GET, ["", "health"]) => health_check().await,	6✔
194	_ => Ok(not_found()),	×
195	}
196	.unwrap_or_else(\|e\| e.to_response());	51✔
197		51✔
198	// Allow CORS for third-party access	51✔
199	response.headers_mut().insert(ACCESS_CONTROL_ALLOW_ORIGIN, HeaderValue::from_static("*"));	51✔
200		51✔
201	Ok(response)	51✔
202	}	51✔
203
204	async fn handle_ohttp_gateway(	38✔
205	body: Incoming,	38✔
206	pool: DbPool,	38✔
207	ohttp: Arc<Mutex<ohttp::Server>>,	38✔
208	) -> Result<Response<BoxBody<Bytes, hyper::Error>>, HandlerError> {	38✔
209	// decapsulate
210	let ohttp_body =	38✔
211	body.collect().await.map_err(\|e\| HandlerError::BadRequest(e.into()))?.to_bytes();	38✔
212	let ohttp_locked = ohttp.lock().await;	38✔
213	let (bhttp_req, res_ctx) = ohttp_locked	38✔
214	.decapsulate(&ohttp_body)	38✔
215	.map_err(\|e\| HandlerError::OhttpKeyRejection(e.into()))?;	38✔
216	drop(ohttp_locked);	37✔
217	let mut cursor = std::io::Cursor::new(bhttp_req);	37✔
218	let req =	37✔
219	bhttp::Message::read_bhttp(&mut cursor).map_err(\|e\| HandlerError::BadRequest(e.into()))?;	37✔
220	let uri = Uri::builder()	37✔
221	.scheme(req.control().scheme().unwrap_or_default())	37✔
222	.authority(req.control().authority().unwrap_or_default())	37✔
223	.path_and_query(req.control().path().unwrap_or_default())	37✔
224	.build()?;	37✔
225	let body = req.content().to_vec();	37✔
226	let mut http_req =	37✔
227	Request::builder().uri(uri).method(req.control().method().unwrap_or_default());	37✔
228	for header in req.header().fields() {	37✔
229	http_req = http_req.header(header.name(), header.value())	×
230	}
231	let request = http_req.body(full(body))?;	37✔
232
233	let response = handle_v2(pool, request).await?;	37✔
234
235	let (parts, body) = response.into_parts();	37✔
236	let mut bhttp_res = bhttp::Message::response(parts.status.as_u16());	37✔
237	for (name, value) in parts.headers.iter() {	37✔
238	bhttp_res.put_header(name.as_str(), value.to_str().unwrap_or_default());	×
239	}	×
240	let full_body =	37✔
241	body.collect().await.map_err(\|e\| HandlerError::InternalServerError(e.into()))?.to_bytes();	37✔
242	bhttp_res.write_content(&full_body);	37✔
243	let mut bhttp_bytes = Vec::new();	37✔
244	bhttp_res	37✔
245	.write_bhttp(bhttp::Mode::KnownLength, &mut bhttp_bytes)	37✔
246	.map_err(\|e\| HandlerError::InternalServerError(e.into()))?;	37✔
247	bhttp_bytes.resize(BHTTP_REQ_BYTES, 0);	37✔
248	let ohttp_res = res_ctx	37✔
249	.encapsulate(&bhttp_bytes)	37✔
250	.map_err(\|e\| HandlerError::InternalServerError(e.into()))?;	37✔
251	assert!(ohttp_res.len() == ENCAPSULATED_MESSAGE_BYTES, "Unexpected OHTTP response size");	37✔
252	Ok(Response::new(full(ohttp_res)))	37✔
253	}	38✔
254
255	async fn handle_v2(	37✔
256	pool: DbPool,	37✔
257	req: Request<BoxBody<Bytes, hyper::Error>>,	37✔
258	) -> Result<Response<BoxBody<Bytes, hyper::Error>>, HandlerError> {	37✔
259	let path = req.uri().path().to_string();	37✔
260	let (parts, body) = req.into_parts();	37✔
261		37✔
262	let path_segments: Vec<&str> = path.split('/').collect();	37✔
263	debug!("handle_v2: {:?}", &path_segments);	37✔
264	match (parts.method, path_segments.as_slice()) {	37✔
265	(Method::POST, &["", id]) => post_subdir(id, body, pool).await,	17✔
266	(Method::GET, &["", id]) => get_subdir(id, pool).await,	19✔
267	(Method::PUT, &["", id]) => put_payjoin_v1(id, body, pool).await,	1✔
268	_ => Ok(not_found()),	×
269	}
270	}	37✔
271
272	async fn health_check() -> Result<Response<BoxBody<Bytes, hyper::Error>>, HandlerError> {	6✔
273	Ok(Response::new(empty()))	6✔
274	}	6✔
275
276	#[derive(Debug)]
277	enum HandlerError {
278	PayloadTooLarge,
279	InternalServerError(anyhow::Error),
280	OhttpKeyRejection(anyhow::Error),
281	BadRequest(anyhow::Error),
282	}
283
284	impl HandlerError {
285	fn to_response(&self) -> Response<BoxBody<Bytes, hyper::Error>> {	1✔
286	let mut res = Response::new(empty());	1✔
287	match self {	1✔
288	HandlerError::PayloadTooLarge => *res.status_mut() = StatusCode::PAYLOAD_TOO_LARGE,	×
289	HandlerError::InternalServerError(e) => {	×
290	error!("Internal server error: {}", e);	×
291	*res.status_mut() = StatusCode::INTERNAL_SERVER_ERROR	×
292	}
293	HandlerError::OhttpKeyRejection(e) => {	1✔
294	const OHTTP_KEY_REJECTION_RES_JSON: &str = r#"{"type":"https://iana.org/assignments/http-problem-types#ohttp-key", "title": "key identifier unknown"}"#;	1✔
295
296	error!("Bad request: Key configuration rejected: {}", e);	1✔
297	*res.status_mut() = StatusCode::BAD_REQUEST;	1✔
298	res.headers_mut()	1✔
299	.insert(CONTENT_TYPE, HeaderValue::from_static("application/problem+json"));	1✔
300	*res.body_mut() = full(OHTTP_KEY_REJECTION_RES_JSON);	1✔
301	}
302	HandlerError::BadRequest(e) => {	×
303	error!("Bad request: {}", e);	×
304	*res.status_mut() = StatusCode::BAD_REQUEST	×
305	}
306	};
307
308	res	1✔
309	}	1✔
310	}
311
312	impl From<hyper::http::Error> for HandlerError {
313	fn from(e: hyper::http::Error) -> Self { HandlerError::InternalServerError(e.into()) }	×
314	}
315
316	impl From<ShortIdError> for HandlerError {
317	fn from(_: ShortIdError) -> Self {	×
318	HandlerError::BadRequest(anyhow::anyhow!("subdirectory ID must be 13 bech32 characters"))	×
319	}	×
320	}
321
322	fn handle_peek(	21✔
323	result: db::Result<Vec<u8>>,	21✔
324	timeout_response: Response<BoxBody<Bytes, hyper::Error>>,	21✔
325	) -> Result<Response<BoxBody<Bytes, hyper::Error>>, HandlerError> {	21✔
326	match result {	21✔
327	Ok(buffered_req) => Ok(Response::new(full(buffered_req))),	18✔
328	Err(e) => match e {	3✔
329	db::Error::Redis(re) => {	×
330	error!("Redis error: {}", re);	×
331	Err(HandlerError::InternalServerError(anyhow::Error::msg("Internal server error")))	×
332	}
333	db::Error::Timeout(_) => Ok(timeout_response),	3✔
334	},
335	}
336	}	21✔
337
338	async fn post_fallback_v1(	2✔
339	id: &str,	2✔
340	query: String,	2✔
341	body: impl Body,	2✔
342	pool: DbPool,	2✔
343	) -> Result<Response<BoxBody<Bytes, hyper::Error>>, HandlerError> {	2✔
344	trace!("Post fallback v1");	2✔
345	let none_response = Response::builder()	2✔
346	.status(StatusCode::SERVICE_UNAVAILABLE)	2✔
347	.body(full(V1_UNAVAILABLE_RES_JSON))?;	2✔
348	let bad_request_body_res =	2✔
349	Response::builder().status(StatusCode::BAD_REQUEST).body(full(V1_REJECT_RES_JSON))?;	2✔
350
351	let body_bytes = match body.collect().await {	2✔
352	Ok(bytes) => bytes.to_bytes(),	2✔
353	Err(_) => return Ok(bad_request_body_res),	×
354	};
355
356	let body_str = match String::from_utf8(body_bytes.to_vec()) {	2✔
357	Ok(body_str) => body_str,	2✔
358	Err(_) => return Ok(bad_request_body_res),	×
359	};
360
361	let v2_compat_body = format!("{}\n{}", body_str, query);	2✔
362	let id = ShortId::from_str(id)?;	2✔
363	pool.push_default(&id, v2_compat_body.into())	2✔
364	.await	2✔
365	.map_err(\|e\| HandlerError::BadRequest(e.into()))?;	2✔
366	handle_peek(pool.peek_v1(&id).await, none_response)	2✔
367	}	2✔
368
369	async fn put_payjoin_v1(	1✔
370	id: &str,	1✔
371	body: BoxBody<Bytes, hyper::Error>,	1✔
372	pool: DbPool,	1✔
373	) -> Result<Response<BoxBody<Bytes, hyper::Error>>, HandlerError> {	1✔
374	trace!("Put_payjoin_v1");	1✔
375	let ok_response = Response::builder().status(StatusCode::OK).body(empty())?;	1✔
376
377	let id = ShortId::from_str(id)?;	1✔
378	let req =	1✔
379	body.collect().await.map_err(\|e\| HandlerError::InternalServerError(e.into()))?.to_bytes();	1✔
380	if req.len() > V1_MAX_BUFFER_SIZE {	1✔
381	return Err(HandlerError::PayloadTooLarge);	×
382	}	1✔
383		1✔
384	match pool.push_v1(&id, req.into()).await {	1✔
385	Ok(_) => Ok(ok_response),	1✔
386	Err(e) => Err(HandlerError::BadRequest(e.into())),	×
387	}
388	}	1✔
389
390	async fn post_subdir(	17✔
391	id: &str,	17✔
392	body: BoxBody<Bytes, hyper::Error>,	17✔
393	pool: DbPool,	17✔
394	) -> Result<Response<BoxBody<Bytes, hyper::Error>>, HandlerError> {	17✔
395	let none_response = Response::builder().status(StatusCode::OK).body(empty())?;	17✔
396	trace!("post_subdir");	17✔
397
398	let id = ShortId::from_str(id)?;	17✔
399
400	let req =	17✔
401	body.collect().await.map_err(\|e\| HandlerError::InternalServerError(e.into()))?.to_bytes();	17✔
402	if req.len() > V1_MAX_BUFFER_SIZE {	17✔
403	return Err(HandlerError::PayloadTooLarge);	×
404	}	17✔
405		17✔
406	match pool.push_default(&id, req.into()).await {	17✔
407	Ok(_) => Ok(none_response),	17✔
408	Err(e) => Err(HandlerError::BadRequest(e.into())),	×
409	}
410	}	17✔
411
412	async fn get_subdir(	19✔
413	id: &str,	19✔
414	pool: DbPool,	19✔
415	) -> Result<Response<BoxBody<Bytes, hyper::Error>>, HandlerError> {	19✔
416	trace!("get_subdir");	19✔
417	let id = ShortId::from_str(id)?;	19✔
418	let timeout_response = Response::builder().status(StatusCode::ACCEPTED).body(empty())?;	19✔
419	handle_peek(pool.peek_default(&id).await, timeout_response)	19✔
420	}	19✔
421
422	fn not_found() -> Response<BoxBody<Bytes, hyper::Error>> {	×
423	let mut res = Response::default();	×
424	*res.status_mut() = StatusCode::NOT_FOUND;	×
425	res	×
426	}	×
427
428	async fn handle_ohttp_gateway_get(	5✔
429	ohttp: &Arc<Mutex<ohttp::Server>>,	5✔
430	query: &str,	5✔
431	) -> Result<Response<BoxBody<Bytes, hyper::Error>>, HandlerError> {	5✔
432	match query {	5✔
433	"allowed_purposes" => Ok(get_ohttp_allowed_purposes().await),	5✔
434	_ => get_ohttp_keys(ohttp).await,	5✔
435	}
436	}	5✔
437
438	async fn get_ohttp_keys(	5✔
439	ohttp: &Arc<Mutex<ohttp::Server>>,	5✔
440	) -> Result<Response<BoxBody<Bytes, hyper::Error>>, HandlerError> {	5✔
441	let ohttp_keys = ohttp	5✔
442	.lock()	5✔
443	.await	5✔
444	.config()	5✔
445	.encode()	5✔
446	.map_err(\|e\| HandlerError::InternalServerError(e.into()))?;	5✔
447	let mut res = Response::new(full(ohttp_keys));	5✔
448	res.headers_mut().insert(CONTENT_TYPE, HeaderValue::from_static("application/ohttp-keys"));	5✔
449	Ok(res)	5✔
450	}	5✔
451
452	async fn get_ohttp_allowed_purposes() -> Response<BoxBody<Bytes, hyper::Error>> {	×
453	// Encode the magic string in the same format as a TLS ALPN protocol list (a	×
454	// U16BE length encoded list of U8 length encoded strings).	×
455	//	×
456	// The string is just "BIP77" followed by a UUID, that signals to relays	×
457	// that this OHTTP gateway will accept any requests associated with this	×
458	// purpose.	×
459	let mut res = Response::new(full(Bytes::from_static(	×
460	b"\x00\x01\x2aBIP77 454403bb-9f7b-4385-b31f-acd2dae20b7e",	×
461	)));	×
462		×
463	res.headers_mut()	×
464	.insert(CONTENT_TYPE, HeaderValue::from_static("application/x-ohttp-allowed-purposes"));	×
465		×
466	res	×
467	}	×
468
469	fn empty() -> BoxBody<Bytes, hyper::Error> {	44✔
470	Empty::<Bytes>::new().map_err(\|never\| match never {}).boxed()	44✔
471	}	44✔
472
473	fn full<T: Into<Bytes>>(chunk: T) -> BoxBody<Bytes, hyper::Error> {	102✔
474	Full::new(chunk.into()).map_err(\|never\| match never {}).boxed()	102✔
475	}	102✔

payjoin / rust-payjoin / 14180251374

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous