Merge pull request #4 from w-henderson/proxy

Implement HTTP proxy

humphrey-server/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "humphrey_server"
-version = "0.1.0"
+version = "0.1.1"
 edition = "2018"
 license = "MIT"
 homepage = "https://github.com/w-henderson/Humphrey"
@@ -11,7 +11,7 @@ keywords = ["http", "server", "http-server"]
 categories = ["web-programming::http-server", "network-programming", "command-line-utilities"]
 
 [dependencies]
-humphrey = { version = "0.1.0", path = "../humphrey" }
+humphrey = { version = "0.1.1", path = "../humphrey" }
 libloading = { version = "0.7", optional = true }
 
 [features]

humphrey-server/README.md

@@ -39,7 +39,7 @@ cache_time = 60               ; Maximium time to cache content for in seconds, d
 plugins = "conf/plugins.txt"  ; A text file containing one plugin library file path on each line
 
 [proxy]
-target = "localhost:8000"  ; The address to proxy traffic to, required if the mode is set to proxy
+target = "127.0.0.1:8000"     ; The address to proxy traffic to, required if the mode is set to proxy
 
 [load_balancer]
 targets = "conf/targets.txt"  ; A text file containing one target on each line to balance traffic between, required if the mode is set to load_balancer

humphrey-server/src/server/load_balancer.rs

@@ -1,14 +1,15 @@
-use humphrey::app::{App, ErrorHandler, WebsocketHandler};
-use humphrey::route::RouteHandler;
+use humphrey::app::App;
+use humphrey::http::headers::ResponseHeader;
+use humphrey::http::proxy::proxy_request;
+use humphrey::http::{Request, Response, StatusCode};
 
 use crate::config::{Config, LoadBalancerMode};
 use crate::logger::Logger;
-use crate::proxy::pipe;
 use crate::server::rand::{Choose, Lcg};
 
-use std::net::TcpStream;
+use std::net::ToSocketAddrs;
 use std::sync::{Arc, Mutex};
-use std::thread::spawn;
+use std::time::Duration;
 
 /// Represents the load balancer application state.
 /// Includes the load balancer instance as well as the logger.
@@ -17,6 +18,7 @@ struct AppState {
     load_balancer: Mutex<LoadBalancer>,
     blacklist: Vec<String>,
     logger: Logger,
+    timeout: Duration,
 }
 
 impl From<&Config> for AppState {
@@ -25,14 +27,15 @@ impl From<&Config> for AppState {
             load_balancer: Mutex::new(LoadBalancer::from(config)),
             blacklist: config.blacklist.clone(),
             logger: Logger::from(config),
+            timeout: Duration::from_secs(5),
         }
     }
 }
 
 /// Main function for the load balancer.
 pub fn main(config: Config) {
-    let app: App<AppState> = App::new_with_config(config.threads, AppState::from(&config))
-        .with_custom_connection_handler(handler);
+    let app: App<AppState> =
+        App::new_with_config(config.threads, AppState::from(&config)).with_route("/*", handler);
 
     let addr = format!("{}:{}", config.address, config.port);
 
@@ -88,68 +91,37 @@ impl From<&Config> for LoadBalancer {
 
 /// Handles individual connections to the server.
 /// Ignores the server's specified routes and error handler.
-fn handler(
-    mut source: TcpStream,
-    _: Arc<Vec<RouteHandler<AppState>>>,
-    _: Arc<ErrorHandler>,
-    _: Arc<WebsocketHandler<AppState>>,
-    state: Arc<AppState>,
-) {
-    let address = source.peer_addr();
-    if address.is_err() {
-        state.logger.warn("Corrupted stream attempted to connect");
-        return;
-    }
-    let address = address.unwrap().ip().to_string();
-
-    // Prevent blacklisted addresses from starting a connection
-    if state.blacklist.contains(&address) {
-        state
-            .logger
-            .warn(&format!("{}: Blacklisted IP tried to connect", &address));
-        return;
-    }
+fn handler(request: Request, state: Arc<AppState>) -> Response {
+    // Return error 403 if the address was blacklisted
+    if state
+        .blacklist
+        .contains(&request.address.origin_addr.to_string())
+    {
+        state.logger.warn(&format!(
+            "{}: Blacklisted IP attempted to request {}",
+            request.address, request.uri
+        ));
+        Response::new(StatusCode::Forbidden)
+            .with_header(ResponseHeader::ContentType, "text/html".into())
+            .with_bytes(b"<h1>403 Forbidden</h1>".to_vec())
+            .with_request_compatibility(&request)
+            .with_generated_headers()
+    } else {
+        // Gets a load balancer target using the thread-safe `Mutex`
+        let mut load_balancer_lock = state.load_balancer.lock().unwrap();
+        let target = load_balancer_lock.select_target();
+        drop(load_balancer_lock);
 
-    // Gets a load balancer target using the thread-safe `Mutex`
-    let mut load_balancer_lock = state.load_balancer.lock().unwrap();
-    let target = load_balancer_lock.select_target();
-    drop(load_balancer_lock);
-
-    if let Ok(mut destination) = TcpStream::connect(&target) {
-        // Logs the connection's success
-        state
-            .logger
-            .info(&format!("{} -> {}: Connection started", &address, &target));
-
-        let mut source_clone = source.try_clone().unwrap();
-        let mut destination_clone = destination.try_clone().unwrap();
-
-        // Pipe data in both directions
-        let forward = spawn(move || pipe(&mut source, &mut destination));
-        let backward = spawn(move || pipe(&mut destination_clone, &mut source_clone));
-
-        // Log any errors
-        if let Err(_) = forward.join().unwrap() {
-            state.logger.error(&format!(
-                "{}: Error proxying data from client to target, connection closed",
-                &address
-            ));
-        }
-        if let Err(_) = backward.join().unwrap() {
-            state.logger.error(&format!(
-                "{}: Error proxying data from target to client, connection closed",
-                &address
-            ));
-        }
+        let target_sock = target.to_socket_addrs().unwrap().next().unwrap();
+        let response = proxy_request(&request, target_sock, state.timeout);
+        let status: u16 = response.status_code.clone().into();
+        let status_string: &str = response.status_code.clone().into();
 
         state.logger.info(&format!(
-            "{} -> {}: Session complete, connection closed",
-            &address, &target
+            "{}: {} {} {}",
+            request.address, status, status_string, request.uri
         ));
-    } else {
-        state.logger.error(&format!(
-            "Could not connect to load balancer target {}",
-            target
-        ))
+
+        response
     }
 }

humphrey-server/src/server/mod.rs

@@ -1,6 +1,7 @@
 pub mod cache;
 pub mod load_balancer;
 pub mod logger;
+pub mod pipe;
 pub mod proxy;
 pub mod rand;
 pub mod route;

humphrey-server/src/server/pipe.rs

@@ -0,0 +1,21 @@
+use std::io::{Read, Write};
+use std::net::{Shutdown, TcpStream};
+
+/// Pipe bytes from one stream to another, up to 1KiB at a time.
+pub fn pipe(source: &mut TcpStream, destination: &mut TcpStream) -> Result<(), ()> {
+    let mut buf: [u8; 1024] = [0; 1024];
+
+    loop {
+        let length = source.read(&mut buf).map_err(|_| ())?;
+
+        if length == 0 {
+            destination.shutdown(Shutdown::Both).map_err(|_| ())?;
+            break;
+        }
+
+        if let Ok(_) = destination.write(&buf[..length]) {
+            destination.flush().map_err(|_| ())?;
+        }
+    }
+    Ok(())
+}

humphrey-server/src/server/proxy.rs

@@ -1,37 +1,46 @@
-use humphrey::app::{App, ErrorHandler, WebsocketHandler};
-use humphrey::route::RouteHandler;
+use humphrey::app::App;
+use humphrey::http::headers::ResponseHeader;
+use humphrey::http::proxy::proxy_request;
+use humphrey::http::{Request, Response, StatusCode};
 
 use crate::config::Config;
 use crate::logger::Logger;
 
-use std::io::{Read, Write};
-use std::net::{Shutdown, TcpStream};
+use std::net::{SocketAddr, ToSocketAddrs};
 use std::sync::Arc;
-use std::thread::spawn;
+use std::time::Duration;
 
 /// Represents the application state.
 /// Includes the proxy target and the logger.
-#[derive(Default)]
 struct AppState {
-    target: String,
+    target: SocketAddr,
     blacklist: Vec<String>,
     logger: Logger,
+    timeout: Duration,
 }
 
 impl From<&Config> for AppState {
     fn from(config: &Config) -> Self {
         Self {
-            target: config.proxy_target.as_ref().unwrap().clone(),
+            target: config
+                .proxy_target
+                .as_ref()
+                .unwrap()
+                .to_socket_addrs()
+                .unwrap()
+                .next()
+                .unwrap(),
             blacklist: config.blacklist.clone(),
             logger: Logger::from(config),
+            timeout: Duration::from_secs(5),
         }
     }
 }
 
 /// Main function for the proxy server.
 pub fn main(config: Config) {
-    let app: App<AppState> = App::new_with_config(config.threads, AppState::from(&config))
-        .with_custom_connection_handler(handler);
+    let app: App<AppState> =
+        App::new_with_config(config.threads, AppState::from(&config)).with_route("/*", handler);
 
     let addr = format!("{}:{}", config.address, config.port);
 
@@ -48,79 +57,31 @@ pub fn main(config: Config) {
 }
 
 /// Handles individual connections to the server.
-/// Ignores the server's specified routes and error handler.
-fn handler(
-    mut source: TcpStream,
-    _: Arc<Vec<RouteHandler<AppState>>>,
-    _: Arc<ErrorHandler>,
-    _: Arc<WebsocketHandler<AppState>>,
-    state: Arc<AppState>,
-) {
-    let address = source.peer_addr();
-    if address.is_err() {
-        state.logger.warn("Corrupted stream attempted to connect");
-        return;
-    }
-    let address = address.unwrap().ip().to_string();
-
-    // Prevent blacklisted addresses from starting a connection
-    if state.blacklist.contains(&address) {
-        state
-            .logger
-            .warn(&format!("{}: Blacklisted IP tried to connect", &address));
-        return;
-    }
-
-    if let Ok(mut destination) = TcpStream::connect(&*state.target) {
-        // The target was successfully connected to
-        let mut source_clone = source.try_clone().unwrap();
-        let mut destination_clone = destination.try_clone().unwrap();
-        state
-            .logger
-            .info(&format!("{}: Connected, proxying data", &address));
-
-        // Pipe data in both directions
-        let forward = spawn(move || pipe(&mut source, &mut destination));
-        let backward = spawn(move || pipe(&mut destination_clone, &mut source_clone));
-
-        // Log any errors
-        if let Err(_) = forward.join().unwrap() {
-            state.logger.error(&format!(
-                "{}: Error proxying data from client to target, connection closed",
-                &address
-            ));
-        }
-        if let Err(_) = backward.join().unwrap() {
-            state.logger.error(&format!(
-                "{}: Error proxying data from target to client, connection closed",
-                &address
-            ));
-        }
-
-        state.logger.info(&format!(
-            "{}: Session complete, connection closed",
-            &address
+fn handler(request: Request, state: Arc<AppState>) -> Response {
+    // Return error 403 if the address was blacklisted
+    if state
+        .blacklist
+        .contains(&request.address.origin_addr.to_string())
+    {
+        state.logger.warn(&format!(
+            "{}: Blacklisted IP attempted to request {}",
+            request.address, request.uri
         ));
+        Response::new(StatusCode::Forbidden)
+            .with_header(ResponseHeader::ContentType, "text/html".into())
+            .with_bytes(b"<h1>403 Forbidden</h1>".to_vec())
+            .with_request_compatibility(&request)
+            .with_generated_headers()
     } else {
-        state.logger.error("Could not connect to target");
-    }
-}
-
-/// Pipe bytes from one stream to another, up to 1KiB at a time.
-pub fn pipe(source: &mut TcpStream, destination: &mut TcpStream) -> Result<(), ()> {
-    let mut buf: [u8; 1024] = [0; 1024];
-
-    loop {
-        let length = source.read(&mut buf).map_err(|_| ())?;
+        let response = proxy_request(&request, state.target, state.timeout);
+        let status: u16 = response.status_code.clone().into();
+        let status_string: &str = response.status_code.clone().into();
 
-        if length == 0 {
-            destination.shutdown(Shutdown::Both).map_err(|_| ())?;
-            break;
-        }
+        state.logger.info(&format!(
+            "{}: {} {} {}",
+            request.address, status, status_string, request.uri
+        ));
 
-        if let Ok(_) = destination.write(&buf[..length]) {
-            destination.flush().map_err(|_| ())?;
-        }
+        response
     }
-    Ok(())
 }

humphrey-server/src/server/static_server.rs

@@ -12,7 +12,7 @@ use crate::cache::Cache;
 use crate::config::{BlacklistMode, Config};
 use crate::logger::Logger;
 use crate::route::try_find_path;
-use crate::server::proxy::pipe;
+use crate::server::pipe::pipe;
 
 use std::fs::File;
 use std::io::{Read, Write};

humphrey/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "humphrey"
-version = "0.1.0"
+version = "0.1.1"
 edition = "2018"
 license = "MIT"
 homepage = "https://github.com/w-henderson/Humphrey"

humphrey/src/http/mod.rs

@@ -3,6 +3,7 @@ pub mod date;
 pub mod headers;
 pub mod method;
 pub mod mime;
+pub mod proxy;
 pub mod request;
 pub mod response;
 pub mod status;