Do http request and response

Author: Ma233

Cargo.lock

@@ -6,6 +6,7 @@ edition = "2021"
 [dependencies]
 clap = "4.5.4"
 futures = "0.3.30"
+httparse = "1.8.0"
 litep2p = "0.5.0"
 # Do not upgrade multiaddr, see: https://github.com/paritytech/litep2p/pull/91
 multiaddr = "0.17.1"

Cargo.toml

@@ -12,6 +12,14 @@ pub enum Error {
     Litep2p(#[from] litep2p::Error),
     #[error("Litep2p request response error: {0:?}")]
     Litep2pRequestResponseError(litep2p::protocol::request_response::RequestResponseError),
+    #[error("Httparse error: {0}")]
+    Httparse(#[from] httparse::Error),
+    #[error("Incomplete http request")]
+    IncompleteHttpRequest,
+    #[error("Protocol not support")]
+    ProtocolNotSupport,
+    #[error("Io error: {0}")]
+    Io(#[from] std::io::Error),
     #[error("Unexpected response type")]
     UnexpectedResponseType,
 }

src/error.rs

@@ -1,14 +1,19 @@
 use std::collections::HashMap;
 use std::net::SocketAddr;
+use std::time::Duration;
 
 use litep2p::protocol::request_response::DialOptions;
 use litep2p::protocol::request_response::RequestResponseEvent;
 use litep2p::types::RequestId;
 use litep2p::PeerId;
 use multiaddr::Multiaddr;
 use multiaddr::Protocol;
+use tokio::io::AsyncReadExt;
+use tokio::io::AsyncWriteExt;
+use tokio::net::TcpStream;
 use tokio::sync::mpsc;
 use tokio::sync::oneshot;
+use tokio::time::timeout;
 use tracing::warn;
 
 use crate::command::proto::AddPeerRequest;
@@ -49,21 +54,23 @@ pub struct PProxy {
     command_rx: mpsc::Receiver<(PProxyCommand, CommandNotifier)>,
     tunnel_notifier: HashMap<RequestId, CommandNotifier>,
     p2p_server: P2pServer,
+    proxy_addr: Option<SocketAddr>,
 }
 
 pub struct PProxyHandle {
     command_tx: mpsc::Sender<(PProxyCommand, CommandNotifier)>,
 }
 
 impl PProxy {
-    pub fn new(server_addr: SocketAddr) -> (Self, PProxyHandle) {
+    pub fn new(server_addr: SocketAddr, proxy_addr: Option<SocketAddr>) -> (Self, PProxyHandle) {
         let (command_tx, command_rx) = mpsc::channel(DEFAULT_CHANNEL_SIZE);
 
         (
             Self {
                 command_rx,
                 tunnel_notifier: HashMap::new(),
                 p2p_server: P2pServer::new(server_addr),
+                proxy_addr,
             },
             PProxyHandle { command_tx },
         )
@@ -97,11 +104,42 @@ impl PProxy {
         match event {
             P2pServerEvent::TunnelEvent(RequestResponseEvent::RequestReceived {
                 request_id,
+                request,
                 ..
-            }) => self
-                .p2p_server
-                .tunnel_handle
-                .send_response(*request_id, vec![1, 2, 3, 4]),
+            }) => {
+                let Some(proxy_addr) = self.proxy_addr else {
+                    return Err(Error::ProtocolNotSupport);
+                };
+
+                let mut headers = [httparse::EMPTY_HEADER; 64];
+                let mut req = httparse::Request::new(&mut headers);
+                if req.parse(request)?.is_partial() {
+                    return Err(Error::IncompleteHttpRequest);
+                }
+
+                let mut stream = tcp_connect_with_timeout(&proxy_addr, 5).await?;
+                stream.write_all(request).await?;
+
+                let mut response = Vec::new();
+                loop {
+                    let mut buf = [0u8; 30000];
+                    match stream.read(&mut buf).await {
+                        Err(_) => break,
+                        Ok(0) => break,
+                        Ok(n) => {
+                            response.extend_from_slice(&buf[..n]);
+                        }
+                    }
+                }
+
+                if response.is_empty() {
+                    response = b"HTTP/1.1 500 Internal Server Error\r\n\r\n".to_vec();
+                }
+
+                self.p2p_server
+                    .tunnel_handle
+                    .send_response(*request_id, response);
+            }
 
             P2pServerEvent::TunnelEvent(RequestResponseEvent::RequestFailed {
                 request_id,
@@ -212,6 +250,12 @@ impl PProxyHandle {
         &self,
         request: RequestHttpServerRequest,
     ) -> Result<RequestHttpServerResponse> {
+        let mut headers = [httparse::EMPTY_HEADER; 64];
+        let mut req = httparse::Request::new(&mut headers);
+        if req.parse(&request.data)?.is_partial() {
+            return Err(Error::IncompleteHttpRequest);
+        }
+
         let (tx, rx) = oneshot::channel();
 
         self.command_tx
@@ -239,3 +283,20 @@ fn extract_peer_id_from_multiaddr(multiaddr: &Multiaddr) -> Result<PeerId> {
     PeerId::from_multihash(multihash)
         .map_err(|_| Error::FailedToExtractPeerIdFromMultiaddr(multiaddr.to_string()))
 }
+
+pub async fn tcp_connect_with_timeout(
+    addr: &SocketAddr,
+    request_timeout_s: u64,
+) -> Result<TcpStream> {
+    match timeout(
+        Duration::from_secs(request_timeout_s),
+        TcpStream::connect(addr),
+    )
+    .await
+    {
+        Ok(result) => result.map_err(Error::Io),
+        Err(_) => Err(Error::Io(std::io::Error::from(
+            std::io::ErrorKind::TimedOut,
+        ))),
+    }
+}

src/lib.rs

@@ -54,12 +54,15 @@ async fn main() {
         .get_one::<String>("COMMANDER_SERVER_ADDR")
         .unwrap()
         .parse()
-        .expect("Invalid server address");
+        .expect("Invalid command server address");
+    let proxy_addr = args
+        .get_one::<String>("PROXY_ADDR")
+        .map(|addr| addr.parse().expect("Invalid proxy address"));
 
     println!("server_addr: {:?}", server_addr);
     println!("commander_server_addr: {:?}", commander_server_addr);
 
-    let (pproxy, pproxy_handle) = PProxy::new(server_addr);
+    let (pproxy, pproxy_handle) = PProxy::new(server_addr, proxy_addr);
 
     let commander = PProxyCommander::new(pproxy_handle);
     let commander_server = Server::builder()