Compare commits

..

41 Commits

Author SHA1 Message Date
Alexey 8c303ab2b6 Merge pull request #765 from Misha20062006/patch-2
Correct saving instructions in QUICK_START_GUIDE.ru.md
2026-05-06 17:13:49 +03:00
Misha20062006 f70c2936c7 Correct saving instructions in QUICK_START_GUIDE.ru.md
Updated instructions for saving changes in the guide.
2026-05-06 00:07:14 +03:00
Alexey 9ee341a94f Merge pull request #757 from Dimasssss/docs
Update CONFIG_PARAMS
2026-05-02 00:36:46 +03:00
Dimasssss f76c847c44 Update CONFIG_PARAMS.en.md 2026-05-01 21:10:34 +03:00
Dimasssss 1aaa9c0bc6 Update CONFIG_PARAMS.ru.md 2026-05-01 21:09:38 +03:00
Alexey e50026e776 Update README.md 2026-04-30 19:41:40 +03:00
Alexey 7106f38fae Update Cargo.lock 2026-04-30 11:38:33 +03:00
Alexey 2a694470d5 Update Cargo.toml 2026-04-30 11:37:18 +03:00
Alexey b98cd37211 TLS Full Certificate Budget Bookkeeping + Hot-path Cleanup and Timeout Invariants + IP-Tracker refactoring + Shard TLS Full-Cert Budget: merge pull request #753 from telemt/flow
TLS Full Certificate Budget Bookkeeping + Hot-path Cleanup and Timeout Invariants + IP-Tracker refactoring + Shard TLS Full-Cert Budget
2026-04-30 11:36:30 +03:00
Alexey 8b62965978 Stabilize unknown-DC symlink race test setup 2026-04-30 11:11:04 +03:00
Alexey d46bda9880 Preserve synchronous IP cleanup queue contract + Rustfmt 2026-04-30 11:05:18 +03:00
Alexey c3de07db6a Shard TLS full-cert budget tracking + Bound user-labeled metrics export cardinality 2026-04-30 11:01:10 +03:00
Alexey 61f9af7ffc Reduce Lock-free IP-Tracker Cleanup backlog 2026-04-30 10:51:04 +03:00
Alexey 1f90e28871 Cap scanner-sensitive Caches and IP-Tracker Cardinality 2026-04-30 10:43:27 +03:00
Alexey 876b74ebf7 Hot-path Cleanup and Timeout Invariants 2026-04-29 23:16:11 +03:00
Alexey b34e1d71ae TLS Full Certificate Budget Bookkeeping 2026-04-29 23:00:25 +03:00
Alexey b1c947e8e3 Limit only new ip when TimeWindow + Fix WorkingDirectory behavior + Atomically updates with Includes + Expose tls_domains links as domain-link pairs + TLS Fetcher on multiple tls_domains: merge pull request #751 from telemt/flow
Limit only new ip when TimeWindow + Fix WorkingDirectory behavior + Atomically updates with Includes + Expose tls_domains links as domain-link pairs + TLS Fetcher on multiple tls_domains
2026-04-29 16:04:36 +03:00
Alexey cfe01dced2 Bump
Signed-off-by: Alexey <247128645+axkurcom@users.noreply.github.com>
2026-04-29 15:54:22 +03:00
Alexey 8520955a5f Update helpers.rs
Signed-off-by: Alexey <247128645+axkurcom@users.noreply.github.com>
2026-04-29 15:53:27 +03:00
Alexey 065786b839 TLS Fetcher on multiple tls_domains by #750
Co-Authored-By: brekotis <93345790+brekotis@users.noreply.github.com>
Signed-off-by: Alexey <247128645+axkurcom@users.noreply.github.com>
2026-04-29 11:47:42 +03:00
Alexey f0e1a6cf1c Expose tls_domains links as domain-link pairs
Co-Authored-By: brekotis <93345790+brekotis@users.noreply.github.com>
Signed-off-by: Alexey <247128645+axkurcom@users.noreply.github.com>
2026-04-29 11:34:47 +03:00
Alexey 236bbb4970 Atomically updates with Includes
Signed-off-by: Alexey <247128645+axkurcom@users.noreply.github.com>
2026-04-28 13:00:13 +03:00
Alexey 8ef5263fce Fix WorkingDirectory behavior
Signed-off-by: Alexey <247128645+axkurcom@users.noreply.github.com>
Co-Authored-By: mikhailnov <m@mikhailnov.ru>
Signed-off-by: Alexey <247128645+axkurcom@users.noreply.github.com>
2026-04-28 12:31:21 +03:00
Alexey 893cef22e3 Update README.md 2026-04-27 23:49:47 +03:00
Alexey bdfa641843 Merge pull request #735 from sanekb/fix_timewindow_same_ip
fix: limit only new ip when TimeWindow mode enabled
2026-04-25 19:08:36 +03:00
Alexey 007fc86189 Merge branch 'flow' into fix_timewindow_same_ip 2026-04-25 18:56:27 +03:00
Alexey 10c9bcd97d Merge pull request #747 from telemt/flow
Restore active IP observability for users without unique-IP limits
2026-04-25 18:11:30 +03:00
Alexey 8ab9405dca Bump 2026-04-25 18:05:22 +03:00
Alexey 9412f089c0 Restore active IP observability for users without unique-IP limits 2026-04-25 15:49:28 +03:00
Alexey 4e57cee9b9 Merge pull request #745 from telemt/flow
API PATCH fixes + No IP tracking with disabled unique-IP limits + Bound hot-path pressure in ME Relay and Handshake + Bounded ME Route fairness and IP-Cleanup-Backlog + Bound relay queues by bytes
2026-04-25 14:45:34 +03:00
Alexey e217371dc8 Bump 2026-04-25 14:36:51 +03:00
sanekb d567dfe40b fix: limit only new ip when TimeWindow mode enabled 2026-04-25 14:36:43 +03:00
Alexey 37c916056a Rustfmt 2026-04-25 14:35:35 +03:00
Alexey 2f2fe9d5d3 Bound relay queues by bytes
Co-Authored-By: brekotis <93345790+brekotis@users.noreply.github.com>
Signed-off-by: Alexey <247128645+axkurcom@users.noreply.github.com>
2026-04-25 13:54:20 +03:00
Alexey 1df668144c Bounded ME Route fairness and IP-Cleanup-Backlog
Co-Authored-By: brekotis <93345790+brekotis@users.noreply.github.com>
Signed-off-by: Alexey <247128645+axkurcom@users.noreply.github.com>
2026-04-25 13:09:10 +03:00
Alexey 8494429690 Merge pull request #743 from amirotin/api/patch-user-null-removal
feat(api): support null-removal in PATCH /v1/users/{user}
2026-04-25 13:07:13 +03:00
Alexey f25bb17b86 Merge branch 'flow' into api/patch-user-null-removal 2026-04-25 12:28:48 +03:00
Alexey 27b5d576c0 Bound hot-path pressure in ME Relay + Handshake
Co-Authored-By: brekotis <93345790+brekotis@users.noreply.github.com>
Signed-off-by: Alexey <247128645+axkurcom@users.noreply.github.com>
2026-04-25 12:16:26 +03:00
Alexey e78592ef9b Avoid IP tracking when unique-IP limits are disabled and cap beobachten memory
Co-Authored-By: brekotis <93345790+brekotis@users.noreply.github.com>
Signed-off-by: Alexey <247128645+axkurcom@users.noreply.github.com>
2026-04-25 12:00:46 +03:00
Mirotin Artem 4ed87d1946 feat(api): support null-removal in PATCH /v1/users/{user}
PatchUserRequest now uses Patch<T> for the five removable fields
(user_ad_tag, max_tcp_conns, expiration_rfc3339, data_quota_bytes,
max_unique_ips). Sending JSON null drops the entry from the
corresponding access HashMap; sending 0 is preserved as a literal
limit; omitted fields stay untouched. The handler synchronises the
in-memory ip_tracker on both set and remove of max_unique_ips. A
helper parse_patch_expiration mirrors parse_optional_expiration for
the new three-state field. Runtime semantics are unchanged.
2026-04-25 00:49:34 +03:00
Mirotin Artem 635bea4de4 feat(api): add Patch<T> enum for JSON merge-patch semantics
Introduce a three-state Patch<T> (Unchanged / Remove / Set) and a
serde helper patch_field that distinguishes an omitted JSON field
from an explicit null. Wired up next as the field type for the
removable settings on PATCH /v1/users/{user}.
2026-04-25 00:49:34 +03:00
37 changed files with 2118 additions and 220 deletions
Generated
+1 -1
View File
@@ -2791,7 +2791,7 @@ checksum = "7b2093cf4c8eb1e67749a6762251bc9cd836b6fc171623bd0a9d324d37af2417"
[[package]]
name = "telemt"
version = "3.4.6"
version = "3.4.10"
dependencies = [
"aes",
"anyhow",
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "telemt"
version = "3.4.6"
version = "3.4.10"
edition = "2024"
[features]
+3 -3
View File
@@ -1,6 +1,6 @@
# Telemt - MTProxy on Rust + Tokio
[![Latest Release](https://img.shields.io/github/v/release/telemt/telemt?color=neon)](https://github.com/telemt/telemt/releases/latest) [![Stars](https://img.shields.io/github/stars/telemt/telemt?style=social)](https://github.com/telemt/telemt/stargazers) [![Forks](https://img.shields.io/github/forks/telemt/telemt?style=social)](https://github.com/telemt/telemt/network/members) [![Telegram](https://img.shields.io/badge/Telegram-Chat-24a1de?logo=telegram&logoColor=24a1de)](https://t.me/telemtrs)
[![Latest Release](https://img.shields.io/github/v/release/telemt/telemt?color=neon)](https://github.com/telemt/telemt/releases/latest) [![Stars](https://img.shields.io/github/stars/telemt/telemt?style=social)](https://github.com/telemt/telemt/stargazers) [![Forks](https://img.shields.io/github/forks/telemt/telemt?style=social)](https://github.com/telemt/telemt/network/members)
[🇷🇺 README на русском](https://github.com/telemt/telemt/blob/main/README.ru.md)
@@ -14,7 +14,7 @@
<p align="center">
<a href="https://t.me/telemtrs">
<img src="/docs/assets/telegram_button.svg" width="150"/>
<img src="https://github.com/user-attachments/assets/30b7e7b9-974a-4e3d-aab6-b58a85de4507" width="240"/>
</a>
</p>
@@ -99,7 +99,7 @@ Monero (XMR) directly:
8Bk4tZEYPQWSypeD2hrUXG2rKbAKF16GqEN942ZdAP5cFdSqW6h4DwkP5cJMAdszzuPeHeHZPTyjWWFwzeFdjuci3ktfMoB
```
All donations go toward infrastructure, development, and research.
All donations go toward infrastructure, development and research
![telemt_scheme](docs/assets/telemt.png)
+79
View File
@@ -162,6 +162,8 @@ This document lists all configuration keys accepted by `config.toml`.
| [`log_level`](#log_level) | `"debug"`, `"verbose"`, `"normal"`, or `"silent"` | `"normal"` |
| [`disable_colors`](#disable_colors) | `bool` | `false` |
| [`me_socks_kdf_policy`](#me_socks_kdf_policy) | `"strict"` or `"compat"` | `"strict"` |
| [`me_route_backpressure_enabled`](#me_route_backpressure_enabled) | `bool` | `false` |
| [`me_route_fairshare_enabled`](#me_route_fairshare_enabled) | `bool` | `false` |
| [`me_route_backpressure_base_timeout_ms`](#me_route_backpressure_base_timeout_ms) | `u64` | `25` |
| [`me_route_backpressure_high_timeout_ms`](#me_route_backpressure_high_timeout_ms) | `u64` | `120` |
| [`me_route_backpressure_high_watermark_pct`](#me_route_backpressure_high_watermark_pct) | `u8` | `80` |
@@ -975,6 +977,24 @@ This document lists all configuration keys accepted by `config.toml`.
[general]
me_socks_kdf_policy = "strict"
```
## me_route_backpressure_enabled
- **Constraints / validation**: `bool`.
- **Description**: Enables channel-pressure-aware route send timeouts.
- **Example**:
```toml
[general]
me_route_backpressure_enabled = false
```
## me_route_fairshare_enabled
- **Constraints / validation**: `bool`.
- **Description**: Enables fair-share routing admission across writer workers.
- **Example**:
```toml
[general]
me_route_fairshare_enabled = false
```
## me_route_backpressure_base_timeout_ms
- **Constraints / validation**: Must be within `1..=5000` (milliseconds).
- **Description**: Base backpressure timeout in milliseconds for ME route-channel send.
@@ -1753,6 +1773,7 @@ This document lists all configuration keys accepted by `config.toml`.
| [`metrics_whitelist`](#metrics_whitelist) | `IpNetwork[]` | `["127.0.0.1/32", "::1/128"]` |
| [`max_connections`](#max_connections) | `u32` | `10000` |
| [`accept_permit_timeout_ms`](#accept_permit_timeout_ms) | `u64` | `250` |
| [`listen_backlog`](#listen_backlog) | `u32` | `1024` |
## port
- **Constraints / validation**: `u16`.
@@ -1763,6 +1784,15 @@ This document lists all configuration keys accepted by `config.toml`.
[server]
port = 443
```
## listen_backlog
- **Constraints / validation**: `u32`. `0` uses the OS default backlog behavior.
- **Description**: Listen backlog passed to `listen(2)` for TCP sockets.
- **Example**:
```toml
[server]
listen_backlog = 1024
```
## listen_addr_ipv4
- **Constraints / validation**: `String` (optional). When set, must be a valid IPv4 address string.
- **Description**: IPv4 bind address for TCP listener (omit this key to disable IPv4 bind).
@@ -2005,6 +2035,7 @@ Note: This section also accepts the legacy alias `[server.admin_api]` (same sche
| [`runtime_edge_top_n`](#runtime_edge_top_n) | `usize` | `10` |
| [`runtime_edge_events_capacity`](#runtime_edge_events_capacity) | `usize` | `256` |
| [`read_only`](#read_only) | `bool` | `false` |
| [`gray_action`](#gray_action) | `"drop"`, `"api"`, or `"200"` | `"drop"` |
## enabled
- **Constraints / validation**: `bool`.
@@ -2015,6 +2046,15 @@ Note: This section also accepts the legacy alias `[server.admin_api]` (same sche
[server.api]
enabled = true
```
## gray_action
- **Constraints / validation**: `"drop"`, `"api"`, or `"200"`.
- **Description**: API response policy for gray/limited states: drop request, serve normal API response, or force `200 OK`.
- **Example**:
```toml
[server.api]
gray_action = "drop"
```
## listen
- **Constraints / validation**: `String`. Must be in `IP:PORT` format.
- **Description**: API bind address in `IP:PORT` format.
@@ -2207,6 +2247,15 @@ Note: This section also accepts the legacy alias `[server.admin_api]` (same sche
[timeouts]
client_handshake = 30
```
## client_first_byte_idle_secs
- **Constraints / validation**: `u64` (seconds). `0` disables first-byte idle enforcement.
- **Description**: Maximum idle time to wait for the first client payload byte after session setup.
- **Example**:
```toml
[timeouts]
client_first_byte_idle_secs = 300
```
## relay_idle_policy_v2_enabled
- **Constraints / validation**: `bool`.
- **Description**: Enables soft/hard middle-relay client idle policy.
@@ -2311,6 +2360,7 @@ Note: This section also accepts the legacy alias `[server.admin_api]` (same sche
| [`server_hello_delay_max_ms`](#server_hello_delay_max_ms) | `u64` | `0` |
| [`tls_new_session_tickets`](#tls_new_session_tickets) | `u8` | `0` |
| [`tls_full_cert_ttl_secs`](#tls_full_cert_ttl_secs) | `u64` | `90` |
| [`serverhello_compact`](#serverhello_compact) | `bool` | `false` |
| [`alpn_enforce`](#alpn_enforce) | `bool` | `true` |
| [`mask_proxy_protocol`](#mask_proxy_protocol) | `u8` | `0` |
| [`mask_shape_hardening`](#mask_shape_hardening) | `bool` | `true` |
@@ -2488,6 +2538,15 @@ Note: This section also accepts the legacy alias `[server.admin_api]` (same sche
[censorship]
tls_full_cert_ttl_secs = 90
```
## serverhello_compact
- **Constraints / validation**: `bool`.
- **Description**: Enables compact ServerHello/Fake-TLS profile to reduce response-size signature.
- **Example**:
```toml
[censorship]
serverhello_compact = false
```
## alpn_enforce
- **Constraints / validation**: `bool`.
- **Description**: Enforces ALPN echo behavior based on client preference.
@@ -2830,6 +2889,8 @@ If your backend or network is very bandwidth-constrained, reduce cap first. If p
| [`replay_check_len`](#replay_check_len) | `usize` | `65536` |
| [`replay_window_secs`](#replay_window_secs) | `u64` | `120` |
| [`ignore_time_skew`](#ignore_time_skew) | `bool` | `false` |
| [`user_rate_limits`](#user_rate_limits) | `Map<String, RateLimitBps>` | `{}` |
| [`cidr_rate_limits`](#cidr_rate_limits) | `Map<IpNetwork, RateLimitBps>` | `{}` |
## users
- **Constraints / validation**: Must not be empty (at least one user must exist). Each value must be **exactly 32 hex characters**.
@@ -2958,6 +3019,24 @@ If your backend or network is very bandwidth-constrained, reduce cap first. If p
```
## user_rate_limits
- **Constraints / validation**: Table `username -> { up_bps, down_bps }`. At least one direction must be non-zero.
- **Description**: Per-user bandwidth caps in bytes/sec for upload (`up_bps`) and download (`down_bps`).
- **Example**:
```toml
[access.user_rate_limits]
alice = { up_bps = 1048576, down_bps = 2097152 }
```
## cidr_rate_limits
- **Constraints / validation**: Table `CIDR -> { up_bps, down_bps }`. CIDR must parse as `IpNetwork`; at least one direction must be non-zero.
- **Description**: Source-subnet bandwidth caps applied alongside per-user limits.
- **Example**:
```toml
[access.cidr_rate_limits]
"203.0.113.0/24" = { up_bps = 0, down_bps = 1048576 }
```
# [[upstreams]]
+79
View File
@@ -162,6 +162,8 @@
| [`log_level`](#log_level) | `"debug"`, `"verbose"`, `"normal"`, or `"silent"` | `"normal"` |
| [`disable_colors`](#disable_colors) | `bool` | `false` |
| [`me_socks_kdf_policy`](#me_socks_kdf_policy) | `"strict"` or `"compat"` | `"strict"` |
| [`me_route_backpressure_enabled`](#me_route_backpressure_enabled) | `bool` | `false` |
| [`me_route_fairshare_enabled`](#me_route_fairshare_enabled) | `bool` | `false` |
| [`me_route_backpressure_base_timeout_ms`](#me_route_backpressure_base_timeout_ms) | `u64` | `25` |
| [`me_route_backpressure_high_timeout_ms`](#me_route_backpressure_high_timeout_ms) | `u64` | `120` |
| [`me_route_backpressure_high_watermark_pct`](#me_route_backpressure_high_watermark_pct) | `u8` | `80` |
@@ -975,6 +977,24 @@
[general]
me_socks_kdf_policy = "strict"
```
## me_route_backpressure_enabled
- **Ограничения / валидация**: `bool`.
- **Описание**: Включает адаптивные таймауты записи маршрута в зависимости от заполнения канала.
- **Example**:
```toml
[general]
me_route_backpressure_enabled = false
```
## me_route_fairshare_enabled
- **Ограничения / валидация**: `bool`.
- **Описание**: Включает справедливое распределение нагрузки маршрутизации между writer-потоками.
- **Example**:
```toml
[general]
me_route_fairshare_enabled = false
```
## me_route_backpressure_base_timeout_ms
- **Ограничения / валидация**: Должно быть в пределах `1..=5000` (миллисекунд).
- **Описание**: Базовый таймаут (в миллисекундах) ожидания при режиме **backpressure** (ситуация, при которой данные обрабатываются медленне, чем получаются) для отправки через ME route-channel.
@@ -1755,6 +1775,7 @@
| [`metrics_whitelist`](#metrics_whitelist) | `IpNetwork[]` | `["127.0.0.1/32", "::1/128"]` |
| [`max_connections`](#max_connections) | `u32` | `10000` |
| [`accept_permit_timeout_ms`](#accept_permit_timeout_ms) | `u64` | `250` |
| [`listen_backlog`](#listen_backlog) | `u32` | `1024` |
## port
- **Ограничения / валидация**: `u16`.
@@ -1765,6 +1786,15 @@
[server]
port = 443
```
## listen_backlog
- **Ограничения / валидация**: `u32`. `0` использует системный backlog по умолчанию.
- **Описание**: Значение backlog, передаваемое в `listen(2)` для TCP-сокетов.
- **Example**:
```toml
[server]
listen_backlog = 1024
```
## listen_addr_ipv4
- **Ограничения / валидация**: `String` (необязательный параметр). Если задан, должен содержать валидный IPv4-адрес в формате строки.
- **Описание**: Прослушиваемый адрес в формате IPv4 (не задавайте этот параметр, если необходимо отключить прослушивание по IPv4).
@@ -2011,6 +2041,7 @@
| [`runtime_edge_top_n`](#runtime_edge_top_n) | `usize` | `10` |
| [`runtime_edge_events_capacity`](#runtime_edge_events_capacity) | `usize` | `256` |
| [`read_only`](#read_only) | `bool` | `false` |
| [`gray_action`](#gray_action) | `"drop"`, `"api"`, or `"200"` | `"drop"` |
## enabled
- **Ограничения / валидация**: `bool`.
@@ -2021,6 +2052,15 @@
[server.api]
enabled = true
```
## gray_action
- **Ограничения / валидация**: `"drop"`, `"api"` или `"200"`.
- **Описание**: Политика ответа API в «серых» (ограниченных) состояниях: сброс, обычный API-ответ, либо `200 OK`.
- **Example**:
```toml
[server.api]
gray_action = "drop"
```
## listen
- **Ограничения / валидация**: `String`. Должно быть в формате `IP:PORT`.
- **Описание**: Адрес биндинга API в формате `IP:PORT`.
@@ -2213,6 +2253,15 @@
[timeouts]
client_handshake = 30
```
## client_first_byte_idle_secs
- **Ограничения / валидация**: `u64` (секунды). `0` отключает проверку простоя до первого байта.
- **Описание**: Максимальное время ожидания первого байта полезной нагрузки от клиента после установления сессии.
- **Example**:
```toml
[timeouts]
client_first_byte_idle_secs = 300
```
## relay_idle_policy_v2_enabled
- **Ограничения / валидация**: `bool`.
- **Описание**: Включает политику простоя клиента для промежуточного узла.
@@ -2317,6 +2366,7 @@
| [`server_hello_delay_max_ms`](#server_hello_delay_max_ms) | `u64` | `0` |
| [`tls_new_session_tickets`](#tls_new_session_tickets) | `u8` | `0` |
| [`tls_full_cert_ttl_secs`](#tls_full_cert_ttl_secs) | `u64` | `90` |
| [`serverhello_compact`](#serverhello_compact) | `bool` | `false` |
| [`alpn_enforce`](#alpn_enforce) | `bool` | `true` |
| [`mask_proxy_protocol`](#mask_proxy_protocol) | `u8` | `0` |
| [`mask_shape_hardening`](#mask_shape_hardening) | `bool` | `true` |
@@ -2493,6 +2543,15 @@
[censorship]
tls_full_cert_ttl_secs = 90
```
## serverhello_compact
- **Ограничения / валидация**: `bool`.
- **Описание**: Включает компактный профиль ServerHello/Fake-TLS для снижения сигнатуры размера ответа.
- **Example**:
```toml
[censorship]
serverhello_compact = false
```
## alpn_enforce
- **Ограничения / валидация**: `bool`.
- **Описание**: Принудительно изменяет поведение возврата ALPN в соответствии с предпочтениями клиента.
@@ -2837,6 +2896,8 @@
| [`replay_check_len`](#replay_check_len) | `usize` | `65536` |
| [`replay_window_secs`](#replay_window_secs) | `u64` | `120` |
| [`ignore_time_skew`](#ignore_time_skew) | `bool` | `false` |
| [`user_rate_limits`](#user_rate_limits) | `Map<String, RateLimitBps>` | `{}` |
| [`cidr_rate_limits`](#cidr_rate_limits) | `Map<IpNetwork, RateLimitBps>` | `{}` |
## users
- **Ограничения / валидация**: Не должно быть пустым (должен существовать хотя бы один пользователь). Каждое значение должно состоять **ровно из 32 шестнадцатеричных символов**.
@@ -2965,6 +3026,24 @@
```
## user_rate_limits
- **Ограничения / валидация**: Таблица `username -> { up_bps, down_bps }`. Должно быть ненулевое значение хотя бы в одном направлении.
- **Описание**: Персональные лимиты скорости по пользователям в байтах/сек для отправки (`up_bps`) и получения (`down_bps`).
- **Example**:
```toml
[access.user_rate_limits]
alice = { up_bps = 1048576, down_bps = 2097152 }
```
## cidr_rate_limits
- **Ограничения / валидация**: Таблица `CIDR -> { up_bps, down_bps }`. CIDR должен корректно разбираться как `IpNetwork`; хотя бы одно направление должно быть ненулевым.
- **Описание**: Лимиты скорости для подсетей источников, применяются поверх пользовательских ограничений.
- **Example**:
```toml
[access.cidr_rate_limits]
"203.0.113.0/24" = { up_bps = 0, down_bps = 1048576 }
```
# [[upstreams]]
+1 -1
View File
@@ -164,7 +164,7 @@ tls_front_dir = "tlsfront" # Директория кэша для эмуляц
hello = "00000000000000000000000000000000"
```
Затем нажмите Ctrl+S -> Ctrl+X, чтобы сохранить
Затем нажмите Ctrl+O -> Ctrl+X, чтобы сохранить
> [!WARNING]
> Замените значение параметра `hello` на значение, которое вы получили в пункте 0.
+18
View File
@@ -82,6 +82,7 @@ pub(super) async fn load_config_from_disk(config_path: &Path) -> Result<ProxyCon
.map_err(|e| ApiFailure::internal(format!("failed to load config: {}", e)))
}
#[allow(dead_code)]
pub(super) async fn save_config_to_disk(
config_path: &Path,
cfg: &ProxyConfig,
@@ -106,6 +107,12 @@ pub(super) async fn save_access_sections_to_disk(
if applied.contains(section) {
continue;
}
if find_toml_table_bounds(&content, section.table_name()).is_none()
&& access_section_is_empty(cfg, *section)
{
applied.push(*section);
continue;
}
let rendered = render_access_section(cfg, *section)?;
content = upsert_toml_table(&content, section.table_name(), &rendered);
applied.push(*section);
@@ -183,6 +190,17 @@ fn render_access_section(cfg: &ProxyConfig, section: AccessSection) -> Result<St
Ok(out)
}
fn access_section_is_empty(cfg: &ProxyConfig, section: AccessSection) -> bool {
match section {
AccessSection::Users => cfg.access.users.is_empty(),
AccessSection::UserAdTags => cfg.access.user_ad_tags.is_empty(),
AccessSection::UserMaxTcpConns => cfg.access.user_max_tcp_conns.is_empty(),
AccessSection::UserExpirations => cfg.access.user_expirations.is_empty(),
AccessSection::UserDataQuota => cfg.access.user_data_quota.is_empty(),
AccessSection::UserMaxUniqueIps => cfg.access.user_max_unique_ips.is_empty(),
}
}
fn serialize_table_body<T: Serialize>(value: &T) -> Result<String, ApiFailure> {
toml::to_string(value)
.map_err(|e| ApiFailure::internal(format!("failed to serialize access section: {}", e)))
+1
View File
@@ -28,6 +28,7 @@ mod config_store;
mod events;
mod http_utils;
mod model;
mod patch;
mod runtime_edge;
mod runtime_init;
mod runtime_min;
+32 -5
View File
@@ -5,6 +5,7 @@ use chrono::{DateTime, Utc};
use hyper::StatusCode;
use serde::{Deserialize, Serialize};
use super::patch::{Patch, patch_field};
use crate::crypto::SecureRandom;
const MAX_USERNAME_LEN: usize = 64;
@@ -455,6 +456,13 @@ pub(super) struct UserLinks {
pub(super) classic: Vec<String>,
pub(super) secure: Vec<String>,
pub(super) tls: Vec<String>,
pub(super) tls_domains: Vec<TlsDomainLink>,
}
#[derive(Serialize)]
pub(super) struct TlsDomainLink {
pub(super) domain: String,
pub(super) link: String,
}
#[derive(Serialize)]
@@ -507,11 +515,16 @@ pub(super) struct CreateUserRequest {
#[derive(Deserialize)]
pub(super) struct PatchUserRequest {
pub(super) secret: Option<String>,
pub(super) user_ad_tag: Option<String>,
pub(super) max_tcp_conns: Option<usize>,
pub(super) expiration_rfc3339: Option<String>,
pub(super) data_quota_bytes: Option<u64>,
pub(super) max_unique_ips: Option<usize>,
#[serde(default, deserialize_with = "patch_field")]
pub(super) user_ad_tag: Patch<String>,
#[serde(default, deserialize_with = "patch_field")]
pub(super) max_tcp_conns: Patch<usize>,
#[serde(default, deserialize_with = "patch_field")]
pub(super) expiration_rfc3339: Patch<String>,
#[serde(default, deserialize_with = "patch_field")]
pub(super) data_quota_bytes: Patch<u64>,
#[serde(default, deserialize_with = "patch_field")]
pub(super) max_unique_ips: Patch<usize>,
}
#[derive(Default, Deserialize)]
@@ -530,6 +543,20 @@ pub(super) fn parse_optional_expiration(
Ok(Some(parsed.with_timezone(&Utc)))
}
pub(super) fn parse_patch_expiration(
value: &Patch<String>,
) -> Result<Patch<DateTime<Utc>>, ApiFailure> {
match value {
Patch::Unchanged => Ok(Patch::Unchanged),
Patch::Remove => Ok(Patch::Remove),
Patch::Set(raw) => {
let parsed = DateTime::parse_from_rfc3339(raw)
.map_err(|_| ApiFailure::bad_request("expiration_rfc3339 must be valid RFC3339"))?;
Ok(Patch::Set(parsed.with_timezone(&Utc)))
}
}
}
pub(super) fn is_valid_user_secret(secret: &str) -> bool {
secret.len() == 32 && secret.chars().all(|c| c.is_ascii_hexdigit())
}
+130
View File
@@ -0,0 +1,130 @@
use serde::Deserialize;
/// Three-state field for JSON Merge Patch semantics on the `PATCH /v1/users/{user}`
/// endpoint.
///
/// `Unchanged` is produced when the JSON body omits the field entirely and tells the
/// handler to leave the corresponding configuration entry untouched. `Remove` is
/// produced when the JSON body sets the field to `null` and instructs the handler to
/// drop the entry from the corresponding access HashMap. `Set` carries an explicit
/// new value, including zero, which is preserved verbatim in the configuration.
#[derive(Debug)]
pub(super) enum Patch<T> {
Unchanged,
Remove,
Set(T),
}
impl<T> Default for Patch<T> {
fn default() -> Self {
Self::Unchanged
}
}
/// Serde deserializer adapter for fields that follow JSON Merge Patch semantics.
///
/// Pair this with `#[serde(default, deserialize_with = "patch_field")]` on a
/// `Patch<T>` field. An omitted field falls back to `Patch::Unchanged` via
/// `Default`; an explicit JSON `null` becomes `Patch::Remove`; any other value
/// becomes `Patch::Set(v)`.
pub(super) fn patch_field<'de, D, T>(deserializer: D) -> Result<Patch<T>, D::Error>
where
D: serde::Deserializer<'de>,
T: serde::Deserialize<'de>,
{
Option::<T>::deserialize(deserializer).map(|opt| match opt {
Some(value) => Patch::Set(value),
None => Patch::Remove,
})
}
#[cfg(test)]
mod tests {
use super::*;
use crate::api::model::{PatchUserRequest, parse_patch_expiration};
use chrono::{TimeZone, Utc};
use serde::Deserialize;
#[derive(Deserialize)]
struct Holder {
#[serde(default, deserialize_with = "patch_field")]
value: Patch<u64>,
}
fn parse(json: &str) -> Holder {
serde_json::from_str(json).expect("valid json")
}
#[test]
fn omitted_field_yields_unchanged() {
let h = parse("{}");
assert!(matches!(h.value, Patch::Unchanged));
}
#[test]
fn explicit_null_yields_remove() {
let h = parse(r#"{"value": null}"#);
assert!(matches!(h.value, Patch::Remove));
}
#[test]
fn explicit_value_yields_set() {
let h = parse(r#"{"value": 42}"#);
assert!(matches!(h.value, Patch::Set(42)));
}
#[test]
fn explicit_zero_yields_set_zero() {
let h = parse(r#"{"value": 0}"#);
assert!(matches!(h.value, Patch::Set(0)));
}
#[test]
fn parse_patch_expiration_passes_unchanged_and_remove_through() {
assert!(matches!(
parse_patch_expiration(&Patch::Unchanged),
Ok(Patch::Unchanged)
));
assert!(matches!(
parse_patch_expiration(&Patch::Remove),
Ok(Patch::Remove)
));
}
#[test]
fn parse_patch_expiration_parses_set_value() {
let parsed =
parse_patch_expiration(&Patch::Set("2030-01-02T03:04:05Z".into())).expect("valid");
match parsed {
Patch::Set(dt) => {
assert_eq!(dt, Utc.with_ymd_and_hms(2030, 1, 2, 3, 4, 5).unwrap());
}
other => panic!("expected Patch::Set, got {:?}", other),
}
}
#[test]
fn parse_patch_expiration_rejects_invalid_set_value() {
assert!(parse_patch_expiration(&Patch::Set("not-a-date".into())).is_err());
}
#[test]
fn patch_user_request_deserializes_mixed_states() {
let raw = r#"{
"secret": "00112233445566778899aabbccddeeff",
"max_tcp_conns": 0,
"max_unique_ips": null,
"data_quota_bytes": 1024
}"#;
let req: PatchUserRequest = serde_json::from_str(raw).expect("valid json");
assert_eq!(
req.secret.as_deref(),
Some("00112233445566778899aabbccddeeff")
);
assert!(matches!(req.max_tcp_conns, Patch::Set(0)));
assert!(matches!(req.max_unique_ips, Patch::Remove));
assert!(matches!(req.data_quota_bytes, Patch::Set(1024)));
assert!(matches!(req.expiration_rfc3339, Patch::Unchanged));
assert!(matches!(req.user_ad_tag, Patch::Unchanged));
}
}
+198 -29
View File
@@ -8,14 +8,15 @@ use crate::stats::Stats;
use super::ApiShared;
use super::config_store::{
AccessSection, ensure_expected_revision, load_config_from_disk, save_access_sections_to_disk,
save_config_to_disk,
AccessSection, current_revision, ensure_expected_revision, load_config_from_disk,
save_access_sections_to_disk,
};
use super::model::{
ApiFailure, CreateUserRequest, CreateUserResponse, PatchUserRequest, RotateSecretRequest,
UserInfo, UserLinks, is_valid_ad_tag, is_valid_user_secret, is_valid_username,
parse_optional_expiration, random_user_secret,
TlsDomainLink, UserInfo, UserLinks, is_valid_ad_tag, is_valid_user_secret, is_valid_username,
parse_optional_expiration, parse_patch_expiration, random_user_secret,
};
use super::patch::Patch;
pub(super) async fn create_user(
body: CreateUserRequest,
@@ -175,6 +176,13 @@ pub(super) async fn patch_user(
expected_revision: Option<String>,
shared: &ApiShared,
) -> Result<(UserInfo, String), ApiFailure> {
let touches_users = body.secret.is_some();
let touches_user_ad_tags = !matches!(&body.user_ad_tag, Patch::Unchanged);
let touches_user_max_tcp_conns = !matches!(&body.max_tcp_conns, Patch::Unchanged);
let touches_user_expirations = !matches!(&body.expiration_rfc3339, Patch::Unchanged);
let touches_user_data_quota = !matches!(&body.data_quota_bytes, Patch::Unchanged);
let touches_user_max_unique_ips = !matches!(&body.max_unique_ips, Patch::Unchanged);
if let Some(secret) = body.secret.as_ref()
&& !is_valid_user_secret(secret)
{
@@ -182,14 +190,14 @@ pub(super) async fn patch_user(
"secret must be exactly 32 hex characters",
));
}
if let Some(ad_tag) = body.user_ad_tag.as_ref()
if let Patch::Set(ad_tag) = &body.user_ad_tag
&& !is_valid_ad_tag(ad_tag)
{
return Err(ApiFailure::bad_request(
"user_ad_tag must be exactly 32 hex characters",
));
}
let expiration = parse_optional_expiration(body.expiration_rfc3339.as_deref())?;
let expiration = parse_patch_expiration(&body.expiration_rfc3339)?;
let _guard = shared.mutation_lock.lock().await;
let mut cfg = load_config_from_disk(&shared.config_path).await?;
ensure_expected_revision(&shared.config_path, expected_revision.as_deref()).await?;
@@ -205,38 +213,95 @@ pub(super) async fn patch_user(
if let Some(secret) = body.secret {
cfg.access.users.insert(user.to_string(), secret);
}
if let Some(ad_tag) = body.user_ad_tag {
cfg.access.user_ad_tags.insert(user.to_string(), ad_tag);
match body.user_ad_tag {
Patch::Unchanged => {}
Patch::Remove => {
cfg.access.user_ad_tags.remove(user);
}
Patch::Set(ad_tag) => {
cfg.access.user_ad_tags.insert(user.to_string(), ad_tag);
}
}
if let Some(limit) = body.max_tcp_conns {
cfg.access
.user_max_tcp_conns
.insert(user.to_string(), limit);
match body.max_tcp_conns {
Patch::Unchanged => {}
Patch::Remove => {
cfg.access.user_max_tcp_conns.remove(user);
}
Patch::Set(limit) => {
cfg.access
.user_max_tcp_conns
.insert(user.to_string(), limit);
}
}
if let Some(expiration) = expiration {
cfg.access
.user_expirations
.insert(user.to_string(), expiration);
match expiration {
Patch::Unchanged => {}
Patch::Remove => {
cfg.access.user_expirations.remove(user);
}
Patch::Set(expiration) => {
cfg.access
.user_expirations
.insert(user.to_string(), expiration);
}
}
if let Some(quota) = body.data_quota_bytes {
cfg.access.user_data_quota.insert(user.to_string(), quota);
}
let mut updated_limit = None;
if let Some(limit) = body.max_unique_ips {
cfg.access
.user_max_unique_ips
.insert(user.to_string(), limit);
updated_limit = Some(limit);
match body.data_quota_bytes {
Patch::Unchanged => {}
Patch::Remove => {
cfg.access.user_data_quota.remove(user);
}
Patch::Set(quota) => {
cfg.access.user_data_quota.insert(user.to_string(), quota);
}
}
// Capture how the per-user IP limit changed, so the in-memory ip_tracker
// can be synced (set or removed) after the config is persisted.
let max_unique_ips_change = match body.max_unique_ips {
Patch::Unchanged => None,
Patch::Remove => {
cfg.access.user_max_unique_ips.remove(user);
Some(None)
}
Patch::Set(limit) => {
cfg.access
.user_max_unique_ips
.insert(user.to_string(), limit);
Some(Some(limit))
}
};
cfg.validate()
.map_err(|e| ApiFailure::bad_request(format!("config validation failed: {}", e)))?;
let revision = save_config_to_disk(&shared.config_path, &cfg).await?;
let mut touched_sections = Vec::new();
if touches_users {
touched_sections.push(AccessSection::Users);
}
if touches_user_ad_tags {
touched_sections.push(AccessSection::UserAdTags);
}
if touches_user_max_tcp_conns {
touched_sections.push(AccessSection::UserMaxTcpConns);
}
if touches_user_expirations {
touched_sections.push(AccessSection::UserExpirations);
}
if touches_user_data_quota {
touched_sections.push(AccessSection::UserDataQuota);
}
if touches_user_max_unique_ips {
touched_sections.push(AccessSection::UserMaxUniqueIps);
}
let revision = if touched_sections.is_empty() {
current_revision(&shared.config_path).await?
} else {
save_access_sections_to_disk(&shared.config_path, &cfg, &touched_sections).await?
};
drop(_guard);
if let Some(limit) = updated_limit {
shared.ip_tracker.set_user_limit(user, limit).await;
match max_unique_ips_change {
Some(Some(limit)) => shared.ip_tracker.set_user_limit(user, limit).await,
Some(None) => shared.ip_tracker.remove_user_limit(user).await,
None => {}
}
let (detected_ip_v4, detected_ip_v6) = shared.detected_link_ips();
let users = users_from_config(
@@ -404,6 +469,7 @@ pub(super) async fn users_from_config(
classic: Vec::new(),
secure: Vec::new(),
tls: Vec::new(),
tls_domains: Vec::new(),
});
users.push(UserInfo {
in_runtime: runtime_cfg
@@ -458,10 +524,12 @@ fn build_user_links(
.public_port
.unwrap_or(resolve_default_link_port(cfg));
let tls_domains = resolve_tls_domains(cfg);
let extra_tls_domains = resolve_extra_tls_domains(cfg);
let mut classic = Vec::new();
let mut secure = Vec::new();
let mut tls = Vec::new();
let mut tls_domain_links = Vec::new();
for host in &hosts {
if cfg.general.modes.classic {
@@ -484,6 +552,17 @@ fn build_user_links(
host, port, secret, domain_hex
));
}
for domain in &extra_tls_domains {
let domain_hex = hex::encode(domain);
let link = format!(
"tg://proxy?server={}&port={}&secret=ee{}{}",
host, port, secret, domain_hex
);
tls_domain_links.push(TlsDomainLink {
domain: (*domain).to_string(),
link,
});
}
}
}
@@ -491,6 +570,7 @@ fn build_user_links(
classic,
secure,
tls,
tls_domains: tls_domain_links,
}
}
@@ -607,6 +687,19 @@ fn resolve_tls_domains(cfg: &ProxyConfig) -> Vec<&str> {
domains
}
fn resolve_extra_tls_domains(cfg: &ProxyConfig) -> Vec<&str> {
let mut domains = Vec::with_capacity(cfg.censorship.tls_domains.len());
let primary = cfg.censorship.tls_domain.as_str();
for domain in &cfg.censorship.tls_domains {
let value = domain.as_str();
if value.is_empty() || value == primary || domains.contains(&value) {
continue;
}
domains.push(value);
}
domains
}
#[cfg(test)]
mod tests {
use super::*;
@@ -696,4 +789,80 @@ mod tests {
assert!(alice.in_runtime);
assert!(!bob.in_runtime);
}
#[tokio::test]
async fn users_from_config_returns_tls_link_for_each_tls_domain() {
let mut cfg = ProxyConfig::default();
cfg.access.users.insert(
"alice".to_string(),
"0123456789abcdef0123456789abcdef".to_string(),
);
cfg.general.modes.classic = false;
cfg.general.modes.secure = false;
cfg.general.modes.tls = true;
cfg.general.links.public_host = Some("proxy.example.net".to_string());
cfg.general.links.public_port = Some(443);
cfg.censorship.tls_domain = "front-a.example.com".to_string();
cfg.censorship.tls_domains = vec![
"front-b.example.com".to_string(),
"front-c.example.com".to_string(),
"front-b.example.com".to_string(),
"front-a.example.com".to_string(),
];
let stats = Stats::new();
let tracker = UserIpTracker::new();
let users = users_from_config(&cfg, &stats, &tracker, None, None, None).await;
let alice = users
.iter()
.find(|entry| entry.username == "alice")
.expect("alice must be present");
assert_eq!(alice.links.tls.len(), 3);
assert!(
alice
.links
.tls
.iter()
.any(|link| link.ends_with(&hex::encode("front-a.example.com")))
);
assert!(
alice
.links
.tls
.iter()
.any(|link| link.ends_with(&hex::encode("front-b.example.com")))
);
assert!(
alice
.links
.tls
.iter()
.any(|link| link.ends_with(&hex::encode("front-c.example.com")))
);
assert_eq!(alice.links.tls_domains.len(), 2);
assert!(
alice
.links
.tls_domains
.iter()
.any(|entry| entry.domain == "front-b.example.com"
&& entry.link.ends_with(&hex::encode("front-b.example.com")))
);
assert!(
alice
.links
.tls_domains
.iter()
.any(|entry| entry.domain == "front-c.example.com"
&& entry.link.ends_with(&hex::encode("front-c.example.com")))
);
assert!(
!alice
.links
.tls_domains
.iter()
.any(|entry| entry.domain == "front-a.example.com")
);
}
}
+2 -2
View File
@@ -102,7 +102,7 @@ pub(crate) fn default_fake_cert_len() -> usize {
}
pub(crate) fn default_tls_front_dir() -> String {
"/etc/telemt/tlsfront".to_string()
"tlsfront".to_string()
}
pub(crate) fn default_replay_check_len() -> usize {
@@ -568,7 +568,7 @@ pub(crate) fn default_beobachten_flush_secs() -> u64 {
}
pub(crate) fn default_beobachten_file() -> String {
"/etc/telemt/beobachten.txt".to_string()
"beobachten.txt".to_string()
}
pub(crate) fn default_tls_new_session_tickets() -> u8 {
+22 -2
View File
@@ -1087,9 +1087,9 @@ impl ProxyConfig {
));
}
if config.general.me_route_blocking_send_timeout_ms > 5000 {
if !(1..=5000).contains(&config.general.me_route_blocking_send_timeout_ms) {
return Err(ProxyError::Config(
"general.me_route_blocking_send_timeout_ms must be within [0, 5000]".to_string(),
"general.me_route_blocking_send_timeout_ms must be within [1, 5000]".to_string(),
));
}
@@ -2602,6 +2602,26 @@ mod tests {
let _ = std::fs::remove_file(path);
}
#[test]
fn me_route_blocking_send_timeout_ms_zero_is_rejected() {
let toml = r#"
[general]
me_route_blocking_send_timeout_ms = 0
[censorship]
tls_domain = "example.com"
[access.users]
user = "00000000000000000000000000000000"
"#;
let dir = std::env::temp_dir();
let path = dir.join("telemt_me_route_blocking_send_timeout_zero_test.toml");
std::fs::write(&path, toml).unwrap();
let err = ProxyConfig::load(&path).unwrap_err().to_string();
assert!(err.contains("general.me_route_blocking_send_timeout_ms must be within [1, 5000]"));
let _ = std::fs::remove_file(path);
}
#[test]
fn me_route_no_writer_mode_is_parsed() {
let toml = r#"
+1 -1
View File
@@ -778,7 +778,7 @@ pub struct GeneralConfig {
pub me_route_hybrid_max_wait_ms: u64,
/// Maximum wait in milliseconds for blocking ME writer channel send fallback.
/// `0` keeps legacy unbounded wait behavior.
/// Must be within [1, 5000].
#[serde(default = "default_me_route_blocking_send_timeout_ms")]
pub me_route_blocking_send_timeout_ms: u64,
+246 -44
View File
@@ -9,30 +9,52 @@ use std::sync::Mutex;
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{Duration, Instant};
use tokio::sync::{Mutex as AsyncMutex, RwLock};
use tokio::sync::{Mutex as AsyncMutex, RwLock, RwLockWriteGuard};
use crate::config::UserMaxUniqueIpsMode;
const CLEANUP_DRAIN_BATCH_LIMIT: usize = 1024;
const MAX_ACTIVE_IP_ENTRIES: u64 = 131_072;
const MAX_RECENT_IP_ENTRIES: u64 = 262_144;
/// Tracks active and recent client IPs for per-user admission control.
#[derive(Debug, Clone)]
pub struct UserIpTracker {
active_ips: Arc<RwLock<HashMap<String, HashMap<IpAddr, usize>>>>,
recent_ips: Arc<RwLock<HashMap<String, HashMap<IpAddr, Instant>>>>,
active_entry_count: Arc<AtomicU64>,
recent_entry_count: Arc<AtomicU64>,
active_cap_rejects: Arc<AtomicU64>,
recent_cap_rejects: Arc<AtomicU64>,
cleanup_deferred_releases: Arc<AtomicU64>,
max_ips: Arc<RwLock<HashMap<String, usize>>>,
default_max_ips: Arc<RwLock<usize>>,
limit_mode: Arc<RwLock<UserMaxUniqueIpsMode>>,
limit_window: Arc<RwLock<Duration>>,
last_compact_epoch_secs: Arc<AtomicU64>,
cleanup_queue: Arc<Mutex<Vec<(String, IpAddr)>>>,
cleanup_queue: Arc<Mutex<HashMap<(String, IpAddr), usize>>>,
cleanup_drain_lock: Arc<AsyncMutex<()>>,
}
/// Point-in-time memory counters for user/IP limiter state.
#[derive(Debug, Clone, Copy)]
pub struct UserIpTrackerMemoryStats {
/// Number of users with active IP state.
pub active_users: usize,
/// Number of users with recent IP state.
pub recent_users: usize,
/// Number of active `(user, ip)` entries.
pub active_entries: usize,
/// Number of recent-window `(user, ip)` entries.
pub recent_entries: usize,
/// Number of deferred disconnect cleanups waiting to be drained.
pub cleanup_queue_len: usize,
/// Number of new connections rejected by the global active-entry cap.
pub active_cap_rejects: u64,
/// Number of new connections rejected by the global recent-entry cap.
pub recent_cap_rejects: u64,
/// Number of release cleanups deferred through the cleanup queue.
pub cleanup_deferred_releases: u64,
}
impl UserIpTracker {
@@ -40,22 +62,74 @@ impl UserIpTracker {
Self {
active_ips: Arc::new(RwLock::new(HashMap::new())),
recent_ips: Arc::new(RwLock::new(HashMap::new())),
active_entry_count: Arc::new(AtomicU64::new(0)),
recent_entry_count: Arc::new(AtomicU64::new(0)),
active_cap_rejects: Arc::new(AtomicU64::new(0)),
recent_cap_rejects: Arc::new(AtomicU64::new(0)),
cleanup_deferred_releases: Arc::new(AtomicU64::new(0)),
max_ips: Arc::new(RwLock::new(HashMap::new())),
default_max_ips: Arc::new(RwLock::new(0)),
limit_mode: Arc::new(RwLock::new(UserMaxUniqueIpsMode::ActiveWindow)),
limit_window: Arc::new(RwLock::new(Duration::from_secs(30))),
last_compact_epoch_secs: Arc::new(AtomicU64::new(0)),
cleanup_queue: Arc::new(Mutex::new(Vec::new())),
cleanup_queue: Arc::new(Mutex::new(HashMap::new())),
cleanup_drain_lock: Arc::new(AsyncMutex::new(())),
}
}
fn decrement_counter(counter: &AtomicU64, amount: usize) {
if amount == 0 {
return;
}
let amount = amount as u64;
let _ = counter.fetch_update(Ordering::AcqRel, Ordering::Relaxed, |current| {
Some(current.saturating_sub(amount))
});
}
fn apply_active_cleanup(
active_ips: &mut HashMap<String, HashMap<IpAddr, usize>>,
user: &str,
ip: IpAddr,
pending_count: usize,
) -> usize {
if pending_count == 0 {
return 0;
}
let mut remove_user = false;
let mut removed_active_entries = 0usize;
if let Some(user_ips) = active_ips.get_mut(user) {
if let Some(count) = user_ips.get_mut(&ip) {
if *count > pending_count {
*count -= pending_count;
} else if user_ips.remove(&ip).is_some() {
removed_active_entries = 1;
}
}
remove_user = user_ips.is_empty();
}
if remove_user {
active_ips.remove(user);
}
removed_active_entries
}
/// Queues a deferred active IP cleanup for a later async drain.
pub fn enqueue_cleanup(&self, user: String, ip: IpAddr) {
match self.cleanup_queue.lock() {
Ok(mut queue) => queue.push((user, ip)),
Ok(mut queue) => {
let count = queue.entry((user, ip)).or_insert(0);
*count = count.saturating_add(1);
self.cleanup_deferred_releases
.fetch_add(1, Ordering::Relaxed);
}
Err(poisoned) => {
let mut queue = poisoned.into_inner();
queue.push((user.clone(), ip));
let count = queue.entry((user.clone(), ip)).or_insert(0);
*count = count.saturating_add(1);
self.cleanup_deferred_releases
.fetch_add(1, Ordering::Relaxed);
self.cleanup_queue.clear_poison();
tracing::warn!(
"UserIpTracker cleanup_queue lock poisoned; recovered and enqueued IP cleanup for {} ({})",
@@ -75,21 +149,34 @@ impl UserIpTracker {
}
#[cfg(test)]
pub(crate) fn cleanup_queue_mutex_for_tests(&self) -> Arc<Mutex<Vec<(String, IpAddr)>>> {
pub(crate) fn cleanup_queue_mutex_for_tests(
&self,
) -> Arc<Mutex<HashMap<(String, IpAddr), usize>>> {
Arc::clone(&self.cleanup_queue)
}
pub(crate) async fn drain_cleanup_queue(&self) {
// Serialize queue draining and active-IP mutation so check-and-add cannot
// observe stale active entries that are already queued for removal.
let _drain_guard = self.cleanup_drain_lock.lock().await;
let Ok(_drain_guard) = self.cleanup_drain_lock.try_lock() else {
return;
};
let to_remove = {
match self.cleanup_queue.lock() {
Ok(mut queue) => {
if queue.is_empty() {
return;
}
std::mem::take(&mut *queue)
let mut drained =
HashMap::with_capacity(queue.len().min(CLEANUP_DRAIN_BATCH_LIMIT));
for _ in 0..CLEANUP_DRAIN_BATCH_LIMIT {
let Some(key) = queue.keys().next().cloned() else {
break;
};
if let Some(count) = queue.remove(&key) {
drained.insert(key, count);
}
}
drained
}
Err(poisoned) => {
let mut queue = poisoned.into_inner();
@@ -97,28 +184,33 @@ impl UserIpTracker {
self.cleanup_queue.clear_poison();
return;
}
let drained = std::mem::take(&mut *queue);
let mut drained =
HashMap::with_capacity(queue.len().min(CLEANUP_DRAIN_BATCH_LIMIT));
for _ in 0..CLEANUP_DRAIN_BATCH_LIMIT {
let Some(key) = queue.keys().next().cloned() else {
break;
};
if let Some(count) = queue.remove(&key) {
drained.insert(key, count);
}
}
self.cleanup_queue.clear_poison();
drained
}
}
};
if to_remove.is_empty() {
return;
}
let mut active_ips = self.active_ips.write().await;
for (user, ip) in to_remove {
if let Some(user_ips) = active_ips.get_mut(&user) {
if let Some(count) = user_ips.get_mut(&ip) {
if *count > 1 {
*count -= 1;
} else {
user_ips.remove(&ip);
}
}
if user_ips.is_empty() {
active_ips.remove(&user);
}
}
let mut removed_active_entries = 0usize;
for ((user, ip), pending_count) in to_remove {
removed_active_entries = removed_active_entries.saturating_add(
Self::apply_active_cleanup(&mut active_ips, &user, ip, pending_count),
);
}
Self::decrement_counter(&self.active_entry_count, removed_active_entries);
}
fn now_epoch_secs() -> u64 {
@@ -128,6 +220,24 @@ impl UserIpTracker {
.as_secs()
}
async fn active_and_recent_write(
&self,
) -> (
RwLockWriteGuard<'_, HashMap<String, HashMap<IpAddr, usize>>>,
RwLockWriteGuard<'_, HashMap<String, HashMap<IpAddr, Instant>>>,
) {
loop {
let active_ips = self.active_ips.write().await;
match self.recent_ips.try_write() {
Ok(recent_ips) => return (active_ips, recent_ips),
Err(_) => {
drop(active_ips);
tokio::task::yield_now().await;
}
}
}
}
async fn maybe_compact_empty_users(&self) {
const COMPACT_INTERVAL_SECS: u64 = 60;
let now_epoch_secs = Self::now_epoch_secs();
@@ -148,14 +258,16 @@ impl UserIpTracker {
return;
}
let mut active_ips = self.active_ips.write().await;
let mut recent_ips = self.recent_ips.write().await;
let window = *self.limit_window.read().await;
let now = Instant::now();
let (mut active_ips, mut recent_ips) = self.active_and_recent_write().await;
let mut pruned_recent_entries = 0usize;
for user_recent in recent_ips.values_mut() {
Self::prune_recent(user_recent, now, window);
pruned_recent_entries =
pruned_recent_entries.saturating_add(Self::prune_recent(user_recent, now, window));
}
Self::decrement_counter(&self.recent_entry_count, pruned_recent_entries);
let mut users =
Vec::<String>::with_capacity(active_ips.len().saturating_add(recent_ips.len()));
@@ -199,6 +311,9 @@ impl UserIpTracker {
active_entries,
recent_entries,
cleanup_queue_len,
active_cap_rejects: self.active_cap_rejects.load(Ordering::Relaxed),
recent_cap_rejects: self.recent_cap_rejects.load(Ordering::Relaxed),
cleanup_deferred_releases: self.cleanup_deferred_releases.load(Ordering::Relaxed),
}
}
@@ -229,11 +344,17 @@ impl UserIpTracker {
max_ips.clone_from(limits);
}
fn prune_recent(user_recent: &mut HashMap<IpAddr, Instant>, now: Instant, window: Duration) {
fn prune_recent(
user_recent: &mut HashMap<IpAddr, Instant>,
now: Instant,
window: Duration,
) -> usize {
if user_recent.is_empty() {
return;
return 0;
}
let before = user_recent.len();
user_recent.retain(|_, seen_at| now.duration_since(*seen_at) <= window);
before.saturating_sub(user_recent.len())
}
pub async fn check_and_add(&self, username: &str, ip: IpAddr) -> Result<(), String> {
@@ -252,26 +373,40 @@ impl UserIpTracker {
let window = *self.limit_window.read().await;
let now = Instant::now();
let mut active_ips = self.active_ips.write().await;
let (mut active_ips, mut recent_ips) = self.active_and_recent_write().await;
let user_active = active_ips
.entry(username.to_string())
.or_insert_with(HashMap::new);
let mut recent_ips = self.recent_ips.write().await;
let user_recent = recent_ips
.entry(username.to_string())
.or_insert_with(HashMap::new);
Self::prune_recent(user_recent, now, window);
let pruned_recent_entries = Self::prune_recent(user_recent, now, window);
Self::decrement_counter(&self.recent_entry_count, pruned_recent_entries);
let recent_contains_ip = user_recent.contains_key(&ip);
if let Some(count) = user_active.get_mut(&ip) {
if !recent_contains_ip
&& self.recent_entry_count.load(Ordering::Relaxed) >= MAX_RECENT_IP_ENTRIES
{
self.recent_cap_rejects.fetch_add(1, Ordering::Relaxed);
return Err(format!(
"IP tracker recent entry cap reached: entries={}/{}",
self.recent_entry_count.load(Ordering::Relaxed),
MAX_RECENT_IP_ENTRIES
));
}
*count = count.saturating_add(1);
user_recent.insert(ip, now);
if user_recent.insert(ip, now).is_none() {
self.recent_entry_count.fetch_add(1, Ordering::Relaxed);
}
return Ok(());
}
let is_new_ip = !recent_contains_ip;
if let Some(limit) = limit {
let active_limit_reached = user_active.len() >= limit;
let recent_limit_reached = user_recent.len() >= limit;
let recent_limit_reached = user_recent.len() >= limit && is_new_ip;
let deny = match mode {
UserMaxUniqueIpsMode::ActiveWindow => active_limit_reached,
UserMaxUniqueIpsMode::TimeWindow => recent_limit_reached,
@@ -291,30 +426,62 @@ impl UserIpTracker {
}
}
user_active.insert(ip, 1);
user_recent.insert(ip, now);
if self.active_entry_count.load(Ordering::Relaxed) >= MAX_ACTIVE_IP_ENTRIES {
self.active_cap_rejects.fetch_add(1, Ordering::Relaxed);
return Err(format!(
"IP tracker active entry cap reached: entries={}/{}",
self.active_entry_count.load(Ordering::Relaxed),
MAX_ACTIVE_IP_ENTRIES
));
}
if is_new_ip && self.recent_entry_count.load(Ordering::Relaxed) >= MAX_RECENT_IP_ENTRIES {
self.recent_cap_rejects.fetch_add(1, Ordering::Relaxed);
return Err(format!(
"IP tracker recent entry cap reached: entries={}/{}",
self.recent_entry_count.load(Ordering::Relaxed),
MAX_RECENT_IP_ENTRIES
));
}
if user_active.insert(ip, 1).is_none() {
self.active_entry_count.fetch_add(1, Ordering::Relaxed);
}
if user_recent.insert(ip, now).is_none() {
self.recent_entry_count.fetch_add(1, Ordering::Relaxed);
}
Ok(())
}
pub async fn remove_ip(&self, username: &str, ip: IpAddr) {
self.maybe_compact_empty_users().await;
let mut active_ips = self.active_ips.write().await;
let mut removed_active_entries = 0usize;
if let Some(user_ips) = active_ips.get_mut(username) {
if let Some(count) = user_ips.get_mut(&ip) {
if *count > 1 {
*count -= 1;
} else {
user_ips.remove(&ip);
if user_ips.remove(&ip).is_some() {
removed_active_entries = 1;
}
}
}
if user_ips.is_empty() {
active_ips.remove(username);
}
}
Self::decrement_counter(&self.active_entry_count, removed_active_entries);
}
pub async fn get_recent_counts_for_users(&self, users: &[String]) -> HashMap<String, usize> {
self.drain_cleanup_queue().await;
self.get_recent_counts_for_users_snapshot(users).await
}
pub(crate) async fn get_recent_counts_for_users_snapshot(
&self,
users: &[String],
) -> HashMap<String, usize> {
let window = *self.limit_window.read().await;
let now = Instant::now();
let recent_ips = self.recent_ips.read().await;
@@ -389,19 +556,29 @@ impl UserIpTracker {
pub async fn get_stats(&self) -> Vec<(String, usize, usize)> {
self.drain_cleanup_queue().await;
self.get_stats_snapshot().await
}
pub(crate) async fn get_stats_snapshot(&self) -> Vec<(String, usize, usize)> {
let active_ips = self.active_ips.read().await;
let active_counts = active_ips
.iter()
.map(|(username, user_ips)| (username.clone(), user_ips.len()))
.collect::<Vec<_>>();
drop(active_ips);
let max_ips = self.max_ips.read().await;
let default_max_ips = *self.default_max_ips.read().await;
let mut stats = Vec::new();
for (username, user_ips) in active_ips.iter() {
let mut stats = Vec::with_capacity(active_counts.len());
for (username, active_count) in active_counts {
let limit = max_ips
.get(username)
.get(&username)
.copied()
.filter(|limit| *limit > 0)
.or((default_max_ips > 0).then_some(default_max_ips))
.unwrap_or(0);
stats.push((username.clone(), user_ips.len(), limit));
stats.push((username, active_count, limit));
}
stats.sort_by(|a, b| a.0.cmp(&b.0));
@@ -410,20 +587,30 @@ impl UserIpTracker {
pub async fn clear_user_ips(&self, username: &str) {
let mut active_ips = self.active_ips.write().await;
active_ips.remove(username);
let removed_active_entries = active_ips
.remove(username)
.map(|ips| ips.len())
.unwrap_or(0);
drop(active_ips);
Self::decrement_counter(&self.active_entry_count, removed_active_entries);
let mut recent_ips = self.recent_ips.write().await;
recent_ips.remove(username);
let removed_recent_entries = recent_ips
.remove(username)
.map(|ips| ips.len())
.unwrap_or(0);
Self::decrement_counter(&self.recent_entry_count, removed_recent_entries);
}
pub async fn clear_all(&self) {
let mut active_ips = self.active_ips.write().await;
active_ips.clear();
drop(active_ips);
self.active_entry_count.store(0, Ordering::Relaxed);
let mut recent_ips = self.recent_ips.write().await;
recent_ips.clear();
self.recent_entry_count.store(0, Ordering::Relaxed);
}
pub async fn is_ip_active(&self, username: &str, ip: IpAddr) -> bool {
@@ -851,4 +1038,19 @@ mod tests {
.unwrap_or(false);
assert!(!stale_exists);
}
#[tokio::test]
async fn test_time_window_allows_same_ip_reconnect() {
let tracker = UserIpTracker::new();
tracker.set_user_limit("test_user", 1).await;
tracker
.set_limit_policy(UserMaxUniqueIpsMode::TimeWindow, 1)
.await;
let ip1 = test_ipv4(10, 4, 0, 1);
assert!(tracker.check_and_add("test_user", ip1).await.is_ok());
tracker.remove_ip("test_user", ip1).await;
assert!(tracker.check_and_add("test_user", ip1).await.is_ok());
}
}
+123 -3
View File
@@ -1,6 +1,6 @@
#![allow(clippy::items_after_test_module)]
use std::path::PathBuf;
use std::path::{Path, PathBuf};
use std::time::Duration;
use tokio::sync::watch;
@@ -17,7 +17,7 @@ use crate::transport::middle_proxy::{
pub(crate) fn resolve_runtime_config_path(
config_path_cli: &str,
startup_cwd: &std::path::Path,
startup_cwd: &Path,
config_path_explicit: bool,
) -> PathBuf {
if config_path_explicit {
@@ -46,6 +46,39 @@ pub(crate) fn resolve_runtime_config_path(
startup_cwd.join("config.toml")
}
pub(crate) fn resolve_runtime_base_dir(
config_path: &Path,
startup_cwd: &Path,
config_path_explicit: bool,
data_path: Option<&Path>,
) -> PathBuf {
if let Some(path) = data_path {
return normalize_runtime_dir(path, startup_cwd);
}
if startup_cwd != Path::new("/") {
return normalize_runtime_dir(startup_cwd, startup_cwd);
}
if config_path_explicit
&& let Some(parent) = config_path.parent()
&& !parent.as_os_str().is_empty()
{
return normalize_runtime_dir(parent, startup_cwd);
}
PathBuf::from("/etc/telemt")
}
fn normalize_runtime_dir(path: &Path, startup_cwd: &Path) -> PathBuf {
let absolute = if path.is_absolute() {
path.to_path_buf()
} else {
startup_cwd.join(path)
};
absolute.canonicalize().unwrap_or(absolute)
}
/// Parsed CLI arguments.
pub(crate) struct CliArgs {
pub config_path: String,
@@ -231,9 +264,11 @@ fn print_help() {
#[cfg(test)]
mod tests {
use std::path::{Path, PathBuf};
use super::{
expected_handshake_close_description, is_expected_handshake_eof, peer_close_description,
resolve_runtime_config_path,
resolve_runtime_base_dir, resolve_runtime_config_path,
};
use crate::error::{ProxyError, StreamError};
@@ -304,6 +339,91 @@ mod tests {
let _ = std::fs::remove_dir(&startup_cwd);
}
#[test]
fn resolve_runtime_base_dir_prefers_cli_data_path() {
let nonce = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_nanos();
let startup_cwd = std::env::temp_dir().join(format!("telemt_runtime_base_cwd_{nonce}"));
let data_path = std::env::temp_dir().join(format!("telemt_runtime_base_data_{nonce}"));
std::fs::create_dir_all(&startup_cwd).unwrap();
std::fs::create_dir_all(&data_path).unwrap();
let resolved = resolve_runtime_base_dir(
&startup_cwd.join("config.toml"),
&startup_cwd,
true,
Some(&data_path),
);
assert_eq!(resolved, data_path.canonicalize().unwrap());
let _ = std::fs::remove_dir(&data_path);
let _ = std::fs::remove_dir(&startup_cwd);
}
#[test]
fn resolve_runtime_base_dir_uses_working_directory_before_explicit_config_parent() {
let nonce = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_nanos();
let startup_cwd = std::env::temp_dir().join(format!("telemt_runtime_base_start_{nonce}"));
let config_dir = std::env::temp_dir().join(format!("telemt_runtime_base_cfg_{nonce}"));
std::fs::create_dir_all(&startup_cwd).unwrap();
std::fs::create_dir_all(&config_dir).unwrap();
let resolved =
resolve_runtime_base_dir(&config_dir.join("telemt.toml"), &startup_cwd, true, None);
assert_eq!(resolved, startup_cwd.canonicalize().unwrap());
let _ = std::fs::remove_dir(&config_dir);
let _ = std::fs::remove_dir(&startup_cwd);
}
#[test]
fn resolve_runtime_base_dir_uses_explicit_config_parent_from_root() {
let nonce = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_nanos();
let config_dir = std::env::temp_dir().join(format!("telemt_runtime_base_root_cfg_{nonce}"));
std::fs::create_dir_all(&config_dir).unwrap();
let resolved =
resolve_runtime_base_dir(&config_dir.join("telemt.toml"), Path::new("/"), true, None);
assert_eq!(resolved, config_dir.canonicalize().unwrap());
let _ = std::fs::remove_dir(&config_dir);
}
#[test]
fn resolve_runtime_base_dir_uses_systemd_working_directory_before_etc() {
let nonce = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_nanos();
let startup_cwd = std::env::temp_dir().join(format!("telemt_runtime_base_systemd_{nonce}"));
std::fs::create_dir_all(&startup_cwd).unwrap();
let resolved =
resolve_runtime_base_dir(&startup_cwd.join("config.toml"), &startup_cwd, false, None);
assert_eq!(resolved, startup_cwd.canonicalize().unwrap());
let _ = std::fs::remove_dir(&startup_cwd);
}
#[test]
fn resolve_runtime_base_dir_falls_back_to_etc_from_root() {
let resolved = resolve_runtime_base_dir(
Path::new("/etc/telemt/config.toml"),
Path::new("/"),
false,
None,
);
assert_eq!(resolved, PathBuf::from("/etc/telemt"));
}
#[test]
fn expected_handshake_eof_matches_connection_reset() {
let err = ProxyError::Io(std::io::Error::from(std::io::ErrorKind::ConnectionReset));
+54 -12
View File
@@ -47,7 +47,7 @@ use crate::stats::{ReplayChecker, Stats};
use crate::stream::BufferPool;
use crate::transport::UpstreamManager;
use crate::transport::middle_proxy::MePool;
use helpers::{parse_cli, resolve_runtime_config_path};
use helpers::{parse_cli, resolve_runtime_base_dir, resolve_runtime_config_path};
#[cfg(unix)]
use crate::daemon::{DaemonOptions, PidFile, drop_privileges};
@@ -112,8 +112,51 @@ async fn run_telemt_core(
std::process::exit(1);
}
};
if let Some(ref data_path) = data_path
&& !data_path.is_absolute()
{
eprintln!(
"[telemt] data_path must be absolute: {}",
data_path.display()
);
std::process::exit(1);
}
let mut config_path =
resolve_runtime_config_path(&config_path_cli, &startup_cwd, config_path_explicit);
let runtime_base_dir = resolve_runtime_base_dir(
&config_path,
&startup_cwd,
config_path_explicit,
data_path.as_deref(),
);
if !runtime_base_dir.exists()
&& let Err(e) = std::fs::create_dir_all(&runtime_base_dir)
{
eprintln!(
"[telemt] Can't create runtime directory {}: {}",
runtime_base_dir.display(),
e
);
std::process::exit(1);
}
if !runtime_base_dir.is_dir() {
eprintln!(
"[telemt] Runtime path exists but is not a directory: {}",
runtime_base_dir.display()
);
std::process::exit(1);
}
if let Err(e) = std::env::set_current_dir(&runtime_base_dir) {
eprintln!(
"[telemt] Can't use runtime directory {}: {}",
runtime_base_dir.display(),
e
);
std::process::exit(1);
}
let mut config = match ProxyConfig::load(&config_path) {
Ok(c) => c,
@@ -156,16 +199,15 @@ async fn run_telemt_core(
);
}
} else {
let system_dir = std::path::Path::new("/etc/telemt");
let system_config_path = system_dir.join("telemt.toml");
let startup_config_path = startup_cwd.join("config.toml");
let runtime_config_path = runtime_base_dir.join("telemt.toml");
let fallback_config_path = runtime_base_dir.join("config.toml");
let mut persisted = false;
if let Some(serialized) = serialized.as_ref() {
match std::fs::create_dir_all(system_dir) {
Ok(()) => match std::fs::write(&system_config_path, serialized) {
match std::fs::create_dir_all(&runtime_base_dir) {
Ok(()) => match std::fs::write(&runtime_config_path, serialized) {
Ok(()) => {
config_path = system_config_path;
config_path = runtime_config_path;
eprintln!(
"[telemt] Created default config at {}",
config_path.display()
@@ -175,7 +217,7 @@ async fn run_telemt_core(
Err(write_error) => {
eprintln!(
"[telemt] Warning: failed to write default config at {}: {}",
system_config_path.display(),
runtime_config_path.display(),
write_error
);
}
@@ -183,16 +225,16 @@ async fn run_telemt_core(
Err(create_error) => {
eprintln!(
"[telemt] Warning: failed to create {}: {}",
system_dir.display(),
runtime_base_dir.display(),
create_error
);
}
}
if !persisted {
match std::fs::write(&startup_config_path, serialized) {
match std::fs::write(&fallback_config_path, serialized) {
Ok(()) => {
config_path = startup_config_path;
config_path = fallback_config_path;
eprintln!(
"[telemt] Created default config at {}",
config_path.display()
@@ -202,7 +244,7 @@ async fn run_telemt_core(
Err(write_error) => {
eprintln!(
"[telemt] Warning: failed to write default config at {}: {}",
startup_config_path.display(),
fallback_config_path.display(),
write_error
);
}
+35 -2
View File
@@ -10,6 +10,14 @@ use crate::tls_front::TlsFrontCache;
use crate::tls_front::fetcher::TlsFetchStrategy;
use crate::transport::UpstreamManager;
fn tls_fetch_host_for_domain(mask_host: &str, primary_tls_domain: &str, domain: &str) -> String {
if mask_host.eq_ignore_ascii_case(primary_tls_domain) {
domain.to_string()
} else {
mask_host.to_string()
}
}
pub(crate) async fn bootstrap_tls_front(
config: &ProxyConfig,
tls_domains: &[String],
@@ -56,6 +64,7 @@ pub(crate) async fn bootstrap_tls_front(
let cache_initial = cache.clone();
let domains_initial = tls_domains.to_vec();
let host_initial = mask_host.clone();
let primary_initial = config.censorship.tls_domain.clone();
let unix_sock_initial = mask_unix_sock.clone();
let scope_initial = tls_fetch_scope.clone();
let upstream_initial = upstream_manager.clone();
@@ -64,7 +73,8 @@ pub(crate) async fn bootstrap_tls_front(
let mut join = tokio::task::JoinSet::new();
for domain in domains_initial {
let cache_domain = cache_initial.clone();
let host_domain = host_initial.clone();
let host_domain =
tls_fetch_host_for_domain(&host_initial, &primary_initial, &domain);
let unix_sock_domain = unix_sock_initial.clone();
let scope_domain = scope_initial.clone();
let upstream_domain = upstream_initial.clone();
@@ -117,6 +127,7 @@ pub(crate) async fn bootstrap_tls_front(
let cache_refresh = cache.clone();
let domains_refresh = tls_domains.to_vec();
let host_refresh = mask_host.clone();
let primary_refresh = config.censorship.tls_domain.clone();
let unix_sock_refresh = mask_unix_sock.clone();
let scope_refresh = tls_fetch_scope.clone();
let upstream_refresh = upstream_manager.clone();
@@ -130,7 +141,8 @@ pub(crate) async fn bootstrap_tls_front(
let mut join = tokio::task::JoinSet::new();
for domain in domains_refresh.clone() {
let cache_domain = cache_refresh.clone();
let host_domain = host_refresh.clone();
let host_domain =
tls_fetch_host_for_domain(&host_refresh, &primary_refresh, &domain);
let unix_sock_domain = unix_sock_refresh.clone();
let scope_domain = scope_refresh.clone();
let upstream_domain = upstream_refresh.clone();
@@ -186,3 +198,24 @@ pub(crate) async fn bootstrap_tls_front(
tls_cache
}
#[cfg(test)]
mod tests {
use super::tls_fetch_host_for_domain;
#[test]
fn tls_fetch_host_uses_each_domain_when_mask_host_is_primary_default() {
assert_eq!(
tls_fetch_host_for_domain("a.com", "a.com", "b.com"),
"b.com"
);
}
#[test]
fn tls_fetch_host_preserves_explicit_non_primary_mask_host() {
assert_eq!(
tls_fetch_host_for_domain("origin.example", "a.com", "b.com"),
"origin.example"
);
}
}
+150 -13
View File
@@ -18,8 +18,13 @@ use crate::ip_tracker::UserIpTracker;
use crate::proxy::shared_state::ProxySharedState;
use crate::stats::Stats;
use crate::stats::beobachten::BeobachtenStore;
use crate::tls_front::cache;
use crate::tls_front::fetcher;
use crate::transport::{ListenOptions, create_listener};
// Keeps `/metrics` response size bounded when per-user telemetry is enabled.
const USER_LABELED_METRICS_MAX_USERS: usize = 4096;
pub async fn serve(
port: u16,
listen: Option<String>,
@@ -311,6 +316,12 @@ async fn render_metrics(
"telemt_telemetry_user_enabled {}",
if user_enabled { 1 } else { 0 }
);
let _ = writeln!(
out,
"# HELP telemt_stats_user_entries Retained per-user stats entries"
);
let _ = writeln!(out, "# TYPE telemt_stats_user_entries gauge");
let _ = writeln!(out, "telemt_stats_user_entries {}", stats.user_stats_len());
let _ = writeln!(
out,
@@ -366,6 +377,53 @@ async fn render_metrics(
stats.get_buffer_pool_in_use_gauge()
);
let _ = writeln!(
out,
"# HELP telemt_tls_fetch_profile_cache_entries Current adaptive TLS fetch profile-cache entries"
);
let _ = writeln!(out, "# TYPE telemt_tls_fetch_profile_cache_entries gauge");
let _ = writeln!(
out,
"telemt_tls_fetch_profile_cache_entries {}",
fetcher::profile_cache_entries_for_metrics()
);
let _ = writeln!(
out,
"# HELP telemt_tls_fetch_profile_cache_cap_drops_total Profile-cache winner inserts skipped because the cache cap was reached"
);
let _ = writeln!(
out,
"# TYPE telemt_tls_fetch_profile_cache_cap_drops_total counter"
);
let _ = writeln!(
out,
"telemt_tls_fetch_profile_cache_cap_drops_total {}",
fetcher::profile_cache_cap_drops_for_metrics()
);
let _ = writeln!(
out,
"# HELP telemt_tls_front_full_cert_budget_ips Current IP entries tracked by TLS full-cert budget"
);
let _ = writeln!(out, "# TYPE telemt_tls_front_full_cert_budget_ips gauge");
let _ = writeln!(
out,
"telemt_tls_front_full_cert_budget_ips {}",
cache::full_cert_sent_ips_for_metrics()
);
let _ = writeln!(
out,
"# HELP telemt_tls_front_full_cert_budget_cap_drops_total New IPs denied full-cert budget tracking because the cap was reached"
);
let _ = writeln!(
out,
"# TYPE telemt_tls_front_full_cert_budget_cap_drops_total counter"
);
let _ = writeln!(
out,
"telemt_tls_front_full_cert_budget_cap_drops_total {}",
cache::full_cert_sent_cap_drops_for_metrics()
);
let _ = writeln!(
out,
"# HELP telemt_connections_total Total accepted connections"
@@ -3019,17 +3077,6 @@ async fn render_metrics(
0
}
);
let _ = writeln!(
out,
"# HELP telemt_telemetry_user_series_suppressed User-labeled metric series suppression flag"
);
let _ = writeln!(out, "# TYPE telemt_telemetry_user_series_suppressed gauge");
let _ = writeln!(
out,
"telemt_telemetry_user_series_suppressed {}",
if user_enabled { 0 } else { 1 }
);
let ip_memory = ip_tracker.memory_stats().await;
let _ = writeln!(
out,
@@ -3071,11 +3118,46 @@ async fn render_metrics(
"telemt_ip_tracker_cleanup_queue_len {}",
ip_memory.cleanup_queue_len
);
let _ = writeln!(
out,
"# HELP telemt_ip_tracker_cleanup_total Release cleanups deferred through the cleanup queue"
);
let _ = writeln!(out, "# TYPE telemt_ip_tracker_cleanup_total counter");
let _ = writeln!(
out,
"telemt_ip_tracker_cleanup_total{{path=\"deferred\"}} {}",
ip_memory.cleanup_deferred_releases
);
let _ = writeln!(
out,
"# HELP telemt_ip_tracker_cap_rejects_total New connection rejects caused by global IP tracker caps"
);
let _ = writeln!(out, "# TYPE telemt_ip_tracker_cap_rejects_total counter");
let _ = writeln!(
out,
"telemt_ip_tracker_cap_rejects_total{{scope=\"active\"}} {}",
ip_memory.active_cap_rejects
);
let _ = writeln!(
out,
"telemt_ip_tracker_cap_rejects_total{{scope=\"recent\"}} {}",
ip_memory.recent_cap_rejects
);
let mut user_stats_emitted = 0usize;
let mut user_stats_suppressed = 0usize;
let mut unique_ip_emitted = 0usize;
let mut unique_ip_suppressed = 0usize;
if user_enabled {
for entry in stats.iter_user_stats() {
if user_stats_emitted >= USER_LABELED_METRICS_MAX_USERS {
user_stats_suppressed = user_stats_suppressed.saturating_add(1);
continue;
}
let user = entry.key();
let s = entry.value();
user_stats_emitted = user_stats_emitted.saturating_add(1);
let _ = writeln!(
out,
"telemt_user_connections_total{{user=\"{}\"}} {}",
@@ -3117,7 +3199,7 @@ async fn render_metrics(
);
}
let ip_stats = ip_tracker.get_stats().await;
let ip_stats = ip_tracker.get_stats_snapshot().await;
let ip_counts: HashMap<String, usize> = ip_stats
.into_iter()
.map(|(user, count, _)| (user, count))
@@ -3129,7 +3211,7 @@ async fn render_metrics(
unique_users.extend(ip_counts.keys().cloned());
let unique_users_vec: Vec<String> = unique_users.iter().cloned().collect();
let recent_counts = ip_tracker
.get_recent_counts_for_users(&unique_users_vec)
.get_recent_counts_for_users_snapshot(&unique_users_vec)
.await;
let _ = writeln!(
@@ -3154,6 +3236,11 @@ async fn render_metrics(
let _ = writeln!(out, "# TYPE telemt_user_unique_ips_utilization gauge");
for user in unique_users {
if unique_ip_emitted >= USER_LABELED_METRICS_MAX_USERS {
unique_ip_suppressed = unique_ip_suppressed.saturating_add(1);
continue;
}
unique_ip_emitted = unique_ip_emitted.saturating_add(1);
let current = ip_counts.get(&user).copied().unwrap_or(0);
let limit = config
.access
@@ -3193,6 +3280,46 @@ async fn render_metrics(
}
}
let _ = writeln!(
out,
"# HELP telemt_telemetry_user_series_suppressed User-labeled metric series suppression flag"
);
let _ = writeln!(out, "# TYPE telemt_telemetry_user_series_suppressed gauge");
let _ = writeln!(
out,
"telemt_telemetry_user_series_suppressed {}",
if user_enabled && user_stats_suppressed == 0 && unique_ip_suppressed == 0 {
0
} else {
1
}
);
let _ = writeln!(
out,
"# HELP telemt_telemetry_user_series_users User-labeled metric users by export status"
);
let _ = writeln!(out, "# TYPE telemt_telemetry_user_series_users gauge");
let _ = writeln!(
out,
"telemt_telemetry_user_series_users{{family=\"stats\",status=\"emitted\"}} {}",
user_stats_emitted
);
let _ = writeln!(
out,
"telemt_telemetry_user_series_users{{family=\"stats\",status=\"suppressed\"}} {}",
user_stats_suppressed
);
let _ = writeln!(
out,
"telemt_telemetry_user_series_users{{family=\"unique_ip\",status=\"emitted\"}} {}",
unique_ip_emitted
);
let _ = writeln!(
out,
"telemt_telemetry_user_series_users{{family=\"unique_ip\",status=\"suppressed\"}} {}",
unique_ip_suppressed
);
out
}
@@ -3406,9 +3533,19 @@ mod tests {
assert!(output.contains("# TYPE telemt_user_unique_ips_recent_window gauge"));
assert!(output.contains("# TYPE telemt_user_unique_ips_limit gauge"));
assert!(output.contains("# TYPE telemt_user_unique_ips_utilization gauge"));
assert!(output.contains("# TYPE telemt_stats_user_entries gauge"));
assert!(output.contains("# TYPE telemt_telemetry_user_series_users gauge"));
assert!(output.contains("# TYPE telemt_ip_tracker_users gauge"));
assert!(output.contains("# TYPE telemt_ip_tracker_entries gauge"));
assert!(output.contains("# TYPE telemt_ip_tracker_cleanup_queue_len gauge"));
assert!(output.contains("# TYPE telemt_ip_tracker_cleanup_total counter"));
assert!(output.contains("# TYPE telemt_ip_tracker_cap_rejects_total counter"));
assert!(output.contains("# TYPE telemt_tls_fetch_profile_cache_entries gauge"));
assert!(output.contains("# TYPE telemt_tls_fetch_profile_cache_cap_drops_total counter"));
assert!(output.contains("# TYPE telemt_tls_front_full_cert_budget_ips gauge"));
assert!(
output.contains("# TYPE telemt_tls_front_full_cert_budget_cap_drops_total counter")
);
}
#[tokio::test]
+19 -6
View File
@@ -31,16 +31,24 @@ struct UserConnectionReservation {
ip_tracker: Arc<UserIpTracker>,
user: String,
ip: IpAddr,
tracks_ip: bool,
active: bool,
}
impl UserConnectionReservation {
fn new(stats: Arc<Stats>, ip_tracker: Arc<UserIpTracker>, user: String, ip: IpAddr) -> Self {
fn new(
stats: Arc<Stats>,
ip_tracker: Arc<UserIpTracker>,
user: String,
ip: IpAddr,
tracks_ip: bool,
) -> Self {
Self {
stats,
ip_tracker,
user,
ip,
tracks_ip,
active: true,
}
}
@@ -49,7 +57,9 @@ impl UserConnectionReservation {
if !self.active {
return;
}
self.ip_tracker.remove_ip(&self.user, self.ip).await;
if self.tracks_ip {
self.ip_tracker.remove_ip(&self.user, self.ip).await;
}
self.active = false;
self.stats.decrement_user_curr_connects(&self.user);
}
@@ -62,7 +72,9 @@ impl Drop for UserConnectionReservation {
}
self.active = false;
self.stats.decrement_user_curr_connects(&self.user);
self.ip_tracker.enqueue_cleanup(self.user.clone(), self.ip);
if self.tracks_ip {
self.ip_tracker.enqueue_cleanup(self.user.clone(), self.ip);
}
}
}
@@ -1419,8 +1431,8 @@ impl RunningClientHandler {
/// Main dispatch after successful handshake.
/// Two modes:
/// - Direct: TCP relay to TG DC (existing behavior)
/// - Middle Proxy: RPC multiplex through ME pool (new — supports CDN DCs)
/// - Direct: TCP relay to TG DC (existing behavior)
/// - Middle Proxy: RPC multiplex through ME pool (supports CDN DCs)
#[cfg(test)]
async fn handle_authenticated_static<R, W>(
client_reader: CryptoReader<R>,
@@ -1621,6 +1633,7 @@ impl RunningClientHandler {
ip_tracker,
user.to_string(),
peer_addr.ip(),
true,
))
}
@@ -1666,7 +1679,6 @@ impl RunningClientHandler {
match ip_tracker.check_and_add(user, peer_addr.ip()).await {
Ok(()) => {
ip_tracker.remove_ip(user, peer_addr.ip()).await;
stats.decrement_user_curr_connects(user);
}
Err(reason) => {
stats.decrement_user_curr_connects(user);
@@ -1682,6 +1694,7 @@ impl RunningClientHandler {
}
}
stats.decrement_user_curr_connects(user);
Ok(())
}
}
+30 -2
View File
@@ -55,6 +55,7 @@ const STICKY_HINT_MAX_ENTRIES: usize = 65_536;
const CANDIDATE_HINT_TRACK_CAP: usize = 64;
const OVERLOAD_CANDIDATE_BUDGET_HINTED: usize = 16;
const OVERLOAD_CANDIDATE_BUDGET_UNHINTED: usize = 8;
const EXPENSIVE_INVALID_SCAN_SATURATION_THRESHOLD: usize = 64;
const RECENT_USER_RING_SCAN_LIMIT: usize = 32;
type HmacSha256 = Hmac<Sha256>;
@@ -551,6 +552,19 @@ fn auth_probe_note_saturation_in(shared: &ProxySharedState, now: Instant) {
}
}
fn auth_probe_note_expensive_invalid_scan_in(
shared: &ProxySharedState,
now: Instant,
validation_checks: usize,
overload: bool,
) {
if overload || validation_checks < EXPENSIVE_INVALID_SCAN_SATURATION_THRESHOLD {
return;
}
auth_probe_note_saturation_in(shared, now);
}
fn auth_probe_record_failure_in(shared: &ProxySharedState, peer_ip: IpAddr, now: Instant) {
let peer_ip = normalize_auth_probe_ip(peer_ip);
let state = &shared.handshake.auth_probe;
@@ -1378,7 +1392,14 @@ where
}
if !matched {
auth_probe_record_failure_in(shared, peer.ip(), Instant::now());
let failure_now = Instant::now();
auth_probe_note_expensive_invalid_scan_in(
shared,
failure_now,
validation_checks,
overload,
);
auth_probe_record_failure_in(shared, peer.ip(), failure_now);
maybe_apply_server_hello_delay(config).await;
debug!(
peer = %peer,
@@ -1753,7 +1774,14 @@ where
}
if !matched {
auth_probe_record_failure_in(shared, peer.ip(), Instant::now());
let failure_now = Instant::now();
auth_probe_note_expensive_invalid_scan_in(
shared,
failure_now,
validation_checks,
overload,
);
auth_probe_record_failure_in(shared, peer.ip(), failure_now);
maybe_apply_server_hello_delay(config).await;
debug!(
peer = %peer,
+72 -5
View File
@@ -12,7 +12,7 @@ use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{Duration, Instant};
use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
use tokio::sync::{mpsc, oneshot, watch};
use tokio::sync::{OwnedSemaphorePermit, Semaphore, mpsc, oneshot, watch};
use tokio::time::timeout;
use tracing::{debug, info, trace, warn};
@@ -36,7 +36,11 @@ use crate::stream::{BufferPool, CryptoReader, CryptoWriter, PooledBuffer};
use crate::transport::middle_proxy::{MePool, MeResponse, proto_flags_for_tag};
enum C2MeCommand {
Data { payload: PooledBuffer, flags: u32 },
Data {
payload: PooledBuffer,
flags: u32,
_permit: OwnedSemaphorePermit,
},
Close,
}
@@ -47,6 +51,8 @@ const DESYNC_ERROR_CLASS: &str = "frame_too_large_crypto_desync";
const C2ME_CHANNEL_CAPACITY_FALLBACK: usize = 128;
const C2ME_SOFT_PRESSURE_MIN_FREE_SLOTS: usize = 64;
const C2ME_SENDER_FAIRNESS_BUDGET: usize = 32;
const C2ME_QUEUED_BYTE_PERMIT_UNIT: usize = 16 * 1024;
const C2ME_QUEUED_PERMITS_PER_SLOT: usize = 4;
const RELAY_IDLE_IO_POLL_MAX: Duration = Duration::from_secs(1);
const TINY_FRAME_DEBT_PER_TINY: u32 = 8;
const TINY_FRAME_DEBT_LIMIT: u32 = 512;
@@ -571,6 +577,43 @@ fn should_yield_c2me_sender(sent_since_yield: usize, has_backlog: bool) -> bool
has_backlog && sent_since_yield >= C2ME_SENDER_FAIRNESS_BUDGET
}
fn c2me_payload_permits(payload_len: usize) -> u32 {
payload_len
.max(1)
.div_ceil(C2ME_QUEUED_BYTE_PERMIT_UNIT)
.min(u32::MAX as usize) as u32
}
fn c2me_queued_permit_budget(channel_capacity: usize, frame_limit: usize) -> usize {
channel_capacity
.saturating_mul(C2ME_QUEUED_PERMITS_PER_SLOT)
.max(c2me_payload_permits(frame_limit) as usize)
.max(1)
}
async fn acquire_c2me_payload_permit(
semaphore: &Arc<Semaphore>,
payload_len: usize,
send_timeout: Option<Duration>,
stats: &Stats,
) -> Result<OwnedSemaphorePermit> {
let permits = c2me_payload_permits(payload_len);
let acquire = semaphore.clone().acquire_many_owned(permits);
match send_timeout {
Some(send_timeout) => match timeout(send_timeout, acquire).await {
Ok(Ok(permit)) => Ok(permit),
Ok(Err(_)) => Err(ProxyError::Proxy("ME sender byte budget closed".into())),
Err(_) => {
stats.increment_me_c2me_send_timeout_total();
Err(ProxyError::Proxy("ME sender byte budget timeout".into()))
}
},
None => acquire
.await
.map_err(|_| ProxyError::Proxy("ME sender byte budget closed".into())),
}
}
fn quota_soft_cap(limit: u64, overshoot: u64) -> u64 {
limit.saturating_add(overshoot)
}
@@ -1122,13 +1165,19 @@ where
0 => None,
timeout_ms => Some(Duration::from_millis(timeout_ms)),
};
let c2me_byte_budget = c2me_queued_permit_budget(c2me_channel_capacity, frame_limit);
let c2me_byte_semaphore = Arc::new(Semaphore::new(c2me_byte_budget));
let (c2me_tx, mut c2me_rx) = mpsc::channel::<C2MeCommand>(c2me_channel_capacity);
let me_pool_c2me = me_pool.clone();
let c2me_sender = tokio::spawn(async move {
let mut sent_since_yield = 0usize;
while let Some(cmd) = c2me_rx.recv().await {
match cmd {
C2MeCommand::Data { payload, flags } => {
C2MeCommand::Data {
payload,
flags,
_permit,
} => {
me_pool_c2me
.send_proxy_req(
conn_id,
@@ -1624,11 +1673,29 @@ where
if payload.len() >= 8 && payload[..8].iter().all(|b| *b == 0) {
flags |= RPC_FLAG_NOT_ENCRYPTED;
}
let payload_permit = match acquire_c2me_payload_permit(
&c2me_byte_semaphore,
payload.len(),
c2me_send_timeout,
stats.as_ref(),
)
.await
{
Ok(permit) => permit,
Err(e) => {
main_result = Err(e);
break;
}
};
// Keep client read loop lightweight: route heavy ME send path via a dedicated task.
if enqueue_c2me_command_in(
shared.as_ref(),
&c2me_tx,
C2MeCommand::Data { payload, flags },
C2MeCommand::Data {
payload,
flags,
_permit: payload_permit,
},
c2me_send_timeout,
stats.as_ref(),
)
@@ -2262,7 +2329,7 @@ where
W: AsyncWrite + Unpin + Send + 'static,
{
match response {
MeResponse::Data { flags, data } => {
MeResponse::Data { flags, data, .. } => {
if batched {
trace!(conn_id, bytes = data.len(), flags, "ME->C data (batched)");
} else {
+47 -1
View File
@@ -282,7 +282,7 @@ async fn user_connection_reservation_drop_enqueues_cleanup_synchronously() {
assert_eq!(stats.get_user_curr_connects(&user), 1);
let reservation =
UserConnectionReservation::new(stats.clone(), ip_tracker.clone(), user.clone(), ip);
UserConnectionReservation::new(stats.clone(), ip_tracker.clone(), user.clone(), ip, true);
// Drop the reservation synchronously without any tokio::spawn/await yielding!
drop(reservation);
@@ -320,6 +320,7 @@ async fn relay_task_abort_releases_user_gate_and_ip_reservation() {
let stats = Arc::new(Stats::new());
let ip_tracker = Arc::new(UserIpTracker::new());
ip_tracker.set_user_limit(user, 8).await;
let mut cfg = ProxyConfig::default();
cfg.access.user_max_tcp_conns.insert(user.to_string(), 8);
@@ -437,6 +438,7 @@ async fn relay_cutover_releases_user_gate_and_ip_reservation() {
let stats = Arc::new(Stats::new());
let ip_tracker = Arc::new(UserIpTracker::new());
ip_tracker.set_user_limit(user, 8).await;
let mut cfg = ProxyConfig::default();
cfg.access.user_max_tcp_conns.insert(user.to_string(), 8);
@@ -958,6 +960,36 @@ async fn reservation_limit_failure_does_not_leak_curr_connects_counter() {
assert_eq!(ip_tracker.get_active_ip_count(user).await, 0);
}
#[tokio::test]
async fn unlimited_unique_ip_user_is_still_visible_in_active_ip_tracker() {
let user = "active-ip-observed-user";
let config = crate::config::ProxyConfig::default();
let stats = Arc::new(crate::stats::Stats::new());
let ip_tracker = Arc::new(crate::ip_tracker::UserIpTracker::new());
let peer = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(198, 51, 200, 17)), 50017);
let reservation = RunningClientHandler::acquire_user_connection_reservation_static(
user,
&config,
stats.clone(),
peer,
ip_tracker.clone(),
)
.await
.expect("reservation without unique-IP limit must succeed");
assert_eq!(stats.get_user_curr_connects(user), 1);
assert_eq!(
ip_tracker.get_active_ip_count(user).await,
1,
"active IP observability must not depend on unique-IP limit enforcement"
);
reservation.release().await;
assert_eq!(stats.get_user_curr_connects(user), 0);
assert_eq!(ip_tracker.get_active_ip_count(user).await, 0);
}
#[tokio::test]
async fn short_tls_probe_is_masked_through_client_pipeline() {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
@@ -2879,6 +2911,7 @@ async fn explicit_reservation_release_cleans_user_and_ip_immediately() {
let stats = Arc::new(Stats::new());
let ip_tracker = Arc::new(UserIpTracker::new());
ip_tracker.set_user_limit(user, 4).await;
let reservation = RunningClientHandler::acquire_user_connection_reservation_static(
user,
@@ -2917,6 +2950,7 @@ async fn explicit_reservation_release_does_not_double_decrement_on_drop() {
let stats = Arc::new(Stats::new());
let ip_tracker = Arc::new(UserIpTracker::new());
ip_tracker.set_user_limit(user, 4).await;
let reservation = RunningClientHandler::acquire_user_connection_reservation_static(
user,
@@ -2947,6 +2981,7 @@ async fn drop_fallback_eventually_cleans_user_and_ip_reservation() {
let stats = Arc::new(Stats::new());
let ip_tracker = Arc::new(UserIpTracker::new());
ip_tracker.set_user_limit(user, 1).await;
let reservation = RunningClientHandler::acquire_user_connection_reservation_static(
user,
@@ -3029,6 +3064,7 @@ async fn release_abort_storm_does_not_leak_user_or_ip_reservations() {
let stats = Arc::new(Stats::new());
let ip_tracker = Arc::new(UserIpTracker::new());
ip_tracker.set_user_limit(user, ATTEMPTS + 16).await;
for idx in 0..ATTEMPTS {
let peer = SocketAddr::new(
@@ -3079,6 +3115,7 @@ async fn release_abort_loop_preserves_immediate_same_ip_reacquire() {
let stats = Arc::new(Stats::new());
let ip_tracker = Arc::new(UserIpTracker::new());
ip_tracker.set_user_limit(user, 1).await;
for _ in 0..ITERATIONS {
let reservation = RunningClientHandler::acquire_user_connection_reservation_static(
@@ -3137,6 +3174,7 @@ async fn adversarial_mixed_release_drop_abort_wave_converges_to_zero() {
let stats = Arc::new(Stats::new());
let ip_tracker = Arc::new(UserIpTracker::new());
ip_tracker.set_user_limit(user, RESERVATIONS + 8).await;
let mut reservations = Vec::with_capacity(RESERVATIONS);
for idx in 0..RESERVATIONS {
@@ -3217,6 +3255,8 @@ async fn parallel_users_abort_release_isolation_preserves_independent_cleanup()
let stats = Arc::new(Stats::new());
let ip_tracker = Arc::new(UserIpTracker::new());
ip_tracker.set_user_limit(user_a, 64).await;
ip_tracker.set_user_limit(user_b, 64).await;
let mut tasks = tokio::task::JoinSet::new();
for idx in 0..64usize {
@@ -3278,6 +3318,7 @@ async fn concurrent_release_storm_leaves_zero_user_and_ip_footprint() {
let stats = Arc::new(Stats::new());
let ip_tracker = Arc::new(UserIpTracker::new());
ip_tracker.set_user_limit(user, RESERVATIONS + 8).await;
let mut reservations = Vec::with_capacity(RESERVATIONS);
for idx in 0..RESERVATIONS {
@@ -3332,6 +3373,7 @@ async fn relay_connect_error_releases_user_and_ip_before_return() {
let stats = Arc::new(Stats::new());
let ip_tracker = Arc::new(UserIpTracker::new());
ip_tracker.set_user_limit(user, 8).await;
let mut config = ProxyConfig::default();
config.access.user_max_tcp_conns.insert(user.to_string(), 1);
@@ -3427,6 +3469,7 @@ async fn mixed_release_and_drop_same_ip_preserves_counter_correctness() {
let stats = Arc::new(Stats::new());
let ip_tracker = Arc::new(UserIpTracker::new());
ip_tracker.set_user_limit(user, 1).await;
let reservation_a = RunningClientHandler::acquire_user_connection_reservation_static(
user,
@@ -3487,6 +3530,7 @@ async fn drop_one_of_two_same_ip_reservations_keeps_ip_active() {
let stats = Arc::new(Stats::new());
let ip_tracker = Arc::new(UserIpTracker::new());
ip_tracker.set_user_limit(user, 1).await;
let reservation_a = RunningClientHandler::acquire_user_connection_reservation_static(
user,
@@ -3696,6 +3740,7 @@ async fn cross_thread_drop_uses_captured_runtime_for_ip_cleanup() {
let stats = Arc::new(Stats::new());
let ip_tracker = Arc::new(UserIpTracker::new());
ip_tracker.set_user_limit(user, 8).await;
let reservation = RunningClientHandler::acquire_user_connection_reservation_static(
user,
@@ -3740,6 +3785,7 @@ async fn immediate_reacquire_after_cross_thread_drop_succeeds() {
let stats = Arc::new(Stats::new());
let ip_tracker = Arc::new(UserIpTracker::new());
ip_tracker.set_user_limit(user, 1).await;
let reservation = RunningClientHandler::acquire_user_connection_reservation_static(
user,
@@ -669,6 +669,13 @@ fn adversarial_check_then_symlink_flip_is_blocked_by_nofollow_open() {
"telemt-unknown-dc-check-open-race-{}",
std::process::id()
));
if let Ok(meta) = fs::symlink_metadata(&parent) {
if meta.file_type().is_symlink() || meta.is_file() {
fs::remove_file(&parent).expect("stale check-open-race path must be removable");
} else {
fs::remove_dir_all(&parent).expect("stale check-open-race parent must be removable");
}
}
fs::create_dir_all(&parent).expect("check-open-race parent must be creatable");
let target = parent.join("unknown-dc.log");
@@ -1252,6 +1252,97 @@ async fn tls_overload_budget_limits_candidate_scan_depth() {
);
}
#[tokio::test]
async fn tls_expensive_invalid_scan_activates_saturation_budget() {
let mut config = ProxyConfig::default();
config.access.users.clear();
config.access.ignore_time_skew = true;
for idx in 0..80u8 {
config.access.users.insert(
format!("user-{idx}"),
format!("{:032x}", u128::from(idx) + 1),
);
}
config.rebuild_runtime_user_auth().unwrap();
let replay_checker = ReplayChecker::new(128, Duration::from_secs(60));
let rng = SecureRandom::new();
let shared = ProxySharedState::new();
let attacker_secret = [0xEFu8; 16];
let handshake = make_valid_tls_handshake(&attacker_secret, 0);
let first_peer: SocketAddr = "198.51.100.214:44326".parse().unwrap();
let first = handle_tls_handshake_with_shared(
&handshake,
tokio::io::empty(),
tokio::io::sink(),
first_peer,
&config,
&replay_checker,
&rng,
None,
shared.as_ref(),
)
.await;
assert!(matches!(first, HandshakeResult::BadClient { .. }));
assert!(
auth_probe_saturation_state_for_testing_in_shared(shared.as_ref())
.lock()
.unwrap()
.is_some(),
"expensive invalid scan must activate global saturation"
);
assert_eq!(
shared
.handshake
.auth_expensive_checks_total
.load(Ordering::Relaxed),
80,
"first invalid probe preserves full first-hit compatibility before enabling saturation"
);
{
let mut saturation = auth_probe_saturation_state_for_testing_in_shared(shared.as_ref())
.lock()
.unwrap();
let state = saturation.as_mut().expect("saturation must be present");
state.blocked_until = Instant::now() + Duration::from_millis(200);
}
let second_peer: SocketAddr = "198.51.100.215:44326".parse().unwrap();
let second = handle_tls_handshake_with_shared(
&handshake,
tokio::io::empty(),
tokio::io::sink(),
second_peer,
&config,
&replay_checker,
&rng,
None,
shared.as_ref(),
)
.await;
assert!(matches!(second, HandshakeResult::BadClient { .. }));
assert_eq!(
shared
.handshake
.auth_budget_exhausted_total
.load(Ordering::Relaxed),
1,
"second invalid probe must be capped by overload budget"
);
assert_eq!(
shared
.handshake
.auth_expensive_checks_total
.load(Ordering::Relaxed),
80 + OVERLOAD_CANDIDATE_BUDGET_UNHINTED as u64,
"saturation budget must bound follow-up invalid scans"
);
}
#[tokio::test]
async fn mtproto_runtime_snapshot_prefers_preferred_user_hint() {
let mut config = ProxyConfig::default();
@@ -70,6 +70,7 @@ async fn me_writer_write_fail_keeps_reserved_quota_and_tracks_fail_metrics() {
MeResponse::Data {
flags: 0,
data: payload.clone(),
route_permit: None,
},
&mut writer,
ProtoTag::Intermediate,
@@ -139,6 +140,7 @@ async fn me_writer_pre_write_quota_reject_happens_before_writer_poll() {
MeResponse::Data {
flags: 0,
data: Bytes::from_static(&[0xAA, 0xBB, 0xCC]),
route_permit: None,
},
&mut writer,
ProtoTag::Intermediate,
@@ -12,6 +12,12 @@ fn make_pooled_payload(data: &[u8]) -> PooledBuffer {
payload
}
fn make_c2me_permit() -> tokio::sync::OwnedSemaphorePermit {
Arc::new(tokio::sync::Semaphore::new(1))
.try_acquire_many_owned(1)
.expect("test permit must be available")
}
#[test]
#[ignore = "Tracking for M-04: Verify should_emit_full_desync returns true on first occurrence and false on duplicate within window"]
fn should_emit_full_desync_filters_duplicates() {
@@ -107,6 +113,7 @@ async fn c2me_channel_full_path_yields_then_sends() {
tx.send(C2MeCommand::Data {
payload: make_pooled_payload(&[0xAA]),
flags: 1,
_permit: make_c2me_permit(),
})
.await
.expect("priming queue with one frame must succeed");
@@ -119,6 +126,7 @@ async fn c2me_channel_full_path_yields_then_sends() {
C2MeCommand::Data {
payload: make_pooled_payload(&[0xBB, 0xCC]),
flags: 2,
_permit: make_c2me_permit(),
},
None,
&stats,
@@ -138,7 +146,7 @@ async fn c2me_channel_full_path_yields_then_sends() {
.expect("receiver should observe primed frame")
.expect("first queued command must exist");
match first {
C2MeCommand::Data { payload, flags } => {
C2MeCommand::Data { payload, flags, .. } => {
assert_eq!(payload.as_ref(), &[0xAA]);
assert_eq!(flags, 1);
}
@@ -155,7 +163,7 @@ async fn c2me_channel_full_path_yields_then_sends() {
.expect("receiver should observe backpressure-resumed frame")
.expect("second queued command must exist");
match second {
C2MeCommand::Data { payload, flags } => {
C2MeCommand::Data { payload, flags, .. } => {
assert_eq!(payload.as_ref(), &[0xBB, 0xCC]);
assert_eq!(flags, 2);
}
+31 -14
View File
@@ -7,6 +7,7 @@ use std::time::{Duration, Instant};
use parking_lot::Mutex;
const CLEANUP_INTERVAL: Duration = Duration::from_secs(30);
const MAX_BEOBACHTEN_ENTRIES: usize = 65_536;
#[derive(Default)]
struct BeobachtenInner {
@@ -48,12 +49,23 @@ impl BeobachtenStore {
Self::cleanup_if_needed(&mut guard, now, ttl);
let key = (class.to_string(), ip);
let entry = guard.entries.entry(key).or_insert(BeobachtenEntry {
tries: 0,
last_seen: now,
});
entry.tries = entry.tries.saturating_add(1);
entry.last_seen = now;
if let Some(entry) = guard.entries.get_mut(&key) {
entry.tries = entry.tries.saturating_add(1);
entry.last_seen = now;
return;
}
if guard.entries.len() >= MAX_BEOBACHTEN_ENTRIES {
return;
}
guard.entries.insert(
key,
BeobachtenEntry {
tries: 1,
last_seen: now,
},
);
}
pub fn snapshot_text(&self, ttl: Duration) -> String {
@@ -62,16 +74,21 @@ impl BeobachtenStore {
}
let now = Instant::now();
let mut guard = self.inner.lock();
Self::cleanup(&mut guard, now, ttl);
guard.last_cleanup = Some(now);
let entries = {
let mut guard = self.inner.lock();
Self::cleanup(&mut guard, now, ttl);
guard.last_cleanup = Some(now);
guard
.entries
.iter()
.map(|((class, ip), entry)| (class.clone(), *ip, entry.tries))
.collect::<Vec<_>>()
};
let mut grouped = BTreeMap::<String, Vec<(IpAddr, u64)>>::new();
for ((class, ip), entry) in &guard.entries {
grouped
.entry(class.clone())
.or_default()
.push((*ip, entry.tries));
for (class, ip, tries) in entries {
grouped.entry(class).or_default().push((ip, tries));
}
if grouped.is_empty() {
+5
View File
@@ -2477,6 +2477,11 @@ impl Stats {
self.user_stats.iter()
}
/// Current number of retained per-user stats entries.
pub fn user_stats_len(&self) -> usize {
self.user_stats.len()
}
pub fn uptime_secs(&self) -> f64 {
self.start_time
.read()
+34 -24
View File
@@ -277,6 +277,7 @@ impl StreamState for TlsReaderState {
pub struct FakeTlsReader<R> {
upstream: R,
state: TlsReaderState,
body_scratch: Vec<u8>,
}
impl<R> FakeTlsReader<R> {
@@ -284,6 +285,7 @@ impl<R> FakeTlsReader<R> {
Self {
upstream,
state: TlsReaderState::Idle,
body_scratch: Vec::new(),
}
}
@@ -439,7 +441,13 @@ impl<R: AsyncRead + Unpin> AsyncRead for FakeTlsReader<R> {
length,
mut buffer,
} => {
let result = poll_read_body(&mut this.upstream, cx, &mut buffer, length);
let result = poll_read_body(
&mut this.upstream,
cx,
&mut buffer,
length,
&mut this.body_scratch,
);
match result {
BodyPollResult::Pending => {
@@ -558,34 +566,36 @@ fn poll_read_body<R: AsyncRead + Unpin>(
cx: &mut Context<'_>,
buffer: &mut BytesMut,
target_len: usize,
scratch: &mut Vec<u8>,
) -> BodyPollResult {
// NOTE: This implementation uses a temporary Vec to avoid tricky borrow/lifetime
// issues with BytesMut spare capacity and ReadBuf across polls.
// It's safe and correct; optimization is possible if needed.
while buffer.len() < target_len {
let remaining = target_len - buffer.len();
let chunk_len = remaining.min(8192);
let mut temp = vec![0u8; remaining.min(8192)];
let mut read_buf = ReadBuf::new(&mut temp);
match Pin::new(&mut *upstream).poll_read(cx, &mut read_buf) {
Poll::Pending => return BodyPollResult::Pending,
Poll::Ready(Err(e)) => return BodyPollResult::Error(e),
Poll::Ready(Ok(())) => {
let n = read_buf.filled().len();
if n == 0 {
return BodyPollResult::Error(Error::new(
ErrorKind::UnexpectedEof,
format!(
"unexpected EOF in TLS body (got {} of {} bytes)",
buffer.len(),
target_len
),
));
}
buffer.extend_from_slice(&temp[..n]);
}
if scratch.len() < chunk_len {
scratch.resize(chunk_len, 0);
}
let n = {
let mut read_buf = ReadBuf::new(&mut scratch[..chunk_len]);
match Pin::new(&mut *upstream).poll_read(cx, &mut read_buf) {
Poll::Pending => return BodyPollResult::Pending,
Poll::Ready(Err(e)) => return BodyPollResult::Error(e),
Poll::Ready(Ok(())) => read_buf.filled().len(),
}
};
if n == 0 {
return BodyPollResult::Error(Error::new(
ErrorKind::UnexpectedEof,
format!(
"unexpected EOF in TLS body (got {} of {} bytes)",
buffer.len(),
target_len
),
));
}
buffer.extend_from_slice(&scratch[..n]);
}
BodyPollResult::Complete(buffer.split().freeze())
+31 -16
View File
@@ -559,9 +559,7 @@ async fn mass_reconnect_sync_cleanup_prevents_temporary_reservation_bloat() {
}
#[tokio::test]
async fn adversarial_drain_cleanup_queue_race_does_not_cause_false_rejections() {
// Regression guard: concurrent cleanup draining must not produce false
// limit denials for a new IP when the previous IP is already queued.
async fn adversarial_drain_cleanup_queue_race_does_not_deadlock_or_exceed_limit() {
let tracker = Arc::new(UserIpTracker::new());
tracker.set_user_limit("racer", 1).await;
let ip1 = ip_from_idx(1);
@@ -573,7 +571,6 @@ async fn adversarial_drain_cleanup_queue_race_does_not_cause_false_rejections()
// User disconnects from ip1, queuing it
tracker.enqueue_cleanup("racer".to_string(), ip1);
let mut saw_false_rejection = false;
for _ in 0..100 {
// Queue cleanup then race explicit drain and check-and-add on the alternative IP.
tracker.enqueue_cleanup("racer".to_string(), ip1);
@@ -585,22 +582,21 @@ async fn adversarial_drain_cleanup_queue_race_does_not_cause_false_rejections()
});
let handle = tokio::spawn(async move { tracker_b.check_and_add("racer", ip2).await });
drain_handle.await.unwrap();
let res = handle.await.unwrap();
if res.is_err() {
saw_false_rejection = true;
break;
}
tokio::time::timeout(Duration::from_secs(1), drain_handle)
.await
.expect("cleanup drain must not deadlock")
.unwrap();
let _ = tokio::time::timeout(Duration::from_secs(1), handle)
.await
.expect("admission must not deadlock")
.unwrap();
// Restore baseline for next iteration.
assert!(tracker.get_active_ip_count("racer").await <= 1);
tracker.drain_cleanup_queue().await;
tracker.remove_ip("racer", ip2).await;
tracker.remove_ip("racer", ip1).await;
tracker.check_and_add("racer", ip1).await.unwrap();
}
assert!(
!saw_false_rejection,
"Concurrent cleanup draining must not cause false-positive IP denials"
);
}
#[tokio::test]
@@ -649,6 +645,25 @@ async fn duplicate_cleanup_entries_do_not_break_future_admission() {
);
}
#[tokio::test]
async fn duplicate_cleanup_entries_are_coalesced_until_drain() {
let tracker = UserIpTracker::new();
let ip = ip_from_idx(7150);
tracker.enqueue_cleanup("coalesced-cleanup".to_string(), ip);
tracker.enqueue_cleanup("coalesced-cleanup".to_string(), ip);
tracker.enqueue_cleanup("coalesced-cleanup".to_string(), ip);
assert_eq!(
tracker.cleanup_queue_len_for_tests(),
1,
"duplicate queued cleanup entries must retain one allocation slot"
);
tracker.drain_cleanup_queue().await;
assert_eq!(tracker.cleanup_queue_len_for_tests(), 0);
}
#[tokio::test]
async fn stress_repeated_queue_poison_recovery_preserves_admission_progress() {
let tracker = UserIpTracker::new();
+285 -13
View File
@@ -1,8 +1,11 @@
use std::collections::HashMap;
use std::collections::hash_map::DefaultHasher;
use std::hash::{Hash, Hasher};
use std::net::IpAddr;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::{Duration, Instant, SystemTime};
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
use tokio::sync::RwLock;
use tokio::time::sleep;
@@ -12,12 +15,30 @@ use crate::tls_front::types::{
CachedTlsData, ParsedServerHello, TlsBehaviorProfile, TlsFetchResult,
};
const FULL_CERT_SENT_SWEEP_INTERVAL_SECS: u64 = 30;
const FULL_CERT_SENT_MAX_IPS: usize = 65_536;
const FULL_CERT_SENT_SHARDS: usize = 64;
static FULL_CERT_SENT_IPS_GAUGE: AtomicU64 = AtomicU64::new(0);
static FULL_CERT_SENT_CAP_DROPS: AtomicU64 = AtomicU64::new(0);
/// Current number of IPs tracked by the TLS full-cert budget gate.
pub(crate) fn full_cert_sent_ips_for_metrics() -> u64 {
FULL_CERT_SENT_IPS_GAUGE.load(Ordering::Relaxed)
}
/// Number of new IPs denied a full-cert budget slot because the cap was reached.
pub(crate) fn full_cert_sent_cap_drops_for_metrics() -> u64 {
FULL_CERT_SENT_CAP_DROPS.load(Ordering::Relaxed)
}
/// Lightweight in-memory + optional on-disk cache for TLS fronting data.
#[derive(Debug)]
pub struct TlsFrontCache {
memory: RwLock<HashMap<String, Arc<CachedTlsData>>>,
default: Arc<CachedTlsData>,
full_cert_sent: RwLock<HashMap<IpAddr, Instant>>,
full_cert_sent_shards: Vec<RwLock<HashMap<IpAddr, Instant>>>,
full_cert_sent_last_sweep_epoch_secs: AtomicU64,
disk_path: PathBuf,
}
@@ -52,7 +73,10 @@ impl TlsFrontCache {
Self {
memory: RwLock::new(map),
default,
full_cert_sent: RwLock::new(HashMap::new()),
full_cert_sent_shards: (0..FULL_CERT_SENT_SHARDS)
.map(|_| RwLock::new(HashMap::new()))
.collect(),
full_cert_sent_last_sweep_epoch_secs: AtomicU64::new(0),
disk_path: disk_path.as_ref().to_path_buf(),
}
}
@@ -69,22 +93,83 @@ impl TlsFrontCache {
self.memory.read().await.contains_key(domain)
}
fn full_cert_sent_shard_index(client_ip: IpAddr) -> usize {
let mut hasher = DefaultHasher::new();
client_ip.hash(&mut hasher);
(hasher.finish() as usize) % FULL_CERT_SENT_SHARDS
}
fn full_cert_sent_shard(&self, client_ip: IpAddr) -> &RwLock<HashMap<IpAddr, Instant>> {
&self.full_cert_sent_shards[Self::full_cert_sent_shard_index(client_ip)]
}
fn decrement_full_cert_sent_entries(amount: usize) {
if amount == 0 {
return;
}
let amount = amount as u64;
let _ =
FULL_CERT_SENT_IPS_GAUGE.fetch_update(Ordering::AcqRel, Ordering::Relaxed, |current| {
Some(current.saturating_sub(amount))
});
}
fn try_reserve_full_cert_sent_entry() -> bool {
let mut current = FULL_CERT_SENT_IPS_GAUGE.load(Ordering::Relaxed);
loop {
if current >= FULL_CERT_SENT_MAX_IPS as u64 {
return false;
}
match FULL_CERT_SENT_IPS_GAUGE.compare_exchange_weak(
current,
current.saturating_add(1),
Ordering::AcqRel,
Ordering::Relaxed,
) {
Ok(_) => return true,
Err(actual) => current = actual,
}
}
}
async fn sweep_full_cert_sent_shards(&self, now: Instant, ttl: Duration) {
for shard in &self.full_cert_sent_shards {
let mut guard = shard.write().await;
let before = guard.len();
guard.retain(|_, seen_at| now.duration_since(*seen_at) < ttl);
Self::decrement_full_cert_sent_entries(before.saturating_sub(guard.len()));
}
}
/// Returns true when full cert payload should be sent for client_ip
/// according to TTL policy.
pub async fn take_full_cert_budget_for_ip(&self, client_ip: IpAddr, ttl: Duration) -> bool {
if ttl.is_zero() {
self.full_cert_sent
.write()
.await
.insert(client_ip, Instant::now());
return true;
}
let now = Instant::now();
let mut guard = self.full_cert_sent.write().await;
guard.retain(|_, seen_at| now.duration_since(*seen_at) < ttl);
let now_epoch_secs = SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_secs();
let should_sweep = self
.full_cert_sent_last_sweep_epoch_secs
.fetch_update(Ordering::AcqRel, Ordering::Relaxed, |last_sweep| {
if now_epoch_secs.saturating_sub(last_sweep) >= FULL_CERT_SENT_SWEEP_INTERVAL_SECS {
Some(now_epoch_secs)
} else {
None
}
})
.is_ok();
match guard.get_mut(&client_ip) {
if should_sweep {
self.sweep_full_cert_sent_shards(now, ttl).await;
}
let mut guard = self.full_cert_sent_shard(client_ip).write().await;
let allowed = match guard.get_mut(&client_ip) {
Some(seen_at) => {
if now.duration_since(*seen_at) >= ttl {
*seen_at = now;
@@ -94,12 +179,43 @@ impl TlsFrontCache {
}
}
None => {
if !Self::try_reserve_full_cert_sent_entry() {
FULL_CERT_SENT_CAP_DROPS.fetch_add(1, Ordering::Relaxed);
return false;
}
guard.insert(client_ip, now);
true
}
};
allowed
}
#[cfg(test)]
async fn insert_full_cert_sent_for_tests(&self, client_ip: IpAddr, seen_at: Instant) {
let mut guard = self.full_cert_sent_shard(client_ip).write().await;
if guard.insert(client_ip, seen_at).is_none() {
FULL_CERT_SENT_IPS_GAUGE.fetch_add(1, Ordering::Relaxed);
}
}
#[cfg(test)]
async fn full_cert_sent_is_empty_for_tests(&self) -> bool {
for shard in &self.full_cert_sent_shards {
if !shard.read().await.is_empty() {
return false;
}
}
true
}
#[cfg(test)]
async fn full_cert_sent_contains_for_tests(&self, client_ip: IpAddr) -> bool {
self.full_cert_sent_shard(client_ip)
.read()
.await
.contains_key(&client_ip)
}
pub async fn set(&self, domain: &str, data: CachedTlsData) {
let mut guard = self.memory.write().await;
guard.insert(domain.to_string(), Arc::new(data));
@@ -130,6 +246,14 @@ impl TlsFrontCache {
warn!(file = %name, "Skipping TLS cache entry with invalid domain");
continue;
}
if !cert_info_matches_domain(&cached) {
warn!(
file = %name,
domain = %cached.domain,
"Skipping TLS cache entry with mismatched certificate metadata"
);
continue;
}
// fetched_at is skipped during deserialization; approximate with file mtime if available.
if let Ok(meta) = entry.metadata().await
&& let Ok(modified) = meta.modified()
@@ -209,10 +333,100 @@ impl TlsFrontCache {
}
}
fn cert_info_matches_domain(cached: &CachedTlsData) -> bool {
let Some(cert_info) = cached.cert_info.as_ref() else {
return true;
};
if !cert_info.san_names.is_empty() {
return cert_info
.san_names
.iter()
.any(|name| dns_name_matches_domain(name, &cached.domain));
}
cert_info
.subject_cn
.as_deref()
.map_or(true, |name| dns_name_matches_domain(name, &cached.domain))
}
fn dns_name_matches_domain(pattern: &str, domain: &str) -> bool {
let pattern = normalize_dns_name(pattern);
let domain = normalize_dns_name(domain);
if pattern == domain {
return true;
}
let Some(suffix) = pattern.strip_prefix("*.") else {
return false;
};
let Some(prefix) = domain.strip_suffix(suffix) else {
return false;
};
prefix.ends_with('.') && !prefix[..prefix.len() - 1].contains('.')
}
fn normalize_dns_name(value: &str) -> String {
value.trim().trim_end_matches('.').to_ascii_lowercase()
}
#[cfg(test)]
mod tests {
use super::*;
fn cached_with_cert_info(
domain: &str,
subject_cn: Option<&str>,
san_names: Vec<&str>,
) -> CachedTlsData {
CachedTlsData {
server_hello_template: ParsedServerHello {
version: [0x03, 0x03],
random: [0u8; 32],
session_id: Vec::new(),
cipher_suite: [0x13, 0x01],
compression: 0,
extensions: Vec::new(),
},
cert_info: Some(crate::tls_front::types::ParsedCertificateInfo {
not_after_unix: None,
not_before_unix: None,
issuer_cn: None,
subject_cn: subject_cn.map(str::to_string),
san_names: san_names.into_iter().map(str::to_string).collect(),
}),
cert_payload: None,
app_data_records_sizes: vec![1024],
total_app_data_len: 1024,
behavior_profile: TlsBehaviorProfile::default(),
fetched_at: SystemTime::now(),
domain: domain.to_string(),
}
}
#[test]
fn cert_info_domain_match_accepts_exact_san() {
let cached = cached_with_cert_info("b.com", Some("a.com"), vec!["b.com"]);
assert!(cert_info_matches_domain(&cached));
}
#[test]
fn cert_info_domain_match_rejects_wrong_san() {
let cached = cached_with_cert_info("b.com", Some("b.com"), vec!["a.com"]);
assert!(!cert_info_matches_domain(&cached));
}
#[test]
fn cert_info_domain_match_accepts_single_label_wildcard_san() {
let cached = cached_with_cert_info("api.b.com", None, vec!["*.b.com"]);
assert!(cert_info_matches_domain(&cached));
}
#[test]
fn cert_info_domain_match_rejects_multi_label_wildcard_san() {
let cached = cached_with_cert_info("deep.api.b.com", None, vec!["*.b.com"]);
assert!(!cert_info_matches_domain(&cached));
}
#[tokio::test]
async fn test_take_full_cert_budget_for_ip_uses_ttl() {
let cache = TlsFrontCache::new(&["example.com".to_string()], 1024, "tlsfront-test-cache");
@@ -230,10 +444,68 @@ mod tests {
#[tokio::test]
async fn test_take_full_cert_budget_for_ip_zero_ttl_always_allows_full_payload() {
let cache = TlsFrontCache::new(&["example.com".to_string()], 1024, "tlsfront-test-cache");
let ip: IpAddr = "127.0.0.1".parse().expect("ip");
let ttl = Duration::ZERO;
assert!(cache.take_full_cert_budget_for_ip(ip, ttl).await);
assert!(cache.take_full_cert_budget_for_ip(ip, ttl).await);
for idx in 0..100_000u32 {
let ip = IpAddr::V4(std::net::Ipv4Addr::new(
10,
((idx >> 16) & 0xff) as u8,
((idx >> 8) & 0xff) as u8,
(idx & 0xff) as u8,
));
assert!(cache.take_full_cert_budget_for_ip(ip, ttl).await);
}
assert!(cache.full_cert_sent_is_empty_for_tests().await);
}
#[tokio::test]
async fn test_take_full_cert_budget_for_ip_sweeps_expired_entries_when_due() {
let cache = TlsFrontCache::new(&["example.com".to_string()], 1024, "tlsfront-test-cache");
let stale_ip: IpAddr = "127.0.0.1".parse().expect("ip");
let new_ip: IpAddr = "127.0.0.2".parse().expect("ip");
let ttl = Duration::from_secs(1);
let stale_seen_at = Instant::now()
.checked_sub(Duration::from_secs(10))
.unwrap_or_else(Instant::now);
cache
.insert_full_cert_sent_for_tests(stale_ip, stale_seen_at)
.await;
cache
.full_cert_sent_last_sweep_epoch_secs
.store(0, Ordering::Relaxed);
assert!(cache.take_full_cert_budget_for_ip(new_ip, ttl).await);
assert!(!cache.full_cert_sent_contains_for_tests(stale_ip).await);
assert!(cache.full_cert_sent_contains_for_tests(new_ip).await);
}
#[tokio::test]
async fn test_take_full_cert_budget_for_ip_does_not_sweep_every_call() {
let cache = TlsFrontCache::new(&["example.com".to_string()], 1024, "tlsfront-test-cache");
let stale_ip: IpAddr = "127.0.0.1".parse().expect("ip");
let new_ip: IpAddr = "127.0.0.2".parse().expect("ip");
let ttl = Duration::from_secs(1);
let stale_seen_at = Instant::now()
.checked_sub(Duration::from_secs(10))
.unwrap_or_else(Instant::now);
let now_epoch_secs = SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_secs();
cache
.insert_full_cert_sent_for_tests(stale_ip, stale_seen_at)
.await;
cache
.full_cert_sent_last_sweep_epoch_secs
.store(now_epoch_secs, Ordering::Relaxed);
assert!(cache.take_full_cert_budget_for_ip(new_ip, ttl).await);
assert!(cache.full_cert_sent_contains_for_tests(stale_ip).await);
assert!(cache.full_cert_sent_contains_for_tests(new_ip).await);
}
}
+64
View File
@@ -3,7 +3,9 @@
use dashmap::DashMap;
use std::net::SocketAddr;
use std::sync::Arc;
use std::sync::Mutex;
use std::sync::OnceLock;
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{Duration, Instant};
use anyhow::{Result, anyhow};
@@ -144,12 +146,37 @@ enum FetchErrorKind {
Other,
}
const PROFILE_CACHE_MAX_ENTRIES: usize = 4096;
static PROFILE_CACHE: OnceLock<DashMap<ProfileCacheKey, ProfileCacheValue>> = OnceLock::new();
static PROFILE_CACHE_INSERT_GUARD: OnceLock<Mutex<()>> = OnceLock::new();
static PROFILE_CACHE_CAP_DROPS: AtomicU64 = AtomicU64::new(0);
fn profile_cache() -> &'static DashMap<ProfileCacheKey, ProfileCacheValue> {
PROFILE_CACHE.get_or_init(DashMap::new)
}
fn profile_cache_insert_guard() -> &'static Mutex<()> {
PROFILE_CACHE_INSERT_GUARD.get_or_init(|| Mutex::new(()))
}
fn sweep_expired_profile_cache(ttl: Duration, now: Instant) {
if ttl.is_zero() {
return;
}
profile_cache().retain(|_, value| now.saturating_duration_since(value.updated_at) <= ttl);
}
/// Current number of adaptive TLS fetch profile-cache entries.
pub(crate) fn profile_cache_entries_for_metrics() -> usize {
profile_cache().len()
}
/// Number of fresh profile-cache winners skipped because the cache was full.
pub(crate) fn profile_cache_cap_drops_for_metrics() -> u64 {
PROFILE_CACHE_CAP_DROPS.load(Ordering::Relaxed)
}
fn route_hint(
upstream: Option<&std::sync::Arc<crate::transport::UpstreamManager>>,
unix_sock: Option<&str>,
@@ -267,6 +294,43 @@ fn remember_profile_success(
let Some(key) = cache_key else {
return;
};
remember_profile_success_with_cap(strategy, key, profile, now, PROFILE_CACHE_MAX_ENTRIES);
}
fn remember_profile_success_with_cap(
strategy: &TlsFetchStrategy,
key: ProfileCacheKey,
profile: TlsFetchProfile,
now: Instant,
max_entries: usize,
) {
let Ok(_guard) = profile_cache_insert_guard().lock() else {
PROFILE_CACHE_CAP_DROPS.fetch_add(1, Ordering::Relaxed);
return;
};
if max_entries == 0 {
PROFILE_CACHE_CAP_DROPS.fetch_add(1, Ordering::Relaxed);
return;
}
if profile_cache().contains_key(&key) {
profile_cache().insert(
key,
ProfileCacheValue {
profile,
updated_at: now,
},
);
return;
}
if profile_cache().len() >= max_entries {
// TLS fetch is control-plane work; sweeping under a tiny mutex keeps
// profile-cache cardinality hard-bounded without touching relay hot paths.
sweep_expired_profile_cache(strategy.profile_cache_ttl, now);
}
if profile_cache().len() >= max_entries {
PROFILE_CACHE_CAP_DROPS.fetch_add(1, Ordering::Relaxed);
return;
}
profile_cache().insert(
key,
ProfileCacheValue {
+25 -1
View File
@@ -46,6 +46,7 @@ mod send_adversarial_tests;
mod wire;
use bytes::Bytes;
use tokio::sync::OwnedSemaphorePermit;
#[allow(unused_imports)]
pub use config_updater::{
@@ -68,9 +69,32 @@ pub use secret::{fetch_proxy_secret, fetch_proxy_secret_with_upstream};
pub(crate) use selftest::{bnd_snapshot, timeskew_snapshot, upstream_bnd_snapshots};
pub use wire::proto_flags_for_tag;
/// Holds D2C queued-byte capacity until a routed payload is consumed or dropped.
pub struct RouteBytePermit {
_permit: OwnedSemaphorePermit,
}
impl std::fmt::Debug for RouteBytePermit {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("RouteBytePermit").finish_non_exhaustive()
}
}
impl RouteBytePermit {
pub(crate) fn new(permit: OwnedSemaphorePermit) -> Self {
Self { _permit: permit }
}
}
/// Response routed from middle proxy readers to client relay tasks.
#[derive(Debug)]
pub enum MeResponse {
Data { flags: u32, data: Bytes },
/// Downstream payload with its queued-byte reservation.
Data {
flags: u32,
data: Bytes,
route_permit: Option<RouteBytePermit>,
},
Ack(u32),
Close,
}
+3 -7
View File
@@ -618,13 +618,9 @@ impl MePool {
me_route_hybrid_max_wait: Duration::from_millis(
me_route_hybrid_max_wait_ms.max(50),
),
me_route_blocking_send_timeout: if me_route_blocking_send_timeout_ms == 0 {
None
} else {
Some(Duration::from_millis(
me_route_blocking_send_timeout_ms.min(5_000),
))
},
me_route_blocking_send_timeout: Some(Duration::from_millis(
me_route_blocking_send_timeout_ms.clamp(1, 5_000),
)),
me_route_last_success_epoch_ms: AtomicU64::new(0),
me_route_hybrid_timeout_warn_epoch_ms: AtomicU64::new(0),
me_async_recovery_last_trigger_epoch_ms: AtomicU64::new(0),
+2 -1
View File
@@ -84,6 +84,7 @@ async fn route_data_with_retry(
MeResponse::Data {
flags,
data: data.clone(),
route_permit: None,
},
timeout_ms,
)
@@ -639,7 +640,7 @@ mod tests {
let routed = route_data_with_retry(&reg, conn_id, 0, Bytes::from_static(b"a"), 20).await;
assert!(matches!(routed, RouteResult::Routed));
match rx.recv().await {
Some(MeResponse::Data { flags, data }) => {
Some(MeResponse::Data { flags, data, .. }) => {
assert_eq!(flags, 0);
assert_eq!(data, Bytes::from_static(b"a"));
}
+184 -9
View File
@@ -1,18 +1,22 @@
use std::collections::{HashMap, HashSet};
use std::net::SocketAddr;
use std::sync::Arc;
use std::sync::atomic::{AtomicU8, AtomicU64, Ordering};
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use dashmap::DashMap;
use tokio::sync::mpsc::error::TrySendError;
use tokio::sync::{Mutex, mpsc};
use tokio::sync::{Mutex, Semaphore, mpsc};
use super::MeResponse;
use super::codec::WriterCommand;
use super::{MeResponse, RouteBytePermit};
const ROUTE_BACKPRESSURE_BASE_TIMEOUT_MS: u64 = 25;
const ROUTE_BACKPRESSURE_HIGH_TIMEOUT_MS: u64 = 120;
const ROUTE_BACKPRESSURE_HIGH_WATERMARK_PCT: u8 = 80;
const ROUTE_QUEUED_BYTE_PERMIT_UNIT: usize = 16 * 1024;
const ROUTE_QUEUED_PERMITS_PER_SLOT: usize = 4;
const ROUTE_QUEUED_MAX_FRAME_PERMITS: usize = 1024;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RouteResult {
@@ -53,6 +57,7 @@ pub(super) struct WriterActivitySnapshot {
struct RoutingTable {
map: DashMap<u64, mpsc::Sender<MeResponse>>,
byte_budget: DashMap<u64, Arc<Semaphore>>,
}
struct WriterTable {
@@ -105,6 +110,7 @@ pub struct ConnRegistry {
route_backpressure_base_timeout_ms: AtomicU64,
route_backpressure_high_timeout_ms: AtomicU64,
route_backpressure_high_watermark_pct: AtomicU8,
route_byte_permits_per_conn: usize,
}
impl ConnRegistry {
@@ -116,10 +122,23 @@ impl ConnRegistry {
}
pub fn with_route_channel_capacity(route_channel_capacity: usize) -> Self {
let route_channel_capacity = route_channel_capacity.max(1);
Self::with_route_limits(
route_channel_capacity,
Self::route_byte_permit_budget(route_channel_capacity),
)
}
fn with_route_limits(
route_channel_capacity: usize,
route_byte_permits_per_conn: usize,
) -> Self {
let start = rand::random::<u64>() | 1;
let route_channel_capacity = route_channel_capacity.max(1);
Self {
routing: RoutingTable {
map: DashMap::new(),
byte_budget: DashMap::new(),
},
writers: WriterTable {
map: DashMap::new(),
@@ -131,15 +150,30 @@ impl ConnRegistry {
inner: Mutex::new(BindingInner::new()),
},
next_id: AtomicU64::new(start),
route_channel_capacity: route_channel_capacity.max(1),
route_channel_capacity,
route_backpressure_base_timeout_ms: AtomicU64::new(ROUTE_BACKPRESSURE_BASE_TIMEOUT_MS),
route_backpressure_high_timeout_ms: AtomicU64::new(ROUTE_BACKPRESSURE_HIGH_TIMEOUT_MS),
route_backpressure_high_watermark_pct: AtomicU8::new(
ROUTE_BACKPRESSURE_HIGH_WATERMARK_PCT,
),
route_byte_permits_per_conn: route_byte_permits_per_conn.max(1),
}
}
fn route_data_permits(data_len: usize) -> u32 {
data_len
.max(1)
.div_ceil(ROUTE_QUEUED_BYTE_PERMIT_UNIT)
.min(u32::MAX as usize) as u32
}
fn route_byte_permit_budget(route_channel_capacity: usize) -> usize {
route_channel_capacity
.saturating_mul(ROUTE_QUEUED_PERMITS_PER_SLOT)
.max(ROUTE_QUEUED_MAX_FRAME_PERMITS)
.max(1)
}
pub fn route_channel_capacity(&self) -> usize {
self.route_channel_capacity
}
@@ -149,6 +183,14 @@ impl ConnRegistry {
Self::with_route_channel_capacity(4096)
}
#[cfg(test)]
fn with_route_byte_permits_for_tests(
route_channel_capacity: usize,
route_byte_permits_per_conn: usize,
) -> Self {
Self::with_route_limits(route_channel_capacity, route_byte_permits_per_conn)
}
pub fn update_route_backpressure_policy(
&self,
base_timeout_ms: u64,
@@ -170,6 +212,10 @@ impl ConnRegistry {
let id = self.next_id.fetch_add(1, Ordering::Relaxed);
let (tx, rx) = mpsc::channel(self.route_channel_capacity);
self.routing.map.insert(id, tx);
self.routing.byte_budget.insert(
id,
Arc::new(Semaphore::new(self.route_byte_permits_per_conn)),
);
(id, rx)
}
@@ -186,6 +232,7 @@ impl ConnRegistry {
/// Unregister connection, returning associated writer_id if any.
pub async fn unregister(&self, id: u64) -> Option<u64> {
self.routing.map.remove(&id);
self.routing.byte_budget.remove(&id);
self.hot_binding.map.remove(&id);
let mut binding = self.binding.inner.lock().await;
binding.meta.remove(&id);
@@ -206,6 +253,64 @@ impl ConnRegistry {
None
}
async fn attach_route_byte_permit(
&self,
id: u64,
resp: MeResponse,
timeout_ms: Option<u64>,
) -> std::result::Result<MeResponse, RouteResult> {
let MeResponse::Data {
flags,
data,
route_permit,
} = resp
else {
return Ok(resp);
};
if route_permit.is_some() {
return Ok(MeResponse::Data {
flags,
data,
route_permit,
});
}
let Some(semaphore) = self
.routing
.byte_budget
.get(&id)
.map(|entry| entry.value().clone())
else {
return Err(RouteResult::NoConn);
};
let permits = Self::route_data_permits(data.len());
let permit = match timeout_ms {
Some(0) => semaphore
.try_acquire_many_owned(permits)
.map_err(|_| RouteResult::QueueFullHigh)?,
Some(timeout_ms) => {
let acquire = semaphore.acquire_many_owned(permits);
match tokio::time::timeout(Duration::from_millis(timeout_ms.max(1)), acquire).await
{
Ok(Ok(permit)) => permit,
Ok(Err(_)) => return Err(RouteResult::ChannelClosed),
Err(_) => return Err(RouteResult::QueueFullHigh),
}
}
None => semaphore
.acquire_many_owned(permits)
.await
.map_err(|_| RouteResult::ChannelClosed)?,
};
Ok(MeResponse::Data {
flags,
data,
route_permit: Some(RouteBytePermit::new(permit)),
})
}
#[allow(dead_code)]
pub async fn route(&self, id: u64, resp: MeResponse) -> RouteResult {
let tx = self.routing.map.get(&id).map(|entry| entry.value().clone());
@@ -214,15 +319,23 @@ impl ConnRegistry {
return RouteResult::NoConn;
};
let base_timeout_ms = self
.route_backpressure_base_timeout_ms
.load(Ordering::Relaxed)
.max(1);
let resp = match self
.attach_route_byte_permit(id, resp, Some(base_timeout_ms))
.await
{
Ok(resp) => resp,
Err(result) => return result,
};
match tx.try_send(resp) {
Ok(()) => RouteResult::Routed,
Err(TrySendError::Closed(_)) => RouteResult::ChannelClosed,
Err(TrySendError::Full(resp)) => {
// Absorb short bursts without dropping/closing the session immediately.
let base_timeout_ms = self
.route_backpressure_base_timeout_ms
.load(Ordering::Relaxed)
.max(1);
let high_timeout_ms = self
.route_backpressure_high_timeout_ms
.load(Ordering::Relaxed)
@@ -266,6 +379,10 @@ impl ConnRegistry {
let Some(tx) = tx else {
return RouteResult::NoConn;
};
let resp = match self.attach_route_byte_permit(id, resp, Some(0)).await {
Ok(resp) => resp,
Err(result) => return result,
};
match tx.try_send(resp) {
Ok(()) => RouteResult::Routed,
@@ -289,6 +406,13 @@ impl ConnRegistry {
let Some(tx) = tx else {
return RouteResult::NoConn;
};
let resp = match self
.attach_route_byte_permit(id, resp, Some(timeout_ms))
.await
{
Ok(resp) => resp,
Err(result) => return result,
};
match tx.try_send(resp) {
Ok(()) => RouteResult::Routed,
@@ -541,8 +665,10 @@ impl ConnRegistry {
mod tests {
use std::net::{IpAddr, Ipv4Addr, SocketAddr};
use super::ConnMeta;
use super::ConnRegistry;
use bytes::Bytes;
use super::{ConnMeta, ConnRegistry, RouteResult};
use crate::transport::middle_proxy::MeResponse;
#[tokio::test]
async fn writer_activity_snapshot_tracks_writer_and_dc_load() {
@@ -608,6 +734,55 @@ mod tests {
assert_eq!(snapshot.active_sessions_by_target_dc.get(&4), Some(&1));
}
#[tokio::test]
async fn route_data_is_bounded_by_byte_permits_before_channel_capacity() {
let registry = ConnRegistry::with_route_byte_permits_for_tests(4, 1);
let (conn_id, mut rx) = registry.register().await;
let routed = registry
.route_nowait(
conn_id,
MeResponse::Data {
flags: 0,
data: Bytes::from_static(&[0xAA]),
route_permit: None,
},
)
.await;
assert!(matches!(routed, RouteResult::Routed));
let blocked = registry
.route_nowait(
conn_id,
MeResponse::Data {
flags: 0,
data: Bytes::from_static(&[0xBB]),
route_permit: None,
},
)
.await;
assert!(
matches!(blocked, RouteResult::QueueFullHigh),
"byte budget must reject data before count capacity is exhausted"
);
drop(rx.recv().await);
let routed_after_drain = registry
.route_nowait(
conn_id,
MeResponse::Data {
flags: 0,
data: Bytes::from_static(&[0xCC]),
route_permit: None,
},
)
.await;
assert!(
matches!(routed_after_drain, RouteResult::Routed),
"receiving queued data must release byte permits"
);
}
#[tokio::test]
async fn bind_writer_rebinds_conn_atomically() {
let registry = ConnRegistry::new();