init project
This commit is contained in:
commit
9f0ed421f9
2
.gitignore
vendored
Normal file
2
.gitignore
vendored
Normal file
@ -0,0 +1,2 @@
|
|||||||
|
/target
|
||||||
|
.idea/
|
||||||
1507
Cargo.lock
generated
Normal file
1507
Cargo.lock
generated
Normal file
File diff suppressed because it is too large
Load Diff
16
Cargo.toml
Normal file
16
Cargo.toml
Normal file
@ -0,0 +1,16 @@
|
|||||||
|
[package]
|
||||||
|
name = "relay_server"
|
||||||
|
version = "0.1.0"
|
||||||
|
edition = "2024"
|
||||||
|
|
||||||
|
[dependencies]
|
||||||
|
anyhow = "1"
|
||||||
|
quinn = { version = "0.11", default-features = true, features = ["runtime-tokio", "rustls"] }
|
||||||
|
rcgen = "0.13"
|
||||||
|
rustls = "0.23"
|
||||||
|
rustls-pemfile = "2"
|
||||||
|
serde = { version = "1", features = ["derive"] }
|
||||||
|
tokio = { version = "1", features = ["full"] }
|
||||||
|
toml = "0.8"
|
||||||
|
tracing = "0.1"
|
||||||
|
tracing-subscriber = { version = "0.3", features = ["env-filter", "fmt"] }
|
||||||
42
README.md
Normal file
42
README.md
Normal file
@ -0,0 +1,42 @@
|
|||||||
|
# relay_server
|
||||||
|
|
||||||
|
Rust QUIC implementation of the RelayKit TCP relay server.
|
||||||
|
|
||||||
|
Each QUIC connection replaces the old UDP peer. Each QUIC bidirectional stream
|
||||||
|
replaces the old `stream_id` and maps to one outbound TCP connection. QUIC
|
||||||
|
provides ordered stream delivery and connection encryption, so the old UDP frame
|
||||||
|
header encryption is intentionally not present.
|
||||||
|
|
||||||
|
## Run
|
||||||
|
|
||||||
|
```sh
|
||||||
|
cargo run -- config.toml.example
|
||||||
|
```
|
||||||
|
|
||||||
|
The server also reads the config path from `RELAY_SERVER_CONFIG`. If neither is
|
||||||
|
provided and `config.toml` does not exist, built-in defaults are used.
|
||||||
|
|
||||||
|
## Stream Protocol
|
||||||
|
|
||||||
|
The client opens a QUIC bidirectional stream and writes one open request using
|
||||||
|
the old binary payload format:
|
||||||
|
|
||||||
|
```text
|
||||||
|
host_len(2), host(host_len), port(2),
|
||||||
|
username_len(2), username(username_len),
|
||||||
|
password_len(2), password(password_len)
|
||||||
|
```
|
||||||
|
|
||||||
|
All integers are unsigned big-endian values. After authentication and a
|
||||||
|
successful TCP connect, the server writes:
|
||||||
|
|
||||||
|
```text
|
||||||
|
0
|
||||||
|
```
|
||||||
|
|
||||||
|
Then both sides relay raw TCP bytes over the QUIC stream. On failure, the server
|
||||||
|
writes:
|
||||||
|
|
||||||
|
```text
|
||||||
|
1, error_len(2), error(error_len)
|
||||||
|
```
|
||||||
14
config.toml.example
Normal file
14
config.toml.example
Normal file
@ -0,0 +1,14 @@
|
|||||||
|
listen = "0.0.0.0:1443"
|
||||||
|
idle_timeout_ms = 60000
|
||||||
|
connect_timeout_ms = 5000
|
||||||
|
max_connections = 1024
|
||||||
|
|
||||||
|
[[users]]
|
||||||
|
username = "admin"
|
||||||
|
password = "v7@Qm!2z#R8$pL4^xT?K"
|
||||||
|
|
||||||
|
[tls]
|
||||||
|
# Leave both paths unset for a generated development certificate.
|
||||||
|
self_signed = true
|
||||||
|
# cert_path = "certs/server.crt"
|
||||||
|
# key_path = "certs/server.key"
|
||||||
86
src/auth.rs
Normal file
86
src/auth.rs
Normal file
@ -0,0 +1,86 @@
|
|||||||
|
use std::{collections::HashMap, sync::Arc};
|
||||||
|
|
||||||
|
use crate::{config::UserConfig, protocol::OpenRequest};
|
||||||
|
|
||||||
|
#[derive(Clone, Debug, Default)]
|
||||||
|
pub struct Authenticator {
|
||||||
|
users: Arc<HashMap<Vec<u8>, Vec<u8>>>,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl Authenticator {
|
||||||
|
pub fn new(users: &[UserConfig]) -> Self {
|
||||||
|
let users = users
|
||||||
|
.iter()
|
||||||
|
.filter(|user| !user.username.is_empty())
|
||||||
|
.map(|user| {
|
||||||
|
(
|
||||||
|
user.username.as_bytes().to_vec(),
|
||||||
|
user.password.as_bytes().to_vec(),
|
||||||
|
)
|
||||||
|
})
|
||||||
|
.collect();
|
||||||
|
|
||||||
|
Self {
|
||||||
|
users: Arc::new(users),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn is_empty(&self) -> bool {
|
||||||
|
self.users.is_empty()
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn authenticate(&self, request: &OpenRequest) -> bool {
|
||||||
|
let Some(expected_password) = self.users.get(request.username.as_slice()) else {
|
||||||
|
return false;
|
||||||
|
};
|
||||||
|
|
||||||
|
secure_equal(request.password.as_slice(), expected_password.as_slice())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn secure_equal(left: &[u8], right: &[u8]) -> bool {
|
||||||
|
if left.len() != right.len() {
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
|
||||||
|
left.iter()
|
||||||
|
.zip(right)
|
||||||
|
.fold(0_u8, |diff, (left, right)| diff | (left ^ right))
|
||||||
|
== 0
|
||||||
|
}
|
||||||
|
|
||||||
|
#[cfg(test)]
|
||||||
|
mod tests {
|
||||||
|
use super::*;
|
||||||
|
|
||||||
|
fn auth() -> Authenticator {
|
||||||
|
Authenticator::new(&[UserConfig {
|
||||||
|
username: "admin".to_owned(),
|
||||||
|
password: "secret".to_owned(),
|
||||||
|
}])
|
||||||
|
}
|
||||||
|
|
||||||
|
fn request(username: &[u8], password: &[u8]) -> OpenRequest {
|
||||||
|
OpenRequest {
|
||||||
|
host: "example.com".to_owned(),
|
||||||
|
port: 443,
|
||||||
|
username: username.to_vec(),
|
||||||
|
password: password.to_vec(),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn authenticates_known_user() {
|
||||||
|
assert!(auth().authenticate(&request(b"admin", b"secret")));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn rejects_bad_password() {
|
||||||
|
assert!(!auth().authenticate(&request(b"admin", b"wrong")));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn rejects_unknown_user() {
|
||||||
|
assert!(!auth().authenticate(&request(b"guest", b"secret")));
|
||||||
|
}
|
||||||
|
}
|
||||||
120
src/config.rs
Normal file
120
src/config.rs
Normal file
@ -0,0 +1,120 @@
|
|||||||
|
use std::{
|
||||||
|
net::SocketAddr,
|
||||||
|
path::{Path, PathBuf},
|
||||||
|
time::Duration,
|
||||||
|
};
|
||||||
|
|
||||||
|
use anyhow::{Context, bail};
|
||||||
|
use serde::Deserialize;
|
||||||
|
|
||||||
|
#[derive(Clone, Debug, Deserialize)]
|
||||||
|
#[serde(default)]
|
||||||
|
pub struct Config {
|
||||||
|
pub listen: String,
|
||||||
|
pub idle_timeout_ms: u64,
|
||||||
|
pub connect_timeout_ms: u64,
|
||||||
|
pub max_connections: usize,
|
||||||
|
pub users: Vec<UserConfig>,
|
||||||
|
pub tls: TlsConfig,
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Clone, Debug, Deserialize)]
|
||||||
|
pub struct UserConfig {
|
||||||
|
pub username: String,
|
||||||
|
pub password: String,
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Clone, Debug, Deserialize)]
|
||||||
|
#[serde(default)]
|
||||||
|
pub struct TlsConfig {
|
||||||
|
pub cert_path: Option<PathBuf>,
|
||||||
|
pub key_path: Option<PathBuf>,
|
||||||
|
pub self_signed: bool,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl Default for Config {
|
||||||
|
fn default() -> Self {
|
||||||
|
Self {
|
||||||
|
listen: "0.0.0.0:1443".to_owned(),
|
||||||
|
idle_timeout_ms: 60_000,
|
||||||
|
connect_timeout_ms: 5_000,
|
||||||
|
max_connections: 1024,
|
||||||
|
users: Vec::new(),
|
||||||
|
tls: TlsConfig::default(),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl Default for TlsConfig {
|
||||||
|
fn default() -> Self {
|
||||||
|
Self {
|
||||||
|
cert_path: None,
|
||||||
|
key_path: None,
|
||||||
|
self_signed: true,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl Config {
|
||||||
|
pub fn load(path: Option<&Path>) -> anyhow::Result<Self> {
|
||||||
|
let Some(path) = path else {
|
||||||
|
return Ok(Self::default());
|
||||||
|
};
|
||||||
|
|
||||||
|
let content = std::fs::read_to_string(path)
|
||||||
|
.with_context(|| format!("failed to read {}", path.display()))?;
|
||||||
|
toml::from_str(&content).with_context(|| format!("failed to parse {}", path.display()))
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn listen_addr(&self) -> anyhow::Result<SocketAddr> {
|
||||||
|
self.listen
|
||||||
|
.parse()
|
||||||
|
.with_context(|| format!("invalid listen address {}", self.listen))
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn idle_timeout(&self) -> Duration {
|
||||||
|
Duration::from_millis(self.idle_timeout_ms)
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn connect_timeout(&self) -> Duration {
|
||||||
|
Duration::from_millis(self.connect_timeout_ms)
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn validate(&self) -> anyhow::Result<()> {
|
||||||
|
if self.max_connections == 0 {
|
||||||
|
bail!("max_connections must be greater than 0");
|
||||||
|
}
|
||||||
|
|
||||||
|
match (
|
||||||
|
&self.tls.cert_path,
|
||||||
|
&self.tls.key_path,
|
||||||
|
self.tls.self_signed,
|
||||||
|
) {
|
||||||
|
(Some(_), Some(_), _) => {}
|
||||||
|
(None, None, true) => {}
|
||||||
|
_ => bail!("tls cert_path and key_path must both be set, or self_signed must be true"),
|
||||||
|
}
|
||||||
|
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[cfg(test)]
|
||||||
|
mod tests {
|
||||||
|
use super::*;
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn parses_default_listen_addr() {
|
||||||
|
assert_eq!(Config::default().listen_addr().unwrap().port(), 1443);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn rejects_zero_connection_limit() {
|
||||||
|
let config = Config {
|
||||||
|
max_connections: 0,
|
||||||
|
..Config::default()
|
||||||
|
};
|
||||||
|
|
||||||
|
assert!(config.validate().is_err());
|
||||||
|
}
|
||||||
|
}
|
||||||
6
src/lib.rs
Normal file
6
src/lib.rs
Normal file
@ -0,0 +1,6 @@
|
|||||||
|
pub mod auth;
|
||||||
|
pub mod config;
|
||||||
|
pub mod protocol;
|
||||||
|
pub mod relay;
|
||||||
|
pub mod server;
|
||||||
|
pub mod tls;
|
||||||
27
src/main.rs
Normal file
27
src/main.rs
Normal file
@ -0,0 +1,27 @@
|
|||||||
|
use std::path::PathBuf;
|
||||||
|
|
||||||
|
use anyhow::Context;
|
||||||
|
use relay_server::{config::Config, server};
|
||||||
|
use tracing_subscriber::{EnvFilter, fmt};
|
||||||
|
|
||||||
|
#[tokio::main]
|
||||||
|
async fn main() -> anyhow::Result<()> {
|
||||||
|
init_logging();
|
||||||
|
|
||||||
|
let config_path = std::env::args_os()
|
||||||
|
.nth(1)
|
||||||
|
.map(PathBuf::from)
|
||||||
|
.or_else(|| std::env::var_os("RELAY_SERVER_CONFIG").map(PathBuf::from))
|
||||||
|
.or_else(|| {
|
||||||
|
let path = PathBuf::from("config.toml");
|
||||||
|
path.exists().then_some(path)
|
||||||
|
});
|
||||||
|
|
||||||
|
let config = Config::load(config_path.as_deref()).context("failed to load config")?;
|
||||||
|
server::run(config).await
|
||||||
|
}
|
||||||
|
|
||||||
|
fn init_logging() {
|
||||||
|
let filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info"));
|
||||||
|
fmt().with_env_filter(filter).init();
|
||||||
|
}
|
||||||
205
src/protocol.rs
Normal file
205
src/protocol.rs
Normal file
@ -0,0 +1,205 @@
|
|||||||
|
use std::{fmt, io, str};
|
||||||
|
|
||||||
|
use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
|
||||||
|
|
||||||
|
pub const RESPONSE_OPEN_ACK: u8 = 0;
|
||||||
|
pub const RESPONSE_ERROR: u8 = 1;
|
||||||
|
const MAX_ERROR_LEN: usize = u16::MAX as usize;
|
||||||
|
|
||||||
|
#[derive(Clone, Debug, Eq, PartialEq)]
|
||||||
|
pub struct OpenRequest {
|
||||||
|
pub host: String,
|
||||||
|
pub port: u16,
|
||||||
|
pub username: Vec<u8>,
|
||||||
|
pub password: Vec<u8>,
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Debug)]
|
||||||
|
pub enum ProtocolError {
|
||||||
|
Io(io::Error),
|
||||||
|
EmptyHost,
|
||||||
|
InvalidHost,
|
||||||
|
InvalidPort,
|
||||||
|
FieldTooLarge(&'static str),
|
||||||
|
}
|
||||||
|
|
||||||
|
impl fmt::Display for ProtocolError {
|
||||||
|
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
||||||
|
match self {
|
||||||
|
Self::Io(err) => write!(f, "io error: {err}"),
|
||||||
|
Self::EmptyHost => f.write_str("host is empty"),
|
||||||
|
Self::InvalidHost => f.write_str("host is not valid utf-8"),
|
||||||
|
Self::InvalidPort => f.write_str("port must be greater than 0"),
|
||||||
|
Self::FieldTooLarge(field) => write!(f, "{field} is too large"),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl std::error::Error for ProtocolError {
|
||||||
|
fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
|
||||||
|
match self {
|
||||||
|
Self::Io(err) => Some(err),
|
||||||
|
_ => None,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl From<io::Error> for ProtocolError {
|
||||||
|
fn from(err: io::Error) -> Self {
|
||||||
|
Self::Io(err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
pub async fn read_open_request<R>(reader: &mut R) -> Result<OpenRequest, ProtocolError>
|
||||||
|
where
|
||||||
|
R: AsyncRead + Unpin,
|
||||||
|
{
|
||||||
|
let host = read_string_field(reader, "host").await?;
|
||||||
|
if host.is_empty() {
|
||||||
|
return Err(ProtocolError::EmptyHost);
|
||||||
|
}
|
||||||
|
|
||||||
|
let port = reader.read_u16().await?;
|
||||||
|
if port == 0 {
|
||||||
|
return Err(ProtocolError::InvalidPort);
|
||||||
|
}
|
||||||
|
|
||||||
|
let username = read_bytes_field(reader, "username").await?;
|
||||||
|
let password = read_bytes_field(reader, "password").await?;
|
||||||
|
|
||||||
|
Ok(OpenRequest {
|
||||||
|
host,
|
||||||
|
port,
|
||||||
|
username,
|
||||||
|
password,
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn encode_open_request(
|
||||||
|
host: &str,
|
||||||
|
port: u16,
|
||||||
|
username: &[u8],
|
||||||
|
password: &[u8],
|
||||||
|
) -> Result<Vec<u8>, ProtocolError> {
|
||||||
|
if host.is_empty() {
|
||||||
|
return Err(ProtocolError::EmptyHost);
|
||||||
|
}
|
||||||
|
if port == 0 {
|
||||||
|
return Err(ProtocolError::InvalidPort);
|
||||||
|
}
|
||||||
|
if host.len() > u16::MAX as usize {
|
||||||
|
return Err(ProtocolError::FieldTooLarge("host"));
|
||||||
|
}
|
||||||
|
if username.len() > u16::MAX as usize {
|
||||||
|
return Err(ProtocolError::FieldTooLarge("username"));
|
||||||
|
}
|
||||||
|
if password.len() > u16::MAX as usize {
|
||||||
|
return Err(ProtocolError::FieldTooLarge("password"));
|
||||||
|
}
|
||||||
|
|
||||||
|
let mut out = Vec::with_capacity(2 + host.len() + 2 + 2 + username.len() + 2 + password.len());
|
||||||
|
write_u16(&mut out, host.len() as u16);
|
||||||
|
out.extend_from_slice(host.as_bytes());
|
||||||
|
write_u16(&mut out, port);
|
||||||
|
write_u16(&mut out, username.len() as u16);
|
||||||
|
out.extend_from_slice(username);
|
||||||
|
write_u16(&mut out, password.len() as u16);
|
||||||
|
out.extend_from_slice(password);
|
||||||
|
Ok(out)
|
||||||
|
}
|
||||||
|
|
||||||
|
pub async fn write_open_ack<W>(writer: &mut W) -> io::Result<()>
|
||||||
|
where
|
||||||
|
W: AsyncWrite + Unpin,
|
||||||
|
{
|
||||||
|
writer.write_all(&[RESPONSE_OPEN_ACK]).await?;
|
||||||
|
writer.flush().await
|
||||||
|
}
|
||||||
|
|
||||||
|
pub async fn write_open_error<W>(writer: &mut W, message: impl AsRef<[u8]>) -> io::Result<()>
|
||||||
|
where
|
||||||
|
W: AsyncWrite + Unpin,
|
||||||
|
{
|
||||||
|
let message = message.as_ref();
|
||||||
|
let message = if message.len() > MAX_ERROR_LEN {
|
||||||
|
&message[..MAX_ERROR_LEN]
|
||||||
|
} else {
|
||||||
|
message
|
||||||
|
};
|
||||||
|
|
||||||
|
writer.write_all(&[RESPONSE_ERROR]).await?;
|
||||||
|
writer.write_u16(message.len() as u16).await?;
|
||||||
|
writer.write_all(message).await?;
|
||||||
|
writer.flush().await
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn read_string_field<R>(reader: &mut R, field: &'static str) -> Result<String, ProtocolError>
|
||||||
|
where
|
||||||
|
R: AsyncRead + Unpin,
|
||||||
|
{
|
||||||
|
let bytes = read_bytes_field(reader, field).await?;
|
||||||
|
let value = str::from_utf8(&bytes).map_err(|_| ProtocolError::InvalidHost)?;
|
||||||
|
Ok(value.to_owned())
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn read_bytes_field<R>(reader: &mut R, _field: &'static str) -> Result<Vec<u8>, ProtocolError>
|
||||||
|
where
|
||||||
|
R: AsyncRead + Unpin,
|
||||||
|
{
|
||||||
|
let len = reader.read_u16().await? as usize;
|
||||||
|
let mut bytes = vec![0_u8; len];
|
||||||
|
reader.read_exact(&mut bytes).await?;
|
||||||
|
Ok(bytes)
|
||||||
|
}
|
||||||
|
|
||||||
|
fn write_u16(out: &mut Vec<u8>, value: u16) {
|
||||||
|
out.extend_from_slice(&value.to_be_bytes());
|
||||||
|
}
|
||||||
|
|
||||||
|
#[cfg(test)]
|
||||||
|
mod tests {
|
||||||
|
use tokio::io::AsyncReadExt;
|
||||||
|
|
||||||
|
use super::*;
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn decodes_old_open_request_payload() {
|
||||||
|
let payload = encode_open_request("example.com", 443, b"admin", b"secret").unwrap();
|
||||||
|
let mut reader = payload.as_slice();
|
||||||
|
|
||||||
|
let request = read_open_request(&mut reader).await.unwrap();
|
||||||
|
|
||||||
|
assert_eq!(request.host, "example.com");
|
||||||
|
assert_eq!(request.port, 443);
|
||||||
|
assert_eq!(request.username, b"admin");
|
||||||
|
assert_eq!(request.password, b"secret");
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn leaves_following_stream_bytes_unconsumed() {
|
||||||
|
let mut payload = encode_open_request("example.com", 443, b"admin", b"secret").unwrap();
|
||||||
|
payload.extend_from_slice(b"GET / HTTP/1.1\r\n\r\n");
|
||||||
|
let mut reader = payload.as_slice();
|
||||||
|
|
||||||
|
let request = read_open_request(&mut reader).await.unwrap();
|
||||||
|
let mut rest = Vec::new();
|
||||||
|
reader.read_to_end(&mut rest).await.unwrap();
|
||||||
|
|
||||||
|
assert_eq!(request.host, "example.com");
|
||||||
|
assert_eq!(rest, b"GET / HTTP/1.1\r\n\r\n");
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn rejects_zero_port() {
|
||||||
|
let payload = encode_open_request("example.com", 1, b"admin", b"secret").unwrap();
|
||||||
|
let mut payload = payload;
|
||||||
|
payload[13] = 0;
|
||||||
|
payload[14] = 0;
|
||||||
|
|
||||||
|
let err = read_open_request(&mut payload.as_slice())
|
||||||
|
.await
|
||||||
|
.unwrap_err();
|
||||||
|
|
||||||
|
assert!(matches!(err, ProtocolError::InvalidPort));
|
||||||
|
}
|
||||||
|
}
|
||||||
119
src/relay.rs
Normal file
119
src/relay.rs
Normal file
@ -0,0 +1,119 @@
|
|||||||
|
use std::{net::SocketAddr, time::Duration};
|
||||||
|
|
||||||
|
use anyhow::Context;
|
||||||
|
use quinn::{RecvStream, SendStream};
|
||||||
|
use tokio::{
|
||||||
|
io::{AsyncWriteExt, copy},
|
||||||
|
net::TcpStream,
|
||||||
|
time::timeout,
|
||||||
|
};
|
||||||
|
use tracing::{debug, warn};
|
||||||
|
|
||||||
|
use crate::{
|
||||||
|
auth::Authenticator,
|
||||||
|
protocol::{self, OpenRequest},
|
||||||
|
};
|
||||||
|
|
||||||
|
#[derive(Clone)]
|
||||||
|
pub struct RelayContext {
|
||||||
|
pub authenticator: Authenticator,
|
||||||
|
pub connect_timeout: Duration,
|
||||||
|
}
|
||||||
|
|
||||||
|
pub async fn handle_stream(
|
||||||
|
remote: SocketAddr,
|
||||||
|
mut send: SendStream,
|
||||||
|
mut recv: RecvStream,
|
||||||
|
context: RelayContext,
|
||||||
|
) {
|
||||||
|
let request = match protocol::read_open_request(&mut recv).await {
|
||||||
|
Ok(request) => request,
|
||||||
|
Err(err) => {
|
||||||
|
warn!(%remote, error = %err, "failed to decode open request");
|
||||||
|
let _ = protocol::write_open_error(&mut send, err.to_string()).await;
|
||||||
|
let _ = send.finish();
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
|
if !context.authenticator.authenticate(&request) {
|
||||||
|
warn!(%remote, host = %request.host, port = request.port, "authentication failed");
|
||||||
|
let _ = protocol::write_open_error(&mut send, "authentication failed").await;
|
||||||
|
let _ = send.finish();
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
match open_tcp(&request, context.connect_timeout).await {
|
||||||
|
Ok(tcp) => {
|
||||||
|
if let Err(err) = protocol::write_open_ack(&mut send).await {
|
||||||
|
warn!(%remote, error = %err, "failed to write open ack");
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
if let Err(err) = relay(remote, request, send, recv, tcp).await {
|
||||||
|
debug!(%remote, error = %err, "relay stream finished with error");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
Err(err) => {
|
||||||
|
warn!(
|
||||||
|
%remote,
|
||||||
|
host = %request.host,
|
||||||
|
port = request.port,
|
||||||
|
error = %err,
|
||||||
|
"failed to connect target"
|
||||||
|
);
|
||||||
|
let _ = protocol::write_open_error(&mut send, format!("connect failed: {err}")).await;
|
||||||
|
let _ = send.finish();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn open_tcp(request: &OpenRequest, connect_timeout: Duration) -> anyhow::Result<TcpStream> {
|
||||||
|
let addr = format!("{}:{}", request.host, request.port);
|
||||||
|
let tcp = timeout(connect_timeout, TcpStream::connect(&addr))
|
||||||
|
.await
|
||||||
|
.with_context(|| format!("connect timeout after {} ms", connect_timeout.as_millis()))?
|
||||||
|
.with_context(|| format!("connect {addr}"))?;
|
||||||
|
tcp.set_nodelay(true).context("set tcp nodelay")?;
|
||||||
|
Ok(tcp)
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn relay(
|
||||||
|
remote: SocketAddr,
|
||||||
|
request: OpenRequest,
|
||||||
|
mut send: SendStream,
|
||||||
|
mut recv: RecvStream,
|
||||||
|
tcp: TcpStream,
|
||||||
|
) -> anyhow::Result<()> {
|
||||||
|
let (mut tcp_read, mut tcp_write) = tcp.into_split();
|
||||||
|
|
||||||
|
let client_to_target = async {
|
||||||
|
let bytes = copy(&mut recv, &mut tcp_write)
|
||||||
|
.await
|
||||||
|
.context("copy client to target")?;
|
||||||
|
tcp_write
|
||||||
|
.shutdown()
|
||||||
|
.await
|
||||||
|
.context("shutdown target write half")?;
|
||||||
|
anyhow::Ok(bytes)
|
||||||
|
};
|
||||||
|
|
||||||
|
let target_to_client = async {
|
||||||
|
let bytes = copy(&mut tcp_read, &mut send)
|
||||||
|
.await
|
||||||
|
.context("copy target to client")?;
|
||||||
|
send.finish().context("finish quic send stream")?;
|
||||||
|
anyhow::Ok(bytes)
|
||||||
|
};
|
||||||
|
|
||||||
|
let (uploaded, downloaded) = tokio::try_join!(client_to_target, target_to_client)?;
|
||||||
|
debug!(
|
||||||
|
%remote,
|
||||||
|
host = %request.host,
|
||||||
|
port = request.port,
|
||||||
|
uploaded,
|
||||||
|
downloaded,
|
||||||
|
"relay stream closed"
|
||||||
|
);
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
90
src/server.rs
Normal file
90
src/server.rs
Normal file
@ -0,0 +1,90 @@
|
|||||||
|
use std::sync::Arc;
|
||||||
|
|
||||||
|
use anyhow::Context;
|
||||||
|
use tokio::sync::{OwnedSemaphorePermit, Semaphore};
|
||||||
|
use tracing::{debug, info, warn};
|
||||||
|
|
||||||
|
use crate::{
|
||||||
|
auth::Authenticator,
|
||||||
|
config::Config,
|
||||||
|
relay::{self, RelayContext},
|
||||||
|
tls,
|
||||||
|
};
|
||||||
|
|
||||||
|
pub async fn run(config: Config) -> anyhow::Result<()> {
|
||||||
|
config.validate()?;
|
||||||
|
let listen_addr = config.listen_addr()?;
|
||||||
|
|
||||||
|
let authenticator = Authenticator::new(&config.users);
|
||||||
|
if authenticator.is_empty() {
|
||||||
|
warn!("no users configured; every open request will be rejected");
|
||||||
|
}
|
||||||
|
|
||||||
|
let mut server_config = tls::build_server_config(&config.tls)?;
|
||||||
|
tls::apply_transport_config(&mut server_config, config.idle_timeout())?;
|
||||||
|
|
||||||
|
let endpoint = quinn::Endpoint::server(server_config, listen_addr)
|
||||||
|
.with_context(|| format!("failed to listen on {listen_addr}"))?;
|
||||||
|
|
||||||
|
info!(%listen_addr, "relay server listening with QUIC");
|
||||||
|
|
||||||
|
let limits = Arc::new(Semaphore::new(config.max_connections));
|
||||||
|
let context = RelayContext {
|
||||||
|
authenticator,
|
||||||
|
connect_timeout: config.connect_timeout(),
|
||||||
|
};
|
||||||
|
|
||||||
|
while let Some(incoming) = endpoint.accept().await {
|
||||||
|
let remote = incoming.remote_address();
|
||||||
|
let limits = Arc::clone(&limits);
|
||||||
|
let context = context.clone();
|
||||||
|
|
||||||
|
tokio::spawn(async move {
|
||||||
|
let Ok(permit) = limits.try_acquire_owned() else {
|
||||||
|
warn!(%remote, "connection rejected because max_connections is reached");
|
||||||
|
match incoming.await {
|
||||||
|
Ok(connection) => connection.close(0_u32.into(), b"server busy"),
|
||||||
|
Err(err) => debug!(%remote, error = %err, "rejected handshake failed"),
|
||||||
|
}
|
||||||
|
return;
|
||||||
|
};
|
||||||
|
|
||||||
|
match incoming.await {
|
||||||
|
Ok(connection) => handle_connection(connection, context, permit).await,
|
||||||
|
Err(err) => debug!(%remote, error = %err, "handshake failed"),
|
||||||
|
}
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn handle_connection(
|
||||||
|
connection: quinn::Connection,
|
||||||
|
context: RelayContext,
|
||||||
|
_permit: OwnedSemaphorePermit,
|
||||||
|
) {
|
||||||
|
let remote = connection.remote_address();
|
||||||
|
info!(%remote, "quic connection accepted");
|
||||||
|
|
||||||
|
loop {
|
||||||
|
match connection.accept_bi().await {
|
||||||
|
Ok((send, recv)) => {
|
||||||
|
let context = context.clone();
|
||||||
|
tokio::spawn(relay::handle_stream(remote, send, recv, context));
|
||||||
|
}
|
||||||
|
Err(quinn::ConnectionError::ApplicationClosed { .. }) => {
|
||||||
|
debug!(%remote, "quic connection closed by peer");
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
Err(quinn::ConnectionError::LocallyClosed) => {
|
||||||
|
debug!(%remote, "quic connection closed locally");
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
Err(err) => {
|
||||||
|
debug!(%remote, error = %err, "failed to accept stream");
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
59
src/tls.rs
Normal file
59
src/tls.rs
Normal file
@ -0,0 +1,59 @@
|
|||||||
|
use std::{fs::File, io::BufReader, sync::Arc};
|
||||||
|
|
||||||
|
use anyhow::{Context, bail};
|
||||||
|
use quinn::ServerConfig;
|
||||||
|
use rustls::pki_types::{CertificateDer, PrivateKeyDer, PrivatePkcs8KeyDer};
|
||||||
|
|
||||||
|
use crate::config::TlsConfig;
|
||||||
|
|
||||||
|
pub fn build_server_config(config: &TlsConfig) -> anyhow::Result<ServerConfig> {
|
||||||
|
let (certs, key) = match (&config.cert_path, &config.key_path) {
|
||||||
|
(Some(cert_path), Some(key_path)) => (
|
||||||
|
load_cert_chain(cert_path).context("failed to load certificate chain")?,
|
||||||
|
load_private_key(key_path).context("failed to load private key")?,
|
||||||
|
),
|
||||||
|
(None, None) if config.self_signed => generate_self_signed()?,
|
||||||
|
_ => bail!("tls cert_path and key_path must both be set, or self_signed must be true"),
|
||||||
|
};
|
||||||
|
|
||||||
|
ServerConfig::with_single_cert(certs, key).context("failed to build QUIC server config")
|
||||||
|
}
|
||||||
|
|
||||||
|
fn load_cert_chain(path: &std::path::Path) -> anyhow::Result<Vec<CertificateDer<'static>>> {
|
||||||
|
let mut reader = BufReader::new(
|
||||||
|
File::open(path).with_context(|| format!("failed to open {}", path.display()))?,
|
||||||
|
);
|
||||||
|
let certs = rustls_pemfile::certs(&mut reader).collect::<Result<Vec<_>, _>>()?;
|
||||||
|
if certs.is_empty() {
|
||||||
|
bail!("{} does not contain any certificates", path.display());
|
||||||
|
}
|
||||||
|
Ok(certs)
|
||||||
|
}
|
||||||
|
|
||||||
|
fn load_private_key(path: &std::path::Path) -> anyhow::Result<PrivateKeyDer<'static>> {
|
||||||
|
let mut reader = BufReader::new(
|
||||||
|
File::open(path).with_context(|| format!("failed to open {}", path.display()))?,
|
||||||
|
);
|
||||||
|
rustls_pemfile::private_key(&mut reader)?
|
||||||
|
.with_context(|| format!("{} does not contain a private key", path.display()))
|
||||||
|
}
|
||||||
|
|
||||||
|
fn generate_self_signed() -> anyhow::Result<(Vec<CertificateDer<'static>>, PrivateKeyDer<'static>)>
|
||||||
|
{
|
||||||
|
let certified = rcgen::generate_simple_self_signed(vec!["localhost".to_owned()])?;
|
||||||
|
let cert = certified.cert.der().clone();
|
||||||
|
let key = PrivatePkcs8KeyDer::from(certified.key_pair.serialize_der()).into();
|
||||||
|
Ok((vec![cert], key))
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn apply_transport_config(
|
||||||
|
server_config: &mut ServerConfig,
|
||||||
|
idle_timeout: std::time::Duration,
|
||||||
|
) -> anyhow::Result<()> {
|
||||||
|
let mut transport = quinn::TransportConfig::default();
|
||||||
|
transport.max_idle_timeout(Some(
|
||||||
|
quinn::IdleTimeout::try_from(idle_timeout).context("invalid idle timeout")?,
|
||||||
|
));
|
||||||
|
server_config.transport_config(Arc::new(transport));
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
Loading…
x
Reference in New Issue
Block a user