22 Commits

Author SHA1 Message Date
fd86162450 Version 0.1.6
All checks were successful
continuous-integration/drone/push Build is passing
continuous-integration/drone/tag Build is passing
Signed-off-by: Jacob Kiers <code@kiers.eu>
2023-08-23 20:52:46 +02:00
a6748f30d9 Make English readme the default
Some checks reported errors
continuous-integration/drone/pr Build was killed
continuous-integration/drone/push Build is passing
Since I'm unable to read Chinese

Signed-off-by: Jacob Kiers <code@kiers.eu>
2023-08-23 20:45:42 +02:00
902b2c0d55 Update build file
All checks were successful
continuous-integration/drone/push Build is passing
Signed-off-by: Jacob Kiers <code@kiers.eu>
2023-08-23 20:33:18 +02:00
fb7a7d9cae Update gitignore
Signed-off-by: Jacob Kiers <code@kiers.eu>
2023-08-23 20:33:07 +02:00
1c325f45b4 Add sample configuration file
Some checks failed
continuous-integration/drone Build is failing
Signed-off-by: Jacob Kiers <code@kiers.eu>
2023-08-23 19:29:49 +02:00
79c931fc38 Add build instructions
Signed-off-by: Jacob Kiers <code@kiers.eu>
2023-08-23 19:29:29 +02:00
915e39b684 Extract DNS address resolution
Signed-off-by: Jacob Kiers <code@kiers.eu>
2023-08-16 09:32:05 +02:00
0c5153bbd6 Rename Proxy::default to ::default_action
Signed-off-by: Jacob Kiers <code@kiers.eu>
2023-08-16 09:31:20 +02:00
01784ee3fd Update dependencies 2023-08-16 09:29:18 +02:00
f4bc441ca8 Enable explicit ipv4 / ipv6 proxying
Signed-off-by: Jacob Kiers <code@kiers.eu>
2023-06-02 17:35:29 +02:00
f010f8c76b Update dependencies
Signed-off-by: Jacob Kiers <code@kiers.eu>
2023-06-02 17:35:13 +02:00
8fbc0c370a Add error messages when failed to start server 2021-12-30 22:05:25 +08:00
bff92738d5 Allow config path from FOURTH_CONFIG 2021-11-01 16:06:47 +08:00
754a5af794 Add publish CI and run fmt 2021-11-01 15:56:57 +08:00
fc7a3038bd Add unknown protocol error 2021-11-01 15:32:08 +08:00
8a96de9666 Update README and minor refactor 2021-11-01 15:25:12 +08:00
0407f4b40c Add config validation 2021-11-01 13:45:47 +08:00
47be2568ba Add upstream scheme support
Need to implement TCP and UDP upstream support.
2021-10-31 19:21:32 +08:00
5944beb6a2 Combine TCP and KCP tests 2021-10-27 08:36:24 +08:00
4363e3f76a Publish 0.1.3 and update README 2021-10-26 23:58:00 +08:00
ee9d0685b3 Refactor TCP and KCP test 2021-10-26 23:52:07 +08:00
421ad8c979 Fix example config 2021-10-26 23:27:03 +08:00
21 changed files with 1152 additions and 520 deletions

3
.cargo/config.toml Normal file
View File

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

92
.drone.jsonnet Normal file
View File

@ -0,0 +1,92 @@
local executableName = 'fourth';
local build_image = 'img.kie.rs/jjkiers/rust-cross:rust1.70-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'],
}

View File

@ -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

1
.gitignore vendored
View File

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

687
Cargo.lock generated

File diff suppressed because it is too large Load Diff

View File

@ -1,6 +1,6 @@
[package]
name = "fourth"
version = "0.1.2"
version = "0.1.6"
edition = "2021"
authors = ["LI Rui <lr_cn@outlook.com>"]
license = "Apache-2.0"
@ -19,12 +19,12 @@ exclude = [".*"]
log = "0.4"
pretty_env_logger = "0.4"
serde = { version = "1.0", features = ["derive"] }
serde_yaml = "0.8"
serde_yaml = "0.9.21"
futures = "0.3"
tls-parser = "0.11"
url = "2.2.2"
time = { version = "0.3.1", features = ["local-offset", "formatting"] }
tokio = { version = "1.0", features = ["full"] }
bytes = "1.1"
kcp = "0.4"
byte_string = "1"

View File

@ -1,69 +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)
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"
other: "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.

80
README-ZH.md Normal file
View File

@ -0,0 +1,80 @@
# Fourth
> 这一波在第四层。
[![](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)
[English](/README-EN.md)
**积极开发中0.1版本迭代可能较快**
Fourth是一个Rust实现的Layer 4代理用于监听指定端口TCP/KCP流量并根据规则转发到指定目标目前只支持TCP
## 功能
- 监听指定端口代理到本地或远端指定端口
- 监听指定端口通过TLS ClientHello消息中的SNI进行分流
- 支持KCP入站警告未测试
## 安装方法
为了确保获得您架构下的最佳性能,请考虑自行编译,首选需要确保您拥有[Rust工具链](https://rustup.rs/)。
```bash
$ cd fourth
$ cargo build --release
```
将在`target/release/fourth`生成二进制文件,您也可以使用`cargo install --path . `来安装二进制文件。
或者您也可以使用Cargo直接安装
```bash
$ cargo install fourth
```
或者您也可以直接从Release中下载二进制文件。
## 配置
Fourth使用yaml格式的配置文件默认情况下会读取`/etc/fourth/config.yaml`,您也可以设置自定义路径到环境变量`FOURTH_CONFIG`,如下是一个最小有效配置:
```yaml
version: 1
log: info
servers:
proxy_server:
listen:
- "127.0.0.1:8081"
default: remote
upstream:
remote: "tcp://www.remote.example.com:8082" # proxy to remote address
```
内置两个的upstreamban立即中断连接、echo返回读到的数据。更详细的配置可以参考[示例配置](./example-config.yaml)。
注意:[::]会默认同时绑定IPv4和IPv6。
## 性能测试
在4C2G的服务器上测试
使用Fourth代理到Nginx直连QPS 120000: ~70000req/s (测试命令:`wrk -t200 -c1000 -d120s --latency http://proxy-server:8081 `
使用Fourth代理到本地iperf38Gbps
## io_uring?
尽管经过了很多尝试我们发现目前一些Rust下面的io_uring实现存在问题我们使用的io_uring库实现尽管在吞吐量上可以做到单线程20Gbps相比之下Tokio仅有8Gbps但在QPS上存在性能损失较大的问题。因此在有成熟的io_uring实现之前我们仍然选择epoll。之后我们会持续关注相关进展。
可能以后会为Linux高内核版本的用户提供可选的io_uring加速。
## 感谢
- [tokio_kcp](https://github.com/Matrix-Zhang/tokio_kcp)
## 协议
Fourth以Apache-2.0协议开源。

View File

@ -1,82 +1,70 @@
# 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)
[English](/README-EN.md)
**Under heavy development, version 0.1 may update frequently**
Fourth是一个Rust实现的Layer 4代理用于监听指定端口TCP/KCP流量并根据规则转发到指定目标目前只支持TCP
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
- 监听指定端口代理到本地或远端指定端口
- 监听指定端口通过TLS ClientHello消息中的SNI进行分流
- 支持KCP入站警告未测试
- Listen on specific port and proxy to local or remote port
- SNI-based rule without terminating TLS connection
- Allow KCP inbound(warning: untested)
## 安装方法
## Installation
为了确保获得您架构下的最佳性能,请考虑自行编译,首选需要确保您拥有[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
$ cargo build --release
```
将在`target/release/fourth`生成二进制文件,您也可以使用`cargo install --path . `来安装二进制文件。
Binary file will be generated at `target/release/fourth`, or you can use `cargo install --path .` to install.
或者您也可以使用Cargo直接安装
Or you can use Cargo to install Fourth:
```bash
$ cargo install fourth
```
## 配置
Or you can download binary file form the Release page.
Fourth使用yaml格式的配置文件默认情况下会读取`/etc/fourth/config.yaml`,如下是一个示例配置。
## Configuration
Fourth will read yaml format configuration file from `/etc/fourth/config.yaml`, and you can set custom path to environment variable `FOURTH_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"
other: "www.remote.example.com:8082" # proxy to remote address
remote: "tcp://www.remote.example.com:8082" # proxy to remote address
```
内置两个的upstreamban立即中断连接、echo返回读到的数据
Built-in two upstreams: ban(terminate connection immediately), echo. For detailed configuration, check [this example](./example-config.yaml).
## io_uring?
## Performance Benchmark
尽管经过了很多尝试我们发现目前一些Rust下面的io_uring实现存在问题我们使用的io_uring库实现尽管在吞吐量上可以做到单线程20Gbps相比之下Tokio仅有8Gbps但在QPS上存在性能损失较大的问题。因此在有成熟的io_uring实现之前我们仍然选择epoll。之后我们会持续关注相关进展。
Tested on 4C2G server:
可能以后会为Linux高内核版本的用户提供可选的io_uring加速。
Use fourth to proxy to Nginx(QPS of direct connection: ~120000): ~70000 req/s (Command: `wrk -t200 -c1000 -d120s --latency http://proxy-server:8081`)
## 感谢
Use fourth to proxy to local iperf3: 8Gbps
## Thanks
- [tokio_kcp](https://github.com/Matrix-Zhang/tokio_kcp)
## 协议
## License
FourthApache-2.0协议开源。
Fourth is available under terms of Apache-2.0.

16
config.yaml.example Normal file
View File

@ -0,0 +1,16 @@
version: 1
log: debug
servers:
example_server:
listen:
- "0.0.0.0:8443"
tls: true # Enable TLS features like SNI filtering
sni:
api.example.org: example-api
www.example.org: gh-proxy
default: ban
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

View File

@ -22,6 +22,6 @@ servers:
default: echo
upstream:
nginx: "127.0.0.1:8080"
proxy: "127.0.0.1:1024"
other: "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

View File

@ -1,12 +1,24 @@
use log::debug;
use crate::servers::upstream_address::UpstreamAddress;
use log::{debug, warn};
use serde::Deserialize;
use std::collections::HashMap;
use std::collections::{HashMap, HashSet};
use std::fs::File;
use std::io::{Error as IOError, Read};
use std::net::SocketAddr;
use tokio::sync::Mutex;
use url::Url;
#[derive(Debug, Clone)]
pub struct Config {
pub base: BaseConfig,
pub base: ParsedConfig,
}
#[derive(Debug, Default, Deserialize, Clone)]
pub struct ParsedConfig {
pub version: i32,
pub log: Option<String>,
pub servers: HashMap<String, ServerConfig>,
pub upstream: HashMap<String, Upstream>,
}
#[derive(Debug, Default, Deserialize, Clone)]
@ -26,6 +38,55 @@ pub struct ServerConfig {
pub default: Option<String>,
}
#[derive(Debug, Clone, Deserialize)]
pub enum Upstream {
Ban,
Echo,
Custom(CustomUpstream),
}
#[derive(Debug)]
struct Addr(Mutex<UpstreamAddress>);
impl Default for Addr {
fn default() -> Self {
Self(Default::default())
}
}
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)]
pub struct CustomUpstream {
pub name: String,
pub addr: String,
pub protocol: String,
#[serde(skip_deserializing)]
addresses: Addr,
}
impl CustomUpstream {
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
}
}
impl Default for CustomUpstream {
fn default() -> Self {
Self {
name: Default::default(),
addr: Default::default(),
protocol: Default::default(),
addresses: Default::default(),
}
}
}
#[derive(Debug)]
pub enum ConfigError {
IO(IOError),
@ -41,29 +102,150 @@ impl Config {
}
}
fn load_config(path: &str) -> Result<BaseConfig, ConfigError> {
fn load_config(path: &str) -> Result<ParsedConfig, 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)?;
let base: BaseConfig = serde_yaml::from_str(&contents)?;
if parsed.version != 1 {
if base.version != 1 {
return Err(ConfigError::Custom(
"Unsupported config version".to_string(),
));
}
let log_level = parsed.log.clone().unwrap_or_else(|| "info".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 {}", parsed.version);
debug!("Config version {}", base.version);
Ok(parsed)
let mut parsed_upstream: HashMap<String, Upstream> = HashMap::new();
for (name, upstream) in base.upstream.iter() {
let upstream_url = match Url::parse(upstream) {
Ok(url) => url,
Err(_) => {
return Err(ConfigError::Custom(format!(
"Invalid upstream url {}",
upstream
)))
}
};
let upstream_host = match upstream_url.host_str() {
Some(host) => host,
None => {
return Err(ConfigError::Custom(format!(
"Invalid upstream url {}",
upstream
)))
}
};
let upsteam_port = match upstream_url.port_or_known_default() {
Some(port) => port,
None => {
return Err(ConfigError::Custom(format!(
"Invalid upstream url {}",
upstream
)))
}
};
match upstream_url.scheme() {
"tcp" | "tcp4" | "tcp6" => {}
_ => {
return Err(ConfigError::Custom(format!(
"Invalid upstream scheme {}",
upstream
)))
}
}
parsed_upstream.insert(
name.to_string(),
Upstream::Custom(CustomUpstream {
name: name.to_string(),
addr: format!("{}:{}", upstream_host, upsteam_port),
protocol: upstream_url.scheme().to_string(),
..Default::default()
}),
);
}
parsed_upstream.insert("ban".to_string(), Upstream::Ban);
parsed_upstream.insert("echo".to_string(), Upstream::Echo);
let parsed = ParsedConfig {
version: base.version,
log: base.log,
servers: base.servers,
upstream: parsed_upstream,
};
verify_config(parsed)
}
fn verify_config(config: ParsedConfig) -> Result<ParsedConfig, 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 {
@ -87,7 +269,7 @@ mod tests {
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(), 4);
assert_eq!(config.base.upstream.len(), 3);
assert_eq!(config.base.servers.len(), 5);
assert_eq!(config.base.upstream.len(), 3 + 2); // Add ban and echo upstreams
}
}

View File

@ -6,9 +6,13 @@ use crate::config::Config;
use crate::servers::Server;
use log::{debug, error};
use std::env;
fn main() {
let config = match Config::new("/etc/fourth/config.yaml") {
let config_path =
env::var("FOURTH_CONFIG").unwrap_or_else(|_| "/etc/fourth/config.yaml".to_string());
let config = match Config::new(&config_path) {
Ok(config) => config,
Err(e) => {
println!("Could not load config: {:?}", e);
@ -20,6 +24,6 @@ fn main() {
let mut server = Server::new(config.base);
debug!("{:?}", server);
let res = server.run();
error!("Server returned an error: {:?}", res);
let _ = server.run();
error!("Server ended with errors");
}

View File

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

View File

@ -5,28 +5,30 @@ 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::{ParsedConfig, Upstream};
use protocol::tcp;
#[derive(Debug)]
pub struct Server {
pub(crate) struct Server {
pub proxies: Vec<Arc<Proxy>>,
pub config: BaseConfig,
pub config: ParsedConfig,
}
#[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(config: ParsedConfig) -> Self {
let mut new_server = Server {
proxies: Vec::new(),
config: config.clone(),
@ -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));
@ -82,11 +85,29 @@ impl Server {
let handle = tokio::spawn(async move {
match config.protocol.as_ref() {
"tcp" => {
let _ = tcp::proxy(config).await;
let res = tcp::proxy(config.clone()).await;
if res.is_err() {
error!("Failed to start {}: {}", config.name, res.err().unwrap());
}
"kcp" => {
let _ = kcp::proxy(config).await;
}
"tcp4" => {
let res = tcp::proxy(config.clone()).await;
if res.is_err() {
error!("Failed to start {}: {}", config.name, res.err().unwrap());
}
}
"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 +124,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,13 +137,24 @@ 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();
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 {
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() {
async fn test_proxy() {
use crate::config::Config;
let config = Config::new("tests/config.yaml").unwrap();
let mut server = Server::new(config.base);
@ -135,44 +166,21 @@ mod test {
let _ = server.run();
});
sleep(Duration::from_secs(1)); // wait for server to start
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];
conn.write(b"hi").await.unwrap();
conn.read(&mut buf).await.unwrap();
assert_eq!(&buf, b"hello");
conn.shutdown().await.unwrap();
}
#[tokio::test]
async fn test_tcp_echo_server() {
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
let mut conn = TcpStream::connect("127.0.0.1:54956").await.unwrap();
let mut buf = [0u8; 1];
for i in 0..=255u8 {
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_echo_server() {
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
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();
// 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 {
conn.write(&[i]).await.unwrap();
@ -180,5 +188,27 @@ mod test {
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"by").await.unwrap();
// conn.read(&mut buf).await.unwrap();
// assert_eq!(&buf, b"hello");
// conn.shutdown().await.unwrap();
}
}

View File

@ -1,3 +1,4 @@
use crate::config::Upstream;
use crate::plugins::kcp::{KcpConfig, KcpListener, KcpStream};
use crate::servers::Proxy;
use futures::future::try_join;
@ -52,25 +53,30 @@ 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).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" {
async fn process(
mut inbound: KcpStream,
upstream: &Upstream,
) -> Result<(), Box<dyn std::error::Error>> {
match upstream {
Upstream::Ban => {
let _ = inbound.shutdown();
return Ok(());
} else if upstream == "echo" {
}
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?;
Upstream::Custom(custom) => match custom.protocol.as_ref() {
"tcp" => {
let outbound = TcpStream::connect(custom.addr.clone()).await?;
let (mut ri, mut wi) = io::split(inbound);
let (mut ro, mut wo) = io::split(outbound);
@ -78,10 +84,16 @@ async fn process(mut inbound: KcpStream, upstream: &str) -> Result<(), Box<dyn s
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?;
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(())
}

View File

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

View File

@ -1,13 +1,14 @@
use crate::config::Upstream;
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::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 std::error::Error>> {
let listener = TcpListener::bind(config.listen).await?;
let config = config.clone();
@ -33,20 +34,20 @@ 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()?);
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 +57,7 @@ async fn accept(inbound: TcpStream, proxy: Arc<Proxy>) -> Result<(), Box<dyn std
}
upstream
}
None => proxy.default.clone(),
None => proxy.default_action.clone(),
}
}
}
@ -69,27 +70,46 @@ 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;
return process(
inbound,
proxy.upstream.get(&proxy.default_action).unwrap().clone(),
)
.await;
// ToDo: Remove unwrap and check default option
}
};
return process(inbound, upstream).await;
return process(inbound, upstream.clone()).await;
}
async fn process(mut inbound: TcpStream, upstream: &str) -> Result<(), Box<dyn std::error::Error>> {
if upstream == "ban" {
async fn process(
mut inbound: TcpStream,
upstream: Upstream,
) -> Result<(), Box<dyn std::error::Error>> {
match upstream {
Upstream::Ban => {
let _ = inbound.shutdown();
return Ok(());
} else if upstream == "echo" {
}
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(());
}
Upstream::Custom(custom) => {
let outbound = match custom.protocol.as_ref() {
"tcp4" | "tcp6" | "tcp" => {
TcpStream::connect(custom.resolve_addresses().await?.as_slice()).await?
}
_ => {
error!("Reached unknown protocol: {:?}", custom.protocol);
return Err("Reached unknown protocol".into());
}
};
let outbound = TcpStream::connect(upstream).await?;
debug!("Connected to {:?}", outbound.peer_addr().unwrap());
let (mut ri, mut wi) = io::split(inbound);
let (mut ro, mut wo) = io::split(outbound);
@ -100,7 +120,8 @@ async fn process(mut inbound: TcpStream, upstream: &str) -> Result<(), Box<dyn s
let (bytes_tx, bytes_rx) = try_join(inbound_to_outbound, outbound_to_inbound).await?;
debug!("Bytes read: {:?} write: {:?}", bytes_tx, bytes_rx);
}
};
Ok(())
}

View File

@ -49,6 +49,7 @@ pub fn get_sni(buf: &[u8]) -> Vec<String> {
}
}
debug!("Found SNIs: {:?}", &snis);
snis
}

View File

@ -0,0 +1,115 @@
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 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.len() > 0
}
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 {}",
&self.resolved_addresses,
self.time_remaining()
);
return Ok(self.resolved_addresses.clone());
}
debug!("Resolving addresses for {}", &self.address);
let lookup_result = tokio::net::lookup_host(&self.address).await;
let resolved_addresses = match lookup_result {
Ok(resolved_addresses) => resolved_addresses,
Err(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);
}
};
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.collect(),
};
debug!("Got addresses for {}: {:?}", &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."),
}
}
}

View File

@ -11,21 +11,26 @@ servers:
proxy.test.com: proxy
www.test.com: web
default: ban
echo_server:
listen:
- "0.0.0.0:54956"
default: echo
tcp_server:
listen:
- "127.0.0.1:54500"
default: tester
tcp_echo_server:
listen:
- "0.0.0.0:54956"
default: echo
kcp_server:
protocol: kcp
listen:
- "127.0.0.1:54958"
default: tester
kcp_echo_server:
protocol: kcp
listen:
- "127.0.0.1:54959"
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"