From 8b9277dabfcdd09a00835534883189beeb995d42 Mon Sep 17 00:00:00 2001 From: Matthew Meszaros Date: Thu, 11 Jun 2026 08:11:05 +0200 Subject: [PATCH] feat: harden tracking service against abuse with per-IP rate limiting, prefetch/scanner filtering, URL length caps, and HMAC-signed click redirects (TRACKING_LINK_SECRET) closing the open-redirect hole --- internal/tasks/template.go | 25 ++++++++ tracking/Cargo.lock | 2 + tracking/Cargo.toml | 2 + tracking/src/abuse.rs | 125 +++++++++++++++++++++++++++++++++++++ tracking/src/config.rs | 29 +++++++++ tracking/src/handlers.rs | 80 +++++++++++++++++++----- tracking/src/main.rs | 3 +- 7 files changed, 251 insertions(+), 15 deletions(-) create mode 100644 tracking/src/abuse.rs diff --git a/internal/tasks/template.go b/internal/tasks/template.go index 9f2099b6..78dc21bd 100644 --- a/internal/tasks/template.go +++ b/internal/tasks/template.go @@ -1,9 +1,13 @@ package tasks import ( + "crypto/hmac" + "crypto/sha256" + "encoding/hex" "fmt" "math/rand" "net/url" + "os" "regexp" "strings" "sync" @@ -15,6 +19,24 @@ import ( "github.com/warmbly/warmbly/internal/pkg/warmpersona" ) +// trackingLinkSecret signs click-tracking redirects so the tracking service +// can refuse forged ?url= values (open-redirect abuse). Shared with the Rust +// service via the same TRACKING_LINK_SECRET env; empty = legacy unsigned. +var trackingLinkSecret = os.Getenv("TRACKING_LINK_SECRET") + +// signTrackingURL returns the hex HMAC-SHA256 tag binding a click redirect to +// (taskID, originalURL), or "" when signing is not configured. +func signTrackingURL(taskID uuid.UUID, originalURL string) string { + if trackingLinkSecret == "" { + return "" + } + mac := hmac.New(sha256.New, []byte(trackingLinkSecret)) + mac.Write([]byte(taskID.String())) + mac.Write([]byte("|")) + mac.Write([]byte(originalURL)) + return hex.EncodeToString(mac.Sum(nil)) +} + // Conversation represents a warmup conversation for AI generation type Conversation struct { ID uuid.UUID @@ -306,6 +328,9 @@ func WrapLinksForTracking(htmlBody string, taskID uuid.UUID, trackingDomain stri trackingDomain, taskID.String(), url.QueryEscape(originalURL)) + if sig := signTrackingURL(taskID, originalURL); sig != "" { + trackingURL += "&s=" + sig + } return fmt.Sprintf(`href="%s"`, trackingURL) }) diff --git a/tracking/Cargo.lock b/tracking/Cargo.lock index 01b4e47b..de2b154a 100644 --- a/tracking/Cargo.lock +++ b/tracking/Cargo.lock @@ -2924,6 +2924,8 @@ dependencies = [ "axum", "base64 0.21.7", "chrono", + "hex", + "hmac", "moka", "rdkafka", "reqwest 0.11.27", diff --git a/tracking/Cargo.toml b/tracking/Cargo.toml index 647b5eb6..9b8ea003 100644 --- a/tracking/Cargo.toml +++ b/tracking/Cargo.toml @@ -27,6 +27,8 @@ serde = { version = "1", features = ["derive"] } serde_json = "1" chrono = { version = "0.4", features = ["serde"] } sha2 = "0.10" +hmac = "0.12" +hex = "0.4" base64 = "0.21" urlencoding = "2" moka = { version = "0.12", features = ["future"] } diff --git a/tracking/src/abuse.rs b/tracking/src/abuse.rs new file mode 100644 index 00000000..f45d30f5 --- /dev/null +++ b/tracking/src/abuse.rs @@ -0,0 +1,125 @@ +//! Anti-abuse layer for the tracking endpoints. +//! +//! Three independent controls, all applied before an event reaches Kafka: +//! - per-source rate limiting (fixed 60s window, bounded cache) +//! - prefetch / scanner filtering (the response is still served so real +//! clients never break; only the analytics event is suppressed) +//! - HMAC verification of click redirects, so the tracking domain cannot be +//! used as an open redirector with forged `?url=` values. + +use axum::http::HeaderMap; +use hmac::{Hmac, Mac}; +use moka::future::Cache; +use sha2::Sha256; +use std::sync::atomic::{AtomicU32, Ordering}; +use std::sync::Arc; +use std::time::Duration; + +/// Fixed-window per-source request counter. Window resets via entry TTL, the +/// cache is hard-capped so a botnet rotating sources cannot exhaust memory. +pub struct RateLimiter { + buckets: Cache>, + limit: u32, +} + +impl RateLimiter { + pub fn new(limit_per_min: u32) -> Self { + Self { + buckets: Cache::builder() + .max_capacity(50_000) + .time_to_live(Duration::from_secs(60)) + .build(), + limit: limit_per_min, + } + } + + /// Returns true while the source is within its per-minute budget. + pub async fn allow(&self, source: &str) -> bool { + let counter = self + .buckets + .get_with(source.to_string(), async { Arc::new(AtomicU32::new(0)) }) + .await; + counter.fetch_add(1, Ordering::Relaxed) < self.limit + } +} + +/// Browser/link-warming prefetches and previews: the fetch is speculative, +/// not a human open/click, so it must not count. +pub fn is_prefetch(headers: &HeaderMap) -> bool { + for name in ["sec-purpose", "purpose", "x-purpose", "x-moz"] { + if let Some(value) = headers.get(name).and_then(|h| h.to_str().ok()) { + let value = value.to_ascii_lowercase(); + if value.contains("prefetch") || value.contains("preview") || value.contains("prerender") { + return true; + } + } + } + false +} + +/// UA markers for crawlers, CLI clients, link-expanding chat apps, uptime +/// monitors, and email security gateways that follow every link in a message. +/// Gmail's image proxy is deliberately NOT listed: it is the only open signal +/// Gmail exposes, and filtering it would zero out opens for Gmail recipients. +const SCANNER_UA_MARKERS: &[&str] = &[ + "bot", + "spider", + "crawl", + "curl/", + "wget/", + "python-requests", + "python/", + "go-http-client", + "okhttp", + "java/", + "headless", + "phantomjs", + "validator", + "pingdom", + "uptime", + "statuscake", + "site24x7", + "bingpreview", + "skypeuripreview", + "whatsapp", + "telegram", + // email security gateways / link rewriters + "urldefense", + "safelinks", + "barracuda", + "mimecast", + "proofpoint", + "forcepoint", + "symantec", + "trendmicro", + "sophos", + "zscaler", +]; + +pub fn is_scanner(user_agent: Option<&str>) -> bool { + let Some(ua) = user_agent else { + // No UA at all is never a real mail client or browser. + return true; + }; + let ua = ua.to_ascii_lowercase(); + SCANNER_UA_MARKERS.iter().any(|marker| ua.contains(marker)) +} + +type HmacSha256 = Hmac; + +/// Verify the `s` query parameter of a click redirect: hex HMAC-SHA256 over +/// `"{task_id}|{original_url}"` with the shared TRACKING_LINK_SECRET. The Go +/// sender signs at link-rewrite time; anything unsigned or mis-signed is a +/// forged redirect. +pub fn verify_signature(secret: &str, task_id: &str, original_url: &str, sig: Option<&str>) -> bool { + let Some(sig) = sig else { return false }; + let Ok(sig_bytes) = hex::decode(sig) else { + return false; + }; + + let mut mac = HmacSha256::new_from_slice(secret.as_bytes()).expect("hmac accepts any key size"); + mac.update(task_id.as_bytes()); + mac.update(b"|"); + mac.update(original_url.as_bytes()); + mac.verify_slice(&sig_bytes).is_ok() +} diff --git a/tracking/src/config.rs b/tracking/src/config.rs index e11566d4..e82d27af 100644 --- a/tracking/src/config.rs +++ b/tracking/src/config.rs @@ -31,6 +31,11 @@ pub struct Config { pub schema_registry_url: String, pub schema_registry_key: Option, pub schema_registry_secret: Option, + /// Shared secret for HMAC-signed click redirects. When set, unsigned or + /// mis-signed /t/c/ requests are refused (open-redirect protection). + pub link_secret: Option, + /// Per-source request budget for both tracking endpoints (default 300/min). + pub rate_limit_per_min: u32, } impl Config { @@ -119,6 +124,21 @@ impl Config { info!("Schema Registry authentication enabled"); } + // Optional signed-link secret (must match the sender's TRACKING_LINK_SECRET) + let link_secret = + Self::get_secret_optional("TRACKING_LINK_SECRET", "tracking/link_secret", &secrets) + .await + .filter(|s| !s.is_empty()); + if link_secret.is_some() { + info!("Signed click redirects enforced"); + } + + let rate_limit_per_min: u32 = env::var("TRACKING_RATE_LIMIT_PER_MIN") + .ok() + .and_then(|v| v.parse().ok()) + .unwrap_or(300); + info!("Per-source rate limit: {}/min", rate_limit_per_min); + Ok(Self { env: env_name, host, @@ -130,6 +150,8 @@ impl Config { schema_registry_url, schema_registry_key, schema_registry_secret, + link_secret, + rate_limit_per_min, }) } @@ -179,6 +201,11 @@ impl Config { info!("Schema Registry authentication enabled"); } + let link_secret = secrets + .get_optional("tracking/link_secret") + .await + .filter(|s| !s.is_empty()); + Ok(Self { env: env.to_string(), host, @@ -190,6 +217,8 @@ impl Config { schema_registry_url, schema_registry_key, schema_registry_secret, + link_secret, + rate_limit_per_min: 300, }) } diff --git a/tracking/src/handlers.rs b/tracking/src/handlers.rs index e5f25a42..7c2554b1 100644 --- a/tracking/src/handlers.rs +++ b/tracking/src/handlers.rs @@ -10,8 +10,15 @@ use std::collections::HashMap; use std::sync::Arc; use std::time::Duration; +use crate::abuse::{is_prefetch, is_scanner, verify_signature, RateLimiter}; +use crate::config::Config; use crate::kafka::{KafkaProducer, TrackingEvent}; +/// Raw (still-encoded) `?url=` values longer than this are rejected before +/// decoding; decoded URLs are capped at the practical browser URL limit. +const MAX_RAW_URL_LEN: usize = 4096; +const MAX_URL_LEN: usize = 2048; + // 1x1 transparent GIF (43 bytes) const TRANSPARENT_GIF: &[u8] = &[ 0x47, 0x49, 0x46, 0x38, 0x39, 0x61, 0x01, 0x00, 0x01, 0x00, 0x80, 0x00, 0x00, 0xFF, 0xFF, 0xFF, @@ -29,10 +36,14 @@ pub struct AppState { /// Cache to deduplicate tracking events /// Each event type + task + IP is cached for 1 hour pub dedupe_cache: Arc, + /// Per-source request budget (anti-flood) + pub rate_limiter: Arc, + /// Shared secret for signed click redirects; None = legacy unsigned links + pub link_secret: Option>, } impl AppState { - pub fn new(kafka: KafkaProducer) -> Self { + pub fn new(kafka: KafkaProducer, config: &Config) -> Self { // Create cache with: // - Max 100k entries // - TTL of 1 hour per entry @@ -46,6 +57,8 @@ impl AppState { Self { kafka, dedupe_cache: Arc::new(dedupe_cache), + rate_limiter: Arc::new(RateLimiter::new(config.rate_limit_per_min)), + link_secret: config.link_secret.clone().map(Arc::new), } } @@ -94,12 +107,13 @@ pub async fn track_open( return pixel_response(); } - // Extract IP hash for deduplication + // Extract IP hash for deduplication + rate limiting let ip_hash = extract_ip_hash(&headers); - // Check for duplicate (same task + IP within 1 hour) - if state.is_duplicate("OPEN", &task_id, &ip_hash).await { - // Still return pixel but don't publish event + // Anti-flood: over-budget sources still get the pixel (real mail clients + // must never see a broken image), but nothing is published. + let source = ip_hash.clone().unwrap_or_else(|| "unknown".to_string()); + if !state.rate_limiter.allow(&source).await { return pixel_response(); } @@ -109,6 +123,17 @@ pub async fn track_open( .and_then(|h| h.to_str().ok()) .map(|s| s.to_string()); + // Speculative fetches and scanners are served but never counted. + if is_prefetch(&headers) || is_scanner(user_agent.as_deref()) { + return pixel_response(); + } + + // Check for duplicate (same task + IP within 1 hour) + if state.is_duplicate("OPEN", &task_id, &ip_hash).await { + // Still return pixel but don't publish event + return pixel_response(); + } + // Publish event asynchronously (fire and forget) let kafka = state.kafka.clone(); tokio::spawn(async move { @@ -138,6 +163,9 @@ pub async fn track_click( // Get original URL from query params first (we need to redirect regardless) let original_url = match params.get("url") { Some(url) => { + if url.len() > MAX_RAW_URL_LEN { + return (StatusCode::BAD_REQUEST, "URL too long").into_response(); + } // Decode URL urlencoding::decode(url) .map(|s| s.into_owned()) @@ -149,19 +177,37 @@ pub async fn track_click( }; // Basic URL validation + if original_url.len() > MAX_URL_LEN { + return (StatusCode::BAD_REQUEST, "URL too long").into_response(); + } if !original_url.starts_with("http://") && !original_url.starts_with("https://") { return (StatusCode::BAD_REQUEST, "Invalid URL").into_response(); } + // Signed-link enforcement: when the shared secret is configured, only + // redirects minted by our own send pipeline are honored. This is what + // stops the tracking domain from being abused as an open redirector. + if let Some(secret) = &state.link_secret { + if !verify_signature(secret, &task_id, &original_url, params.get("s").map(String::as_str)) { + return (StatusCode::NOT_FOUND, "Unknown link").into_response(); + } + } + + // Anti-flood: refuse the redirect outright over budget. Unlike the pixel + // there is no rendering concern, and serving unlimited redirects would + // keep the redirector attractive to abusers even with events suppressed. + let ip_hash = extract_ip_hash(&headers); + let source = ip_hash.clone().unwrap_or_else(|| "unknown".to_string()); + if !state.rate_limiter.allow(&source).await { + return (StatusCode::TOO_MANY_REQUESTS, "Slow down").into_response(); + } + // Validate task_id is a valid UUID format if uuid::Uuid::parse_str(&task_id).is_err() { // Still redirect but don't track return Redirect::temporary(&original_url).into_response(); } - // Extract IP hash for deduplication - let ip_hash = extract_ip_hash(&headers); - // Create a unique key for this specific link click (task + URL + IP) let url_hash = { let mut hasher = Sha256::new(); @@ -172,18 +218,24 @@ pub async fn track_click( let dedupe_key = format!("{}:{}", task_id, url_hash); - // Check for duplicate (same task + URL + IP within 1 hour) - if state.is_duplicate("CLICK", &dedupe_key, &ip_hash).await { - // Still redirect but don't publish event - return Redirect::temporary(&original_url).into_response(); - } - // Extract metadata from request let user_agent = headers .get(header::USER_AGENT) .and_then(|h| h.to_str().ok()) .map(|s| s.to_string()); + // Security gateways and link previewers follow every URL in a message; + // serve them the destination but never count a click. + if is_prefetch(&headers) || is_scanner(user_agent.as_deref()) { + return Redirect::temporary(&original_url).into_response(); + } + + // Check for duplicate (same task + URL + IP within 1 hour) + if state.is_duplicate("CLICK", &dedupe_key, &ip_hash).await { + // Still redirect but don't publish event + return Redirect::temporary(&original_url).into_response(); + } + // Publish event asynchronously (fire and forget) let kafka = state.kafka.clone(); let original_url_clone = original_url.clone(); diff --git a/tracking/src/main.rs b/tracking/src/main.rs index fbfb8a05..a8d72f21 100644 --- a/tracking/src/main.rs +++ b/tracking/src/main.rs @@ -1,3 +1,4 @@ +mod abuse; mod aws; mod config; mod handlers; @@ -49,7 +50,7 @@ async fn main() { } }; - let state = AppState::new(kafka); + let state = AppState::new(kafka, &config); // Build router let app = Router::new()