From 0a9174a696d4592f6f07d9f12608c2fc09e206ae Mon Sep 17 00:00:00 2001 From: Zhang Jingqiang Date: Fri, 24 May 2024 17:45:14 +0800 Subject: [PATCH] g3tiles: add s2n-tls proxy server --- .github/workflows/linux.yml | 36 ++- Cargo.lock | 50 ++- Cargo.toml | 3 + g3tiles/Cargo.toml | 4 + .../examples/benchmark/server.d/s2n-tls.yaml | 17 + g3tiles/src/config/server/mod.rs | 14 + .../src/config/server/s2n_tls_proxy/host.rs | 215 +++++++++++++ .../src/config/server/s2n_tls_proxy/mod.rs | 219 +++++++++++++ g3tiles/src/main.rs | 3 + g3tiles/src/serve/mod.rs | 2 + g3tiles/src/serve/ops.rs | 4 + g3tiles/src/serve/s2n_tls_proxy/host.rs | 151 +++++++++ g3tiles/src/serve/s2n_tls_proxy/mod.rs | 24 ++ g3tiles/src/serve/s2n_tls_proxy/server.rs | 294 ++++++++++++++++++ .../src/serve/s2n_tls_proxy/task/accept.rs | 234 ++++++++++++++ .../src/serve/s2n_tls_proxy/task/common.rs | 41 +++ g3tiles/src/serve/s2n_tls_proxy/task/mod.rs | 24 ++ g3tiles/src/serve/s2n_tls_proxy/task/relay.rs | 240 ++++++++++++++ lib/g3-types/src/net/tls/cert_pair.rs | 53 ++++ lib/g3-types/src/net/tls/mod.rs | 3 + lib/g3-yaml/src/value/mod.rs | 3 + lib/g3-yaml/src/value/s2n_tls.rs | 109 +++++++ 22 files changed, 1724 insertions(+), 19 deletions(-) create mode 100644 g3tiles/examples/benchmark/server.d/s2n-tls.yaml create mode 100644 g3tiles/src/config/server/s2n_tls_proxy/host.rs create mode 100644 g3tiles/src/config/server/s2n_tls_proxy/mod.rs create mode 100644 g3tiles/src/serve/s2n_tls_proxy/host.rs create mode 100644 g3tiles/src/serve/s2n_tls_proxy/mod.rs create mode 100644 g3tiles/src/serve/s2n_tls_proxy/server.rs create mode 100644 g3tiles/src/serve/s2n_tls_proxy/task/accept.rs create mode 100644 g3tiles/src/serve/s2n_tls_proxy/task/common.rs create mode 100644 g3tiles/src/serve/s2n_tls_proxy/task/mod.rs create mode 100644 g3tiles/src/serve/s2n_tls_proxy/task/relay.rs create mode 100644 lib/g3-types/src/net/tls/cert_pair.rs create mode 100644 lib/g3-yaml/src/value/s2n_tls.rs diff --git a/.github/workflows/linux.yml b/.github/workflows/linux.yml index 73c388e66..4a71058df 100644 --- a/.github/workflows/linux.yml +++ b/.github/workflows/linux.yml @@ -29,9 +29,9 @@ jobs: strategy: matrix: rust: - - stable - - beta - - nightly + - stable + - beta + - nightly steps: - name: Checkout sources uses: actions/checkout@v4 @@ -54,20 +54,20 @@ jobs: name: Clippy runs-on: ubuntu-latest steps: - - name: Checkout sources - uses: actions/checkout@v4 - with: - submodules: true - - name: Install stable toolchain - uses: dtolnay/rust-toolchain@stable - with: - components: clippy - - name: Install dependencies - run: | - sudo apt-get update - sudo apt-get install capnproto libc-ares-dev libssl-dev liblua5.4-dev - - name: Cargo clippy - run: cargo clippy --tests -- --deny warnings + - name: Checkout sources + uses: actions/checkout@v4 + with: + submodules: true + - name: Install stable toolchain + uses: dtolnay/rust-toolchain@stable + with: + components: clippy + - name: Install dependencies + run: | + sudo apt-get update + sudo apt-get install capnproto libc-ares-dev libssl-dev liblua5.4-dev + - name: Cargo clippy + run: cargo clippy --tests -- --deny warnings build-vendored: name: Build vendored @@ -89,6 +89,8 @@ jobs: include: - feature: vendored-c-ares component: g3proxy + - feature: s2n-tls + component: g3tiles steps: - name: Checkout sources uses: actions/checkout@v4 diff --git a/Cargo.lock b/Cargo.lock index ceb16eef7..a1ad6aa2a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -224,6 +224,7 @@ dependencies = [ "aws-lc-sys", "mirai-annotations", "paste", + "untrusted 0.7.1", "zeroize", ] @@ -2053,6 +2054,8 @@ dependencies = [ "rustc-hash 2.0.0", "rustls", "rustls-pki-types", + "s2n-tls", + "s2n-tls-tokio", "serde_json", "slog", "thiserror", @@ -3251,7 +3254,7 @@ dependencies = [ "getrandom", "libc", "spin", - "untrusted", + "untrusted 0.9.0", "windows-sys 0.52.0", ] @@ -3381,7 +3384,7 @@ dependencies = [ "aws-lc-rs", "ring", "rustls-pki-types", - "untrusted", + "untrusted 0.9.0", ] [[package]] @@ -3390,6 +3393,43 @@ version = "1.0.18" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f3cb5ba0dc43242ce17de99c180e96db90b235b8a9fdc9543c96d2209116bd9f" +[[package]] +name = "s2n-tls" +version = "0.3.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8511c46a4faf19ec94702457f9dc0270408661bd11e356bdf2aac4ee21cf411b" +dependencies = [ + "errno", + "hex", + "libc", + "pin-project-lite", + "s2n-tls-sys", +] + +[[package]] +name = "s2n-tls-sys" +version = "0.3.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6cf8a5d07b81f0f37d6e16d91812f94cafb0bf9f0e9866a59a383280a45abda8" +dependencies = [ + "aws-lc-rs", + "cc", + "libc", +] + +[[package]] +name = "s2n-tls-tokio" +version = "0.3.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3e43fccad80380810925d663fbe52f7b22c0b5e16471c0e692c0681eb9c94eda" +dependencies = [ + "errno", + "libc", + "pin-project-lite", + "s2n-tls", + "tokio", +] + [[package]] name = "schannel" version = "0.1.26" @@ -3863,6 +3903,12 @@ version = "0.1.14" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7dd6e30e90baa6f72411720665d41d89b9a3d039dc45b8faea1ddd07f617f6af" +[[package]] +name = "untrusted" +version = "0.7.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a156c684c91ea7d62626509bce3cb4e1d9ed5c4d978f7b4352658f96a4c26b4a" + [[package]] name = "untrusted" version = "0.9.0" diff --git a/Cargo.toml b/Cargo.toml index f36a4f284..525c08bea 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -159,6 +159,9 @@ openssl = { package = "variant-ssl", version = "0.15.5" } openssl-sys = { package = "variant-ssl-sys", version = "0.15.5" } openssl-probe = "0.1" # +s2n-tls = "0.3" +s2n-tls-tokio = "0.3" +# flume = { version = "0.11", default-features = false } # c-ares = { version = "11.0", default-features = false } diff --git a/g3tiles/Cargo.toml b/g3tiles/Cargo.toml index 2db570c3a..9deeb9469 100644 --- a/g3tiles/Cargo.toml +++ b/g3tiles/Cargo.toml @@ -32,6 +32,8 @@ openssl.workspace = true openssl-probe = { workspace = true, optional = true } rustls.workspace = true rustls-pki-types.workspace = true +s2n-tls = { workspace = true, optional = true } +s2n-tls-tokio = { workspace = true, optional = true } quinn = { workspace = true, optional = true, features = ["rustls"] } tokio-rustls.workspace = true governor = { workspace = true, features = ["std", "jitter"] } @@ -58,8 +60,10 @@ g3-build-env.workspace = true [features] default = ["quic"] quic = ["g3-daemon/quic", "g3-yaml/quinn", "g3-types/quinn", "dep:quinn"] +s2n-tls = ["dep:s2n-tls", "dep:s2n-tls-tokio", "vendored-aws-lc"] rustls-aws-lc = ["rustls/aws-lc-rs"] vendored-openssl = ["openssl/vendored", "openssl-probe"] vendored-tongsuo = ["openssl/tongsuo", "openssl-probe", "g3-yaml/tongsuo", "g3-types/tongsuo"] vendored-aws-lc = ["rustls-aws-lc", "openssl/aws-lc", "openssl-probe", "g3-types/aws-lc", "g3-openssl/aws-lc"] vendored-boringssl = ["openssl/boringssl", "openssl-probe", "g3-types/boringssl", "g3-openssl/boringssl"] + diff --git a/g3tiles/examples/benchmark/server.d/s2n-tls.yaml b/g3tiles/examples/benchmark/server.d/s2n-tls.yaml new file mode 100644 index 000000000..86108ed92 --- /dev/null +++ b/g3tiles/examples/benchmark/server.d/s2n-tls.yaml @@ -0,0 +1,17 @@ +--- + +name: s2n-tls +type: S2nTlsProxy +listen: + address: "[::]:9097" +listen_in_worker: true + +virtual_hosts: + name: bench + exact_match: bench.example.net + use_session_ticket: true + cert_pairs: + certificate: bench.example.net-ec256.crt + private_key: bench.example.net-ec256.key + backends: + - http diff --git a/g3tiles/src/config/server/mod.rs b/g3tiles/src/config/server/mod.rs index afcd64645..d06c551d7 100644 --- a/g3tiles/src/config/server/mod.rs +++ b/g3tiles/src/config/server/mod.rs @@ -36,6 +36,8 @@ pub(crate) mod plain_tcp_port; pub(crate) mod keyless_proxy; pub(crate) mod openssl_proxy; pub(crate) mod rustls_proxy; +#[cfg(feature = "s2n-tls")] +pub(crate) mod s2n_tls_proxy; mod registry; @@ -86,6 +88,8 @@ pub(crate) enum AnyServerConfig { PlainQuicPort(Box), OpensslProxy(openssl_proxy::OpensslProxyServerConfig), RustlsProxy(rustls_proxy::RustlsProxyServerConfig), + #[cfg(feature = "s2n-tls")] + S2nTlsProxy(s2n_tls_proxy::S2nTlsProxyServerConfig), KeylessProxy(keyless_proxy::KeylessProxyServerConfig), } @@ -99,6 +103,8 @@ macro_rules! impl_transparent0 { AnyServerConfig::PlainQuicPort(s) => s.$f(), AnyServerConfig::OpensslProxy(s) => s.$f(), AnyServerConfig::RustlsProxy(s) => s.$f(), + #[cfg(feature = "s2n-tls")] + AnyServerConfig::S2nTlsProxy(s) => s.$f(), AnyServerConfig::KeylessProxy(s) => s.$f(), } } @@ -115,6 +121,8 @@ macro_rules! impl_transparent1 { AnyServerConfig::PlainQuicPort(s) => s.$f(p), AnyServerConfig::OpensslProxy(s) => s.$f(p), AnyServerConfig::RustlsProxy(s) => s.$f(p), + #[cfg(feature = "s2n-tls")] + AnyServerConfig::S2nTlsProxy(s) => s.$f(p), AnyServerConfig::KeylessProxy(s) => s.$f(p), } } @@ -184,6 +192,12 @@ fn load_server( .context("failed to load this RustlsProxy server")?; Ok(AnyServerConfig::RustlsProxy(server)) } + #[cfg(feature = "s2n-tls")] + "s2n_tls_proxy" | "s2ntlsproxy" => { + let server = s2n_tls_proxy::S2nTlsProxyServerConfig::parse(map, position) + .context("failed to load this S2nTlsProxy server")?; + Ok(AnyServerConfig::S2nTlsProxy(server)) + } "keyless_proxy" | "keylessproxy" => { let server = keyless_proxy::KeylessProxyServerConfig::parse(map, position) .context("failed to load this KeylessProxy server")?; diff --git a/g3tiles/src/config/server/s2n_tls_proxy/host.rs b/g3tiles/src/config/server/s2n_tls_proxy/host.rs new file mode 100644 index 000000000..00f937307 --- /dev/null +++ b/g3tiles/src/config/server/s2n_tls_proxy/host.rs @@ -0,0 +1,215 @@ +/* + * Copyright 2024 ByteDance and/or its affiliates. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +use std::time::SystemTime; + +use anyhow::{anyhow, Context}; +use s2n_tls::enums::ClientAuthType; +use yaml_rust::Yaml; + +use g3_types::collection::NamedValue; +use g3_types::limit::RateLimitQuotaConfig; +use g3_types::metrics::MetricsName; +use g3_types::net::{TcpSockSpeedLimitConfig, UnparsedTlsCertPair}; +use g3_types::route::AlpnMatch; +use g3_yaml::{YamlDocPosition, YamlMapCallback}; + +#[derive(Clone, Debug, PartialEq)] +pub(crate) struct S2nTlsHostConfig { + name: String, + cert_pairs: Vec, + client_auth: bool, + client_auth_certs: Vec, + use_session_ticket: bool, + pub(crate) request_alive_max: Option, + pub(crate) request_rate_limit: Option, + pub(crate) tcp_sock_speed_limit: Option, + pub(crate) task_idle_max_count: Option, + pub(crate) backends: AlpnMatch, +} + +impl Default for S2nTlsHostConfig { + fn default() -> Self { + S2nTlsHostConfig { + name: String::new(), + cert_pairs: Vec::with_capacity(1), + client_auth: false, + client_auth_certs: Vec::new(), + use_session_ticket: false, + request_alive_max: None, + request_rate_limit: None, + tcp_sock_speed_limit: None, + task_idle_max_count: None, + backends: AlpnMatch::default(), + } + } +} + +impl NamedValue for S2nTlsHostConfig { + type Name = str; + type NameOwned = String; + + fn name(&self) -> &Self::Name { + self.name.as_str() + } + + fn name_owned(&self) -> Self::NameOwned { + self.name.clone() + } +} + +impl S2nTlsHostConfig { + pub(crate) fn build_tls_config(&self) -> anyhow::Result { + let mut builder = s2n_tls::config::Builder::new(); + builder + .set_security_policy(&s2n_tls::security::DEFAULT_TLS13) + .map_err(|e| anyhow!("failed to set security policy: {e}"))?; + + if self.client_auth { + builder + .set_client_auth_type(ClientAuthType::Required) + .map_err(|e| anyhow!("failed to enable client auth: {e}"))?; + if !self.client_auth_certs.is_empty() { + builder + .wipe_trust_store() + .map_err(|e| anyhow!("failed to wipe default trusted CA certs: {e}"))?; + } + for cert in &self.client_auth_certs { + builder + .trust_pem(cert.as_bytes()) + .map_err(|e| anyhow!("failed to add client auth CA cert: {e}"))?; + } + } else { + builder.with_system_certs(false).map_err(|e| { + anyhow!("failed to disable the load of system default ca certs: {e}") + })?; + } + + for (i, pair) in self.cert_pairs.iter().enumerate() { + builder + .load_pem(pair.cert_chain(), pair.private_key()) + .map_err(|e| anyhow!("failed to load cert and key pair {i}: {e}"))?; + } + + // TODO set session storage + if self.use_session_ticket { + // TODO rotate session ticket key + builder + .add_session_ticket_key(b"test", b"1234567890abcdef", SystemTime::now()) + .map_err(|e| anyhow!("failed to add session ticket key: {e}"))?; + } + + if !self.backends.is_empty() { + builder + .set_application_protocol_preference(self.backends.protocols().clone()) + .map_err(|e| anyhow!("failed to set ALPN list: {e}"))?; + } + + builder + .build() + .map_err(|e| anyhow!("failed to build tls config: {e}")) + } +} + +impl YamlMapCallback for S2nTlsHostConfig { + fn type_name(&self) -> &'static str { + "S2nTlsHostConfig" + } + + fn parse_kv( + &mut self, + key: &str, + value: &Yaml, + doc: Option<&YamlDocPosition>, + ) -> anyhow::Result<()> { + match g3_yaml::key::normalize(key).as_str() { + "name" => { + self.name = g3_yaml::value::as_string(value)?; + Ok(()) + } + "cert_pairs" => { + let lookup_dir = g3_daemon::config::get_lookup_dir(doc)?; + self.cert_pairs = g3_yaml::value::as_list(value, |v| { + g3_yaml::value::as_s2n_tls_certificate_pair(v, Some(lookup_dir)) + }) + .context(format!( + "invalid s2n tls cert pair list value for key {key}" + ))?; + Ok(()) + } + "enable_client_auth" => { + self.client_auth = g3_yaml::value::as_bool(value) + .context(format!("invalid value for key {key}"))?; + Ok(()) + } + "use_session_ticket" => { + self.use_session_ticket = g3_yaml::value::as_bool(value) + .context(format!("invalid value for key {key}"))?; + Ok(()) + } + "ca_certificate" | "ca_cert" | "client_auth_certificate" | "client_auth_cert" => { + let lookup_dir = g3_daemon::config::get_lookup_dir(doc)?; + let certs = g3_yaml::value::as_s2n_tls_certificates(value, Some(lookup_dir)) + .context(format!("invalid certificate(s) value for key {key}"))?; + self.client_auth_certs.extend(certs); + Ok(()) + } + "request_rate_limit" | "request_limit_quota" => { + let quota = g3_yaml::value::as_rate_limit_quota(value) + .context(format!("invalid request quota value for key {key}"))?; + self.request_rate_limit = Some(quota); + Ok(()) + } + "request_max_alive" | "request_alive_max" => { + let alive_max = g3_yaml::value::as_usize(value) + .context(format!("invalid usize value for key {key}"))?; + self.request_alive_max = Some(alive_max); + Ok(()) + } + "tcp_sock_speed_limit" | "tcp_conn_speed_limit" => { + let limit = g3_yaml::value::as_tcp_sock_speed_limit(value).context(format!( + "invalid tcp socket speed limit value for key {key}" + ))?; + self.tcp_sock_speed_limit = Some(limit); + Ok(()) + } + "task_idle_max_count" => { + let max_count = g3_yaml::value::as_i32(value) + .context(format!("invalid i32 value for key {key}"))?; + self.task_idle_max_count = Some(max_count); + Ok(()) + } + "backends" => { + self.backends = g3_yaml::value::as_alpn_matched_backends(value)?; + Ok(()) + } + _ => Err(anyhow!("invalid key {key}")), + } + } + + fn check(&mut self) -> anyhow::Result<()> { + if self.name.is_empty() { + return Err(anyhow!("no name set")); + } + if self.cert_pairs.is_empty() { + return Err(anyhow!("no certificate set")); + } + if self.backends.is_empty() { + return Err(anyhow!("no backend service set")); + } + Ok(()) + } +} diff --git a/g3tiles/src/config/server/s2n_tls_proxy/mod.rs b/g3tiles/src/config/server/s2n_tls_proxy/mod.rs new file mode 100644 index 000000000..e06783981 --- /dev/null +++ b/g3tiles/src/config/server/s2n_tls_proxy/mod.rs @@ -0,0 +1,219 @@ +/* + * Copyright 2024 ByteDance and/or its affiliates. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +use std::sync::Arc; +use std::time::Duration; + +use anyhow::{anyhow, Context}; +use ascii::AsciiString; +use yaml_rust::{yaml, Yaml}; + +use g3_io_ext::LimitedCopyConfig; +use g3_types::acl::AclNetworkRuleBuilder; +use g3_types::metrics::{MetricsName, StaticMetricsTags}; +use g3_types::net::{TcpListenConfig, TcpMiscSockOpts, TcpSockSpeedLimitConfig}; +use g3_types::route::HostMatch; +use g3_yaml::YamlDocPosition; + +use super::{ + AnyServerConfig, ServerConfig, ServerConfigDiffAction, IDLE_CHECK_DEFAULT_DURATION, + IDLE_CHECK_MAXIMUM_DURATION, +}; + +mod host; +pub(crate) use host::S2nTlsHostConfig; + +const SERVER_CONFIG_TYPE: &str = "S2nTlsProxy"; + +#[derive(Clone, Debug, PartialEq)] +pub(crate) struct S2nTlsProxyServerConfig { + name: MetricsName, + position: Option, + pub(crate) shared_logger: Option, + pub(crate) listen: TcpListenConfig, + pub(crate) listen_in_worker: bool, + pub(crate) ingress_net_filter: Option, + pub(crate) extra_metrics_tags: Option>, + pub(crate) accept_timeout: Duration, + pub(crate) hosts: HostMatch>, + pub(crate) tcp_sock_speed_limit: TcpSockSpeedLimitConfig, + pub(crate) task_idle_check_duration: Duration, + pub(crate) task_idle_max_count: i32, + pub(crate) tcp_copy: LimitedCopyConfig, + pub(crate) tcp_misc_opts: TcpMiscSockOpts, + pub(crate) spawn_task_unconstrained: bool, +} + +impl S2nTlsProxyServerConfig { + pub(crate) fn new(position: Option) -> Self { + S2nTlsProxyServerConfig { + name: MetricsName::default(), + position, + shared_logger: None, + listen: TcpListenConfig::default(), + listen_in_worker: false, + ingress_net_filter: None, + extra_metrics_tags: None, + accept_timeout: Duration::from_secs(10), + hosts: HostMatch::default(), + tcp_sock_speed_limit: TcpSockSpeedLimitConfig::default(), + task_idle_check_duration: IDLE_CHECK_DEFAULT_DURATION, + task_idle_max_count: 1, + tcp_copy: Default::default(), + tcp_misc_opts: Default::default(), + spawn_task_unconstrained: false, + } + } + + pub(super) fn parse( + map: &yaml::Hash, + position: Option, + ) -> anyhow::Result { + let mut server = S2nTlsProxyServerConfig::new(position); + + g3_yaml::foreach_kv(map, |k, v| server.set(k, v))?; + + server.check()?; + Ok(server) + } + + fn check(&mut self) -> anyhow::Result<()> { + if self.name.is_empty() { + return Err(anyhow!("name is not set")); + } + if self.hosts.is_empty() { + return Err(anyhow!("no host config set")); + } + if self.task_idle_check_duration > IDLE_CHECK_MAXIMUM_DURATION { + self.task_idle_check_duration = IDLE_CHECK_MAXIMUM_DURATION; + } + Ok(()) + } + + fn set(&mut self, k: &str, v: &Yaml) -> anyhow::Result<()> { + match g3_yaml::key::normalize(k).as_str() { + super::CONFIG_KEY_SERVER_TYPE => Ok(()), + super::CONFIG_KEY_SERVER_NAME => { + self.name = g3_yaml::value::as_metrics_name(v)?; + Ok(()) + } + "shared_logger" => { + let name = g3_yaml::value::as_ascii(v)?; + self.shared_logger = Some(name); + Ok(()) + } + "extra_metrics_tags" => { + let tags = g3_yaml::value::as_static_metrics_tags(v) + .context(format!("invalid static metrics tags value for key {k}"))?; + self.extra_metrics_tags = Some(Arc::new(tags)); + Ok(()) + } + "listen" => { + self.listen = g3_yaml::value::as_tcp_listen_config(v) + .context(format!("invalid tcp listen config value for key {k}"))?; + Ok(()) + } + "listen_in_worker" => { + self.listen_in_worker = g3_yaml::value::as_bool(v)?; + Ok(()) + } + "ingress_network_filter" | "ingress_net_filter" => { + let filter = g3_yaml::value::acl::as_ingress_network_rule_builder(v).context( + format!("invalid ingress network acl rule value for key {k}"), + )?; + self.ingress_net_filter = Some(filter); + Ok(()) + } + "accept_timeout" => { + let timeout = g3_yaml::humanize::as_duration(v) + .context(format!("invalid humanize duration value for key {k}"))?; + self.accept_timeout = timeout; + Ok(()) + } + "virtual_hosts" | "hosts" => { + self.hosts = g3_yaml::value::as_host_matched_obj(v, self.position.as_ref())?; + Ok(()) + } + "tcp_sock_speed_limit" | "tcp_conn_speed_limit" => { + self.tcp_sock_speed_limit = g3_yaml::value::as_tcp_sock_speed_limit(v) + .context(format!("invalid tcp socket speed limit value for key {k}"))?; + Ok(()) + } + "task_idle_check_duration" => { + self.task_idle_check_duration = g3_yaml::humanize::as_duration(v) + .context(format!("invalid humanize duration value for key {k}"))?; + Ok(()) + } + "task_idle_max_count" => { + self.task_idle_max_count = + g3_yaml::value::as_i32(v).context(format!("invalid i32 value for key {k}"))?; + Ok(()) + } + "tcp_copy_buffer_size" => { + let buffer_size = g3_yaml::humanize::as_usize(v) + .context(format!("invalid humanize usize value for key {k}"))?; + self.tcp_copy.set_buffer_size(buffer_size); + Ok(()) + } + "tcp_copy_yield_size" => { + let yield_size = g3_yaml::humanize::as_usize(v) + .context(format!("invalid humanize usize value for key {k}"))?; + self.tcp_copy.set_yield_size(yield_size); + Ok(()) + } + "tcp_misc_opts" => { + self.tcp_misc_opts = g3_yaml::value::as_tcp_misc_sock_opts(v) + .context(format!("invalid tcp misc sock opts value for key {k}"))?; + Ok(()) + } + "spawn_task_unconstrained" | "task_unconstrained" => { + self.spawn_task_unconstrained = g3_yaml::value::as_bool(v)?; + Ok(()) + } + _ => Err(anyhow!("invalid key {k}")), + } + } +} + +impl ServerConfig for S2nTlsProxyServerConfig { + fn name(&self) -> &MetricsName { + &self.name + } + + fn position(&self) -> Option { + self.position.clone() + } + + fn server_type(&self) -> &'static str { + SERVER_CONFIG_TYPE + } + + fn diff_action(&self, new: &AnyServerConfig) -> ServerConfigDiffAction { + let AnyServerConfig::S2nTlsProxy(new) = new else { + return ServerConfigDiffAction::SpawnNew; + }; + + if self.eq(new) { + return ServerConfigDiffAction::NoAction; + } + + if self.listen != new.listen { + return ServerConfigDiffAction::ReloadAndRespawn; + } + + ServerConfigDiffAction::ReloadOnlyConfig + } +} diff --git a/g3tiles/src/main.rs b/g3tiles/src/main.rs index 705ff6c53..151e17428 100644 --- a/g3tiles/src/main.rs +++ b/g3tiles/src/main.rs @@ -35,6 +35,9 @@ fn main() -> anyhow::Result<()> { .install_default() .unwrap(); + #[cfg(feature = "s2n-tls")] + s2n_tls::init::init(); + let Some(proc_args) = g3tiles::opts::parse_clap().context("failed to parse command line options")? else { diff --git a/g3tiles/src/serve/mod.rs b/g3tiles/src/serve/mod.rs index 4fddb2ad9..76e7ef478 100644 --- a/g3tiles/src/serve/mod.rs +++ b/g3tiles/src/serve/mod.rs @@ -44,6 +44,8 @@ mod plain_tcp_port; mod keyless_proxy; mod openssl_proxy; mod rustls_proxy; +#[cfg(feature = "s2n-tls")] +mod s2n_tls_proxy; mod ops; pub(crate) use ops::{ diff --git a/g3tiles/src/serve/ops.rs b/g3tiles/src/serve/ops.rs index eb2a9471f..925661251 100644 --- a/g3tiles/src/serve/ops.rs +++ b/g3tiles/src/serve/ops.rs @@ -37,6 +37,8 @@ use super::plain_tcp_port::PlainTcpPort; use super::keyless_proxy::KeylessProxyServer; use super::openssl_proxy::OpensslProxyServer; use super::rustls_proxy::RustlsProxyServer; +#[cfg(feature = "s2n-tls")] +use super::s2n_tls_proxy::S2nTlsProxyServer; static SERVER_OPS_LOCK: Mutex<()> = Mutex::const_new(()); @@ -227,6 +229,8 @@ fn spawn_new_unlocked(config: AnyServerConfig) -> anyhow::Result<()> { AnyServerConfig::PlainQuicPort(c) => PlainQuicPort::prepare_initial(*c)?, AnyServerConfig::OpensslProxy(c) => OpensslProxyServer::prepare_initial(c)?, AnyServerConfig::RustlsProxy(c) => RustlsProxyServer::prepare_initial(c)?, + #[cfg(feature = "s2n-tls")] + AnyServerConfig::S2nTlsProxy(c) => S2nTlsProxyServer::prepare_initial(c)?, AnyServerConfig::KeylessProxy(c) => KeylessProxyServer::prepare_initial(c)?, }; registry::add(name.clone(), server)?; diff --git a/g3tiles/src/serve/s2n_tls_proxy/host.rs b/g3tiles/src/serve/s2n_tls_proxy/host.rs new file mode 100644 index 000000000..badb65178 --- /dev/null +++ b/g3tiles/src/serve/s2n_tls_proxy/host.rs @@ -0,0 +1,151 @@ +/* + * Copyright 2024 ByteDance and/or its affiliates. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +use std::sync::Arc; + +use arc_swap::ArcSwap; +use governor::{clock::DefaultClock, state::InMemoryState, state::NotKeyed, RateLimiter}; + +use g3_types::collection::NamedValue; +use g3_types::limit::{GaugeSemaphore, GaugeSemaphorePermit}; +use g3_types::metrics::MetricsName; +use g3_types::route::AlpnMatch; + +use crate::backend::ArcBackend; +use crate::config::server::s2n_tls_proxy::S2nTlsHostConfig; + +pub(crate) struct S2nTlsHost { + pub(super) config: Arc, + pub(super) tls_config: s2n_tls::config::Config, + req_alive_sem: Option, + request_rate_limit: Option>>, + pub(crate) backends: Arc>>, +} + +impl S2nTlsHost { + pub(super) fn try_build(config: &Arc) -> anyhow::Result { + let tls_config = config.build_tls_config()?; + + let backends = config.backends.build(crate::backend::get_or_insert_default); + + let request_rate_limit = config + .request_rate_limit + .as_ref() + .map(|quota| Arc::new(RateLimiter::direct(quota.get_inner()))); + let req_alive_sem = config.request_alive_max.map(GaugeSemaphore::new); + + Ok(S2nTlsHost { + config: config.clone(), + tls_config, + req_alive_sem, + request_rate_limit, + backends: Arc::new(ArcSwap::new(Arc::new(backends))), + }) + } + + pub(super) fn new_for_reload(&self, config: Arc) -> anyhow::Result { + let tls_config = config.build_tls_config()?; + + let request_rate_limit = if let Some(quota) = &config.request_rate_limit { + if let Some(old_limiter) = &self.request_rate_limit { + if let Some(old_quota) = &self.config.request_rate_limit { + if quota.eq(old_quota) { + // always use the old rate limiter when possible + Some(Arc::clone(old_limiter)) + } else { + Some(Arc::new(RateLimiter::direct(quota.get_inner()))) + } + } else { + unreachable!() + } + } else { + Some(Arc::new(RateLimiter::direct(quota.get_inner()))) + } + } else { + None + }; + let req_alive_sem = if let Some(p) = &config.request_alive_max { + let sema = self + .req_alive_sem + .as_ref() + .map(|sema| sema.new_updated(*p)) + .unwrap_or_else(|| GaugeSemaphore::new(*p)); + Some(sema) + } else { + None + }; + + let new_host = S2nTlsHost { + config, + tls_config, + req_alive_sem, + request_rate_limit, + backends: self.backends.clone(), // use the old container + }; + new_host.update_backends(); // update backends using the new config + Ok(new_host) + } + + pub(super) fn check_rate_limit(&self) -> Result<(), ()> { + if let Some(limit) = &self.request_rate_limit { + if limit.check().is_err() { + // TODO add stats + return Err(()); + } + } + Ok(()) + } + + pub(super) fn acquire_request_semaphore(&self) -> Result, ()> { + self.req_alive_sem + .as_ref() + .map(|sem| sem.try_acquire().map_err(|_| {})) + .transpose() + } + + pub(super) fn get_backend(&self, protocol: &str) -> Option { + self.backends.load().get(protocol).cloned() + } + + pub(super) fn get_default_backend(&self) -> Option { + self.backends.load().get_default().cloned() + } + + pub(super) fn use_backend(&self, name: &MetricsName) -> bool { + self.config.backends.contains_value(name) + } + + pub(super) fn update_backends(&self) { + let backends = self + .config + .backends + .build(crate::backend::get_or_insert_default); + self.backends.store(Arc::new(backends)); + } +} + +impl NamedValue for S2nTlsHost { + type Name = str; + type NameOwned = String; + + fn name(&self) -> &Self::Name { + self.config.name() + } + + fn name_owned(&self) -> Self::NameOwned { + self.config.name_owned() + } +} diff --git a/g3tiles/src/serve/s2n_tls_proxy/mod.rs b/g3tiles/src/serve/s2n_tls_proxy/mod.rs new file mode 100644 index 000000000..14d0a7113 --- /dev/null +++ b/g3tiles/src/serve/s2n_tls_proxy/mod.rs @@ -0,0 +1,24 @@ +/* + * Copyright 2024 ByteDance and/or its affiliates. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +mod server; +pub(crate) use server::S2nTlsProxyServer; + +mod task; +use task::{CommonTaskContext, S2nTlsAcceptTask}; + +mod host; +use host::S2nTlsHost; diff --git a/g3tiles/src/serve/s2n_tls_proxy/server.rs b/g3tiles/src/serve/s2n_tls_proxy/server.rs new file mode 100644 index 000000000..8bdac08c8 --- /dev/null +++ b/g3tiles/src/serve/s2n_tls_proxy/server.rs @@ -0,0 +1,294 @@ +/* + * Copyright 2024 ByteDance and/or its affiliates. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +use std::net::SocketAddr; +use std::sync::Arc; + +use ahash::AHashMap; +use anyhow::{anyhow, Context}; +use async_trait::async_trait; +#[cfg(feature = "quic")] +use quinn::Connection; +use slog::Logger; +use tokio::net::TcpStream; +use tokio::sync::broadcast; + +use g3_daemon::listen::{AcceptQuicServer, AcceptTcpServer, ListenStats, ListenTcpRuntime}; +use g3_daemon::server::{BaseServer, ClientConnectionInfo, ServerReloadCommand}; +use g3_types::acl::{AclAction, AclNetworkRule}; +use g3_types::metrics::MetricsName; +use g3_types::route::HostMatch; + +use super::{CommonTaskContext, S2nTlsAcceptTask, S2nTlsHost}; +use crate::config::server::s2n_tls_proxy::S2nTlsProxyServerConfig; +use crate::config::server::{AnyServerConfig, ServerConfig}; +use crate::module::stream::StreamServerStats; +use crate::serve::{ + ArcServer, ArcServerStats, Server, ServerInternal, ServerQuitPolicy, ServerStats, WrapArcServer, +}; + +pub(crate) struct S2nTlsProxyServer { + config: Arc, + server_stats: Arc, + listen_stats: Arc, + ingress_net_filter: Option, + reload_sender: broadcast::Sender, + task_logger: Logger, + hosts: Arc>>, + tls_accept_config: s2n_tls::config::Config, + + quit_policy: Arc, + reload_version: usize, +} + +impl S2nTlsProxyServer { + fn new( + config: Arc, + server_stats: Arc, + listen_stats: Arc, + hosts: Arc>>, + version: usize, + ) -> anyhow::Result { + let reload_sender = crate::serve::new_reload_notify_channel(); + + let tls_accept_config = + S2nTlsAcceptTask::build_accept_config().context("failed to build tls accept config")?; + + let ingress_net_filter = config + .ingress_net_filter + .as_ref() + .map(|builder| builder.build()); + + let task_logger = config.get_task_logger(); + + // always update extra metrics tags + server_stats.set_extra_tags(config.extra_metrics_tags.clone()); + + Ok(S2nTlsProxyServer { + config, + server_stats, + listen_stats, + ingress_net_filter, + reload_sender, + task_logger, + hosts, + tls_accept_config, + quit_policy: Arc::new(ServerQuitPolicy::default()), + reload_version: version, + }) + } + + pub(crate) fn prepare_initial(config: S2nTlsProxyServerConfig) -> anyhow::Result { + let config = Arc::new(config); + let server_stats = Arc::new(StreamServerStats::new(config.name())); + let listen_stats = Arc::new(ListenStats::new(config.name())); + + let hosts = config.hosts.try_build_arc(S2nTlsHost::try_build)?; + + let server = + S2nTlsProxyServer::new(config, server_stats, listen_stats, Arc::new(hosts), 1)?; + Ok(Arc::new(server)) + } + + fn prepare_reload(&self, config: AnyServerConfig) -> anyhow::Result { + if let AnyServerConfig::S2nTlsProxy(config) = config { + let config = Arc::new(config); + let server_stats = Arc::clone(&self.server_stats); + let listen_stats = Arc::clone(&self.listen_stats); + + let old_hosts_map = self.hosts.get_all_values(); + let new_conf_map = config.hosts.get_all_values(); + let mut new_hosts_map = AHashMap::with_capacity(new_conf_map.len()); + for (name, conf) in new_conf_map { + let host = if let Some(old_host) = old_hosts_map.get(&name) { + old_host.new_for_reload(conf)? + } else { + S2nTlsHost::try_build(&conf)? + }; + new_hosts_map.insert(name, Arc::new(host)); + } + let hosts = config.hosts.build_from(new_hosts_map); + + S2nTlsProxyServer::new( + config, + server_stats, + listen_stats, + Arc::new(hosts), + self.reload_version + 1, + ) + } else { + Err(anyhow!( + "config type mismatch: expect {}, actual {}", + self.config.server_type(), + config.server_type() + )) + } + } + + fn drop_early(&self, client_addr: SocketAddr) -> bool { + if let Some(ingress_net_filter) = &self.ingress_net_filter { + let (_, action) = ingress_net_filter.check(client_addr.ip()); + match action { + AclAction::Permit | AclAction::PermitAndLog => {} + AclAction::Forbid | AclAction::ForbidAndLog => { + self.listen_stats.add_dropped(); + return true; + } + } + } + + // TODO add cps limit + + false + } + + async fn run_task(&self, stream: TcpStream, cc_info: ClientConnectionInfo) { + let ctx = CommonTaskContext { + server_config: Arc::clone(&self.config), + server_stats: Arc::clone(&self.server_stats), + server_quit_policy: Arc::clone(&self.quit_policy), + cc_info, + task_logger: self.task_logger.clone(), + }; + + if self.config.spawn_task_unconstrained { + tokio::task::unconstrained(S2nTlsAcceptTask::new(ctx).into_running( + stream, + &self.hosts, + self.tls_accept_config.clone(), + )) + .await + } else { + S2nTlsAcceptTask::new(ctx) + .into_running(stream, &self.hosts, self.tls_accept_config.clone()) + .await; + } + } +} + +impl ServerInternal for S2nTlsProxyServer { + fn _clone_config(&self) -> AnyServerConfig { + AnyServerConfig::S2nTlsProxy(self.config.as_ref().clone()) + } + + fn _update_config_in_place(&self, _flags: u64, _config: AnyServerConfig) -> anyhow::Result<()> { + Ok(()) + } + + fn _depend_on_server(&self, _name: &MetricsName) -> bool { + false + } + + fn _reload_config_notify_runtime(&self) { + let cmd = ServerReloadCommand::ReloadVersion(self.reload_version); + let _ = self.reload_sender.send(cmd); + } + + fn _update_next_servers_in_place(&self) {} + + fn _reload_with_old_notifier(&self, config: AnyServerConfig) -> anyhow::Result { + let mut server = self.prepare_reload(config)?; + server.reload_sender = self.reload_sender.clone(); + Ok(Arc::new(server)) + } + + fn _reload_with_new_notifier(&self, config: AnyServerConfig) -> anyhow::Result { + let server = self.prepare_reload(config)?; + Ok(Arc::new(server)) + } + + fn _start_runtime(&self, server: &ArcServer) -> anyhow::Result<()> { + let runtime = + ListenTcpRuntime::new(WrapArcServer(server.clone()), server.get_listen_stats()); + runtime + .run_all_instances( + &self.config.listen, + self.config.listen_in_worker, + &self.reload_sender, + ) + .map(|_| self.server_stats.set_online()) + } + + fn _abort_runtime(&self) { + let _ = self.reload_sender.send(ServerReloadCommand::QuitRuntime); + self.server_stats.set_offline(); + } +} + +impl BaseServer for S2nTlsProxyServer { + #[inline] + fn name(&self) -> &MetricsName { + self.config.name() + } + + #[inline] + fn server_type(&self) -> &'static str { + self.config.server_type() + } + + #[inline] + fn version(&self) -> usize { + self.reload_version + } +} + +#[async_trait] +impl AcceptTcpServer for S2nTlsProxyServer { + async fn run_tcp_task(&self, stream: TcpStream, cc_info: ClientConnectionInfo) { + let client_addr = cc_info.client_addr(); + self.server_stats.add_conn(client_addr); + if self.drop_early(client_addr) { + return; + } + + self.run_task(stream, cc_info).await + } +} + +#[async_trait] +impl AcceptQuicServer for S2nTlsProxyServer { + #[cfg(feature = "quic")] + async fn run_quic_task(&self, _connection: Connection, _cc_info: ClientConnectionInfo) {} +} + +#[async_trait] +impl Server for S2nTlsProxyServer { + fn get_server_stats(&self) -> Option { + Some(Arc::clone(&self.server_stats) as _) + } + + fn get_listen_stats(&self) -> Arc { + Arc::clone(&self.listen_stats) + } + + fn alive_count(&self) -> i32 { + self.server_stats.alive_count() + } + + #[inline] + fn quit_policy(&self) -> &Arc { + &self.quit_policy + } + + fn update_backend(&self, name: &MetricsName) { + let host_map = self.hosts.get_all_values(); + for host in host_map.values() { + if host.use_backend(name) { + host.update_backends(); + } + } + } +} diff --git a/g3tiles/src/serve/s2n_tls_proxy/task/accept.rs b/g3tiles/src/serve/s2n_tls_proxy/task/accept.rs new file mode 100644 index 000000000..ea0c88e14 --- /dev/null +++ b/g3tiles/src/serve/s2n_tls_proxy/task/accept.rs @@ -0,0 +1,234 @@ +/* + * Copyright 2024 ByteDance and/or its affiliates. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +use std::pin::Pin; +use std::str::FromStr; +use std::sync::Arc; + +use anyhow::anyhow; +use log::debug; +use s2n_tls::callbacks::{ClientHelloCallback, ConnectionFuture}; +use s2n_tls::config::Config; +use s2n_tls::connection::{Connection, ModifiedBuilder}; +use s2n_tls::error::Error; +use s2n_tls_tokio::{TlsAcceptor, TlsStream}; +use thiserror::Error; +use tokio::io::{AsyncRead, AsyncWrite, AsyncWriteExt}; +use tokio::net::TcpStream; +use tokio::time::Instant; + +use g3_daemon::stat::task::TcpStreamConnectionStats; +use g3_io_ext::LimitedStream; +use g3_types::limit::GaugeSemaphorePermit; +use g3_types::net::Host; +use g3_types::route::HostMatch; + +use super::{CommonTaskContext, S2nTlsRelayTask}; +use crate::module::stream::StreamAcceptTaskCltWrapperStats; +use crate::serve::s2n_tls_proxy::S2nTlsHost; + +pub(crate) struct S2nTlsAcceptTask { + ctx: CommonTaskContext, + alive_permit: Option, +} + +impl S2nTlsAcceptTask { + pub(crate) fn new(ctx: CommonTaskContext) -> Self { + S2nTlsAcceptTask { + ctx, + alive_permit: None, + } + } + + pub(crate) async fn into_running( + mut self, + stream: TcpStream, + hosts: &Arc>>, + accept_config: Config, + ) { + let time_accepted = Instant::now(); + + let pre_handshake_stats = Arc::new(TcpStreamConnectionStats::default()); + let wrapper_stats = + StreamAcceptTaskCltWrapperStats::new(&self.ctx.server_stats, &pre_handshake_stats); + + let limit_config = self.ctx.server_config.tcp_sock_speed_limit; + let stream = LimitedStream::local_limited( + stream, + limit_config.shift_millis, + limit_config.max_north, + limit_config.max_south, + Arc::new(wrapper_stats), + ); + + if let Some((mut tls_stream, host)) = self.handshake(stream, hosts, accept_config).await { + if tls_stream.as_ref().resumed() { + // Quick ACK is needed with session resumption + self.ctx.cc_info.tcp_sock_try_quick_ack(); + } + + let backend = if let Some(alpn) = tls_stream.as_ref().application_protocol() { + let protocol = unsafe { std::str::from_utf8_unchecked(alpn) }; + host.get_backend(protocol) + } else { + host.get_default_backend() + }; + let Some(backend) = backend else { + let _ = tls_stream.shutdown().await; + return; + }; + + S2nTlsRelayTask::new( + self.ctx, + host, + backend.clone(), + time_accepted.elapsed(), + pre_handshake_stats, + self.alive_permit, + ) + .into_running(tls_stream) + .await; + } + } + + async fn handshake( + &mut self, + stream: S, + hosts: &Arc>>, + accept_config: Config, + ) -> Option<(TlsStream, Arc)> + where + S: AsyncRead + AsyncWrite + Unpin, + { + let conn_builder = ModifiedBuilder::new(accept_config, |conn| { + conn.set_application_context(ClientHelloContext::new(hosts.clone())); + Ok(conn) + }); + let tls_acceptor = TlsAcceptor::new(conn_builder); + match tokio::time::timeout( + self.ctx.server_config.accept_timeout, + tls_acceptor.accept(stream), + ) + .await + { + Ok(Ok(mut tls_stream)) => { + let ctx = tls_stream + .as_mut() + .application_context_mut::()?; + self.alive_permit = ctx.sema.take(); + + let host = ctx.host.take()?; + Some((tls_stream, host)) + } + Ok(Err(e)) => { + debug!("failed to recv client hello: {e}"); + None + } + Err(_) => { + debug!("timeout to recv client hello"); + None + } + } + } + + pub(crate) fn build_accept_config() -> anyhow::Result { + let mut builder = Config::builder(); + builder + .set_security_policy(&s2n_tls::security::DEFAULT_TLS13) + .map_err(|e| anyhow!("failed to set security policy: {e}"))?; + builder + .with_system_certs(false) + .map_err(|e| anyhow!("failed to disable load of system certs: {e}"))?; + builder + .set_client_hello_callback(AcceptClientHelloCallback {}) + .map_err(|e| anyhow!("failed to set client hello callback: {e}"))?; + builder + .build() + .map_err(|e| anyhow!("failed to build accept tls config: {e}")) + } +} + +#[derive(Debug, Error)] +enum ClientHelloApplicationError { + #[error("rate limited")] + RateLimited, + #[error("max alive quota reached")] + AliveLimited, +} + +struct ClientHelloContext { + hosts: Arc>>, + sema: Option, + host: Option>, +} + +impl ClientHelloContext { + fn new(hosts: Arc>>) -> Self { + ClientHelloContext { + hosts, + sema: None, + host: None, + } + } +} + +struct AcceptClientHelloCallback {} + +impl ClientHelloCallback for AcceptClientHelloCallback { + fn on_client_hello( + &self, + connection: &mut Connection, + ) -> Result>>, Error> { + let ch = connection.client_hello()?; + let sni = ch.server_name()?; + + let Some(ctx) = connection.application_context_mut::() else { + return Ok(None); + }; + + let host = if !sni.is_empty() { + let server_name = std::str::from_utf8(sni.as_slice()).unwrap(); + let host = Host::from_str(server_name).unwrap(); + ctx.hosts.get(&host) + } else { + ctx.hosts.get_default() + }; + + match host { + Some(host) => { + ctx.host = Some(host.clone()); + if host.check_rate_limit().is_err() { + return Err(Error::application(Box::new( + ClientHelloApplicationError::RateLimited, + ))); + } + // we do not check request alive sema here + let Ok(sema) = host.acquire_request_semaphore() else { + return Err(Error::application(Box::new( + ClientHelloApplicationError::AliveLimited, + ))); + }; + ctx.sema = sema; + + let config = host.tls_config.clone(); + connection.set_config(config)?; + connection.server_name_extension_used(); + Ok(None) + } + None => Ok(None), + } + } +} diff --git a/g3tiles/src/serve/s2n_tls_proxy/task/common.rs b/g3tiles/src/serve/s2n_tls_proxy/task/common.rs new file mode 100644 index 000000000..45f3fba8e --- /dev/null +++ b/g3tiles/src/serve/s2n_tls_proxy/task/common.rs @@ -0,0 +1,41 @@ +/* + * Copyright 2023 ByteDance and/or its affiliates. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +use std::net::SocketAddr; +use std::sync::Arc; + +use slog::Logger; + +use g3_daemon::server::ClientConnectionInfo; + +use crate::config::server::s2n_tls_proxy::S2nTlsProxyServerConfig; +use crate::module::stream::StreamServerStats; +use crate::serve::ServerQuitPolicy; + +pub(crate) struct CommonTaskContext { + pub server_config: Arc, + pub server_stats: Arc, + pub server_quit_policy: Arc, + pub cc_info: ClientConnectionInfo, + pub task_logger: Logger, +} + +impl CommonTaskContext { + #[inline] + pub(super) fn client_addr(&self) -> SocketAddr { + self.cc_info.client_addr() + } +} diff --git a/g3tiles/src/serve/s2n_tls_proxy/task/mod.rs b/g3tiles/src/serve/s2n_tls_proxy/task/mod.rs new file mode 100644 index 000000000..a25e1136a --- /dev/null +++ b/g3tiles/src/serve/s2n_tls_proxy/task/mod.rs @@ -0,0 +1,24 @@ +/* + * Copyright 2023 ByteDance and/or its affiliates. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +mod common; +pub(super) use common::CommonTaskContext; + +mod accept; +pub(super) use accept::S2nTlsAcceptTask; + +mod relay; +use relay::S2nTlsRelayTask; diff --git a/g3tiles/src/serve/s2n_tls_proxy/task/relay.rs b/g3tiles/src/serve/s2n_tls_proxy/task/relay.rs new file mode 100644 index 000000000..89189d82c --- /dev/null +++ b/g3tiles/src/serve/s2n_tls_proxy/task/relay.rs @@ -0,0 +1,240 @@ +/* + * Copyright 2024 ByteDance and/or its affiliates. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + *ru + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +use std::sync::Arc; +use std::time::Duration; + +use log::debug; +use s2n_tls_tokio::TlsStream; +use tokio::io::{AsyncRead, AsyncWrite}; +use tokio::time::Instant; + +use g3_daemon::stat::task::{TcpStreamConnectionStats, TcpStreamTaskStats}; +use g3_io_ext::{LimitedCopy, LimitedCopyConfig, LimitedCopyError, LimitedStream}; +use g3_types::limit::GaugeSemaphorePermit; + +use super::CommonTaskContext; +use crate::backend::ArcBackend; +use crate::config::server::ServerConfig; +use crate::log::task::tcp_connect::TaskLogForTcpConnect; +use crate::module::stream::StreamRelayTaskCltWrapperStats; +use crate::serve::s2n_tls_proxy::S2nTlsHost; +use crate::serve::{ServerTaskError, ServerTaskNotes, ServerTaskResult, ServerTaskStage}; + +pub(crate) struct S2nTlsRelayTask { + ctx: CommonTaskContext, + host: Arc, + backend: ArcBackend, + task_notes: ServerTaskNotes, + task_stats: Arc, + alive_permit: Option, +} + +impl S2nTlsRelayTask { + pub(crate) fn new( + ctx: CommonTaskContext, + host: Arc, + backend: ArcBackend, + wait_time: Duration, + pre_handshake_stats: Arc, + alive_permit: Option, + ) -> Self { + let task_notes = ServerTaskNotes::new(ctx.cc_info.clone(), wait_time); + S2nTlsRelayTask { + ctx, + host, + backend, + task_notes, + task_stats: Arc::new(TcpStreamTaskStats::with_clt_stats( + pre_handshake_stats.as_ref().clone(), + )), + alive_permit, + } + } + + fn get_log_context(&self) -> TaskLogForTcpConnect { + TaskLogForTcpConnect { + task_notes: &self.task_notes, + client_rd_bytes: self.task_stats.clt.read.get_bytes(), + client_wr_bytes: self.task_stats.clt.write.get_bytes(), + remote_rd_bytes: self.task_stats.ups.read.get_bytes(), + remote_wr_bytes: self.task_stats.ups.write.get_bytes(), + } + } + + pub(crate) async fn into_running(mut self, tls_stream: TlsStream>) + where + S: AsyncRead + AsyncWrite + Unpin, + { + self.pre_start(); + if let Err(e) = self.run(tls_stream).await { + self.get_log_context().log(&self.ctx.task_logger, &e) + } + self.pre_stop(); + } + + fn pre_start(&self) { + debug!( + "S2nTlsProxy: new client from {} to {} server {}", + self.ctx.client_addr(), + self.ctx.server_config.server_type(), + self.ctx.server_config.name(), + ); + self.ctx.server_stats.add_task(); + self.ctx.server_stats.inc_alive_task(); + } + + fn pre_stop(&mut self) { + if let Some(permit) = self.alive_permit.take() { + drop(permit); + } + self.ctx.server_stats.dec_alive_task(); + } + + async fn run(&mut self, tls_stream: TlsStream>) -> ServerTaskResult<()> + where + S: AsyncRead + AsyncWrite + Unpin, + { + self.task_notes.stage = ServerTaskStage::Preparing; + + // set client side socket options + self.ctx + .cc_info + .tcp_sock_set_raw_opts(&self.ctx.server_config.tcp_misc_opts, true) + .map_err(|_| { + ServerTaskError::InternalServerError("failed to set client socket options") + })?; + + self.task_notes.stage = ServerTaskStage::Connecting; + + let (ups_r, ups_w) = self.backend.stream_connect(&self.task_notes).await?; + + self.task_notes.stage = ServerTaskStage::Connected; + + self.run_connected(tls_stream, ups_r, ups_w).await + } + + async fn run_connected( + &mut self, + tls_stream: TlsStream>, + ups_r: UR, + ups_w: UW, + ) -> ServerTaskResult<()> + where + S: AsyncRead + AsyncWrite + Unpin, + UR: AsyncRead + Unpin, + UW: AsyncWrite + Unpin, + { + self.task_notes.mark_relaying(); + self.relay(tls_stream, ups_r, ups_w).await + } + + async fn relay( + &mut self, + mut tls_stream: TlsStream>, + mut ups_r: UR, + mut ups_w: UW, + ) -> ServerTaskResult<()> + where + S: AsyncRead + AsyncWrite + Unpin, + UR: AsyncRead + Unpin, + UW: AsyncWrite + Unpin, + { + self.reset_clt_limit_and_stats(&mut tls_stream); + let (mut clt_r, mut clt_w) = tokio::io::split(tls_stream); + + let copy_config = LimitedCopyConfig::default(); + let mut clt_to_ups = LimitedCopy::new(&mut clt_r, &mut ups_w, ©_config); + let mut ups_to_clt = LimitedCopy::new(&mut ups_r, &mut clt_w, ©_config); + + let idle_duration = self.ctx.server_config.task_idle_check_duration; + let task_idle_max_count = self + .host + .config + .task_idle_max_count + .unwrap_or(self.ctx.server_config.task_idle_max_count); + let mut idle_interval = + tokio::time::interval_at(Instant::now() + idle_duration, idle_duration); + let mut idle_count = 0; + loop { + tokio::select! { + biased; + + r = &mut clt_to_ups => { + let _ = ups_to_clt.write_flush().await; + return match r { + Ok(_) => Err(ServerTaskError::ClosedByClient), + Err(LimitedCopyError::ReadFailed(e)) => Err(ServerTaskError::ClientTcpReadFailed(e)), + Err(LimitedCopyError::WriteFailed(e)) => Err(ServerTaskError::UpstreamWriteFailed(e)), + }; + } + r = &mut ups_to_clt => { + let _ = clt_to_ups.write_flush().await; + return match r { + Ok(_) => Err(ServerTaskError::ClosedByUpstream), + Err(LimitedCopyError::ReadFailed(e)) => Err(ServerTaskError::UpstreamReadFailed(e)), + Err(LimitedCopyError::WriteFailed(e)) => Err(ServerTaskError::ClientTcpWriteFailed(e)), + }; + } + _ = idle_interval.tick() => { + if clt_to_ups.is_idle() && ups_to_clt.is_idle() { + idle_count += 1; + + if idle_count >= task_idle_max_count { + return Err(ServerTaskError::Idle(idle_duration, idle_count)); + } + } else { + idle_count = 0; + + clt_to_ups.reset_active(); + ups_to_clt.reset_active(); + } + + if self.ctx.server_quit_policy.force_quit() { + return Err(ServerTaskError::CanceledAsServerQuit) + } + } + }; + } + } + + fn reset_clt_limit_and_stats(&self, tls_stream: &mut TlsStream>) + where + S: AsyncRead + AsyncWrite + Unpin, + { + // reset io limit + if let Some(limit) = &self.host.config.tcp_sock_speed_limit { + let limit = self + .ctx + .server_config + .tcp_sock_speed_limit + .shrink_as_smaller(limit); + tls_stream.get_mut().reset_local_limit( + limit.shift_millis, + limit.max_north, + limit.max_south, + ); + } + + // reset io stats + // TODO add host level stats + let clt_wrapper_stats = + StreamRelayTaskCltWrapperStats::new(&self.ctx.server_stats, &self.task_stats); + tls_stream + .get_mut() + .reset_stats(Arc::new(clt_wrapper_stats)); + } +} diff --git a/lib/g3-types/src/net/tls/cert_pair.rs b/lib/g3-types/src/net/tls/cert_pair.rs new file mode 100644 index 000000000..6f0202852 --- /dev/null +++ b/lib/g3-types/src/net/tls/cert_pair.rs @@ -0,0 +1,53 @@ +/* + * Copyright 2024 ByteDance and/or its affiliates. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +use anyhow::anyhow; + +#[derive(Clone, Default, Debug, PartialEq, Eq)] +pub struct UnparsedTlsCertPair { + cert_chain: String, + private_key: String, +} + +impl UnparsedTlsCertPair { + pub fn check(&self) -> anyhow::Result<()> { + if self.cert_chain.is_empty() { + return Err(anyhow!("no certificate set")); + } + if self.private_key.is_empty() { + return Err(anyhow!("no private key set")); + } + Ok(()) + } + + #[inline] + pub fn cert_chain(&self) -> &[u8] { + self.cert_chain.as_bytes() + } + + pub fn set_cert_chain(&mut self, chain: String) { + self.cert_chain = chain; + } + + #[inline] + pub fn private_key(&self) -> &[u8] { + self.private_key.as_bytes() + } + + pub fn set_private_key(&mut self, key: String) { + self.private_key = key; + } +} diff --git a/lib/g3-types/src/net/tls/mod.rs b/lib/g3-types/src/net/tls/mod.rs index 65cb17895..7e178b988 100644 --- a/lib/g3-types/src/net/tls/mod.rs +++ b/lib/g3-types/src/net/tls/mod.rs @@ -37,3 +37,6 @@ pub use ticketer::{ mod version; pub use version::TlsVersion; + +mod cert_pair; +pub use cert_pair::UnparsedTlsCertPair; diff --git a/lib/g3-yaml/src/value/mod.rs b/lib/g3-yaml/src/value/mod.rs index 1765c658f..aec455939 100644 --- a/lib/g3-yaml/src/value/mod.rs +++ b/lib/g3-yaml/src/value/mod.rs @@ -102,6 +102,9 @@ mod quinn; #[cfg(feature = "quinn")] pub use quinn::as_quinn_transport_config; +mod s2n_tls; +pub use s2n_tls::*; + #[cfg(all(unix, not(target_os = "openbsd"), feature = "sched"))] mod sched; #[cfg(all(unix, not(target_os = "openbsd"), feature = "sched"))] diff --git a/lib/g3-yaml/src/value/s2n_tls.rs b/lib/g3-yaml/src/value/s2n_tls.rs new file mode 100644 index 000000000..003f5f9ff --- /dev/null +++ b/lib/g3-yaml/src/value/s2n_tls.rs @@ -0,0 +1,109 @@ +/* + * Copyright 2024 ByteDance and/or its affiliates. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +use std::io::Read; +use std::path::Path; + +use anyhow::{anyhow, Context}; +use yaml_rust::Yaml; + +use g3_types::net::UnparsedTlsCertPair; + +fn as_certificates_from_single_element( + value: &Yaml, + lookup_dir: Option<&Path>, +) -> anyhow::Result { + const MAX_FILE_SIZE: usize = 4_000_000; // 4MB + + if let Yaml::String(s) = value { + if s.trim_start().starts_with("--") { + return Ok(s.to_string()); + } + } + + let (file, path) = crate::value::as_file(value, lookup_dir).context("invalid file")?; + let mut contents = String::with_capacity(MAX_FILE_SIZE); + file.take(MAX_FILE_SIZE as u64) + .read_to_string(&mut contents) + .map_err(|e| anyhow!("failed to read contents of file {}: {e}", path.display()))?; + Ok(contents) +} + +pub fn as_s2n_tls_certificates( + value: &Yaml, + lookup_dir: Option<&Path>, +) -> anyhow::Result> { + if let Yaml::Array(seq) = value { + let mut certs = Vec::new(); + for (i, v) in seq.iter().enumerate() { + let this_certs = as_certificates_from_single_element(v, lookup_dir) + .context(format!("invalid certificates value for element #{i}"))?; + certs.push(this_certs); + } + Ok(certs) + } else { + as_certificates_from_single_element(value, lookup_dir).map(|cert| vec![cert]) + } +} + +pub fn as_s2n_tls_private_key(value: &Yaml, lookup_dir: Option<&Path>) -> anyhow::Result { + const MAX_FILE_SIZE: usize = 256_000; // 256KB + + if let Yaml::String(s) = value { + if s.trim_start().starts_with("--") { + return Ok(s.to_string()); + } + } + + let (file, path) = crate::value::as_file(value, lookup_dir).context("invalid file")?; + let mut contents = String::with_capacity(MAX_FILE_SIZE); + file.take(MAX_FILE_SIZE as u64) + .read_to_string(&mut contents) + .map_err(|e| anyhow!("failed to read contents of file {}: {e}", path.display()))?; + Ok(contents) +} + +pub fn as_s2n_tls_certificate_pair( + value: &Yaml, + lookup_dir: Option<&Path>, +) -> anyhow::Result { + if let Yaml::Hash(map) = value { + let mut pair = UnparsedTlsCertPair::default(); + + crate::foreach_kv(map, |k, v| match crate::key::normalize(k).as_str() { + "certificate" | "cert" => { + let cert = as_certificates_from_single_element(v, lookup_dir) + .context(format!("invalid certificates value for key {k}"))?; + pair.set_cert_chain(cert); + Ok(()) + } + "private_key" | "key" => { + let key = as_s2n_tls_private_key(v, lookup_dir) + .context(format!("invalid private key value for key {k}"))?; + pair.set_private_key(key); + Ok(()) + } + _ => Err(anyhow!("invalid key {k}")), + })?; + + pair.check()?; + Ok(pair) + } else { + Err(anyhow!( + "yaml value type for s2n tls certificate pair should be 'map'" + )) + } +}