Improve config file handling

Signed-off-by: Jacob Kiers <code@kiers.eu>

.cargo/config.toml

@@ -0,0 +1,3 @@
+[profile.release]
+lto = "thin"
+strip = true

.drone.jsonnet

@@ -0,0 +1,92 @@
+local executableName = 'l4p';
+local build_image = 'img.kie.rs/jjkiers/rust-cross:rust1.71.1-zig';
+
+local archs = [
+  { target: 'aarch64-unknown-linux-musl', short: 'arm64-musl' },
+  { target: 'x86_64-pc-windows-gnu', short: 'windows' },
+  { target: 'x86_64-unknown-linux-musl', short: 'amd64-musl' },
+];
+
+local getStepName(arch) = 'Build for ' + arch.short;
+
+local builtExecutableName(arch) = executableName + if std.length(std.findSubstr(arch.short, 'windows')) > 0 then '.exe' else '';
+local targetExecutableName(arch) = executableName + '-' + arch.target + if std.length(std.findSubstr(arch.short, 'windows')) > 0 then '.exe' else '';
+
+local getVolumeName(arch) = 'target-' + arch.target;
+local getLocalVolumes(arch) = [
+  {
+    name: getVolumeName(arch),
+    temp: {},
+  }
+  for arch in archs
+];
+
+local add_build_steps() = [
+  {
+    name: getStepName(arch),
+    image: build_image,
+    commands: [
+      'echo Hello World from Jsonnet on ' + arch.target + '!',
+      'cargo zigbuild --release --target ' + arch.target,
+      'cp target/' + arch.target + '/release/' + builtExecutableName(arch) + ' artifacts/' + targetExecutableName(arch),
+      'rm -rf target/' + arch.target + '/release/*',
+    ],
+    depends_on: ['Prepare'],
+    volumes: [{
+      name: getVolumeName(arch),
+      path: '/drone/src/target',
+    }],
+  }
+  for arch in archs
+];
+
+{
+  kind: 'pipeline',
+  type: 'docker',
+  name: 'default',
+  platform: {
+    arch: 'amd64',
+  },
+  steps:
+    [{
+      name: 'Prepare',
+      image: build_image,
+      commands: [
+        'mkdir artifacts',
+        'echo Using image: ' + build_image,
+        'cargo --version',
+        'rustc --version',
+      ],
+    }] +
+    add_build_steps() +
+    [
+      {
+        name: 'Show built artifacts',
+        image: build_image,
+        commands: [
+          'ls -lah artifacts',
+        ],
+        depends_on: [getStepName(a) for a in archs],
+      },
+      {
+        name: 'Create release on gitea',
+        image: 'plugins/gitea-release',
+        settings: {
+          api_key: {
+            from_secret: 'gitea_token',
+          },
+          base_url: 'https://code.kiers.eu',
+          files: 'artifacts/*',
+          checksum: 'sha256',
+        },
+        when: {
+          event: ['tag', 'promote'],
+        },
+        depends_on: ['Show built artifacts'],
+      },
+    ],
+
+  volumes: getLocalVolumes(archs),
+
+  image_pull_secrets: ['docker_private_repo'],
+}

.github/workflows/rust.yml

@@ -1,24 +0,0 @@
-name: Rust
-
-on:
-  push:
-    branches: [ main ]
-  pull_request:
-    branches: [ main ]
-
-env:
-  CARGO_TERM_COLOR: always
-
-jobs:
-  build:
-
-    runs-on: ubuntu-latest
-
-    steps:
-    - uses: actions/checkout@v2
-    - name: Upgrade Rust
-      run: rustup update
-    - name: Build
-      run: cargo build --verbose
-    - name: Run tests
-      run: cargo test --verbose

.gitignore

@@ -1 +1,2 @@
 /target
+config.yaml

Cargo.toml

@@ -1,13 +1,13 @@
 [package]
-name = "fourth"
-version = "0.1.3"
+name = "layer4-proxy"
+version = "0.1.7"
 edition = "2021"
-authors = ["LI Rui <lr_cn@outlook.com>"]
+authors = ["Jacob Kiers <code@kiers.eu>"]
 license = "Apache-2.0"
 description = "Simple and fast layer 4 proxy in Rust"
 readme = "README.md"
-homepage = "https://github.com/KernelErr/fourth"
-repository = "https://github.com/KernelErr/fourth"
+homepage = "https://code.kiers.eu/jjkiers/layer4-proxy"
+repository = "https://code.kiers.eu/jjkiers/layer4-proxy"
 keywords = ["proxy", "network"]
 categories = ["web-programming"]
 
@@ -15,16 +15,20 @@ categories = ["web-programming"]
 
 exclude = [".*"]
 
+[[bin]]
+name = "l4p"
+path = "src/main.rs"
+
 [dependencies]
-log = "0.4"
-pretty_env_logger = "0.4"
-serde = { version = "1.0", features = ["derive"] }
-serde_yaml = "0.8"
-futures = "0.3"
-tls-parser = "0.11"
-
-tokio = { version = "1.0", features = ["full"] }
-
+async-trait = "0.1.73"
+byte_string = "1"
 bytes = "1.1"
-kcp = "0.4"
-byte_string = "1"
+futures = "0.3"
+log = "0.4"
+pretty_env_logger = "0.5"
+serde = { version = "1.0", features = ["derive"] }
+serde_yaml = "0.9.21"
+time = { version = "0.3.1", features = ["local-offset", "formatting"] }
+tls-parser = "0.11"
+tokio = { version = "1.0", features = ["full"] }
+url = "2.2.2"

README-EN.md

@@ -1,71 +0,0 @@
-# Fourth
-
-> Hey, now we are on level 4!
-
-[![](https://img.shields.io/crates/v/fourth)](https://crates.io/crates/fourth) [![CI](https://img.shields.io/github/workflow/status/kernelerr/fourth/Rust)](https://github.com/KernelErr/fourth/actions/workflows/rust.yml)
-
-**Under heavy development, version 0.1 may update frequently**
-
-Fourth is a layer 4 proxy implemented by Rust to listen on specific ports and transfer TCP/KCP data to remote addresses(only TCP) according to configuration.
-
-## Features
-
-- Listen on specific port and proxy to local or remote port
-- SNI-based rule without terminating TLS connection
-- Allow KCP inbound(warning: untested)
-
-## Installation
-
-To gain best performance on your computer's architecture, please consider build the source code. First, you may need [Rust tool chain](https://rustup.rs/).
-
-```bash
-$ cd fourth
-$ cargo build --release
-```
-
-Binary file will be generated at `target/release/fourth`, or you can use `cargo install --path .` to install.
-
-Or you can use Cargo to install Fourth:
-
-```bash
-$ cargo install fourth
-```
-
-## Configuration
-
-Fourth will read yaml format configuration file from `/etc/fourth/config.yaml`, here is an example:
-
-```yaml
-version: 1
-log: info
-
-servers:
-  example_server:
-    listen:
-      - "0.0.0.0:443"
-      - "[::]:443"
-    tls: true # Enable TLS features like SNI
-    sni:
-      proxy.example.com: proxy
-      www.example.com: nginx
-    default: ban
-  relay_server:
-    listen:
-      - "127.0.0.1:8081"
-    default: remote
-
-upstream:
-  nginx: "127.0.0.1:8080"
-  proxy: "127.0.0.1:1024"
-  remote: "www.remote.example.com:8082" # proxy to remote address
-```
-
-Built-in two upstreams: ban(terminate connection immediately), echo
-
-## Thanks
-
-- [tokio_kcp](https://github.com/Matrix-Zhang/tokio_kcp)
-
-## License
-
-Fourth is available under terms of Apache-2.0.

README.md

@@ -1,84 +1,64 @@
-# Fourth
+# l4p
 
-> 这一波在第四层。
+> Hey, now we are on level 4!
 
-[![](https://img.shields.io/crates/v/fourth)](https://crates.io/crates/fourth) [![CI](https://img.shields.io/github/workflow/status/kernelerr/fourth/Rust)](https://github.com/KernelErr/fourth/actions/workflows/rust.yml)
+![CI](https://drone-ci.kiers.eu/api/badges/jjkiers/layer4-proxy/status.svg)
 
-[English](/README-EN.md)
+`l4p` is a layer 4 proxy implemented by Rust to listen on specific ports and transfer TCP/KCP data to remote addresses(only TCP) according to configuration.
 
-**积极开发中，0.1版本迭代可能较快**
+## Features
 
-Fourth是一个Rust实现的Layer 4代理，用于监听指定端口TCP/KCP流量，并根据规则转发到指定目标（目前只支持TCP）。
+- Listen on specific port and proxy to local or remote port
+- SNI-based rule without terminating TLS connection
+- DNS-based backend with periodic resolution
 
-## 功能
+## Installation
 
-- 监听指定端口代理到本地或远端指定端口
-- 监听指定端口，通过TLS ClientHello消息中的SNI进行分流
-- 支持KCP入站（警告：未测试）
-
-## 安装方法
-
-为了确保获得您架构下的最佳性能，请考虑自行编译，首选需要确保您拥有[Rust工具链](https://rustup.rs/)。
+To gain best performance on your computer's architecture, please consider build the source code. First, you may need [Rust tool chain](https://rustup.rs/).
 
 ```bash
-$ cd fourth
+$ cd l4p
 $ cargo build --release
 ```
 
-将在`target/release/fourth`生成二进制文件，您也可以使用`cargo install --path . `来安装二进制文件。
+Binary file will be generated at `target/release/l4p`, or you can use `cargo install --path .` to install.
 
-或者您也可以使用Cargo直接安装：
+Or you can use Cargo to install `l4p`:
 
 ```bash
-$ cargo install fourth
+$ cargo install l4p
 ```
 
-## 配置
+Or you can download binary file form the Release page.
 
-Fourth使用yaml格式的配置文件，默认情况下会读取`/etc/fourth/config.yaml`，如下是一个示例配置。
+## Configuration
+
+`l4p` will read yaml format configuration file from `/etc/l4p/l4p.yaml`, and you can set custom path to environment variable `L4P_CONFIG`, here is an minimal viable example:
 
 ```yaml
 version: 1
 log: info
 
 servers:
-  example_server:
-    listen:
-      - "0.0.0.0:443"
-      - "[::]:443"
-    tls: true # Enable TLS features like SNI filtering
-    sni:
-      proxy.example.com: proxy
-      www.example.com: nginx
-    default: ban
   proxy_server:
     listen:
       - "127.0.0.1:8081"
     default: remote
-  kcp_server:
-    protocol: kcp # default TCP
-    listen:
-      - "127.0.0.1:8082"
-    default: echo
 
 upstream:
-  nginx: "127.0.0.1:8080"
-  proxy: "127.0.0.1:1024"
-  remote: "www.remote.example.com:8082" # proxy to remote address
+  remote: "tcp://www.remote.example.com:8082" # proxy to remote address
 ```
 
-内置两个的upstream：ban（立即中断连接）、echo（返回读到的数据）。
+There are two upstreams built in:
+* Ban, which terminates the connection immediately
+* Echo, which reflects back with the input
 
-## io_uring?
+For detailed configuration, check [this example](./config.yaml.example).
 
-尽管经过了很多尝试，我们发现目前一些Rust下面的io_uring实现存在问题，我们使用的io_uring库实现尽管在吞吐量上可以做到单线程20Gbps（相比之下Tokio仅有8Gbps），但在QPS上存在性能损失较大的问题。因此在有成熟的io_uring实现之前，我们仍然选择epoll。之后我们会持续关注相关进展。
+## Thanks
 
-可能以后会为Linux高内核版本的用户提供可选的io_uring加速。
+- [`l4p`](https://crates.io/crates/`l4p`), of which this is a heavily modified fork.
 
-## 感谢
+## License
 
-- [tokio_kcp](https://github.com/Matrix-Zhang/tokio_kcp)
-
-## 协议
-
-Fourth以Apache-2.0协议开源。
+`l4p` is available under terms of Apache-2.0.

config.yaml.example

@@ -0,0 +1,21 @@
+version: 1
+log: debug
+
+servers:
+  first_server:
+    listen:
+      - "0.0.0.0:8443"
+      - "[::]:8443"
+    tls: true # Enable TLS features like SNI filtering
+    sni:
+      api.example.org: example-api
+      www.example.org: proxy
+    default: ban
+
+  second-server:
+    listen: [ "127.0.0.1:8080" ]
+    default: echo
+
+upstream:
+  proxy: "tcp://new-www.example.org:443" # Connect over IPv4 or IPv6 to new-www.example.org:443
+  example-api: "tcp6://api-v1.example.com:443" # Connect over IPv6 to api-v1.example.com:443

example-config.yaml

@@ -22,6 +22,6 @@ servers:
     default: echo
 
 upstream:
-  nginx: "127.0.0.1:8080"
-  proxy: "127.0.0.1:1024"
-  remote: "www.remote.example.com:8082" # proxy to remote address
+  nginx: "tcp://127.0.0.1:8080"
+  proxy: "tcp://127.0.0.1:1024"
+  remote: "tcp://www.remote.example.com:8082" # proxy to remote address

fourth.service

@@ -0,0 +1,51 @@
+[Unit]
+Description=Fourth - Layer 4 proxy
+
+After=network-online.target
+Wants=network-online.target
+
+[Install]
+WantedBy=default.target
+
+[Service]
+Type=simple
+
+# Allow read-only access to the config directory
+ReadOnlyPaths=/etc/fourth
+# Path to the binary
+ExecStart=/usr/local/bin/fourth
+
+# Needs CAP_NET_BIND_SERVICE in order to bind to lower ports
+# When using ports above 1024, these should be made empty
+AmbientCapabilities=CAP_NET_BIND_SERVICE
+CapabilityBoundingSet=CAP_NET_BIND_SERVICE
+
+# Run as a dynamic user
+DynamicUser=yes
+
+# Security
+PrivateTmp=yes
+PrivateDevices=yes
+ProtectSystem=strict
+ProtectHome=yes
+SystemCallFilter=@basic-io @file-system @network-io @system-service
+SystemCallFilter=~@privileged
+SystemCallFilter=~@resources
+NoNewPrivileges=yes
+ProtectProc=invisible
+RemoveIPC=yes
+RestrictAddressFamilies=AF_INET AF_INET6
+RestrictNamespaces=yes
+ProtectHostname=yes
+ProtectClock=yes
+ProtectKernelModules=yes
+ProtectKernelLogs=yes
+ProtectControlGroups=yes
+LockPersonality=yes
+MemoryDenyWriteExecute=yes
+RestrictRealtime=yes
+ProcSubset=pid
+UMask=0077
+SystemCallArchitectures=native
+RestrictSUIDSGID=yes
+ProtectKernelTunables=yes

src/config.rs

@@ -1,93 +0,0 @@
-use log::debug;
-use serde::Deserialize;
-use std::collections::HashMap;
-use std::fs::File;
-use std::io::{Error as IOError, Read};
-
-#[derive(Debug, Clone)]
-pub struct Config {
-    pub base: BaseConfig,
-}
-
-#[derive(Debug, Default, Deserialize, Clone)]
-pub struct BaseConfig {
-    pub version: i32,
-    pub log: Option<String>,
-    pub servers: HashMap<String, ServerConfig>,
-    pub upstream: HashMap<String, String>,
-}
-
-#[derive(Debug, Default, Deserialize, Clone)]
-pub struct ServerConfig {
-    pub listen: Vec<String>,
-    pub protocol: Option<String>,
-    pub tls: Option<bool>,
-    pub sni: Option<HashMap<String, String>>,
-    pub default: Option<String>,
-}
-
-#[derive(Debug)]
-pub enum ConfigError {
-    IO(IOError),
-    Yaml(serde_yaml::Error),
-    Custom(String),
-}
-
-impl Config {
-    pub fn new(path: &str) -> Result<Config, ConfigError> {
-        let base = (load_config(path))?;
-
-        Ok(Config { base })
-    }
-}
-
-fn load_config(path: &str) -> Result<BaseConfig, ConfigError> {
-    let mut contents = String::new();
-    let mut file = (File::open(path))?;
-    (file.read_to_string(&mut contents))?;
-
-    let parsed: BaseConfig = serde_yaml::from_str(&contents)?;
-
-    if parsed.version != 1 {
-        return Err(ConfigError::Custom(
-            "Unsupported config version".to_string(),
-        ));
-    }
-
-    let log_level = parsed.log.clone().unwrap_or_else(|| "info".to_string());
-    if !log_level.eq("disable") {
-        std::env::set_var("FOURTH_LOG", log_level.clone());
-        pretty_env_logger::init_custom_env("FOURTH_LOG");
-        debug!("Set log level to {}", log_level);
-    }
-
-    debug!("Config version {}", parsed.version);
-
-    Ok(parsed)
-}
-
-impl From<IOError> for ConfigError {
-    fn from(err: IOError) -> ConfigError {
-        ConfigError::IO(err)
-    }
-}
-
-impl From<serde_yaml::Error> for ConfigError {
-    fn from(err: serde_yaml::Error) -> ConfigError {
-        ConfigError::Yaml(err)
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-
-    #[test]
-    fn test_load_config() {
-        let config = Config::new("tests/config.yaml").unwrap();
-        assert_eq!(config.base.version, 1);
-        assert_eq!(config.base.log.unwrap(), "disable");
-        assert_eq!(config.base.servers.len(), 5);
-        assert_eq!(config.base.upstream.len(), 3);
-    }
-}

src/config/config.rs

@@ -0,0 +1,227 @@
+use crate::upstreams::ProxyToUpstream;
+use crate::upstreams::Upstream;
+use log::{debug, warn};
+use serde::Deserialize;
+use std::collections::{HashMap, HashSet};
+use std::fs::File;
+use std::io::{Error as IOError, Read};
+use url::Url;
+
+#[derive(Debug, Clone)]
+pub struct ConfigV1 {
+    pub base: ParsedConfigV1,
+}
+
+#[derive(Debug, Default, Deserialize, Clone)]
+pub struct ParsedConfigV1 {
+    pub version: i32,
+    pub log: Option<String>,
+    pub servers: HashMap<String, ServerConfig>,
+    pub upstream: HashMap<String, Upstream>,
+}
+
+#[derive(Debug, Default, Deserialize, Clone)]
+pub struct BaseConfig {
+    pub version: i32,
+    pub log: Option<String>,
+    pub servers: HashMap<String, ServerConfig>,
+    pub upstream: HashMap<String, String>,
+}
+
+#[derive(Debug, Default, Deserialize, Clone)]
+pub struct ServerConfig {
+    pub listen: Vec<String>,
+    pub protocol: Option<String>,
+    pub tls: Option<bool>,
+    pub sni: Option<HashMap<String, String>>,
+    pub default: Option<String>,
+}
+impl TryInto<ProxyToUpstream> for &str {
+    type Error = ConfigError;
+
+    fn try_into(self) -> Result<ProxyToUpstream, Self::Error> {
+        let upstream_url = match Url::parse(self) {
+            Ok(url) => url,
+            Err(_) => {
+                return Err(ConfigError::Custom(format!(
+                    "Invalid upstream url {}",
+                    self
+                )))
+            }
+        };
+
+        let upstream_host = match upstream_url.host_str() {
+            Some(host) => host,
+            None => {
+                return Err(ConfigError::Custom(format!(
+                    "Invalid upstream url {}",
+                    self
+                )))
+            }
+        };
+
+        let upstream_port = match upstream_url.port_or_known_default() {
+            Some(port) => port,
+            None => {
+                return Err(ConfigError::Custom(format!(
+                    "Invalid upstream url {}",
+                    self
+                )))
+            }
+        };
+
+        match upstream_url.scheme() {
+            "tcp" | "tcp4" | "tcp6" => {}
+            _ => {
+                return Err(ConfigError::Custom(format!(
+                    "Invalid upstream scheme {}",
+                    self
+                )))
+            }
+        }
+
+        Ok(ProxyToUpstream::new(
+            format!("{}:{}", upstream_host, upstream_port),
+            upstream_url.scheme().to_string(),
+        ))
+    }
+}
+
+#[derive(Debug)]
+pub enum ConfigError {
+    IO(IOError),
+    Yaml(serde_yaml::Error),
+    Custom(String),
+}
+
+impl ConfigV1 {
+    pub fn new(path: &str) -> Result<ConfigV1, ConfigError> {
+        let base = load_config(path)?;
+
+        Ok(ConfigV1 { base })
+    }
+}
+
+fn load_config(path: &str) -> Result<ParsedConfigV1, ConfigError> {
+    let mut contents = String::new();
+    let mut file = File::open(path)?;
+    file.read_to_string(&mut contents)?;
+
+    let base: BaseConfig = serde_yaml::from_str(&contents)?;
+
+    if base.version != 1 {
+        return Err(ConfigError::Custom(
+            "Unsupported config version".to_string(),
+        ));
+    }
+
+    let log_level = base.log.clone().unwrap_or_else(|| "info".to_string());
+    if !log_level.eq("disable") {
+        std::env::set_var("FOURTH_LOG", log_level.clone());
+        pretty_env_logger::init_custom_env("FOURTH_LOG");
+        debug!("Set log level to {}", log_level);
+    }
+
+    debug!("Config version {}", base.version);
+
+    let mut parsed_upstream: HashMap<String, Upstream> = HashMap::new();
+
+    parsed_upstream.insert("ban".to_string(), Upstream::Ban);
+    parsed_upstream.insert("echo".to_string(), Upstream::Echo);
+
+    for (name, upstream) in base.upstream.iter() {
+        let ups = upstream.as_str().try_into()?;
+        parsed_upstream.insert(name.to_string(), Upstream::Proxy(ups));
+    }
+
+    let parsed = ParsedConfigV1 {
+        version: base.version,
+        log: base.log,
+        servers: base.servers,
+        upstream: parsed_upstream,
+    };
+
+    verify_config(parsed)
+}
+
+fn verify_config(config: ParsedConfigV1) -> Result<ParsedConfigV1, ConfigError> {
+    let mut used_upstreams: HashSet<String> = HashSet::new();
+    let mut upstream_names: HashSet<String> = HashSet::new();
+    let mut listen_addresses: HashSet<String> = HashSet::new();
+
+    // Check for duplicate upstream names
+    for (name, _) in config.upstream.iter() {
+        if upstream_names.contains(name) {
+            return Err(ConfigError::Custom(format!(
+                "Duplicate upstream name {}",
+                name
+            )));
+        }
+
+        upstream_names.insert(name.to_string());
+    }
+
+    for (_, server) in config.servers.clone() {
+        // check for duplicate listen addresses
+        for listen in server.listen {
+            if listen_addresses.contains(&listen) {
+                return Err(ConfigError::Custom(format!(
+                    "Duplicate listen address {}",
+                    listen
+                )));
+            }
+
+            listen_addresses.insert(listen.to_string());
+        }
+
+        if server.tls.unwrap_or_default() && server.sni.is_some() {
+            for (_, val) in server.sni.unwrap() {
+                used_upstreams.insert(val.to_string());
+            }
+        }
+
+        if server.default.is_some() {
+            used_upstreams.insert(server.default.unwrap().to_string());
+        }
+
+        for key in &used_upstreams {
+            if !config.upstream.contains_key(key) {
+                return Err(ConfigError::Custom(format!("Upstream {} not found", key)));
+            }
+        }
+    }
+
+    for key in &upstream_names {
+        if !used_upstreams.contains(key) && !key.eq("echo") && !key.eq("ban") {
+            warn!("Upstream {} not used", key);
+        }
+    }
+
+    Ok(config)
+}
+
+impl From<IOError> for ConfigError {
+    fn from(err: IOError) -> ConfigError {
+        ConfigError::IO(err)
+    }
+}
+
+impl From<serde_yaml::Error> for ConfigError {
+    fn from(err: serde_yaml::Error) -> ConfigError {
+        ConfigError::Yaml(err)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_load_config() {
+        let config = ConfigV1::new("tests/config.yaml").unwrap();
+        assert_eq!(config.base.version, 1);
+        assert_eq!(config.base.log.unwrap(), "disable");
+        assert_eq!(config.base.servers.len(), 5);
+        assert_eq!(config.base.upstream.len(), 3 + 2); // Add ban and echo upstreams
+    }
+}

src/config/mod.rs

@@ -0,0 +1,3 @@
+mod config;
+pub(crate) use config::ConfigV1;
+pub(crate) use config::ParsedConfigV1;

src/main.rs

@@ -1,14 +1,27 @@
 mod config;
 mod plugins;
 mod servers;
+mod upstreams;
 
-use crate::config::Config;
+use crate::config::ConfigV1;
 use crate::servers::Server;
 
 use log::{debug, error};
+use std::path::PathBuf;
 
 fn main() {
-    let config = match Config::new("/etc/fourth/config.yaml") {
+    let config_path = match find_config() {
+        Ok(p) => p,
+        Err(paths) => {
+            println!("Could not find config file. Tried paths:");
+            for p in paths {
+                println!("- {}", p);
+            }
+            std::process::exit(1);
+        }
+    };
+
+    let config = match ConfigV1::new(&config_path) {
         Ok(config) => config,
         Err(e) => {
             println!("Could not load config: {:?}", e);
@@ -17,9 +30,35 @@ fn main() {
     };
     debug!("{:?}", config);
 
-    let mut server = Server::new(config.base);
+    let mut server = Server::new_from_v1_config(config.base);
     debug!("{:?}", server);
 
-    let res = server.run();
-    error!("Server returned an error: {:?}", res);
+    let _ = server.run();
+    error!("Server ended with errors");
+}
+
+fn find_config() -> Result<String, Vec<String>> {
+    let possible_paths = ["/etc/l4p", ""];
+    let possible_names = ["l4p.yaml", "config.yaml"];
+
+    let mut tried_paths = Vec::<String>::new();
+
+    for path in possible_paths
+        .iter()
+        .flat_map(|&path| {
+            possible_names
+                .iter()
+                .map(move |&file| PathBuf::new().join(path).join(file))
+        })
+        .collect::<Vec<PathBuf>>()
+    {
+        let path_str = path.to_string_lossy().to_string();
+        if path.exists() {
+            return Ok(path_str);
+        }
+
+        tried_paths.push(path_str);
+    }
+
+    Err(tried_paths)
 }

src/plugins/mod.rs

@@ -1 +1 @@
-pub mod kcp;
+//pub mod kcp;

src/servers/mod.rs

@@ -2,34 +2,36 @@ use log::{error, info};
 use std::collections::{HashMap, HashSet};
 use std::net::SocketAddr;
 use std::sync::Arc;
+
 use tokio::task::JoinHandle;
 
 mod protocol;
-use crate::config::BaseConfig;
-use protocol::{kcp, tcp};
+pub(crate) mod upstream_address;
+
+use crate::config::ParsedConfigV1;
+use crate::upstreams::Upstream;
+use protocol::tcp;
 
 #[derive(Debug)]
-pub struct Server {
+pub(crate) struct Server {
     pub proxies: Vec<Arc<Proxy>>,
-    pub config: BaseConfig,
 }
 
 #[derive(Debug, Clone)]
-pub struct Proxy {
+pub(crate) struct Proxy {
     pub name: String,
     pub listen: SocketAddr,
     pub protocol: String,
     pub tls: bool,
     pub sni: Option<HashMap<String, String>>,
-    pub default: String,
-    pub upstream: HashMap<String, String>,
+    pub default_action: String,
+    pub upstream: HashMap<String, Upstream>,
 }
 
 impl Server {
-    pub fn new(config: BaseConfig) -> Self {
+    pub fn new_from_v1_config(config: ParsedConfigV1) -> Self {
         let mut new_server = Server {
             proxies: Vec::new(),
-            config: config.clone(),
         };
 
         for (name, proxy) in config.servers.iter() {
@@ -53,13 +55,14 @@ impl Server {
                         continue;
                     }
                 };
+
                 let proxy = Proxy {
                     name: name.clone(),
                     listen: listen_addr,
                     protocol: protocol.clone(),
                     tls,
                     sni: sni.clone(),
-                    default: default.clone(),
+                    default_action: default.clone(),
                     upstream: upstream.clone(),
                 };
                 new_server.proxies.push(Arc::new(proxy));
@@ -81,12 +84,18 @@ impl Server {
             );
             let handle = tokio::spawn(async move {
                 match config.protocol.as_ref() {
-                    "tcp" => {
-                        let _ = tcp::proxy(config).await;
-                    }
-                    "kcp" => {
-                        let _ = kcp::proxy(config).await;
+                    "tcp" | "tcp4" | "tcp6" => {
+                        let res = tcp::proxy(config.clone()).await;
+                        if res.is_err() {
+                            error!("Failed to start {}: {}", config.name, res.err().unwrap());
+                        }
                     }
+                    // "kcp" => {
+                    //     let res = kcp::proxy(config.clone()).await;
+                    //     if res.is_err() {
+                    //         error!("Failed to start {}: {}", config.name, res.err().unwrap());
+                    //     }
+                    // }
                     _ => {
                         error!("Invalid protocol: {}", config.protocol)
                     }
@@ -103,13 +112,12 @@ impl Server {
 }
 
 #[cfg(test)]
-mod test {
-    use crate::plugins::kcp::{KcpConfig, KcpStream};
-    use std::net::SocketAddr;
+mod tests {
+    //use crate::plugins::kcp::{KcpConfig, KcpStream};
     use std::thread::{self, sleep};
     use std::time::Duration;
     use tokio::io::{AsyncReadExt, AsyncWriteExt};
-    use tokio::net::{TcpListener, TcpStream};
+    use tokio::net::TcpListener;
 
     use super::*;
 
@@ -117,19 +125,27 @@ mod test {
     async fn tcp_mock_server() {
         let server_addr: SocketAddr = "127.0.0.1:54599".parse().unwrap();
         let listener = TcpListener::bind(server_addr).await.unwrap();
-        let (mut stream, _) = listener.accept().await.unwrap();
-        let mut buf = [0u8; 1024];
-        let n = stream.read(&mut buf).await.unwrap();
-        if n > 0 {
-            stream.write(b"hello").await.unwrap();
+        loop {
+            let (mut stream, _) = listener.accept().await.unwrap();
+            let mut buf = [0u8; 2];
+            let mut n = stream.read(&mut buf).await.unwrap();
+            while n > 0 {
+                let _ = stream.write(b"hello").await.unwrap();
+                if buf.eq(b"by") {
+                    stream.shutdown().await.unwrap();
+                    break;
+                }
+                n = stream.read(&mut buf).await.unwrap();
+            }
+            stream.shutdown().await.unwrap();
         }
     }
 
     #[tokio::test]
-    async fn test_tcp_proxy() {
-        use crate::config::Config;
-        let config = Config::new("tests/config.yaml").unwrap();
-        let mut server = Server::new(config.base);
+    async fn test_proxy() {
+        use crate::config::ConfigV1;
+        let config = ConfigV1::new("tests/config.yaml").unwrap();
+        let mut server = Server::new_from_v1_config(config.base);
         thread::spawn(move || {
             tcp_mock_server();
         });
@@ -139,54 +155,48 @@ mod test {
         });
         sleep(Duration::from_secs(1)); // wait for server to start
 
-        // // test proxy
-        // let mut conn = TcpStream::connect("127.0.0.1:54500").await.unwrap();
+        // test TCP proxy
+        let mut conn = tokio::net::TcpStream::connect("127.0.0.1:54500")
+            .await
+            .unwrap();
+        let mut buf = [0u8; 5];
+        let _ = conn.write(b"hi").await.unwrap();
+        let _ = conn.read(&mut buf).await.unwrap();
+        assert_eq!(&buf, b"hello");
+        conn.shutdown().await.unwrap();
+
+        // test TCP echo
+        let mut conn = tokio::net::TcpStream::connect("127.0.0.1:54956")
+            .await
+            .unwrap();
+        let mut buf = [0u8; 1];
+        for i in 0..=10u8 {
+            let _ = conn.write(&[i]).await.unwrap();
+            let _ = conn.read(&mut buf).await.unwrap();
+            assert_eq!(&buf, &[i]);
+        }
+        conn.shutdown().await.unwrap();
+
+        // test KCP echo
+        // let kcp_config = KcpConfig::default();
+        // let server_addr: SocketAddr = "127.0.0.1:54959".parse().unwrap();
+        // let mut conn = KcpStream::connect(&kcp_config, server_addr).await.unwrap();
+        // let mut buf = [0u8; 1];
+        // for i in 0..=10u8 {
+        //     conn.write(&[i]).await.unwrap();
+        //     conn.read(&mut buf).await.unwrap();
+        //     assert_eq!(&buf, &[i]);
+        // }
+        // conn.shutdown().await.unwrap();
+        //
+        // // test KCP proxy and close mock server
+        // let kcp_config = KcpConfig::default();
+        // let server_addr: SocketAddr = "127.0.0.1:54958".parse().unwrap();
+        // let mut conn = KcpStream::connect(&kcp_config, server_addr).await.unwrap();
         // let mut buf = [0u8; 5];
-        // conn.write(b"hi").await.unwrap();
+        // conn.write(b"by").await.unwrap();
         // conn.read(&mut buf).await.unwrap();
         // assert_eq!(&buf, b"hello");
         // conn.shutdown().await.unwrap();
-
-        // test echo
-        let mut conn = TcpStream::connect("127.0.0.1:54956").await.unwrap();
-        let mut buf = [0u8; 1];
-        for i in 0..=10u8 {
-            conn.write(&[i]).await.unwrap();
-            conn.read(&mut buf).await.unwrap();
-            assert_eq!(&buf, &[i]);
-        }
-        conn.shutdown().await.unwrap();
-    }
-
-    #[tokio::test]
-    async fn test_kcp_proxy() {
-        use crate::config::Config;
-        let config = Config::new("tests/config.yaml").unwrap();
-        let mut server = Server::new(config.base);
-        thread::spawn(move || {
-            let _ = server.run();
-        });
-        sleep(Duration::from_secs(1)); // wait for server to start
-
-        // test proxy
-        let kcp_config = KcpConfig::default();
-        let server_addr: SocketAddr = "127.0.0.1:54958".parse().unwrap();
-        let mut conn = KcpStream::connect(&kcp_config, server_addr).await.unwrap();
-        let mut buf = [0u8; 5];
-        conn.write(b"hi").await.unwrap();
-        conn.read(&mut buf).await.unwrap();
-        assert_eq!(&buf, b"hello");
-
-        // test echo
-        let kcp_config = KcpConfig::default();
-        let server_addr: SocketAddr = "127.0.0.1:54959".parse().unwrap();
-        let mut conn = KcpStream::connect(&kcp_config, server_addr).await.unwrap();
-        let mut buf = [0u8; 1];
-        for i in 0..=10u8 {
-            conn.write(&[i]).await.unwrap();
-            conn.read(&mut buf).await.unwrap();
-            assert_eq!(&buf, &[i]);
-        }
-        conn.shutdown().await.unwrap();
     }
 }

src/servers/protocol/kcp.rs

@@ -1,5 +1,6 @@
+use crate::config::Upstream;
 use crate::plugins::kcp::{KcpConfig, KcpListener, KcpStream};
-use crate::servers::Proxy;
+use crate::servers::{copy, Proxy};
 use futures::future::try_join;
 use log::{debug, error, warn};
 use std::net::SocketAddr;
@@ -41,7 +42,7 @@ async fn accept(
 ) -> Result<(), Box<dyn std::error::Error>> {
     debug!("New connection from {:?}", peer);
 
-    let upstream_name = proxy.default.clone();
+    let upstream_name = proxy.default_action.clone();
 
     debug!("Upstream: {}", upstream_name);
 
@@ -52,49 +53,46 @@ async fn accept(
                 "No upstream named {:?} on server {:?}",
                 proxy.default, proxy.name
             );
-            return process(inbound, &proxy.default).await;
+            return process(inbound, proxy.upstream.get(&proxy.default_action).unwrap()).await;
+            // ToDo: Remove unwrap and check default option
         }
     };
     return process(inbound, upstream).await;
 }
 
-async fn process(mut inbound: KcpStream, upstream: &str) -> Result<(), Box<dyn std::error::Error>> {
-    if upstream == "ban" {
-        let _ = inbound.shutdown();
-        return Ok(());
-    } else if upstream == "echo" {
-        let (mut ri, mut wi) = io::split(inbound);
-        let inbound_to_inbound = copy(&mut ri, &mut wi);
-        let bytes_tx = inbound_to_inbound.await;
-        debug!("Bytes read: {:?}", bytes_tx);
-        return Ok(());
-    }
+async fn process(
+    mut inbound: KcpStream,
+    upstream: &Upstream,
+) -> Result<(), Box<dyn std::error::Error>> {
+    match upstream {
+        Upstream::Ban => {
+            let _ = inbound.shutdown();
+        }
+        Upstream::Echo => {
+            let (mut ri, mut wi) = io::split(inbound);
+            let inbound_to_inbound = copy(&mut ri, &mut wi);
+            let bytes_tx = inbound_to_inbound.await;
+            debug!("Bytes read: {:?}", bytes_tx);
+        }
+        Upstream::Custom(custom) => match custom.protocol.as_ref() {
+            "tcp" => {
+                let outbound = TcpStream::connect(custom.addr.clone()).await?;
 
-    let outbound = TcpStream::connect(upstream).await?;
+                let (mut ri, mut wi) = io::split(inbound);
+                let (mut ro, mut wo) = io::split(outbound);
 
-    let (mut ri, mut wi) = io::split(inbound);
-    let (mut ro, mut wo) = io::split(outbound);
+                let inbound_to_outbound = copy(&mut ri, &mut wo);
+                let outbound_to_inbound = copy(&mut ro, &mut wi);
 
-    let inbound_to_outbound = copy(&mut ri, &mut wo);
-    let outbound_to_inbound = copy(&mut ro, &mut wi);
-
-    let (bytes_tx, bytes_rx) = try_join(inbound_to_outbound, outbound_to_inbound).await?;
-
-    debug!("Bytes read: {:?} write: {:?}", bytes_tx, bytes_rx);
+                let (bytes_tx, bytes_rx) =
+                    try_join(inbound_to_outbound, outbound_to_inbound).await?;
 
+                debug!("Bytes read: {:?} write: {:?}", bytes_tx, bytes_rx);
+            }
+            _ => {
+                error!("Reached unknown protocol: {:?}", custom.protocol);
+            }
+        },
+    };
     Ok(())
 }
-
-async fn copy<'a, R, W>(reader: &'a mut R, writer: &'a mut W) -> io::Result<u64>
-where
-    R: AsyncRead + Unpin + ?Sized,
-    W: AsyncWrite + Unpin + ?Sized,
-{
-    match io::copy(reader, writer).await {
-        Ok(u64) => {
-            let _ = writer.shutdown().await;
-            Ok(u64)
-        }
-        Err(_) => Ok(0),
-    }
-}

src/servers/protocol/mod.rs

@@ -1,3 +1,3 @@
-pub mod kcp;
+//pub mod kcp;
 pub mod tcp;
 pub mod tls;

src/servers/protocol/tcp.rs

@@ -1,13 +1,11 @@
 use crate::servers::protocol::tls::get_sni;
 use crate::servers::Proxy;
-use futures::future::try_join;
-use log::{debug, error, warn};
+use log::{debug, error, info, warn};
+use std::error::Error;
 use std::sync::Arc;
-use tokio::io;
-use tokio::io::{AsyncRead, AsyncWrite, AsyncWriteExt};
 use tokio::net::{TcpListener, TcpStream};
 
-pub async fn proxy(config: Arc<Proxy>) -> Result<(), Box<dyn std::error::Error>> {
+pub(crate) async fn proxy(config: Arc<Proxy>) -> Result<(), Box<dyn Error>> {
     let listener = TcpListener::bind(config.listen).await?;
     let config = config.clone();
 
@@ -32,21 +30,21 @@ pub async fn proxy(config: Arc<Proxy>) -> Result<(), Box<dyn std::error::Error>>
     }
 }
 
-async fn accept(inbound: TcpStream, proxy: Arc<Proxy>) -> Result<(), Box<dyn std::error::Error>> {
-    debug!("New connection from {:?}", inbound.peer_addr()?);
+async fn accept(inbound: TcpStream, proxy: Arc<Proxy>) -> Result<(), Box<dyn Error>> {
+    info!("New connection from {:?}", inbound.peer_addr()?);
 
     let upstream_name = match proxy.tls {
-        false => proxy.default.clone(),
+        false => proxy.default_action.clone(),
         true => {
             let mut hello_buf = [0u8; 1024];
             inbound.peek(&mut hello_buf).await?;
             let snis = get_sni(&hello_buf);
             if snis.is_empty() {
-                proxy.default.clone()
+                proxy.default_action.clone()
             } else {
                 match proxy.sni.clone() {
                     Some(sni_map) => {
-                        let mut upstream = proxy.default.clone();
+                        let mut upstream = proxy.default_action.clone();
                         for sni in snis {
                             let m = sni_map.get(&sni);
                             if m.is_some() {
@@ -56,7 +54,7 @@ async fn accept(inbound: TcpStream, proxy: Arc<Proxy>) -> Result<(), Box<dyn std
                         }
                         upstream
                     }
-                    None => proxy.default.clone(),
+                    None => proxy.default_action.clone(),
                 }
             }
         }
@@ -69,51 +67,11 @@ async fn accept(inbound: TcpStream, proxy: Arc<Proxy>) -> Result<(), Box<dyn std
         None => {
             warn!(
                 "No upstream named {:?} on server {:?}",
-                proxy.default, proxy.name
+                proxy.default_action, proxy.name
             );
-            return process(inbound, &proxy.default).await;
+            proxy.upstream.get(&proxy.default_action).unwrap()
         }
     };
-    return process(inbound, upstream).await;
-}
-
-async fn process(mut inbound: TcpStream, upstream: &str) -> Result<(), Box<dyn std::error::Error>> {
-    if upstream == "ban" {
-        let _ = inbound.shutdown();
-        return Ok(());
-    } else if upstream == "echo" {
-        let (mut ri, mut wi) = io::split(inbound);
-        let inbound_to_inbound = copy(&mut ri, &mut wi);
-        let bytes_tx = inbound_to_inbound.await;
-        debug!("Bytes read: {:?}", bytes_tx);
-        return Ok(());
-    }
-
-    let outbound = TcpStream::connect(upstream).await?;
-
-    let (mut ri, mut wi) = io::split(inbound);
-    let (mut ro, mut wo) = io::split(outbound);
-
-    let inbound_to_outbound = copy(&mut ri, &mut wo);
-    let outbound_to_inbound = copy(&mut ro, &mut wi);
-
-    let (bytes_tx, bytes_rx) = try_join(inbound_to_outbound, outbound_to_inbound).await?;
-
-    debug!("Bytes read: {:?} write: {:?}", bytes_tx, bytes_rx);
-
-    Ok(())
-}
-
-async fn copy<'a, R, W>(reader: &'a mut R, writer: &'a mut W) -> io::Result<u64>
-where
-    R: AsyncRead + Unpin + ?Sized,
-    W: AsyncWrite + Unpin + ?Sized,
-{
-    match io::copy(reader, writer).await {
-        Ok(u64) => {
-            let _ = writer.shutdown().await;
-            Ok(u64)
-        }
-        Err(_) => Ok(0),
-    }
+
+    upstream.process(inbound).await
 }

src/servers/protocol/tls.rs

@@ -49,6 +49,7 @@ pub fn get_sni(buf: &[u8]) -> Vec<String> {
         }
     }
 
+    debug!("Found SNIs: {:?}", &snis);
     snis
 }
 
@@ -98,6 +99,6 @@ mod tests {
             0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
         ];
         let sni = get_sni(&BUF);
-        assert!(sni[0] == "www.lirui.tech".to_string());
+        assert!(sni[0] == *"www.lirui.tech");
     }
 }

src/servers/upstream_address.rs

@@ -0,0 +1,141 @@
+use log::debug;
+use std::fmt::{Display, Formatter};
+use std::io::Result;
+use std::net::SocketAddr;
+use time::{Duration, Instant, OffsetDateTime};
+
+#[derive(Debug, Clone, Default)]
+pub(crate) struct UpstreamAddress {
+    address: String,
+    resolved_addresses: Vec<SocketAddr>,
+    resolved_time: Option<Instant>,
+    ttl: Option<Duration>,
+}
+
+impl Display for UpstreamAddress {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        self.address.fmt(f)
+    }
+}
+
+impl UpstreamAddress {
+    pub fn new(address: String) -> Self {
+        UpstreamAddress {
+            address,
+            ..Default::default()
+        }
+    }
+
+    pub fn is_valid(&self) -> bool {
+        if let Some(resolved) = self.resolved_time {
+            if let Some(ttl) = self.ttl {
+                return resolved.elapsed() < ttl;
+            }
+        }
+
+        false
+    }
+
+    fn is_resolved(&self) -> bool {
+        !self.resolved_addresses.is_empty()
+    }
+
+    fn time_remaining(&self) -> Duration {
+        if !self.is_valid() {
+            return Duration::seconds(0);
+        }
+
+        self.ttl.unwrap() - self.resolved_time.unwrap().elapsed()
+    }
+
+    pub async fn resolve(&mut self, mode: ResolutionMode) -> Result<Vec<SocketAddr>> {
+        if self.is_resolved() && self.is_valid() {
+            debug!(
+                "Already got address {:?}, still valid for {:.3}s",
+                &self.resolved_addresses,
+                self.time_remaining().as_seconds_f64()
+            );
+            return Ok(self.resolved_addresses.clone());
+        }
+
+        debug!(
+            "Resolving addresses for {} with mode {:?}",
+            &self.address, &mode
+        );
+
+        let lookup_result = tokio::net::lookup_host(&self.address).await;
+
+        let resolved_addresses: Vec<SocketAddr> = match lookup_result {
+            Ok(resolved_addresses) => resolved_addresses.into_iter().collect(),
+            Err(e) => {
+                debug!("Failed looking up {}: {}", &self.address, &e);
+                // Protect against DNS flooding. Cache the result for 1 second.
+                self.resolved_time = Some(Instant::now());
+                self.ttl = Some(Duration::seconds(3));
+                return Err(e);
+            }
+        };
+
+        debug!("Resolved addresses: {:?}", &resolved_addresses);
+
+        let addresses: Vec<SocketAddr> = match mode {
+            ResolutionMode::Ipv4 => resolved_addresses
+                .into_iter()
+                .filter(|a| a.is_ipv4())
+                .collect(),
+
+            ResolutionMode::Ipv6 => resolved_addresses
+                .into_iter()
+                .filter(|a| a.is_ipv6())
+                .collect(),
+
+            _ => resolved_addresses,
+        };
+
+        debug!(
+            "Got {} addresses for {}: {:?}",
+            &mode, &self.address, &addresses
+        );
+        debug!(
+            "Resolved at {}",
+            OffsetDateTime::now_utc()
+                .format(&time::format_description::well_known::Rfc3339)
+                .expect("Format")
+        );
+
+        self.resolved_addresses = addresses;
+        self.resolved_time = Some(Instant::now());
+        self.ttl = Some(Duration::minutes(1));
+
+        Ok(self.resolved_addresses.clone())
+    }
+}
+
+#[derive(Debug, Default, Clone)]
+pub(crate) enum ResolutionMode {
+    #[default]
+    Ipv4AndIpv6,
+    Ipv4,
+    Ipv6,
+}
+
+impl From<&str> for ResolutionMode {
+    fn from(value: &str) -> Self {
+        match value {
+            "tcp4" => ResolutionMode::Ipv4,
+            "tcp6" => ResolutionMode::Ipv6,
+            "tcp" => ResolutionMode::Ipv4AndIpv6,
+            _ => panic!("This should never happen. Please check configuration parser."),
+        }
+    }
+}
+
+impl Display for ResolutionMode {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        match self {
+            ResolutionMode::Ipv4 => write!(f, "IPv4Only"),
+            ResolutionMode::Ipv6 => write!(f, "IPv6Only"),
+            ResolutionMode::Ipv4AndIpv6 => write!(f, "IPv4 and IPv6"),
+        }
+    }
+}

src/upstreams/mod.rs

@@ -0,0 +1,51 @@
+mod proxy_to_upstream;
+
+use log::debug;
+use serde::Deserialize;
+use std::error::Error;
+use tokio::io;
+use tokio::io::{AsyncRead, AsyncWrite, AsyncWriteExt};
+use tokio::net::TcpStream;
+
+pub use crate::upstreams::proxy_to_upstream::ProxyToUpstream;
+
+#[derive(Debug, Clone, Deserialize)]
+pub enum Upstream {
+    Ban,
+    Echo,
+    Proxy(ProxyToUpstream),
+}
+
+impl Upstream {
+    pub(crate) async fn process(&self, mut inbound: TcpStream) -> Result<(), Box<dyn Error>> {
+        match self {
+            Upstream::Ban => {
+                inbound.shutdown().await?;
+            }
+            Upstream::Echo => {
+                let (mut ri, mut wi) = io::split(inbound);
+                let inbound_to_inbound = copy(&mut ri, &mut wi);
+                let bytes_tx = inbound_to_inbound.await;
+                debug!("Bytes read: {:?}", bytes_tx);
+            }
+            Upstream::Proxy(config) => {
+                config.proxy(inbound).await?;
+            }
+        };
+        Ok(())
+    }
+}
+
+async fn copy<'a, R, W>(reader: &'a mut R, writer: &'a mut W) -> io::Result<u64>
+where
+    R: AsyncRead + Unpin + ?Sized,
+    W: AsyncWrite + Unpin + ?Sized,
+{
+    match io::copy(reader, writer).await {
+        Ok(u64) => {
+            let _ = writer.shutdown().await;
+            Ok(u64)
+        }
+        Err(_) => Ok(0),
+    }
+}

src/upstreams/proxy_to_upstream.rs

@@ -0,0 +1,68 @@
+use crate::servers::upstream_address::UpstreamAddress;
+
+use crate::upstreams::copy;
+use futures::future::try_join;
+use log::{debug, error};
+use serde::Deserialize;
+use std::net::SocketAddr;
+use tokio::io;
+use tokio::net::TcpStream;
+use tokio::sync::Mutex;
+
+#[derive(Debug, Default)]
+struct Addr(Mutex<UpstreamAddress>);
+
+impl Clone for Addr {
+    fn clone(&self) -> Self {
+        tokio::task::block_in_place(|| Self(Mutex::new(self.0.blocking_lock().clone())))
+    }
+}
+
+#[derive(Debug, Clone, Deserialize, Default)]
+pub struct ProxyToUpstream {
+    pub addr: String,
+    pub protocol: String,
+    #[serde(skip_deserializing)]
+    addresses: Addr,
+}
+
+impl ProxyToUpstream {
+    pub async fn resolve_addresses(&self) -> std::io::Result<Vec<SocketAddr>> {
+        let mut addr = self.addresses.0.lock().await;
+        addr.resolve((*self.protocol).into()).await
+    }
+
+    pub fn new(address: String, protocol: String) -> Self {
+        Self {
+            addr: address.clone(),
+            protocol,
+            addresses: Addr(Mutex::new(UpstreamAddress::new(address))),
+        }
+    }
+
+    pub(crate) async fn proxy(&self, inbound: TcpStream) -> Result<(), Box<dyn std::error::Error>> {
+        let outbound = match self.protocol.as_ref() {
+            "tcp4" | "tcp6" | "tcp" => {
+                TcpStream::connect(self.resolve_addresses().await?.as_slice()).await?
+            }
+            _ => {
+                error!("Reached unknown protocol: {:?}", self.protocol);
+                return Err("Reached unknown protocol".into());
+            }
+        };
+
+        debug!("Connected to {:?}", outbound.peer_addr().unwrap());
+
+        let (mut ri, mut wi) = io::split(inbound);
+        let (mut ro, mut wo) = io::split(outbound);
+
+        let inbound_to_outbound = copy(&mut ri, &mut wo);
+        let outbound_to_inbound = copy(&mut ro, &mut wi);
+
+        let (bytes_tx, bytes_rx) = try_join(inbound_to_outbound, outbound_to_inbound).await?;
+
+        debug!("Bytes read: {:?} write: {:?}", bytes_tx, bytes_rx);
+
+        Ok(())
+    }
+}

tests/config.yaml

@@ -31,6 +31,6 @@ servers:
     default: echo
 
 upstream:
-  web: "127.0.0.1:8080"
-  proxy: "www.example.com:1024"
-  tester: "127.0.0.1:54599"
+  web: "tcp://127.0.0.1:8080"
+  proxy: "tcp://www.example.com:1024"
+  tester: "tcp://127.0.0.1:54599"