Merge branch 'main' of https://git.meex.lol/MeexReay/flowgate

README.md

@@ -3,22 +3,32 @@ HTTP requests redirection system
 
 Features:
 - Request redirection
-- TLS support
+- TLS support (via Rustls!)
 - Keep-alive connections
 - Sending IP in header (X-Real-IP)
+- Multiple ip forwarding methods
+- Accepts incoming ip forwarding
 
 TODO:
-- Remove panics
-- Creating trees of flowgate
 - Filter by headers
 - Modify response headers
-- HTTP/3 full support
+- HTTP/3 full support (quic/udp)
 
 ## How to use
 
 Firstly, download it from releases. or build from sources (read BUILD.md) \
 Just run it and configure in `conf.yml` file.
 
+### Logging
+
+To get all logs (with debug ones), set this env var:
+
+```
+RUST_LOG=debug
+```
+
+Read more: [env_logger](https://docs.rs/env_logger/latest/env_logger/#enabling-logging)
+
 ## Configuration
 
 ### IP forwarding methods

conf.yml

@@ -1,15 +1,16 @@
-http_host: localhost:80     # Http server host
+http_host: localhost:80     # Http server host (optional)
-https_host: localhost:443   # Https server host
+https_host: localhost:443   # Https server host (optional)
 
 connection_timeout: 10         # Read and write timeout of connections in seconds (optional, default - 10)
 incoming_ip_forwarding: none   # Read IP forwarding on incoming connections (optional, default - none)
+threadpool_size: 10            # Size of the global threadpool (optional, default - 10)
 
 sites:
-  - domain: localhost                                # Site domain (use wildcard matching)
+ - domain: localhost                                # Site domain (use wildcard matching)
-    host: localhost:8080                             # Http server host
+   host: localhost:8080                             # Http server host
-    ip_forwarding: simple                            # IP forwarding method type (optional, default - header)
+   ip_forwarding: simple                            # IP forwarding method type (optional, default - header)
-    enable_keep_alive: true                          # Enable keep-alive connections (optional, default - true)
+   enable_keep_alive: true                          # Enable keep-alive connections (optional, default - true)
-    support_keep_alive: true                         # Does server supports keep-alive connections (optional, default - true)
+   support_keep_alive: true                         # Does server supports keep-alive connections (optional, default - true)
-    # ssl_cert: "/path/to/public/certificate.txt"    # Ssl public certificate file (optional)
+   # ssl_cert: "/path/to/public/certificate.txt"    # Ssl public certificate file (optional)
-    # ssl_key: "/path/to/private/key.txt"            # Ssl private key file (optional)
+   # ssl_key: "/path/to/private/key.txt"            # Ssl private key file (optional)
-    replace_host: "meex.lol"                         # Replace Host header in requests to server (optional)
+   replace_host: "meex.lol"                         # Replace Host header in requests to server (optional)

src/config.rs

@@ -51,10 +51,11 @@ impl IpForwarding {
 #[derive(Clone, Debug)]
 pub struct Config {
     pub sites: Vec<SiteConfig>,
-    pub http_host: String,
+    pub http_host: Option<String>,
-    pub https_host: String,
+    pub https_host: Option<String>,
     pub connection_timeout: Duration,
-    pub incoming_ip_forwarding: IpForwarding
+    pub incoming_ip_forwarding: IpForwarding,
+    pub threadpool_size: usize
 }
 
 impl Config {
@@ -62,8 +63,10 @@ impl Config {
         let file_content = fs::read_to_string(filename).ok()?;
         let doc = serde_yml::from_str::<Value>(file_content.as_str()).ok()?;
 
-        let http_host = doc["http_host"].as_str()?.to_string();
+        let http_host = doc.get("http_host").and_then(|o| Some(o.as_str()?.to_string()));
-        let https_host = doc["https_host"].as_str()?.to_string();
+        let https_host = doc.get("https_host").and_then(|o| Some(o.as_str()?.to_string()));
+
+        let threadpool_size = doc.get("threadpool_size").and_then(|o| Some(o.as_u64()? as usize + 2)).unwrap_or(12);
 
         let connection_timeout = Duration::from_secs(doc.get("connection_timeout")
             .unwrap_or(&Value::Number(Number::from(10))).as_u64()?);
@@ -115,7 +118,8 @@ impl Config {
             http_host,
             https_host,
             connection_timeout,
-            incoming_ip_forwarding
+            incoming_ip_forwarding,
+            threadpool_size
         }.clone())
     }
 

src/server.rs

@@ -41,7 +41,7 @@ impl FlowgateServer {
     pub fn start(self) -> ThreadPool {
         let local_self = Arc::new(self);
 
-        let threadpool = ThreadPool::new(10);
+        let threadpool = ThreadPool::new(local_self.config.threadpool_size);
 
         let mut handles = Vec::new();
 
@@ -68,25 +68,29 @@ impl FlowgateServer {
     }
 
     pub fn run_http(self: Arc<Self>, threadpool: &ThreadPool) -> Result<(), Box<dyn Error>> {
-        let listener = TcpListener::bind(&self.config.http_host)?;
+        if let Some(host) = self.config.http_host.clone() {
+            let listener = TcpListener::bind(&host)?;
 
-        info!("HTTP server runned on {}", &self.config.http_host);
+            info!("HTTP server runned on {}", &host);
 
-        for stream in listener.incoming() {
+            for stream in listener.incoming() {
-            if let Ok(mut stream) = stream {
+                if let Ok(mut stream) = stream {
-                let local_self = self.clone();
+                    let local_self = self.clone();
 
-                let Ok(addr) = stream.peer_addr() else { continue };
+                    let Ok(addr) = stream.peer_addr() else { continue };
-                let Ok(_) = stream.set_write_timeout(Some(local_self.config.connection_timeout)) else { continue };
+                    let Ok(_) = stream.set_write_timeout(Some(local_self.config.connection_timeout)) else { continue };
-                let Ok(_) = stream.set_read_timeout(Some(local_self.config.connection_timeout)) else { continue };
+                    let Ok(_) = stream.set_read_timeout(Some(local_self.config.connection_timeout)) else { continue };
 
-                threadpool.execute(move || {
+                    threadpool.execute(move || {
-                    local_self.accept_stream(
+                        local_self.accept_stream(
-                        &mut stream,
+                            &mut stream,
-                        addr,
+                            addr,
-                        false
+                            false
-                    );
+                        );
-                });
+                        
+                        debug!("{} close connection", addr);
+                    });
+                }
             }
         }
 
@@ -94,36 +98,38 @@ impl FlowgateServer {
     }
 
     pub fn run_https(self: Arc<Self>, threadpool: &ThreadPool) -> Result<(), Box<dyn Error>> {
-        let listener = TcpListener::bind(&self.config.https_host)?;
+        if let Some(host) = self.config.https_host.clone() {
+            let listener = TcpListener::bind(&host)?;
 
-        info!("HTTPS server runned on {}", &self.config.https_host);
+            info!("HTTPS server runned on {}", &host);
 
-        let config = Arc::new(create_server_config(self.config.clone()));
+            let config = Arc::new(create_server_config(self.config.clone()));
 
-        for stream in listener.incoming() {
+            for stream in listener.incoming() {
-            if let Ok(stream) = stream {
+                if let Ok(stream) = stream {
-                let local_self = self.clone();
+                    let local_self = self.clone();
-                let config = config.clone();
+                    let config = config.clone();
 
-                let Ok(addr) = stream.peer_addr() else { continue };
+                    let Ok(addr) = stream.peer_addr() else { continue };
-                let Ok(_) = stream.set_write_timeout(Some(local_self.config.connection_timeout)) else { continue };
+                    let Ok(_) = stream.set_write_timeout(Some(local_self.config.connection_timeout)) else { continue };
-                let Ok(_) = stream.set_read_timeout(Some(local_self.config.connection_timeout)) else { continue };
+                    let Ok(_) = stream.set_read_timeout(Some(local_self.config.connection_timeout)) else { continue };
 
-                threadpool.execute(move || {
+                    threadpool.execute(move || {
-                    let Ok(connection) = ServerConnection::new(config) else { return };
+                        let Ok(connection) = ServerConnection::new(config) else { return };
-                    debug!("{} open connection", addr);
+                        debug!("{} open connection", addr);
-                    let mut stream = StreamOwned::new(connection, stream);
+                        let mut stream = StreamOwned::new(connection, stream);
 
-                    let Ok(_) = stream.conn.complete_io(&mut stream.sock) else { debug!("{} close connection", addr); return };
+                        let Ok(_) = stream.conn.complete_io(&mut stream.sock) else { debug!("{} close connection", addr); return };
+
+                        local_self.accept_stream(
+                            &mut stream,
+                            addr,
+                            true
+                        );
 
-                    local_self.accept_stream(
-                        &mut stream,
-                        addr,
-                        true
-                    ).map(|_| {
                         debug!("{} close connection", addr);
                     });
-                });
+                }
             }
         }
 
@@ -362,6 +368,11 @@ impl FlowgateServer {
             }
         } else if is_chunked {
             transfer_chunked(&mut stream, conn.stream.get_mut())?;
+        } else {
+            let buffer = stream.buffer().to_vec();
+    
+            conn.stream.get_mut().write_all(&buffer).ok()?;
+            stream.consume(buffer.len());
         }
 
         debug!("{} {} send body to server", addr, status[1]);
@@ -432,6 +443,11 @@ impl FlowgateServer {
                 }
             } else if is_chunked {
                 transfer_chunked(&mut conn.stream, stream.get_mut())?;
+            } else {
+                let buffer = conn.stream.buffer().to_vec();
+    
+                stream.get_mut().write_all(&buffer).ok()?;
+                conn.stream.consume(buffer.len());
             }
 
             debug!("{} {} send response body to clientr", addr, status[1]);

tests/half_request.py

@@ -0,0 +1,32 @@
+import socket
+
+HOST = '172.16.1.32'
+PORT = 80
+
+with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
+    s.connect((HOST, PORT))
+
+    s.send(b"""GET / HTTP/1.1\r
+Accept: text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7\r
+Accept-Encoding: gzip, deflate, br, zstd\r
+Accept-Language: en-US,en;q=0.9\r
+Cache-Control: no-cache\r
+Connection: keep-alive\r
+Host: git.meex.lol\r
+Pragma: no-cache\r
+Sec-Fetch-Dest: document\r
+Sec-Fetch-Mode: navigate\r
+Sec-Fetch-Site: none\r
+Sec-Fetch-User: ?1\r
+Upgrade-Insecure-Requests: 1\r
+User-Agent: Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/135.0.0.0 Safari/537.36\r
+sec-ch-ua: \"Chromium\";v=\"135\", \"Not-A.Brand\";v=\"8\"\r
+sec-ch-ua-mobile: ?0\r
+sec-ch-ua-platform: \"Linux\"\r
+\r
+""")
+
+    content_length = 0
+    while True:
+        data = s.recv(1024)
+        print(data)