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

@@ -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)
 	})

tracking/Cargo.lock

@@ -2924,6 +2924,8 @@ dependencies = [
  "axum",
  "base64 0.21.7",
  "chrono",
+ "hex",
+ "hmac",
  "moka",
  "rdkafka",
  "reqwest 0.11.27",

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"] }

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<String, Arc<AtomicU32>>,
+    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<Sha256>;
+
+/// 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()
+}

tracking/src/config.rs

@@ -31,6 +31,11 @@ pub struct Config {
     pub schema_registry_url: String,
     pub schema_registry_key: Option<String>,
     pub schema_registry_secret: Option<String>,
+    /// 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<String>,
+    /// 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,
         })
     }
 

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<DedupeCache>,
+    /// Per-source request budget (anti-flood)
+    pub rate_limiter: Arc<RateLimiter>,
+    /// Shared secret for signed click redirects; None = legacy unsigned links
+    pub link_secret: Option<Arc<String>>,
 }
 
 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();

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()