diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..3a8a1de --- /dev/null +++ b/.env.example @@ -0,0 +1,66 @@ +# ========================================== +# Telemt Easy Configuration +# ========================================== +# This file lets you configure your Telegram MTProto proxy easily. +# If you are using Docker, Railway, or Dokploy, you can copy these variables +# directly into your environment settings. + +# --- 1. Your Proxy Address (Share Link) --- +# This is the IP address or domain name you will give to your users so they can connect. +# Example: If your server IP is 123.45.67.89, put that here. +# TELEMT_GENERAL__LINKS__PUBLIC_HOST=123.45.67.89 + +# The port your users will connect to. +# On Dokploy, the proxy runs on port 8443 by default (port 443 is used by +# Traefik for HTTP services). Set this to match PROXY_PUBLIC_PORT below. +# TELEMT_GENERAL__LINKS__PUBLIC_PORT=8443 + + +# --- 2. User Passwords (Secrets) --- +# To connect to your proxy, your users need a "secret" (password). +# A secret MUST be exactly 32 characters long, using only numbers (0-9) and letters (a-f). +# Generate a random one with: openssl rand -hex 16 +# +# Format: TELEMT_ACCESS__USERS__=<32_CHARACTER_SECRET> +# +# Example user "myfamily" with a 32-character secret: +# TELEMT_ACCESS__USERS__MYFAMILY=11112222333344445555666677778888 + + +# --- 3. Anti-Censorship (Fake Domain) --- +# To hide your proxy from internet censors, telemt disguises the traffic to look like +# someone is just browsing a normal website. +# Enter a popular, unblocked website domain in your country (e.g. google.com, or a local site). +# TELEMT_CENSORSHIP__TLS_DOMAIN=petrovich.ru + +# Keep these enabled for maximum protection against deep packet inspection (DPI). +# TELEMT_CENSORSHIP__MASK=true +# TELEMT_CENSORSHIP__TLS_EMULATION=true + + +# --- 4. Sponsored Channel (Optional) --- +# If you registered your proxy with @MTProxybot on Telegram to show a sponsored channel, +# paste the 32-character "ad_tag" they gave you here. Leave commented out if you don't have one. +# TELEMT_GENERAL__AD_TAG=00000000000000000000000000000000 + + +# --- 5. Advanced Server Settings --- +# The port the proxy actually runs on internally. Usually matches your PUBLIC_PORT. +# TELEMT_SERVER__PORT=443 + +# How much detail to print in the logs (silent, normal, debug) +# TELEMT_GENERAL__LOG_LEVEL=normal + + +# --- 6. Dokploy (docker-compose-dokploy.yml only) --- +# These are used by the Dokploy compose file, NOT Telemt config vars. +# Do not prefix them with TELEMT_. + +# The host port to publish for MTProto traffic. The container always +# listens on 443 internally; this maps it to a host port that doesn't +# conflict with Traefik (which owns port 443 for HTTP services). +# Default is 8443 if not set. +# PROXY_PUBLIC_PORT=8443 + +# Log level for the Rust process itself (not Telemt config). +# RUST_LOG=info diff --git a/.gitignore b/.gitignore index bc782ca..86c4b3a 100644 --- a/.gitignore +++ b/.gitignore @@ -21,4 +21,10 @@ target #.idea/ proxy-secret -coverage-html/ \ No newline at end of file +coverage-html/ + +# OS +.DS_Store + +# Editor +.vscode/ \ No newline at end of file diff --git a/Cargo.toml b/Cargo.toml index 7eb4c79..374ac4a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -44,6 +44,7 @@ shadowsocks = { version = "1.24.0", features = ["aead-cipher-2022"] } serde = { version = "1.0.228", features = ["derive"] } serde_json = "1.0.150" toml = "1.1" +figment = { version = "0.10.19", features = ["toml", "env"] } x509-parser = "0.18.1" # Utils diff --git a/Dockerfile b/Dockerfile index 9bd83e8..d4168ac 100644 --- a/Dockerfile +++ b/Dockerfile @@ -3,6 +3,50 @@ ARG TELEMT_REPOSITORY=telemt/telemt ARG TELEMT_VERSION=latest +# ========================== +# Rust Builder (compiles from source, static musl binary) +# ========================== +FROM rust:1.88-bookworm AS builder + +ARG TARGETARCH + +RUN set -eux; \ + apt-get update; \ + apt-get install -y --no-install-recommends musl-tools; \ + rm -rf /var/lib/apt/lists/* + +# Determine the musl target triple based on the build architecture. +RUN set -eux; \ + case "${TARGETARCH:-amd64}" in \ + amd64) echo "x86_64-unknown-linux-musl" > /tmp/target.txt ;; \ + arm64) echo "aarch64-unknown-linux-musl" > /tmp/target.txt ;; \ + *) echo "Unsupported TARGETARCH: ${TARGETARCH}" >&2; exit 1 ;; \ + esac + +RUN rustup target add "$(cat /tmp/target.txt)" + +WORKDIR /build + +# Copy dependency manifests first for layer caching. +COPY Cargo.toml Cargo.lock ./ +COPY .cargo .cargo/ + +# Create a dummy src/main.rs so cargo can fetch and compile dependencies +# without the full source tree. This layer is cached as long as +# Cargo.toml/Cargo.lock don't change. +RUN mkdir -p src && echo "fn main() {}" > src/main.rs && \ + cargo build --release --target "$(cat /tmp/target.txt)" || true + +# Copy the actual source and build the real binary. +COPY src src +COPY benches benches + +RUN touch src/main.rs && cargo build --release --target "$(cat /tmp/target.txt)" + +# Place the binary at a fixed path so downstream stages can COPY it +# without knowing the target triple. +RUN cp "target/$(cat /tmp/target.txt)/release/telemt" /telemt-binary + # ========================== # Minimal Image # ========================== @@ -16,44 +60,11 @@ RUN set -eux; \ apt-get update; \ apt-get install -y --no-install-recommends \ binutils \ - ca-certificates \ - curl \ - tar; \ + ca-certificates; \ rm -rf /var/lib/apt/lists/* -RUN set -eux; \ - case "${TARGETARCH}" in \ - amd64) ASSET="telemt-x86_64-linux-musl.tar.gz" ;; \ - arm64) ASSET="telemt-aarch64-linux-musl.tar.gz" ;; \ - *) echo "Unsupported TARGETARCH: ${TARGETARCH}" >&2; exit 1 ;; \ - esac; \ - VERSION="${TELEMT_VERSION#refs/tags/}"; \ - if [ -z "${VERSION}" ] || [ "${VERSION}" = "latest" ]; then \ - BASE_URL="https://github.com/${TELEMT_REPOSITORY}/releases/latest/download"; \ - else \ - BASE_URL="https://github.com/${TELEMT_REPOSITORY}/releases/download/${VERSION}"; \ - fi; \ - curl -fL \ - --retry 5 \ - --retry-delay 3 \ - --connect-timeout 10 \ - --max-time 120 \ - -o "/tmp/${ASSET}" \ - "${BASE_URL}/${ASSET}"; \ - curl -fL \ - --retry 5 \ - --retry-delay 3 \ - --connect-timeout 10 \ - --max-time 120 \ - -o "/tmp/${ASSET}.sha256" \ - "${BASE_URL}/${ASSET}.sha256"; \ - cd /tmp; \ - sha256sum -c "${ASSET}.sha256"; \ - tar -xzf "${ASSET}" -C /tmp; \ - test -f /tmp/telemt; \ - install -m 0755 /tmp/telemt /telemt; \ - strip --strip-unneeded /telemt || true; \ - rm -f "/tmp/${ASSET}" "/tmp/${ASSET}.sha256" /tmp/telemt +COPY --from=builder /telemt-binary /telemt +RUN strip --strip-unneeded /telemt || true RUN --mount=type=bind,target=/tmp \ mkdir -p /app && \ @@ -62,7 +73,7 @@ RUN --mount=type=bind,target=/tmp \ elif [ -f /tmp/config/config.toml ]; then \ cp /tmp/config/config.toml /app/config.toml; \ else \ - echo "Config file not found" && exit 1; \ + echo "No config.toml provided; creating empty config.toml for ENV overriding" && touch /app/config.toml; \ fi # ========================== diff --git a/docker-compose-dokploy.yml b/docker-compose-dokploy.yml new file mode 100644 index 0000000..50704b7 --- /dev/null +++ b/docker-compose-dokploy.yml @@ -0,0 +1,71 @@ +# Dokploy-ready compose file. +# +# MTProto is not HTTP, so Traefik (HTTP reverse proxy) cannot route it. +# Dokploy's Traefik is configured with HTTP TLS termination on port 443, +# which conflicts with MTProto's TLS passthrough requirement. +# +# Solution: publish a dedicated port (8443) directly from the container, +# bypassing Traefik entirely for MTProto traffic. Port 443 stays with +# Traefik for other HTTP services on the server. +# +# Set TELEMT_GENERAL__LINKS__PUBLIC_PORT=8443 in your Dokploy env vars +# so generated share links use the correct public port. +# +# Metrics and API endpoints remain internal (expose only, no public ports). +# To access them, use `docker exec` or uncomment the ports below. +# +# All env vars are documented in .env.example. Set them in the Dokploy UI +# environment section, not in this file. + +services: + telemt: + image: ghcr.io/telemt/telemt:latest + build: + context: . + target: prod + restart: unless-stopped + ports: + - "${PROXY_PUBLIC_PORT:-8443}:443" + expose: + - "9090" + - "9091" + working_dir: /run/telemt + command: ["/app/config.toml"] + # volumes: + # - ./config:/etc/telemt:rw + tmpfs: + - /run/telemt:rw,mode=1777,size=4m + # env_file passes all vars from Dokploy's .env (TELEMT_*, RUST_LOG, etc.) + # to the container. environment below provides defaults for RUST_LOG. + env_file: + - .env + environment: + - RUST_LOG=${RUST_LOG:-info} + healthcheck: + test: [ "CMD", "/app/telemt", "healthcheck", "/app/config.toml", "--mode", "liveness" ] + interval: 30s + timeout: 5s + retries: 3 + start_period: 20s + cap_drop: + - ALL + cap_add: + - NET_BIND_SERVICE + read_only: true + security_opt: + - no-new-privileges:true + ulimits: + nofile: + soft: 65536 + hard: 262144 + logging: + driver: json-file + options: + max-size: "50m" + max-file: "5" + networks: + - dokploy-network + +networks: + dokploy-network: + external: true diff --git a/docker-compose.yml b/docker-compose.yml index a90efde..44f14c8 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -14,15 +14,22 @@ services: # Config is mounted as a directory (not a single file) so the API can # atomically update config.toml via write-temp → rename within the same FS. working_dir: /run/telemt - command: ["/etc/telemt/config.toml"] - volumes: - - ./config:/etc/telemt:rw + # Optional: Mount config directory if you want to use the API to mutate config.toml + # In ENV-driven mode, the config is read from /app/config.toml and ENV vars + # override config values. To use file-based/API-edit mode instead, uncomment + # the volume below AND change the command to ["/etc/telemt/config.toml"]. + command: ["/app/config.toml"] + # volumes: + # - ./config:/etc/telemt:rw tmpfs: - /run/telemt:rw,mode=1777,size=4m environment: - RUST_LOG=info + # ENV variables can be used to override config values: + # - TELEMT_GENERAL__PORT=443 + # - TELEMT_GENERAL__AD_TAG=1234567890abcdef1234567890abcdef healthcheck: - test: [ "CMD", "/app/telemt", "healthcheck", "/etc/telemt/config.toml", "--mode", "liveness" ] + test: [ "CMD", "/app/telemt", "healthcheck", "/app/config.toml", "--mode", "liveness" ] interval: 30s timeout: 5s retries: 3 diff --git a/src/config/hot_reload.rs b/src/config/hot_reload.rs index 1a4ab2c..fc0a45d 100644 --- a/src/config/hot_reload.rs +++ b/src/config/hot_reload.rs @@ -613,6 +613,7 @@ fn overlay_hot_fields(old: &ProxyConfig, new: &ProxyConfig) -> ProxyConfig { if cfg.rebuild_runtime_user_auth().is_err() { cfg.runtime_user_auth = None; } + cfg.rebuild_runtime_ad_tags(); cfg } diff --git a/src/config/load.rs b/src/config/load.rs index aa97b58..d483226 100644 --- a/src/config/load.rs +++ b/src/config/load.rs @@ -5,6 +5,7 @@ use std::net::{IpAddr, SocketAddr}; use std::path::{Path, PathBuf}; use std::sync::Arc; +use figment::providers::Format; use rand::RngExt; use serde::{Deserialize, Serialize}; use tracing::warn; @@ -48,6 +49,117 @@ const MIN_MAX_CLIENT_FRAME_BYTES: usize = 4 * 1024; const MAX_MAX_CLIENT_FRAME_BYTES: usize = 16 * 1024 * 1024; const MAX_API_REQUEST_BODY_LIMIT_BYTES: usize = 1024 * 1024; +// Applies TELEMT_-prefixed environment variables to the parsed TOML config. +// Figment's Env provider creates uppercase keys that don't match lowercase +// TOML keys when extracting to toml::Value, so we override manually. +fn apply_telemt_env_overrides(toml: &mut toml::Value) { + for (key, value) in std::env::vars() { + let Some(rest) = key.strip_prefix("TELEMT_") else { + continue; + }; + let segments: Vec<&str> = rest.split("__").collect(); + if segments.is_empty() { + continue; + } + let path: Vec = segments + .iter() + .enumerate() + .map(|(i, seg)| { + if is_map_key_position(&segments[..i]) { + seg.to_string() + } else { + seg.to_lowercase() + } + }) + .collect(); + set_nested_toml_value(toml, &path, &value); + } +} + +fn is_map_key_position(prefix: &[&str]) -> bool { + match prefix { + [a, b] => { + a.eq_ignore_ascii_case("access") + && (b.eq_ignore_ascii_case("users") || b.eq_ignore_ascii_case("user_ad_tags")) + } + _ => false, + } +} + +// Recursively sets a value in a nested TOML table, inferring the type from +// the existing value when present (bool, integer, or string fallback). +fn set_nested_toml_value(root: &mut toml::Value, path: &[String], raw_value: &str) { + if path.is_empty() || !root.is_table() { + return; + } + let table = root.as_table_mut().unwrap(); + if path.len() == 1 { + let coerced = match table.get(&path[0]) { + Some(toml::Value::Boolean(_)) => match raw_value.parse::() { + Ok(b) => toml::Value::Boolean(b), + Err(_) => toml::Value::String(raw_value.to_string()), + }, + Some(toml::Value::Integer(_)) => match raw_value.parse::() { + Ok(i) => toml::Value::Integer(i), + Err(_) => toml::Value::String(raw_value.to_string()), + }, + Some(toml::Value::Float(_)) => match raw_value.parse::() { + Ok(f) => toml::Value::Float(f), + Err(_) => toml::Value::String(raw_value.to_string()), + }, + None => { + if let Ok(i) = raw_value.parse::() { + toml::Value::Integer(i) + } else if let Ok(b) = raw_value.parse::() { + toml::Value::Boolean(b) + } else { + toml::Value::String(raw_value.to_string()) + } + } + _ => toml::Value::String(raw_value.to_string()), + }; + table.insert(path[0].clone(), coerced); + } else { + let entry = table + .entry(path[0].clone()) + .or_insert_with(|| toml::Value::Table(toml::Table::new())); + set_nested_toml_value(entry, &path[1..], raw_value); + } +} + +/// Precomputed per-user and global ad_tag bytes for the middle-proxy hot path. +/// +/// Decoding happens once at config load and on hot-reload via +/// [`ProxyConfig::rebuild_runtime_ad_tags`], so per-session relay setup performs +/// a cache lookup instead of repeated `hex::decode` allocations. +#[derive(Debug, Clone, Default)] +pub(crate) struct AdTagCache { + global: Option<[u8; 16]>, + per_user: HashMap, +} + +impl AdTagCache { + fn decode_tag(tag: &str) -> Option<[u8; 16]> { + let bytes = hex::decode(tag).ok()?; + <[u8; 16]>::try_from(bytes.as_slice()).ok() + } + + fn build(global_ad_tag: &Option, user_ad_tags: &HashMap) -> Self { + let global = global_ad_tag.as_deref().and_then(Self::decode_tag); + let mut per_user = HashMap::with_capacity(user_ad_tags.len()); + for (user, tag) in user_ad_tags { + if let Some(decoded) = Self::decode_tag(tag) { + per_user.insert(user.clone(), decoded); + } + } + Self { global, per_user } + } + + pub(crate) fn effective(&self, user: &str) -> Option<[u8; 16]> { + self.per_user.get(user).copied().or(self.global) + } +} + #[derive(Debug, Clone)] pub(crate) struct LoadedConfig { pub(crate) config: ProxyConfig, @@ -112,6 +224,10 @@ pub struct ProxyConfig { /// Precomputed authentication snapshot for handshake hot paths. #[serde(skip)] pub(crate) runtime_user_auth: Option>, + + /// Precomputed ad_tag cache for middle-relay hot paths. + #[serde(skip)] + pub(crate) runtime_ad_tags: Option>, } impl ProxyConfig { @@ -129,8 +245,11 @@ impl ProxyConfig { source_files.insert(normalize_config_path(path)); let processed = preprocess_includes(&content, base_dir, 0, &mut source_files)?; - let parsed_toml: toml::Value = - toml::from_str(&processed).map_err(|e| ProxyError::Config(e.to_string()))?; + let mut parsed_toml: toml::Value = figment::Figment::new() + .merge(figment::providers::Toml::string(&processed)) + .extract() + .map_err(|e| ProxyError::Config(e.to_string()))?; + apply_telemt_env_overrides(&mut parsed_toml); handle_unknown_config_keys(&parsed_toml)?; let general_table = parsed_toml .get("general") @@ -1382,6 +1501,7 @@ impl ProxyConfig { validate_logging_config(&config.logging)?; validate_upstreams(&config)?; + config.rebuild_runtime_ad_tags(); config.rebuild_runtime_user_auth()?; Ok(LoadedConfig { @@ -1401,6 +1521,33 @@ impl ProxyConfig { self.runtime_user_auth.as_deref() } + /// Rebuilds the precomputed ad_tag cache from the current `general.ad_tag` + /// and `access.user_ad_tags`. Invoked at load and on hot-reload. + pub(crate) fn rebuild_runtime_ad_tags(&mut self) { + let cache = AdTagCache::build(&self.general.ad_tag, &self.access.user_ad_tags); + self.runtime_ad_tags = Some(Arc::new(cache)); + } + + /// Returns the effective ad_tag bytes for `user` from the precomputed cache. + /// + /// Falls back to decoding on demand if the runtime cache is absent (e.g. in tests). + pub(crate) fn effective_ad_tag(&self, user: &str) -> Option<[u8; 16]> { + match self.runtime_ad_tags.as_deref() { + Some(cache) => cache.effective(user), + None => self + .access + .user_ad_tags + .get(user) + .and_then(|tag| AdTagCache::decode_tag(tag)) + .or_else(|| { + self.general + .ad_tag + .as_deref() + .and_then(AdTagCache::decode_tag) + }), + } + } + /// Validates cross-field configuration invariants after deserialization. pub fn validate(&self) -> Result<()> { if self.access.users.is_empty() { diff --git a/src/metrics/mod.rs b/src/metrics/mod.rs new file mode 100644 index 0000000..aaee846 --- /dev/null +++ b/src/metrics/mod.rs @@ -0,0 +1,272 @@ +// Metrics HTTP server, request dispatch, and beobachten renderer. +// Submodules: +// - render: Prometheus metrics rendering (render_metrics) +// - tls_front: TLS front profile health rendering helpers +// - tests: integration tests for the metrics endpoint + +mod render; +mod tls_front; + +#[cfg(test)] +mod tests; + +use std::convert::Infallible; +use std::net::SocketAddr; +use std::sync::Arc; +use std::time::Duration; + +use arc_swap::ArcSwap; +use http_body_util::Full; +use hyper::body::Bytes; +use hyper::server::conn::http1; +use hyper::service::service_fn; +use hyper::{Request, Response, StatusCode}; +use tokio::net::TcpListener; +use tokio::sync::Semaphore; +use tokio::time::timeout; +use tracing::{debug, info, warn}; + +use crate::config::ProxyConfig; +use crate::ip_tracker::UserIpTracker; +use crate::maestro::generation::RuntimeGeneration; +use crate::proxy::shared_state::ProxySharedState; +use crate::stats::Stats; +use crate::stats::beobachten::BeobachtenStore; +use crate::tls_front::TlsFrontCache; +use crate::transport::{ListenOptions, create_listener}; + +use render::render_metrics; + +const METRICS_MAX_CONTROL_CONNECTIONS: usize = 512; +const METRICS_HTTP_CONNECTION_TIMEOUT: Duration = Duration::from_secs(15); + +pub async fn serve( + port: u16, + listen: Option, + listen_backlog: u32, + active_runtime: Arc>, +) { + // If `metrics_listen` is set, bind on that single address only. + if let Some(ref listen_addr) = listen { + let addr: SocketAddr = match listen_addr.parse() { + Ok(a) => a, + Err(e) => { + warn!(error = %e, "Invalid metrics_listen address: {}", listen_addr); + return; + } + }; + // Match `server.api.listen`: `[::]:port` is a dual-stack wildcard + // on Linux when `net.ipv6.bindv6only=0`. + let ipv6_only = addr.is_ipv6() && !addr.ip().is_unspecified(); + match bind_metrics_listener(addr, ipv6_only, listen_backlog) { + Ok(listener) => { + info!("Metrics endpoint: http://{}/metrics and /beobachten", addr); + serve_listener(listener, active_runtime).await; + } + Err(e) => { + warn!(error = %e, "Failed to bind metrics on {}", addr); + } + } + return; + } + + // Fallback: keep metrics local unless an explicit metrics_listen is configured. + let mut listener_v4 = None; + let mut listener_v6 = None; + + let addr_v4 = SocketAddr::from(([127, 0, 0, 1], port)); + match bind_metrics_listener(addr_v4, false, listen_backlog) { + Ok(listener) => { + info!( + "Metrics endpoint: http://{}/metrics and /beobachten", + addr_v4 + ); + listener_v4 = Some(listener); + } + Err(e) => { + warn!(error = %e, "Failed to bind metrics on {}", addr_v4); + } + } + + let addr_v6 = SocketAddr::from(([0, 0, 0, 0, 0, 0, 0, 1], port)); + match bind_metrics_listener(addr_v6, true, listen_backlog) { + Ok(listener) => { + info!( + "Metrics endpoint: http://[::1]:{}/metrics and /beobachten", + port + ); + listener_v6 = Some(listener); + } + Err(e) => { + warn!(error = %e, "Failed to bind metrics on {}", addr_v6); + } + } + + match (listener_v4, listener_v6) { + (None, None) => { + warn!("Metrics listener is unavailable on both IPv4 and IPv6"); + } + (Some(listener), None) | (None, Some(listener)) => { + serve_listener(listener, active_runtime).await; + } + (Some(listener4), Some(listener6)) => { + let active_runtime_v6 = active_runtime.clone(); + tokio::spawn(async move { + serve_listener(listener6, active_runtime_v6).await; + }); + serve_listener(listener4, active_runtime).await; + } + } +} + +fn bind_metrics_listener( + addr: SocketAddr, + ipv6_only: bool, + listen_backlog: u32, +) -> std::io::Result { + let options = ListenOptions { + reuse_port: false, + ipv6_only, + backlog: listen_backlog, + ..Default::default() + }; + let socket = create_listener(addr, &options)?; + TcpListener::from_std(socket.into()) +} + +async fn serve_listener(listener: TcpListener, active_runtime: Arc>) { + let connection_permits = Arc::new(Semaphore::new(METRICS_MAX_CONTROL_CONNECTIONS)); + + loop { + let (stream, peer) = match listener.accept().await { + Ok(v) => v, + Err(e) => { + warn!(error = %e, "Metrics accept error"); + continue; + } + }; + + let runtime = active_runtime.load_full(); + let config = runtime.config(); + if !config.server.metrics_whitelist.is_empty() + && !config + .server + .metrics_whitelist + .iter() + .any(|net| net.contains(peer.ip())) + { + debug!(peer = %peer, "Metrics request denied by whitelist"); + continue; + } + + let connection_permit = match connection_permits.clone().try_acquire_owned() { + Ok(permit) => permit, + Err(_) => { + debug!( + peer = %peer, + max_connections = METRICS_MAX_CONTROL_CONNECTIONS, + "Dropping metrics connection: control-plane connection budget exhausted" + ); + continue; + } + }; + + let active_runtime = active_runtime.clone(); + tokio::spawn(async move { + let _connection_permit = connection_permit; + let svc = service_fn(move |req| { + let runtime = active_runtime.load_full(); + let stats = runtime.stats.clone(); + let beobachten = runtime.beobachten.clone(); + let shared_state = runtime.proxy_shared.clone(); + let ip_tracker = runtime.ip_tracker.clone(); + let tls_cache = runtime.tls_cache.clone(); + let config = runtime.config(); + async move { + handle( + req, + &stats, + &beobachten, + &shared_state, + &ip_tracker, + tls_cache.as_deref(), + &config, + ) + .await + } + }); + match timeout( + METRICS_HTTP_CONNECTION_TIMEOUT, + http1::Builder::new().serve_connection(hyper_util::rt::TokioIo::new(stream), svc), + ) + .await + { + Ok(Ok(())) => {} + Ok(Err(e)) => { + debug!(error = %e, "Metrics connection error"); + } + Err(_) => { + debug!( + peer = %peer, + timeout_ms = METRICS_HTTP_CONNECTION_TIMEOUT.as_millis() as u64, + "Metrics connection timed out" + ); + } + } + }); + } +} + +async fn handle( + req: Request, + stats: &Stats, + beobachten: &BeobachtenStore, + shared_state: &ProxySharedState, + ip_tracker: &UserIpTracker, + tls_cache: Option<&TlsFrontCache>, + config: &ProxyConfig, +) -> Result>, Infallible> { + if req.uri().path() == "/metrics" { + let body = render_metrics(stats, shared_state, config, ip_tracker, tls_cache).await; + let resp = Response::builder() + .status(StatusCode::OK) + .header("content-type", "text/plain; version=0.0.4; charset=utf-8") + .body(Full::new(Bytes::from(body))) + .unwrap(); + return Ok(resp); + } + + if req.uri().path() == "/beobachten" { + let body = render_beobachten(stats, beobachten, config); + let resp = Response::builder() + .status(StatusCode::OK) + .header("content-type", "text/plain; charset=utf-8") + .body(Full::new(Bytes::from(body))) + .unwrap(); + return Ok(resp); + } + + let resp = Response::builder() + .status(StatusCode::NOT_FOUND) + .body(Full::new(Bytes::from("Not Found\n"))) + .unwrap(); + Ok(resp) +} + +fn render_beobachten(stats: &Stats, beobachten: &BeobachtenStore, config: &ProxyConfig) -> String { + if !config.general.beobachten { + return "beobachten disabled\n".to_string(); + } + + let ttl = Duration::from_secs(config.general.beobachten_minutes.saturating_mul(60)); + let mut body = beobachten.snapshot_text(ttl); + let tls_text = stats.tls_fingerprint_snapshot_text(ttl, 20); + if !tls_text.is_empty() { + if !body.ends_with('\n') { + body.push('\n'); + } + body.push('\n'); + body.push_str(&tls_text); + } + body +} diff --git a/src/metrics.rs b/src/metrics/render.rs similarity index 73% rename from src/metrics.rs rename to src/metrics/render.rs index fb35b84..7015502 100644 --- a/src/metrics.rs +++ b/src/metrics/render.rs @@ -1,432 +1,19 @@ use std::collections::{BTreeSet, HashMap}; -use std::convert::Infallible; -use std::net::SocketAddr; -use std::sync::Arc; -use std::time::Duration; - -use arc_swap::ArcSwap; -use http_body_util::Full; -use hyper::body::Bytes; -use hyper::server::conn::http1; -use hyper::service::service_fn; -use hyper::{Request, Response, StatusCode}; -use tokio::net::TcpListener; -use tokio::sync::Semaphore; -use tokio::time::timeout; -use tracing::{debug, info, warn}; use crate::config::ProxyConfig; use crate::ip_tracker::UserIpTracker; -use crate::maestro::generation::RuntimeGeneration; use crate::proxy::shared_state::ProxySharedState; use crate::stats::Stats; -use crate::stats::beobachten::BeobachtenStore; use crate::tls_front::TlsFrontCache; use crate::tls_front::cache; use crate::tls_front::fetcher; -use crate::transport::{ListenOptions, create_listener}; + +use super::tls_front::render_tls_front_profile_health; // Keeps `/metrics` response size bounded when per-user telemetry is enabled. const USER_LABELED_METRICS_MAX_USERS: usize = 4096; -// Keeps TLS-front per-domain health series bounded for large generated configs. -const TLS_FRONT_PROFILE_HEALTH_MAX_DOMAINS: usize = 256; -const METRICS_MAX_CONTROL_CONNECTIONS: usize = 512; -const METRICS_HTTP_CONNECTION_TIMEOUT: Duration = Duration::from_secs(15); -pub async fn serve( - port: u16, - listen: Option, - listen_backlog: u32, - active_runtime: Arc>, -) { - // If `metrics_listen` is set, bind on that single address only. - if let Some(ref listen_addr) = listen { - let addr: SocketAddr = match listen_addr.parse() { - Ok(a) => a, - Err(e) => { - warn!(error = %e, "Invalid metrics_listen address: {}", listen_addr); - return; - } - }; - // Match `server.api.listen`: `[::]:port` is a dual-stack wildcard - // on Linux when `net.ipv6.bindv6only=0`. - let ipv6_only = addr.is_ipv6() && !addr.ip().is_unspecified(); - match bind_metrics_listener(addr, ipv6_only, listen_backlog) { - Ok(listener) => { - info!("Metrics endpoint: http://{}/metrics and /beobachten", addr); - serve_listener(listener, active_runtime).await; - } - Err(e) => { - warn!(error = %e, "Failed to bind metrics on {}", addr); - } - } - return; - } - - // Fallback: keep metrics local unless an explicit metrics_listen is configured. - let mut listener_v4 = None; - let mut listener_v6 = None; - - let addr_v4 = SocketAddr::from(([127, 0, 0, 1], port)); - match bind_metrics_listener(addr_v4, false, listen_backlog) { - Ok(listener) => { - info!( - "Metrics endpoint: http://{}/metrics and /beobachten", - addr_v4 - ); - listener_v4 = Some(listener); - } - Err(e) => { - warn!(error = %e, "Failed to bind metrics on {}", addr_v4); - } - } - - let addr_v6 = SocketAddr::from(([0, 0, 0, 0, 0, 0, 0, 1], port)); - match bind_metrics_listener(addr_v6, true, listen_backlog) { - Ok(listener) => { - info!( - "Metrics endpoint: http://[::1]:{}/metrics and /beobachten", - port - ); - listener_v6 = Some(listener); - } - Err(e) => { - warn!(error = %e, "Failed to bind metrics on {}", addr_v6); - } - } - - match (listener_v4, listener_v6) { - (None, None) => { - warn!("Metrics listener is unavailable on both IPv4 and IPv6"); - } - (Some(listener), None) | (None, Some(listener)) => { - serve_listener(listener, active_runtime).await; - } - (Some(listener4), Some(listener6)) => { - let active_runtime_v6 = active_runtime.clone(); - tokio::spawn(async move { - serve_listener(listener6, active_runtime_v6).await; - }); - serve_listener(listener4, active_runtime).await; - } - } -} - -fn bind_metrics_listener( - addr: SocketAddr, - ipv6_only: bool, - listen_backlog: u32, -) -> std::io::Result { - let options = ListenOptions { - reuse_port: false, - ipv6_only, - backlog: listen_backlog, - ..Default::default() - }; - let socket = create_listener(addr, &options)?; - TcpListener::from_std(socket.into()) -} - -async fn serve_listener(listener: TcpListener, active_runtime: Arc>) { - let connection_permits = Arc::new(Semaphore::new(METRICS_MAX_CONTROL_CONNECTIONS)); - - loop { - let (stream, peer) = match listener.accept().await { - Ok(v) => v, - Err(e) => { - warn!(error = %e, "Metrics accept error"); - continue; - } - }; - - let runtime = active_runtime.load_full(); - let config = runtime.config(); - if !config.server.metrics_whitelist.is_empty() - && !config - .server - .metrics_whitelist - .iter() - .any(|net| net.contains(peer.ip())) - { - debug!(peer = %peer, "Metrics request denied by whitelist"); - continue; - } - - let connection_permit = match connection_permits.clone().try_acquire_owned() { - Ok(permit) => permit, - Err(_) => { - debug!( - peer = %peer, - max_connections = METRICS_MAX_CONTROL_CONNECTIONS, - "Dropping metrics connection: control-plane connection budget exhausted" - ); - continue; - } - }; - - let active_runtime = active_runtime.clone(); - tokio::spawn(async move { - let _connection_permit = connection_permit; - let svc = service_fn(move |req| { - let runtime = active_runtime.load_full(); - let stats = runtime.stats.clone(); - let beobachten = runtime.beobachten.clone(); - let shared_state = runtime.proxy_shared.clone(); - let ip_tracker = runtime.ip_tracker.clone(); - let tls_cache = runtime.tls_cache.clone(); - let config = runtime.config(); - async move { - handle( - req, - &stats, - &beobachten, - &shared_state, - &ip_tracker, - tls_cache.as_deref(), - &config, - ) - .await - } - }); - match timeout( - METRICS_HTTP_CONNECTION_TIMEOUT, - http1::Builder::new().serve_connection(hyper_util::rt::TokioIo::new(stream), svc), - ) - .await - { - Ok(Ok(())) => {} - Ok(Err(e)) => { - debug!(error = %e, "Metrics connection error"); - } - Err(_) => { - debug!( - peer = %peer, - timeout_ms = METRICS_HTTP_CONNECTION_TIMEOUT.as_millis() as u64, - "Metrics connection timed out" - ); - } - } - }); - } -} - -async fn handle( - req: Request, - stats: &Stats, - beobachten: &BeobachtenStore, - shared_state: &ProxySharedState, - ip_tracker: &UserIpTracker, - tls_cache: Option<&TlsFrontCache>, - config: &ProxyConfig, -) -> Result>, Infallible> { - if req.uri().path() == "/metrics" { - let body = render_metrics(stats, shared_state, config, ip_tracker, tls_cache).await; - let resp = Response::builder() - .status(StatusCode::OK) - .header("content-type", "text/plain; version=0.0.4; charset=utf-8") - .body(Full::new(Bytes::from(body))) - .unwrap(); - return Ok(resp); - } - - if req.uri().path() == "/beobachten" { - let body = render_beobachten(stats, beobachten, config); - let resp = Response::builder() - .status(StatusCode::OK) - .header("content-type", "text/plain; charset=utf-8") - .body(Full::new(Bytes::from(body))) - .unwrap(); - return Ok(resp); - } - - let resp = Response::builder() - .status(StatusCode::NOT_FOUND) - .body(Full::new(Bytes::from("Not Found\n"))) - .unwrap(); - Ok(resp) -} - -fn render_beobachten(stats: &Stats, beobachten: &BeobachtenStore, config: &ProxyConfig) -> String { - if !config.general.beobachten { - return "beobachten disabled\n".to_string(); - } - - let ttl = Duration::from_secs(config.general.beobachten_minutes.saturating_mul(60)); - let mut body = beobachten.snapshot_text(ttl); - let tls_text = stats.tls_fingerprint_snapshot_text(ttl, 20); - if !tls_text.is_empty() { - if !body.ends_with('\n') { - body.push('\n'); - } - body.push('\n'); - body.push_str(&tls_text); - } - body -} - -fn tls_front_domains(config: &ProxyConfig) -> Vec { - let mut domains = Vec::with_capacity(1 + config.censorship.tls_domains.len()); - if !config.censorship.tls_domain.is_empty() { - domains.push(config.censorship.tls_domain.clone()); - } - for domain in &config.censorship.tls_domains { - if !domain.is_empty() && !domains.contains(domain) { - domains.push(domain.clone()); - } - } - domains -} - -fn prometheus_label_value(value: &str) -> String { - value.replace('\\', "\\\\").replace('"', "\\\"") -} - -async fn render_tls_front_profile_health( - out: &mut String, - config: &ProxyConfig, - tls_cache: Option<&TlsFrontCache>, -) { - use std::fmt::Write; - - let domains = tls_front_domains(config); - let (health, suppressed) = match (config.censorship.tls_emulation, tls_cache) { - (true, Some(cache)) => { - cache - .profile_health_snapshot(&domains, TLS_FRONT_PROFILE_HEALTH_MAX_DOMAINS) - .await - } - _ => (Vec::new(), domains.len()), - }; - - let _ = writeln!( - out, - "# HELP telemt_tls_front_profile_domains TLS front configured profile domains by export status" - ); - let _ = writeln!(out, "# TYPE telemt_tls_front_profile_domains gauge"); - let _ = writeln!( - out, - "telemt_tls_front_profile_domains{{status=\"configured\"}} {}", - domains.len() - ); - let _ = writeln!( - out, - "telemt_tls_front_profile_domains{{status=\"emitted\"}} {}", - health.len() - ); - let _ = writeln!( - out, - "telemt_tls_front_profile_domains{{status=\"suppressed\"}} {}", - suppressed - ); - let _ = writeln!( - out, - "# HELP telemt_tls_front_profile_info TLS front profile source and feature flags per configured domain" - ); - let _ = writeln!(out, "# TYPE telemt_tls_front_profile_info gauge"); - let _ = writeln!( - out, - "# HELP telemt_tls_front_profile_quality_info TLS front profile quality and key-share group per configured domain" - ); - let _ = writeln!(out, "# TYPE telemt_tls_front_profile_quality_info gauge"); - let _ = writeln!( - out, - "# HELP telemt_tls_front_profile_age_seconds Age of cached TLS front profile data per configured domain" - ); - let _ = writeln!(out, "# TYPE telemt_tls_front_profile_age_seconds gauge"); - let _ = writeln!( - out, - "# HELP telemt_tls_front_profile_server_hello_bytes TLS front cached ServerHello record body bytes per configured domain" - ); - let _ = writeln!( - out, - "# TYPE telemt_tls_front_profile_server_hello_bytes gauge" - ); - let _ = writeln!( - out, - "# HELP telemt_tls_front_profile_server_hello_extensions TLS front cached visible ServerHello extension count per configured domain" - ); - let _ = writeln!( - out, - "# TYPE telemt_tls_front_profile_server_hello_extensions gauge" - ); - let _ = writeln!( - out, - "# HELP telemt_tls_front_profile_app_data_records TLS front cached app-data record count per configured domain" - ); - let _ = writeln!( - out, - "# TYPE telemt_tls_front_profile_app_data_records gauge" - ); - let _ = writeln!( - out, - "# HELP telemt_tls_front_profile_ticket_records TLS front cached ticket-like tail record count per configured domain" - ); - let _ = writeln!(out, "# TYPE telemt_tls_front_profile_ticket_records gauge"); - let _ = writeln!( - out, - "# HELP telemt_tls_front_profile_change_cipher_spec_records TLS front cached ChangeCipherSpec record count per configured domain" - ); - let _ = writeln!( - out, - "# TYPE telemt_tls_front_profile_change_cipher_spec_records gauge" - ); - let _ = writeln!( - out, - "# HELP telemt_tls_front_profile_app_data_bytes TLS front cached total app-data bytes per configured domain" - ); - let _ = writeln!(out, "# TYPE telemt_tls_front_profile_app_data_bytes gauge"); - - for item in health { - let domain = prometheus_label_value(&item.domain); - let _ = writeln!( - out, - "telemt_tls_front_profile_info{{domain=\"{}\",source=\"{}\",is_default=\"{}\",has_cert_info=\"{}\",has_cert_payload=\"{}\"}} 1", - domain, item.source, item.is_default, item.has_cert_info, item.has_cert_payload - ); - let _ = writeln!( - out, - "telemt_tls_front_profile_quality_info{{domain=\"{}\",quality=\"{}\",key_share_group=\"{}\"}} 1", - domain, item.quality, item.key_share_group - ); - let _ = writeln!( - out, - "telemt_tls_front_profile_age_seconds{{domain=\"{}\"}} {}", - domain, item.age_seconds - ); - let _ = writeln!( - out, - "telemt_tls_front_profile_server_hello_bytes{{domain=\"{}\"}} {}", - domain, item.server_hello_record_len - ); - let _ = writeln!( - out, - "telemt_tls_front_profile_server_hello_extensions{{domain=\"{}\"}} {}", - domain, item.server_hello_extensions - ); - let _ = writeln!( - out, - "telemt_tls_front_profile_app_data_records{{domain=\"{}\"}} {}", - domain, item.app_data_records - ); - let _ = writeln!( - out, - "telemt_tls_front_profile_ticket_records{{domain=\"{}\"}} {}", - domain, item.ticket_records - ); - let _ = writeln!( - out, - "telemt_tls_front_profile_change_cipher_spec_records{{domain=\"{}\"}} {}", - domain, item.change_cipher_spec_count - ); - let _ = writeln!( - out, - "telemt_tls_front_profile_app_data_bytes{{domain=\"{}\"}} {}", - domain, item.total_app_data_len - ); - } -} - -async fn render_metrics( +pub(super) async fn render_metrics( stats: &Stats, shared_state: &ProxySharedState, config: &ProxyConfig, @@ -3858,436 +3445,3 @@ async fn render_metrics( out } - -#[cfg(test)] -mod tests { - use super::*; - use http_body_util::BodyExt; - use std::net::IpAddr; - use std::time::SystemTime; - - use crate::tls_front::types::{ - CachedTlsData, ParsedServerHello, TlsBehaviorProfile, TlsCertPayload, TlsProfileSource, - }; - - #[tokio::test] - async fn test_render_metrics_format() { - let stats = Arc::new(Stats::new()); - let shared_state = ProxySharedState::new(); - let tracker = UserIpTracker::new(); - let mut config = ProxyConfig::default(); - config - .access - .user_max_unique_ips - .insert("alice".to_string(), 4); - - stats.increment_connects_all(); - stats.increment_connects_all(); - stats.increment_connects_bad_with_class("tls_handshake_bad_client"); - stats.increment_handshake_timeouts(); - stats.increment_handshake_failure_class("timeout"); - shared_state - .handshake - .auth_expensive_checks_total - .fetch_add(9, std::sync::atomic::Ordering::Relaxed); - shared_state - .handshake - .auth_budget_exhausted_total - .fetch_add(2, std::sync::atomic::Ordering::Relaxed); - stats.increment_upstream_connect_attempt_total(); - stats.increment_upstream_connect_attempt_total(); - stats.increment_upstream_connect_success_total(); - stats.increment_upstream_connect_fail_total(); - stats.increment_upstream_connect_failfast_hard_error_total(); - stats.observe_upstream_connect_attempts_per_request(2); - stats.observe_upstream_connect_duration_ms(220, true); - stats.observe_upstream_connect_duration_ms(1500, false); - stats.increment_me_rpc_proxy_req_signal_sent_total(); - stats.increment_me_rpc_proxy_req_signal_failed_total(); - stats.increment_me_rpc_proxy_req_signal_skipped_no_meta_total(); - stats.increment_me_rpc_proxy_req_signal_response_total(); - stats.increment_me_rpc_proxy_req_signal_close_sent_total(); - stats.increment_me_idle_close_by_peer_total(); - stats.increment_relay_idle_soft_mark_total(); - stats.increment_relay_idle_hard_close_total(); - stats.increment_relay_pressure_evict_total(); - stats.increment_relay_protocol_desync_close_total(); - stats.increment_me_d2c_batches_total(); - stats.add_me_d2c_batch_frames_total(3); - stats.add_me_d2c_batch_bytes_total(2048); - stats.increment_me_d2c_flush_reason(crate::stats::MeD2cFlushReason::AckImmediate); - stats.increment_me_d2c_data_frames_total(); - stats.increment_me_d2c_ack_frames_total(); - stats.add_me_d2c_payload_bytes_total(1800); - stats.increment_me_d2c_write_mode(crate::stats::MeD2cWriteMode::Coalesced); - stats.increment_me_d2c_quota_reject_total(crate::stats::MeD2cQuotaRejectStage::PostWrite); - stats.observe_me_d2c_frame_buf_shrink(4096); - stats.increment_me_endpoint_quarantine_total(); - stats.increment_me_endpoint_quarantine_unexpected_total(); - stats.increment_me_endpoint_quarantine_draining_suppressed_total(); - stats.increment_user_connects("alice"); - stats.increment_user_curr_connects("alice"); - stats.add_user_octets_from("alice", 1024); - stats.add_user_octets_to("alice", 2048); - stats.increment_user_msgs_from("alice"); - stats.increment_user_msgs_to("alice"); - stats.increment_user_msgs_to("alice"); - tracker - .check_and_add("alice", "203.0.113.10".parse().unwrap()) - .await - .unwrap(); - - let output = render_metrics(&stats, shared_state.as_ref(), &config, &tracker, None).await; - - assert!(output.contains(&format!( - "telemt_build_info{{version=\"{}\"}} 1", - env!("CARGO_PKG_VERSION") - ))); - assert!(output.contains("telemt_connections_total 2")); - assert!(output.contains("telemt_connections_bad_total 1")); - assert!(output.contains( - "telemt_connections_bad_by_class_total{class=\"tls_handshake_bad_client\"} 1" - )); - assert!(output.contains("telemt_handshake_timeouts_total 1")); - assert!(output.contains("telemt_handshake_failures_by_class_total{class=\"timeout\"} 1")); - assert!(output.contains("telemt_auth_expensive_checks_total 9")); - assert!(output.contains("telemt_auth_budget_exhausted_total 2")); - assert!(output.contains("telemt_upstream_connect_attempt_total 2")); - assert!(output.contains("telemt_upstream_connect_success_total 1")); - assert!(output.contains("telemt_upstream_connect_fail_total 1")); - assert!(output.contains("telemt_upstream_connect_failfast_hard_error_total 1")); - assert!(output.contains("telemt_upstream_connect_attempts_per_request{bucket=\"2\"} 1")); - assert!( - output - .contains("telemt_upstream_connect_duration_success_total{bucket=\"101_500ms\"} 1") - ); - assert!( - output.contains("telemt_upstream_connect_duration_fail_total{bucket=\"gt_1000ms\"} 1") - ); - assert!(output.contains("telemt_me_rpc_proxy_req_signal_sent_total 1")); - assert!(output.contains("telemt_me_rpc_proxy_req_signal_failed_total 1")); - assert!(output.contains("telemt_me_rpc_proxy_req_signal_skipped_no_meta_total 1")); - assert!(output.contains("telemt_me_rpc_proxy_req_signal_response_total 1")); - assert!(output.contains("telemt_me_rpc_proxy_req_signal_close_sent_total 1")); - assert!(output.contains("telemt_me_idle_close_by_peer_total 1")); - assert!(output.contains("telemt_relay_idle_soft_mark_total 1")); - assert!(output.contains("telemt_relay_idle_hard_close_total 1")); - assert!(output.contains("telemt_relay_pressure_evict_total 1")); - assert!(output.contains("telemt_relay_protocol_desync_close_total 1")); - assert!(output.contains("telemt_me_d2c_batches_total 1")); - assert!(output.contains("telemt_me_d2c_batch_frames_total 3")); - assert!(output.contains("telemt_me_d2c_batch_bytes_total 2048")); - assert!(output.contains("telemt_me_d2c_flush_reason_total{reason=\"ack_immediate\"} 1")); - assert!(output.contains("telemt_me_d2c_data_frames_total 1")); - assert!(output.contains("telemt_me_d2c_ack_frames_total 1")); - assert!(output.contains("telemt_me_d2c_payload_bytes_total 1800")); - assert!(output.contains("telemt_me_d2c_write_mode_total{mode=\"coalesced\"} 1")); - assert!(output.contains("telemt_me_d2c_quota_reject_total{stage=\"post_write\"} 1")); - assert!(output.contains("telemt_me_d2c_frame_buf_shrink_total 1")); - assert!(output.contains("telemt_me_d2c_frame_buf_shrink_bytes_total 4096")); - assert!(output.contains("telemt_me_endpoint_quarantine_total 1")); - assert!(output.contains("telemt_me_endpoint_quarantine_unexpected_total 1")); - assert!(output.contains("telemt_me_endpoint_quarantine_draining_suppressed_total 1")); - assert!(output.contains("telemt_user_connections_total{user=\"alice\"} 1")); - assert!(output.contains("telemt_user_connections_current{user=\"alice\"} 1")); - assert!(output.contains("telemt_user_octets_from_client{user=\"alice\"} 1024")); - assert!(output.contains("telemt_user_octets_to_client{user=\"alice\"} 2048")); - assert!(output.contains("telemt_user_msgs_from_client{user=\"alice\"} 1")); - assert!(output.contains("telemt_user_msgs_to_client{user=\"alice\"} 2")); - assert!(output.contains("telemt_user_unique_ips_current{user=\"alice\"} 1")); - assert!(output.contains("telemt_user_unique_ips_recent_window{user=\"alice\"} 1")); - assert!(output.contains("telemt_user_unique_ips_limit{user=\"alice\"} 4")); - assert!(output.contains("telemt_user_unique_ips_utilization{user=\"alice\"} 0.250000")); - assert!(output.contains("telemt_ip_tracker_users{scope=\"active\"} 1")); - assert!(output.contains("telemt_ip_tracker_entries{scope=\"active\"} 1")); - assert!(output.contains("telemt_ip_tracker_cleanup_queue_len 0")); - } - - #[tokio::test] - async fn test_render_tls_front_profile_health() { - let stats = Stats::new(); - let shared_state = ProxySharedState::new(); - let tracker = UserIpTracker::new(); - let mut config = ProxyConfig::default(); - config.censorship.tls_domain = "primary.example".to_string(); - config.censorship.tls_domains = vec!["fallback.example".to_string()]; - - let cache = TlsFrontCache::new( - &[ - "primary.example".to_string(), - "fallback.example".to_string(), - ], - 1024, - "tlsfront-profile-health-test", - ); - cache - .set( - "primary.example", - CachedTlsData { - server_hello_template: ParsedServerHello { - version: [0x03, 0x03], - random: [0u8; 32], - session_id: Vec::new(), - cipher_suite: [0x13, 0x01], - compression: 0, - extensions: { - let mut key_share = vec![0x00, 0x1d, 0x00, 0x20]; - key_share.resize(36, 0x42); - vec![ - crate::tls_front::types::TlsExtension { - ext_type: 0x002b, - data: vec![0x03, 0x04], - }, - crate::tls_front::types::TlsExtension { - ext_type: 0x0033, - data: key_share, - }, - ] - }, - }, - cert_info: None, - cert_payload: Some(TlsCertPayload { - cert_chain_der: vec![vec![0x30, 0x01]], - certificate_message: vec![0x0b, 0x00, 0x00, 0x00], - }), - app_data_records_sizes: vec![1024, 512], - total_app_data_len: 1536, - behavior_profile: TlsBehaviorProfile { - change_cipher_spec_count: 1, - app_data_record_sizes: vec![1024, 512], - ticket_record_sizes: vec![69], - source: TlsProfileSource::Merged, - ..TlsBehaviorProfile::default() - }, - fetched_at: SystemTime::now(), - domain: "primary.example".to_string(), - }, - ) - .await; - - let output = render_metrics(&stats, &shared_state, &config, &tracker, Some(&cache)).await; - - assert!(output.contains("telemt_tls_front_profile_domains{status=\"configured\"} 2")); - assert!(output.contains("telemt_tls_front_profile_domains{status=\"emitted\"} 2")); - assert!(output.contains("telemt_tls_front_profile_domains{status=\"suppressed\"} 0")); - assert!( - output.contains("telemt_tls_front_profile_info{domain=\"primary.example\",source=\"merged\",is_default=\"false\",has_cert_info=\"false\",has_cert_payload=\"true\"} 1") - ); - assert!( - output.contains("telemt_tls_front_profile_info{domain=\"fallback.example\",source=\"default\",is_default=\"true\",has_cert_info=\"false\",has_cert_payload=\"false\"} 1") - ); - assert!( - output.contains("telemt_tls_front_profile_quality_info{domain=\"primary.example\",quality=\"raw_strict\",key_share_group=\"x25519\"} 1") - ); - assert!( - output.contains("telemt_tls_front_profile_quality_info{domain=\"fallback.example\",quality=\"fallback\",key_share_group=\"none\"} 1") - ); - assert!(output.contains( - "telemt_tls_front_profile_server_hello_bytes{domain=\"primary.example\"} 90" - )); - assert!(output.contains( - "telemt_tls_front_profile_server_hello_extensions{domain=\"primary.example\"} 2" - )); - assert!( - output.contains( - "telemt_tls_front_profile_app_data_records{domain=\"primary.example\"} 2" - ) - ); - assert!( - output - .contains("telemt_tls_front_profile_ticket_records{domain=\"primary.example\"} 1") - ); - assert!(output.contains( - "telemt_tls_front_profile_change_cipher_spec_records{domain=\"primary.example\"} 1" - )); - assert!( - output.contains( - "telemt_tls_front_profile_app_data_bytes{domain=\"primary.example\"} 1536" - ) - ); - } - - #[tokio::test] - async fn test_render_empty_stats() { - let stats = Stats::new(); - let shared_state = ProxySharedState::new(); - let tracker = UserIpTracker::new(); - let config = ProxyConfig::default(); - let output = render_metrics(&stats, &shared_state, &config, &tracker, None).await; - assert!(output.contains("telemt_connections_total 0")); - assert!(output.contains("telemt_connections_bad_total 0")); - assert!(output.contains("telemt_handshake_timeouts_total 0")); - assert!(output.contains("telemt_auth_expensive_checks_total 0")); - assert!(output.contains("telemt_auth_budget_exhausted_total 0")); - assert!(output.contains("telemt_user_unique_ips_current{user=")); - assert!(output.contains("telemt_user_unique_ips_recent_window{user=")); - } - - #[tokio::test] - async fn test_render_uses_global_each_unique_ip_limit() { - let stats = Stats::new(); - let shared_state = ProxySharedState::new(); - stats.increment_user_connects("alice"); - stats.increment_user_curr_connects("alice"); - let tracker = UserIpTracker::new(); - tracker - .check_and_add("alice", "203.0.113.10".parse().unwrap()) - .await - .unwrap(); - let mut config = ProxyConfig::default(); - config.access.user_max_unique_ips_global_each = 2; - - let output = render_metrics(&stats, &shared_state, &config, &tracker, None).await; - - assert!(output.contains("telemt_user_unique_ips_limit{user=\"alice\"} 2")); - assert!(output.contains("telemt_user_unique_ips_utilization{user=\"alice\"} 0.500000")); - } - - #[tokio::test] - async fn test_render_has_type_annotations() { - let stats = Stats::new(); - let shared_state = ProxySharedState::new(); - let tracker = UserIpTracker::new(); - let config = ProxyConfig::default(); - let output = render_metrics(&stats, &shared_state, &config, &tracker, None).await; - assert!(output.contains("# TYPE telemt_uptime_seconds gauge")); - assert!(output.contains("# TYPE telemt_connections_total counter")); - assert!(output.contains("# TYPE telemt_connections_bad_total counter")); - assert!(output.contains("# TYPE telemt_connections_bad_by_class_total counter")); - assert!(output.contains("# TYPE telemt_handshake_timeouts_total counter")); - assert!(output.contains("# TYPE telemt_handshake_failures_by_class_total counter")); - assert!(output.contains("# TYPE telemt_auth_expensive_checks_total counter")); - assert!(output.contains("# TYPE telemt_auth_budget_exhausted_total counter")); - assert!(output.contains("# TYPE telemt_upstream_connect_attempt_total counter")); - assert!(output.contains("# TYPE telemt_me_rpc_proxy_req_signal_sent_total counter")); - assert!(output.contains("# TYPE telemt_me_idle_close_by_peer_total counter")); - assert!(output.contains("# TYPE telemt_relay_idle_soft_mark_total counter")); - assert!(output.contains("# TYPE telemt_relay_idle_hard_close_total counter")); - assert!(output.contains("# TYPE telemt_relay_pressure_evict_total counter")); - assert!(output.contains("# TYPE telemt_relay_protocol_desync_close_total counter")); - assert!(output.contains("# TYPE telemt_me_d2c_batches_total counter")); - assert!(output.contains("# TYPE telemt_me_d2c_flush_reason_total counter")); - assert!(output.contains("# TYPE telemt_me_d2c_write_mode_total counter")); - assert!(output.contains("# TYPE telemt_me_d2c_batch_frames_bucket_total counter")); - assert!(output.contains("# TYPE telemt_me_d2c_flush_duration_us_bucket_total counter")); - assert!(output.contains("# TYPE telemt_me_endpoint_quarantine_total counter")); - assert!(output.contains("# TYPE telemt_me_endpoint_quarantine_unexpected_total counter")); - assert!( - output - .contains("# TYPE telemt_me_endpoint_quarantine_draining_suppressed_total counter") - ); - assert!(output.contains("# TYPE telemt_me_writer_removed_total counter")); - assert!( - output - .contains("# TYPE telemt_me_writer_removed_unexpected_minus_restored_total gauge") - ); - assert!(output.contains("# TYPE telemt_user_unique_ips_current gauge")); - 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") - ); - assert!(output.contains("# TYPE telemt_tls_front_profile_domains gauge")); - assert!(output.contains("# TYPE telemt_tls_front_profile_info gauge")); - assert!(output.contains("# TYPE telemt_tls_front_profile_quality_info gauge")); - assert!(output.contains("# TYPE telemt_tls_front_profile_age_seconds gauge")); - assert!(output.contains("# TYPE telemt_tls_front_profile_server_hello_bytes gauge")); - assert!(output.contains("# TYPE telemt_tls_front_profile_server_hello_extensions gauge")); - assert!(output.contains("# TYPE telemt_tls_front_profile_app_data_records gauge")); - assert!(output.contains("# TYPE telemt_tls_front_profile_ticket_records gauge")); - assert!( - output.contains("# TYPE telemt_tls_front_profile_change_cipher_spec_records gauge") - ); - assert!(output.contains("# TYPE telemt_tls_front_profile_app_data_bytes gauge")); - } - - #[tokio::test] - async fn test_endpoint_integration() { - let stats = Arc::new(Stats::new()); - let beobachten = Arc::new(BeobachtenStore::new()); - let shared_state = ProxySharedState::new(); - let tracker = UserIpTracker::new(); - let mut config = ProxyConfig::default(); - stats.increment_connects_all(); - stats.increment_connects_all(); - stats.increment_connects_all(); - - let req = Request::builder().uri("/metrics").body(()).unwrap(); - let resp = handle( - req, - &stats, - &beobachten, - shared_state.as_ref(), - &tracker, - None, - &config, - ) - .await - .unwrap(); - assert_eq!(resp.status(), StatusCode::OK); - let body = resp.into_body().collect().await.unwrap().to_bytes(); - assert!( - std::str::from_utf8(body.as_ref()) - .unwrap() - .contains("telemt_connections_total 3") - ); - assert!( - std::str::from_utf8(body.as_ref()) - .unwrap() - .contains(&format!( - "telemt_build_info{{version=\"{}\"}} 1", - env!("CARGO_PKG_VERSION") - )) - ); - - config.general.beobachten = true; - config.general.beobachten_minutes = 10; - beobachten.record( - "TLS-scanner", - "203.0.113.10".parse::().unwrap(), - Duration::from_secs(600), - ); - let req_beob = Request::builder().uri("/beobachten").body(()).unwrap(); - let resp_beob = handle( - req_beob, - &stats, - &beobachten, - shared_state.as_ref(), - &tracker, - None, - &config, - ) - .await - .unwrap(); - assert_eq!(resp_beob.status(), StatusCode::OK); - let body_beob = resp_beob.into_body().collect().await.unwrap().to_bytes(); - let beob_text = std::str::from_utf8(body_beob.as_ref()).unwrap(); - assert!(beob_text.contains("[TLS-scanner]")); - assert!(beob_text.contains("203.0.113.10-1")); - - let req404 = Request::builder().uri("/other").body(()).unwrap(); - let resp404 = handle( - req404, - &stats, - &beobachten, - shared_state.as_ref(), - &tracker, - None, - &config, - ) - .await - .unwrap(); - assert_eq!(resp404.status(), StatusCode::NOT_FOUND); - } -} diff --git a/src/metrics/tests.rs b/src/metrics/tests.rs new file mode 100644 index 0000000..8db5fdf --- /dev/null +++ b/src/metrics/tests.rs @@ -0,0 +1,417 @@ +use super::render::render_metrics; +use super::*; +use http_body_util::BodyExt; +use std::net::IpAddr; +use std::time::SystemTime; + +use crate::tls_front::types::{ + CachedTlsData, ParsedServerHello, TlsBehaviorProfile, TlsCertPayload, TlsProfileSource, +}; + +#[tokio::test] +async fn test_render_metrics_format() { + let stats = Arc::new(Stats::new()); + let shared_state = ProxySharedState::new(); + let tracker = UserIpTracker::new(); + let mut config = ProxyConfig::default(); + config + .access + .user_max_unique_ips + .insert("alice".to_string(), 4); + + stats.increment_connects_all(); + stats.increment_connects_all(); + stats.increment_connects_bad_with_class("tls_handshake_bad_client"); + stats.increment_handshake_timeouts(); + stats.increment_handshake_failure_class("timeout"); + shared_state + .handshake + .auth_expensive_checks_total + .fetch_add(9, std::sync::atomic::Ordering::Relaxed); + shared_state + .handshake + .auth_budget_exhausted_total + .fetch_add(2, std::sync::atomic::Ordering::Relaxed); + stats.increment_upstream_connect_attempt_total(); + stats.increment_upstream_connect_attempt_total(); + stats.increment_upstream_connect_success_total(); + stats.increment_upstream_connect_fail_total(); + stats.increment_upstream_connect_failfast_hard_error_total(); + stats.observe_upstream_connect_attempts_per_request(2); + stats.observe_upstream_connect_duration_ms(220, true); + stats.observe_upstream_connect_duration_ms(1500, false); + stats.increment_me_rpc_proxy_req_signal_sent_total(); + stats.increment_me_rpc_proxy_req_signal_failed_total(); + stats.increment_me_rpc_proxy_req_signal_skipped_no_meta_total(); + stats.increment_me_rpc_proxy_req_signal_response_total(); + stats.increment_me_rpc_proxy_req_signal_close_sent_total(); + stats.increment_me_idle_close_by_peer_total(); + stats.increment_relay_idle_soft_mark_total(); + stats.increment_relay_idle_hard_close_total(); + stats.increment_relay_pressure_evict_total(); + stats.increment_relay_protocol_desync_close_total(); + stats.increment_me_d2c_batches_total(); + stats.add_me_d2c_batch_frames_total(3); + stats.add_me_d2c_batch_bytes_total(2048); + stats.increment_me_d2c_flush_reason(crate::stats::MeD2cFlushReason::AckImmediate); + stats.increment_me_d2c_data_frames_total(); + stats.increment_me_d2c_ack_frames_total(); + stats.add_me_d2c_payload_bytes_total(1800); + stats.increment_me_d2c_write_mode(crate::stats::MeD2cWriteMode::Coalesced); + stats.increment_me_d2c_quota_reject_total(crate::stats::MeD2cQuotaRejectStage::PostWrite); + stats.observe_me_d2c_frame_buf_shrink(4096); + stats.increment_me_endpoint_quarantine_total(); + stats.increment_me_endpoint_quarantine_unexpected_total(); + stats.increment_me_endpoint_quarantine_draining_suppressed_total(); + stats.increment_user_connects("alice"); + stats.increment_user_curr_connects("alice"); + stats.add_user_octets_from("alice", 1024); + stats.add_user_octets_to("alice", 2048); + stats.increment_user_msgs_from("alice"); + stats.increment_user_msgs_to("alice"); + stats.increment_user_msgs_to("alice"); + tracker + .check_and_add("alice", "203.0.113.10".parse().unwrap()) + .await + .unwrap(); + + let output = render_metrics(&stats, shared_state.as_ref(), &config, &tracker, None).await; + + assert!(output.contains(&format!( + "telemt_build_info{{version=\"{}\"}} 1", + env!("CARGO_PKG_VERSION") + ))); + assert!(output.contains("telemt_connections_total 2")); + assert!(output.contains("telemt_connections_bad_total 1")); + assert!( + output.contains("telemt_connections_bad_by_class_total{class=\"tls_handshake_bad_client\"} 1") + ); + assert!(output.contains("telemt_handshake_timeouts_total 1")); + assert!(output.contains("telemt_handshake_failures_by_class_total{class=\"timeout\"} 1")); + assert!(output.contains("telemt_auth_expensive_checks_total 9")); + assert!(output.contains("telemt_auth_budget_exhausted_total 2")); + assert!(output.contains("telemt_upstream_connect_attempt_total 2")); + assert!(output.contains("telemt_upstream_connect_success_total 1")); + assert!(output.contains("telemt_upstream_connect_fail_total 1")); + assert!(output.contains("telemt_upstream_connect_failfast_hard_error_total 1")); + assert!(output.contains("telemt_upstream_connect_attempts_per_request{bucket=\"2\"} 1")); + assert!( + output.contains("telemt_upstream_connect_duration_success_total{bucket=\"101_500ms\"} 1") + ); + assert!(output.contains("telemt_upstream_connect_duration_fail_total{bucket=\"gt_1000ms\"} 1")); + assert!(output.contains("telemt_me_rpc_proxy_req_signal_sent_total 1")); + assert!(output.contains("telemt_me_rpc_proxy_req_signal_failed_total 1")); + assert!(output.contains("telemt_me_rpc_proxy_req_signal_skipped_no_meta_total 1")); + assert!(output.contains("telemt_me_rpc_proxy_req_signal_response_total 1")); + assert!(output.contains("telemt_me_rpc_proxy_req_signal_close_sent_total 1")); + assert!(output.contains("telemt_me_idle_close_by_peer_total 1")); + assert!(output.contains("telemt_relay_idle_soft_mark_total 1")); + assert!(output.contains("telemt_relay_idle_hard_close_total 1")); + assert!(output.contains("telemt_relay_pressure_evict_total 1")); + assert!(output.contains("telemt_relay_protocol_desync_close_total 1")); + assert!(output.contains("telemt_me_d2c_batches_total 1")); + assert!(output.contains("telemt_me_d2c_batch_frames_total 3")); + assert!(output.contains("telemt_me_d2c_batch_bytes_total 2048")); + assert!(output.contains("telemt_me_d2c_flush_reason_total{reason=\"ack_immediate\"} 1")); + assert!(output.contains("telemt_me_d2c_data_frames_total 1")); + assert!(output.contains("telemt_me_d2c_ack_frames_total 1")); + assert!(output.contains("telemt_me_d2c_payload_bytes_total 1800")); + assert!(output.contains("telemt_me_d2c_write_mode_total{mode=\"coalesced\"} 1")); + assert!(output.contains("telemt_me_d2c_quota_reject_total{stage=\"post_write\"} 1")); + assert!(output.contains("telemt_me_d2c_frame_buf_shrink_total 1")); + assert!(output.contains("telemt_me_d2c_frame_buf_shrink_bytes_total 4096")); + assert!(output.contains("telemt_me_endpoint_quarantine_total 1")); + assert!(output.contains("telemt_me_endpoint_quarantine_unexpected_total 1")); + assert!(output.contains("telemt_me_endpoint_quarantine_draining_suppressed_total 1")); + assert!(output.contains("telemt_user_connections_total{user=\"alice\"} 1")); + assert!(output.contains("telemt_user_connections_current{user=\"alice\"} 1")); + assert!(output.contains("telemt_user_octets_from_client{user=\"alice\"} 1024")); + assert!(output.contains("telemt_user_octets_to_client{user=\"alice\"} 2048")); + assert!(output.contains("telemt_user_msgs_from_client{user=\"alice\"} 1")); + assert!(output.contains("telemt_user_msgs_to_client{user=\"alice\"} 2")); + assert!(output.contains("telemt_user_unique_ips_current{user=\"alice\"} 1")); + assert!(output.contains("telemt_user_unique_ips_recent_window{user=\"alice\"} 1")); + assert!(output.contains("telemt_user_unique_ips_limit{user=\"alice\"} 4")); + assert!(output.contains("telemt_user_unique_ips_utilization{user=\"alice\"} 0.250000")); + assert!(output.contains("telemt_ip_tracker_users{scope=\"active\"} 1")); + assert!(output.contains("telemt_ip_tracker_entries{scope=\"active\"} 1")); + assert!(output.contains("telemt_ip_tracker_cleanup_queue_len 0")); +} + +#[tokio::test] +async fn test_render_tls_front_profile_health() { + let stats = Stats::new(); + let shared_state = ProxySharedState::new(); + let tracker = UserIpTracker::new(); + let mut config = ProxyConfig::default(); + config.censorship.tls_domain = "primary.example".to_string(); + config.censorship.tls_domains = vec!["fallback.example".to_string()]; + + let cache = TlsFrontCache::new( + &[ + "primary.example".to_string(), + "fallback.example".to_string(), + ], + 1024, + "tlsfront-profile-health-test", + ); + cache + .set( + "primary.example", + CachedTlsData { + server_hello_template: ParsedServerHello { + version: [0x03, 0x03], + random: [0u8; 32], + session_id: Vec::new(), + cipher_suite: [0x13, 0x01], + compression: 0, + extensions: { + let mut key_share = vec![0x00, 0x1d, 0x00, 0x20]; + key_share.resize(36, 0x42); + vec![ + crate::tls_front::types::TlsExtension { + ext_type: 0x002b, + data: vec![0x03, 0x04], + }, + crate::tls_front::types::TlsExtension { + ext_type: 0x0033, + data: key_share, + }, + ] + }, + }, + cert_info: None, + cert_payload: Some(TlsCertPayload { + cert_chain_der: vec![vec![0x30, 0x01]], + certificate_message: vec![0x0b, 0x00, 0x00, 0x00], + }), + app_data_records_sizes: vec![1024, 512], + total_app_data_len: 1536, + behavior_profile: TlsBehaviorProfile { + change_cipher_spec_count: 1, + app_data_record_sizes: vec![1024, 512], + ticket_record_sizes: vec![69], + source: TlsProfileSource::Merged, + ..TlsBehaviorProfile::default() + }, + fetched_at: SystemTime::now(), + domain: "primary.example".to_string(), + }, + ) + .await; + + let output = render_metrics(&stats, &shared_state, &config, &tracker, Some(&cache)).await; + + assert!(output.contains("telemt_tls_front_profile_domains{status=\"configured\"} 2")); + assert!(output.contains("telemt_tls_front_profile_domains{status=\"emitted\"} 2")); + assert!(output.contains("telemt_tls_front_profile_domains{status=\"suppressed\"} 0")); + assert!( + output.contains("telemt_tls_front_profile_info{domain=\"primary.example\",source=\"merged\",is_default=\"false\",has_cert_info=\"false\",has_cert_payload=\"true\"} 1") + ); + assert!( + output.contains("telemt_tls_front_profile_info{domain=\"fallback.example\",source=\"default\",is_default=\"true\",has_cert_info=\"false\",has_cert_payload=\"false\"} 1") + ); + assert!( + output.contains("telemt_tls_front_profile_quality_info{domain=\"primary.example\",quality=\"raw_strict\",key_share_group=\"x25519\"} 1") + ); + assert!( + output.contains("telemt_tls_front_profile_quality_info{domain=\"fallback.example\",quality=\"fallback\",key_share_group=\"none\"} 1") + ); + assert!( + output + .contains("telemt_tls_front_profile_server_hello_bytes{domain=\"primary.example\"} 90") + ); + assert!(output.contains( + "telemt_tls_front_profile_server_hello_extensions{domain=\"primary.example\"} 2" + )); + assert!( + output.contains("telemt_tls_front_profile_app_data_records{domain=\"primary.example\"} 2") + ); + assert!( + output.contains("telemt_tls_front_profile_ticket_records{domain=\"primary.example\"} 1") + ); + assert!(output.contains( + "telemt_tls_front_profile_change_cipher_spec_records{domain=\"primary.example\"} 1" + )); + assert!( + output.contains("telemt_tls_front_profile_app_data_bytes{domain=\"primary.example\"} 1536") + ); +} + +#[tokio::test] +async fn test_render_empty_stats() { + let stats = Stats::new(); + let shared_state = ProxySharedState::new(); + let tracker = UserIpTracker::new(); + let config = ProxyConfig::default(); + let output = render_metrics(&stats, &shared_state, &config, &tracker, None).await; + assert!(output.contains("telemt_connections_total 0")); + assert!(output.contains("telemt_connections_bad_total 0")); + assert!(output.contains("telemt_handshake_timeouts_total 0")); + assert!(output.contains("telemt_auth_expensive_checks_total 0")); + assert!(output.contains("telemt_auth_budget_exhausted_total 0")); + assert!(output.contains("telemt_user_unique_ips_current{user=")); + assert!(output.contains("telemt_user_unique_ips_recent_window{user=")); +} + +#[tokio::test] +async fn test_render_uses_global_each_unique_ip_limit() { + let stats = Stats::new(); + let shared_state = ProxySharedState::new(); + stats.increment_user_connects("alice"); + stats.increment_user_curr_connects("alice"); + let tracker = UserIpTracker::new(); + tracker + .check_and_add("alice", "203.0.113.10".parse().unwrap()) + .await + .unwrap(); + let mut config = ProxyConfig::default(); + config.access.user_max_unique_ips_global_each = 2; + + let output = render_metrics(&stats, &shared_state, &config, &tracker, None).await; + + assert!(output.contains("telemt_user_unique_ips_limit{user=\"alice\"} 2")); + assert!(output.contains("telemt_user_unique_ips_utilization{user=\"alice\"} 0.500000")); +} + +#[tokio::test] +async fn test_render_has_type_annotations() { + let stats = Stats::new(); + let shared_state = ProxySharedState::new(); + let tracker = UserIpTracker::new(); + let config = ProxyConfig::default(); + let output = render_metrics(&stats, &shared_state, &config, &tracker, None).await; + assert!(output.contains("# TYPE telemt_uptime_seconds gauge")); + assert!(output.contains("# TYPE telemt_connections_total counter")); + assert!(output.contains("# TYPE telemt_connections_bad_total counter")); + assert!(output.contains("# TYPE telemt_connections_bad_by_class_total counter")); + assert!(output.contains("# TYPE telemt_handshake_timeouts_total counter")); + assert!(output.contains("# TYPE telemt_handshake_failures_by_class_total counter")); + assert!(output.contains("# TYPE telemt_auth_expensive_checks_total counter")); + assert!(output.contains("# TYPE telemt_auth_budget_exhausted_total counter")); + assert!(output.contains("# TYPE telemt_upstream_connect_attempt_total counter")); + assert!(output.contains("# TYPE telemt_me_rpc_proxy_req_signal_sent_total counter")); + assert!(output.contains("# TYPE telemt_me_idle_close_by_peer_total counter")); + assert!(output.contains("# TYPE telemt_relay_idle_soft_mark_total counter")); + assert!(output.contains("# TYPE telemt_relay_idle_hard_close_total counter")); + assert!(output.contains("# TYPE telemt_relay_pressure_evict_total counter")); + assert!(output.contains("# TYPE telemt_relay_protocol_desync_close_total counter")); + assert!(output.contains("# TYPE telemt_me_d2c_batches_total counter")); + assert!(output.contains("# TYPE telemt_me_d2c_flush_reason_total counter")); + assert!(output.contains("# TYPE telemt_me_d2c_write_mode_total counter")); + assert!(output.contains("# TYPE telemt_me_d2c_batch_frames_bucket_total counter")); + assert!(output.contains("# TYPE telemt_me_d2c_flush_duration_us_bucket_total counter")); + assert!(output.contains("# TYPE telemt_me_endpoint_quarantine_total counter")); + assert!(output.contains("# TYPE telemt_me_endpoint_quarantine_unexpected_total counter")); + assert!( + output.contains("# TYPE telemt_me_endpoint_quarantine_draining_suppressed_total counter") + ); + assert!(output.contains("# TYPE telemt_me_writer_removed_total counter")); + assert!( + output.contains("# TYPE telemt_me_writer_removed_unexpected_minus_restored_total gauge") + ); + assert!(output.contains("# TYPE telemt_user_unique_ips_current gauge")); + 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")); + assert!(output.contains("# TYPE telemt_tls_front_profile_domains gauge")); + assert!(output.contains("# TYPE telemt_tls_front_profile_info gauge")); + assert!(output.contains("# TYPE telemt_tls_front_profile_quality_info gauge")); + assert!(output.contains("# TYPE telemt_tls_front_profile_age_seconds gauge")); + assert!(output.contains("# TYPE telemt_tls_front_profile_server_hello_bytes gauge")); + assert!(output.contains("# TYPE telemt_tls_front_profile_server_hello_extensions gauge")); + assert!(output.contains("# TYPE telemt_tls_front_profile_app_data_records gauge")); + assert!(output.contains("# TYPE telemt_tls_front_profile_ticket_records gauge")); + assert!(output.contains("# TYPE telemt_tls_front_profile_change_cipher_spec_records gauge")); + assert!(output.contains("# TYPE telemt_tls_front_profile_app_data_bytes gauge")); +} + +#[tokio::test] +async fn test_endpoint_integration() { + let stats = Arc::new(Stats::new()); + let beobachten = Arc::new(BeobachtenStore::new()); + let shared_state = ProxySharedState::new(); + let tracker = UserIpTracker::new(); + let mut config = ProxyConfig::default(); + stats.increment_connects_all(); + stats.increment_connects_all(); + stats.increment_connects_all(); + + let req = Request::builder().uri("/metrics").body(()).unwrap(); + let resp = handle( + req, + &stats, + &beobachten, + shared_state.as_ref(), + &tracker, + None, + &config, + ) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + let body = resp.into_body().collect().await.unwrap().to_bytes(); + assert!( + std::str::from_utf8(body.as_ref()) + .unwrap() + .contains("telemt_connections_total 3") + ); + assert!( + std::str::from_utf8(body.as_ref()) + .unwrap() + .contains(&format!( + "telemt_build_info{{version=\"{}\"}} 1", + env!("CARGO_PKG_VERSION") + )) + ); + + config.general.beobachten = true; + config.general.beobachten_minutes = 10; + beobachten.record( + "TLS-scanner", + "203.0.113.10".parse::().unwrap(), + Duration::from_secs(600), + ); + let req_beob = Request::builder().uri("/beobachten").body(()).unwrap(); + let resp_beob = handle( + req_beob, + &stats, + &beobachten, + shared_state.as_ref(), + &tracker, + None, + &config, + ) + .await + .unwrap(); + assert_eq!(resp_beob.status(), StatusCode::OK); + let body_beob = resp_beob.into_body().collect().await.unwrap().to_bytes(); + let beob_text = std::str::from_utf8(body_beob.as_ref()).unwrap(); + assert!(beob_text.contains("[TLS-scanner]")); + assert!(beob_text.contains("203.0.113.10-1")); + + let req404 = Request::builder().uri("/other").body(()).unwrap(); + let resp404 = handle( + req404, + &stats, + &beobachten, + shared_state.as_ref(), + &tracker, + None, + &config, + ) + .await + .unwrap(); + assert_eq!(resp404.status(), StatusCode::NOT_FOUND); +} diff --git a/src/metrics/tls_front.rs b/src/metrics/tls_front.rs new file mode 100644 index 0000000..1a4db5c --- /dev/null +++ b/src/metrics/tls_front.rs @@ -0,0 +1,167 @@ +use crate::config::ProxyConfig; +use crate::tls_front::TlsFrontCache; + +// Keeps TLS-front per-domain health series bounded for large generated configs. +const TLS_FRONT_PROFILE_HEALTH_MAX_DOMAINS: usize = 256; + +fn tls_front_domains(config: &ProxyConfig) -> Vec { + let mut domains = Vec::with_capacity(1 + config.censorship.tls_domains.len()); + if !config.censorship.tls_domain.is_empty() { + domains.push(config.censorship.tls_domain.clone()); + } + for domain in &config.censorship.tls_domains { + if !domain.is_empty() && !domains.contains(domain) { + domains.push(domain.clone()); + } + } + domains +} + +fn prometheus_label_value(value: &str) -> String { + value.replace('\\', "\\\\").replace('"', "\\\"") +} + +pub(super) async fn render_tls_front_profile_health( + out: &mut String, + config: &ProxyConfig, + tls_cache: Option<&TlsFrontCache>, +) { + use std::fmt::Write; + + let domains = tls_front_domains(config); + let (health, suppressed) = match (config.censorship.tls_emulation, tls_cache) { + (true, Some(cache)) => { + cache + .profile_health_snapshot(&domains, TLS_FRONT_PROFILE_HEALTH_MAX_DOMAINS) + .await + } + _ => (Vec::new(), domains.len()), + }; + + let _ = writeln!( + out, + "# HELP telemt_tls_front_profile_domains TLS front configured profile domains by export status" + ); + let _ = writeln!(out, "# TYPE telemt_tls_front_profile_domains gauge"); + let _ = writeln!( + out, + "telemt_tls_front_profile_domains{{status=\"configured\"}} {}", + domains.len() + ); + let _ = writeln!( + out, + "telemt_tls_front_profile_domains{{status=\"emitted\"}} {}", + health.len() + ); + let _ = writeln!( + out, + "telemt_tls_front_profile_domains{{status=\"suppressed\"}} {}", + suppressed + ); + let _ = writeln!( + out, + "# HELP telemt_tls_front_profile_info TLS front profile source and feature flags per configured domain" + ); + let _ = writeln!(out, "# TYPE telemt_tls_front_profile_info gauge"); + let _ = writeln!( + out, + "# HELP telemt_tls_front_profile_quality_info TLS front profile quality and key-share group per configured domain" + ); + let _ = writeln!(out, "# TYPE telemt_tls_front_profile_quality_info gauge"); + let _ = writeln!( + out, + "# HELP telemt_tls_front_profile_age_seconds Age of cached TLS front profile data per configured domain" + ); + let _ = writeln!(out, "# TYPE telemt_tls_front_profile_age_seconds gauge"); + let _ = writeln!( + out, + "# HELP telemt_tls_front_profile_server_hello_bytes TLS front cached ServerHello record body bytes per configured domain" + ); + let _ = writeln!( + out, + "# TYPE telemt_tls_front_profile_server_hello_bytes gauge" + ); + let _ = writeln!( + out, + "# HELP telemt_tls_front_profile_server_hello_extensions TLS front cached visible ServerHello extension count per configured domain" + ); + let _ = writeln!( + out, + "# TYPE telemt_tls_front_profile_server_hello_extensions gauge" + ); + let _ = writeln!( + out, + "# HELP telemt_tls_front_profile_app_data_records TLS front cached app-data record count per configured domain" + ); + let _ = writeln!( + out, + "# TYPE telemt_tls_front_profile_app_data_records gauge" + ); + let _ = writeln!( + out, + "# HELP telemt_tls_front_profile_ticket_records TLS front cached ticket-like tail record count per configured domain" + ); + let _ = writeln!(out, "# TYPE telemt_tls_front_profile_ticket_records gauge"); + let _ = writeln!( + out, + "# HELP telemt_tls_front_profile_change_cipher_spec_records TLS front cached ChangeCipherSpec record count per configured domain" + ); + let _ = writeln!( + out, + "# TYPE telemt_tls_front_profile_change_cipher_spec_records gauge" + ); + let _ = writeln!( + out, + "# HELP telemt_tls_front_profile_app_data_bytes TLS front cached total app-data bytes per configured domain" + ); + let _ = writeln!(out, "# TYPE telemt_tls_front_profile_app_data_bytes gauge"); + + for item in health { + let domain = prometheus_label_value(&item.domain); + let _ = writeln!( + out, + "telemt_tls_front_profile_info{{domain=\"{}\",source=\"{}\",is_default=\"{}\",has_cert_info=\"{}\",has_cert_payload=\"{}\"}} 1", + domain, item.source, item.is_default, item.has_cert_info, item.has_cert_payload + ); + let _ = writeln!( + out, + "telemt_tls_front_profile_quality_info{{domain=\"{}\",quality=\"{}\",key_share_group=\"{}\"}} 1", + domain, item.quality, item.key_share_group + ); + let _ = writeln!( + out, + "telemt_tls_front_profile_age_seconds{{domain=\"{}\"}} {}", + domain, item.age_seconds + ); + let _ = writeln!( + out, + "telemt_tls_front_profile_server_hello_bytes{{domain=\"{}\"}} {}", + domain, item.server_hello_record_len + ); + let _ = writeln!( + out, + "telemt_tls_front_profile_server_hello_extensions{{domain=\"{}\"}} {}", + domain, item.server_hello_extensions + ); + let _ = writeln!( + out, + "telemt_tls_front_profile_app_data_records{{domain=\"{}\"}} {}", + domain, item.app_data_records + ); + let _ = writeln!( + out, + "telemt_tls_front_profile_ticket_records{{domain=\"{}\"}} {}", + domain, item.ticket_records + ); + let _ = writeln!( + out, + "telemt_tls_front_profile_change_cipher_spec_records{{domain=\"{}\"}} {}", + domain, item.change_cipher_spec_count + ); + let _ = writeln!( + out, + "telemt_tls_front_profile_app_data_bytes{{domain=\"{}\"}} {}", + domain, item.total_app_data_len + ); + } +} diff --git a/src/proxy/middle_relay/session.rs b/src/proxy/middle_relay/session.rs index 371a5c5..22477ee 100644 --- a/src/proxy/middle_relay/session.rs +++ b/src/proxy/middle_relay/session.rs @@ -78,25 +78,11 @@ where return Err(ProxyError::RouteSwitched); } - // Per-user ad_tag from access.user_ad_tags; fallback to general.ad_tag (hot-reloadable) - let user_tag: Option> = config - .access - .user_ad_tags - .get(&user) - .and_then(|s| hex::decode(s).ok()) - .filter(|v| v.len() == 16); - let global_tag: Option> = config - .general - .ad_tag - .as_ref() - .and_then(|s| hex::decode(s).ok()) - .filter(|v| v.len() == 16); - let effective_tag = user_tag.or(global_tag); - - let proto_flags = proto_flags_for_tag(proto_tag, effective_tag.is_some()); - let effective_tag_array = effective_tag - .as_deref() - .and_then(|tag| <[u8; 16]>::try_from(tag).ok()); + // Per-user ad_tag from access.user_ad_tags with fallback to general.ad_tag. + // Decoded once at config load / hot-reload (see ProxyConfig::effective_ad_tag), + // so this is a cache lookup rather than a per-session hex decode + allocation. + let effective_tag_array = config.effective_ad_tag(&user); + let proto_flags = proto_flags_for_tag(proto_tag, effective_tag_array.is_some()); debug!( trace_id = format_args!("0x{:016x}", trace_id), user = %user,