feat: address the Greptile review on website tracking by believing forwarded-IP headers only from TRACKING_TRUSTED_PROXIES (socket peer otherwise, proxy-appended last X-Forwarded-For entry, applied to pixel, click and page-hit paths), making IdentifyVisitor report whether it claimed the row so a lost identification race re-reads the visitor and splits onto a fresh record instead of attributing the hit to the wrong contact, forgetting the edge dedupe entry when a forward to the backend fails so the retry is counted, and trimming the new Go and Rust comments to the one-line style

deploy/config/env.example

@@ -230,6 +230,9 @@ TRACKING_PORT=3000
 TRACKING_RATE_LIMIT_PER_MIN=300
 # Website page views accepted per source per minute (the snippet posts to /p).
 TRACKING_PAGEHIT_RATE_LIMIT_PER_MIN=60
+# CIDRs whose forwarded-IP headers the tracking service believes. Empty uses the
+# socket peer. Set it behind a reverse proxy, like TRUSTED_PROXIES above.
+# TRACKING_TRUSTED_PROXIES=10.0.0.0/8
 
 # === Realtime service (Elixir/Phoenix) ===
 PHX_HOST=localhost

docker-compose.yml

@@ -411,6 +411,8 @@ services:
       KAFKA_BOOTSTRAP_SERVERS: ${KAFKA_BOOTSTRAP_SERVERS:-}
       KAFKA_TRACKING_TOPIC: ${KAFKA_TRACKING_TOPIC:-tracking-events}
       TRACKING_RATE_LIMIT_PER_MIN: ${TRACKING_RATE_LIMIT_PER_MIN:-}
+      TRACKING_PAGEHIT_RATE_LIMIT_PER_MIN: ${TRACKING_PAGEHIT_RATE_LIMIT_PER_MIN:-}
+      TRACKING_TRUSTED_PROXIES: ${TRACKING_TRUSTED_PROXIES:-}
       SENTRY_DSN: ${SENTRY_DSN:-}
       # Overridable so `make dev` (native backend) can point at
       # host.docker.internal:8080 while `docker compose up` uses the container.

docs/content/docs/development/configuration.mdx

@@ -384,6 +384,7 @@ The Rust open and click service. It reads its own environment, so these have to
 | `INTERNAL_API_TOKEN` | Bearer token for that lookup. **Required**: the service exits at boot on an empty value | none |
 | `TRACKING_RATE_LIMIT_PER_MIN` | Counted pixel and click requests per source per minute. Over budget, pixels are still served but not counted, and click redirects get `429` | `300` |
 | `TRACKING_PAGEHIT_RATE_LIMIT_PER_MIN` | Website page views accepted per source per minute, on top of the shared budget above. Over budget, the snippet gets `429` | `60` |
+| `TRACKING_TRUSTED_PROXIES` | CIDRs whose `X-Forwarded-For` / `CF-Connecting-IP` headers the tracking service believes (the proxy-appended, last `X-Forwarded-For` entry). Empty trusts nothing and uses the socket peer, which is correct for a directly exposed service; set it behind a reverse proxy or the per-source rate limits and the location stored with page views are caller-controlled. Same convention as the backend's `TRUSTED_PROXIES` | empty |
 | `EVENTBUS_PROVIDER` | `nats` or `kafka`. Kafka needs an image built with `CARGO_FEATURES=kafka` | `nats` |
 | `NATS_URL`, `NATS_SUBJECT_PREFIX` | JetStream address and subject prefix. The publish subject is `<prefix>.<topic>` | `nats://localhost:4222`, `warmbly` |
 | `KAFKA_TRACKING_TOPIC` | Event topic, read by the Rust publisher **and** the Go subscriber | `tracking-events` |

docs/content/docs/guides/website-tracking.mdx

@@ -77,4 +77,4 @@ Page views show in the **Activity** tab of a contact under the **Website** filte
 
 ## Abuse controls
 
-The ingest endpoint on the tracking service is public, so it runs behind the same controls as the open pixel and click redirects: a per-source request budget (with a tighter one for page views on top), prefetch and crawler filtering, an 8 KB body cap with limits on every field, a short window that ignores reloads and double-fires, a per-source budget for unknown site keys, and a circuit breaker toward the backend. Over budget requests get `429`; everything else that is declined is acknowledged quietly so a probe learns nothing about which keys exist.
+The ingest endpoint on the tracking service is public, so it runs behind the same controls as the open pixel and click redirects: a per-source request budget (with a tighter one for page views on top), prefetch and crawler filtering, an 8 KB body cap with limits on every field, a short window that ignores reloads and double-fires, a per-source budget for unknown site keys, and a circuit breaker toward the backend. Forwarded-IP headers are believed only from proxies the operator lists in `TRACKING_TRUSTED_PROXIES`, so a caller cannot choose its own rate-limit bucket or the location stored with a view. Over budget requests get `429`; everything else that is declined is acknowledged quietly so a probe learns nothing about which keys exist.

internal/api/handler/website_tracking.go

@@ -12,8 +12,7 @@ import (
 	"github.com/warmbly/warmbly/internal/models"
 )
 
-// Website tracking settings (JWT only, MANAGE_SETTINGS): the snippet's site
-// key, consent mode, location precision, allowed hosts and retention.
+// Website tracking settings (JWT only, MANAGE_SETTINGS).
 
 func (h *Handler) GetWebsiteTrackingSettings(c *gin.Context) {
 	orgID := middleware.GetOrganizationID(c)
@@ -56,8 +55,7 @@ func (h *Handler) UpdateWebsiteTrackingSettings(c *gin.Context) {
 	c.JSON(http.StatusOK, settings)
 }
 
-// RotateWebsiteTrackingKey issues a new site key. Snippets carrying the old
-// one are refused from the next hit on.
+// RotateWebsiteTrackingKey issues a new site key; the old one stops at once.
 func (h *Handler) RotateWebsiteTrackingKey(c *gin.Context) {
 	orgID := middleware.GetOrganizationID(c)
 	if orgID == nil {
@@ -81,15 +79,11 @@ func (h *Handler) RotateWebsiteTrackingKey(c *gin.Context) {
 	c.JSON(http.StatusOK, settings)
 }
 
-// InternalIngestPageHit is where the tracking service forwards a page view
-// after its own rate limiting, filtering and payload caps.
+// InternalIngestPageHit receives a page view from the tracking service.
 //
 //	POST /api/v1/internal/page-hits
 //	  -> 204 accepted or quietly rejected | 200 {"new_visitor_key":...}
-//	  -> 400 malformed | 404 unknown site key
-//
-// Unknown keys 404 so the edge can cache them negatively and cut off a
-// probing source, exactly like an unknown click ticket.
+//	  -> 400 malformed | 404 unknown site key (cached negatively at the edge)
 func (h *Handler) InternalIngestPageHit(c *gin.Context) {
 	var req models.WebsiteHitRequest
 	if err := c.ShouldBindJSON(&req); err != nil {
@@ -113,8 +107,7 @@ func (h *Handler) InternalIngestPageHit(c *gin.Context) {
 	}
 }
 
-// websiteIdentifyForLink is the click redirect's question: may this
-// destination carry the visitor-identification ticket?
+// websiteIdentifyForLink: may this click destination carry the identification ticket?
 func (h *Handler) websiteIdentifyForLink(c *gin.Context, campaignID uuid.UUID, destination string) bool {
 	if h.WebsiteTrackingService == nil {
 		return false

internal/app/websitetracking/service.go

@@ -1,9 +1,6 @@
-// Package websitetracking is the control plane behind the website tracking
-// snippet: the workspace's settings, and the ingest path the Rust tracking
-// service forwards page views to. Device and location are derived here from
-// the request facts the edge saw; nothing about them is trusted from the
-// browser, and a hit only reaches a contact through a click ticket the
-// contact's own email carried.
+// Package websitetracking owns the tracking snippet's settings and the page
+// view ingest path. Device and location are derived server-side; a hit reaches
+// a contact only through a click ticket. See docs/guides/website-tracking.
 package websitetracking
 
 import (
@@ -28,9 +25,7 @@ import (
 	"github.com/warmbly/warmbly/internal/repository"
 )
 
-// Payload caps. The tracking service enforces the same limits before
-// forwarding; these are the backstop for anything that reaches the backend
-// another way.
+// Payload caps, mirrored by the tracking service; the backstop here.
 const (
 	maxURLLen      = 2048
 	maxTitleLen    = 512
@@ -45,19 +40,15 @@ const (
 )
 
 var (
-	// ErrUnknownSite is a site key no workspace owns. The tracking service
-	// caches this negatively and budgets misses per source.
+	// ErrUnknownSite: no workspace owns the key (the edge caches this negatively).
 	ErrUnknownSite = errors.New("unknown site key")
-	// ErrRejected is a well-formed hit the workspace's policy does not accept
-	// (tracking off, consent not met, host not allowed, automated client).
-	// The browser gets a quiet 204 either way.
+	// ErrRejected: well-formed but declined by policy (off, consent, host, bot).
 	ErrRejected = errors.New("hit rejected")
 	// ErrMalformed is a payload that fails validation.
 	ErrMalformed = errors.New("malformed hit")
 )
 
-// identifyParam is the query parameter the click redirect appends and the
-// snippet strips. Kept out of stored URLs even if a snippet forgets to.
+// identifyParam is what the click redirect appends; never stored in a URL.
 const identifyParam = "wbly_t"
 
 type Service interface {
@@ -65,13 +56,10 @@ type Service interface {
 	UpdateSettings(ctx context.Context, orgID, userID uuid.UUID, req *models.UpdateWebsiteTrackingSettingsRequest) (*models.WebsiteTrackingSettings, *errx.Error)
 	RotateSiteKey(ctx context.Context, orgID, userID uuid.UUID) (*models.WebsiteTrackingSettings, *errx.Error)
 
-	// ShouldIdentify reports whether a click redirect for this campaign may
-	// append the ticket to the destination: only when the workspace has
-	// tracking on and the destination host is one it registered.
+	// ShouldIdentify: may the click redirect append the ticket to this destination?
 	ShouldIdentify(ctx context.Context, campaignID uuid.UUID, destination string) bool
 
-	// Ingest records one page view. Returns ErrUnknownSite, ErrRejected or
-	// ErrMalformed for the tracking service to map to a status.
+	// Ingest records one page view; ErrUnknownSite/ErrRejected/ErrMalformed map to statuses.
 	Ingest(ctx context.Context, req *models.WebsiteHitRequest) (*models.WebsiteHitResult, error)
 }
 
@@ -85,8 +73,7 @@ func NewService(repo repository.WebsiteTrackingRepository, geoClient *geo.Client
 	return &service{repo: repo, geo: geoClient, publisher: publisher}
 }
 
-// newKey is 128 bits of randomness as hex: unguessable, URL-safe, and the
-// same shape for site keys and server-issued visitor ids.
+// newKey is 128 random bits as hex, used for site keys and visitor ids.
 func newKey() string {
 	b := make([]byte, 16)
 	if _, err := rand.Read(b); err != nil {
@@ -160,8 +147,7 @@ func (s *service) RotateSiteKey(ctx context.Context, orgID, userID uuid.UUID) (*
 	return s.GetSettings(ctx, orgID)
 }
 
-// normalizeHosts reduces whatever was typed (URLs, mixed case, blank lines)
-// to unique bare hostnames.
+// normalizeHosts reduces pasted URLs and mixed case to unique bare hostnames.
 func normalizeHosts(raw []string) ([]string, bool) {
 	out := make([]string, 0, len(raw))
 	seen := map[string]bool{}
@@ -185,9 +171,7 @@ func normalizeHosts(raw []string) ([]string, bool) {
 	return out, true
 }
 
-// hostAllowed matches a request host against the registered list. Ports are
-// ignored and a registered apex covers its subdomains, so "example.com"
-// admits "www.example.com" and "app.example.com".
+// hostAllowed ignores ports; a registered apex covers its subdomains.
 func hostAllowed(allowed []string, host string) bool {
 	host = hostOnly(host)
 	if host == "" {
@@ -266,8 +250,7 @@ func (s *service) Ingest(ctx context.Context, req *models.WebsiteHitRequest) (*m
 		return nil, ErrRejected
 	}
 
-	// The server, not the snippet, decides whether consent was sufficient. A
-	// stale snippet attribute cannot downgrade an explicit-mode workspace.
+	// The server decides whether consent sufficed; a stale snippet cannot downgrade it.
 	switch site.ConsentMode {
 	case models.WebsiteConsentExplicit:
 		if req.Consent != "granted" {
@@ -310,8 +293,7 @@ func (s *service) Ingest(ctx context.Context, req *models.WebsiteHitRequest) (*m
 		DeviceType:     deviceType(ua),
 	}
 
-	// Strip the identification ticket from the stored URL so it never shows
-	// up in the timeline or travels in an archive.
+	// The ticket never reaches the stored URL.
 	q := pageURL.Query()
 	hit.UTMSource = clip(q.Get("utm_source"), 256)
 	hit.UTMMedium = clip(q.Get("utm_medium"), 256)
@@ -354,24 +336,7 @@ func (s *service) Ingest(ctx context.Context, req *models.WebsiteHitRequest) (*m
 			}
 			// A ticket from another workspace's campaign proves nothing here.
 			if terr == nil && ok && ticketOrg == site.OrganizationID {
-				switch {
-				case visitor.ContactID == nil:
-					if ierr := s.repo.IdentifyVisitor(ctx, visitor.ID, contactID, "email_link", now); ierr != nil {
-						db.CaptureError(ierr, "", nil, "websitetracking IdentifyVisitor")
-					} else {
-						visitor.ContactID = &contactID
-					}
-				case *visitor.ContactID != contactID:
-					// A different person on the same browser: split rather than
-					// merge, so neither contact inherits the other's history.
-					fresh, cerr := s.repo.CreateVisitor(ctx, site.OrganizationID, newKey(), contactID, "email_link", now)
-					if cerr != nil {
-						db.CaptureError(cerr, "", nil, "websitetracking CreateVisitor")
-					} else {
-						visitor = fresh
-						result.NewVisitorKey = fresh.VisitorKey
-					}
-				}
+				visitor, result.NewVisitorKey = s.attach(ctx, site.OrganizationID, visitor, contactID, now)
 			}
 		}
 	}
@@ -393,9 +358,41 @@ func (s *service) Ingest(ctx context.Context, req *models.WebsiteHitRequest) (*m
 	return result, nil
 }
 
-// locate fills the location columns from the request IP, trimmed to the
-// workspace's precision. Best-effort: no database or a private address leaves
-// them empty. The IP itself goes no further than this call.
+// attach ties the browser to the ticket's contact. An anonymous row is
+// claimed with a guarded update; if a concurrent hit won that race for a
+// different contact, or the row already belonged to someone else, the browser
+// is split onto a fresh row so neither contact inherits the other's history.
+func (s *service) attach(ctx context.Context, orgID uuid.UUID, visitor *models.WebsiteVisitor, contactID uuid.UUID, now time.Time) (*models.WebsiteVisitor, string) {
+	if visitor.ContactID == nil {
+		updated, err := s.repo.IdentifyVisitor(ctx, visitor.ID, contactID, "email_link", now)
+		if err != nil {
+			db.CaptureError(err, "", nil, "websitetracking IdentifyVisitor")
+			return visitor, ""
+		}
+		if updated {
+			visitor.ContactID = &contactID
+			return visitor, ""
+		}
+		current, err := s.repo.GetVisitorByID(ctx, visitor.ID)
+		if err != nil {
+			db.CaptureError(err, "", nil, "websitetracking GetVisitorByID")
+			return visitor, ""
+		}
+		visitor = current
+	}
+	if visitor.ContactID != nil && *visitor.ContactID == contactID {
+		return visitor, ""
+	}
+	fresh, err := s.repo.CreateVisitor(ctx, orgID, newKey(), contactID, "email_link", now)
+	if err != nil {
+		db.CaptureError(err, "", nil, "websitetracking CreateVisitor")
+		return visitor, ""
+	}
+	return fresh, fresh.VisitorKey
+}
+
+// locate fills location from the request IP at the workspace's precision;
+// best-effort, and the IP goes no further than this call.
 func (s *service) locate(ip string, precision models.WebsiteLocationPrecision, hit *models.WebsitePageHit) {
 	if precision == models.WebsiteLocationNone || s.geo == nil || ip == "" {
 		return

internal/jobs/website_tracking_retention.go

@@ -8,9 +8,7 @@ import (
 	"github.com/warmbly/warmbly/internal/repository"
 )
 
-// WebsiteTrackingRetentionJob prunes page views past each workspace's own
-// retention window. The window is the promise the workspace makes on its
-// tracking settings page, so the sweep is what keeps that promise.
+// WebsiteTrackingRetentionJob prunes page views past each workspace's own window.
 type WebsiteTrackingRetentionJob struct {
 	repo repository.WebsiteTrackingRepository
 }
@@ -19,8 +17,7 @@ func NewWebsiteTrackingRetentionJob(repo repository.WebsiteTrackingRepository) *
 	return &WebsiteTrackingRetentionJob{repo: repo}
 }
 
-// Run executes one pruning pass over every workspace. A failure on one
-// workspace is reported and the pass continues with the next.
+// Run prunes every workspace; one failure does not stop the pass.
 func (j *WebsiteTrackingRetentionJob) Run(ctx context.Context) error {
 	if j.repo == nil {
 		return nil

internal/repository/pg_website_tracking.go

@@ -11,10 +11,8 @@ import (
 	"github.com/warmbly/warmbly/internal/models"
 )
 
-// WebsiteTrackingRepository is the store behind the website tracking snippet:
-// per-workspace settings, the browsers the snippet has seen, and their page
-// views. Writes on the hit path come only from the backend's internal ingest
-// endpoint; the dashboard only reads hits.
+// WebsiteTrackingRepository stores tracking settings, the browsers the snippet
+// has seen, and their page views. Only the internal ingest path writes hits.
 type WebsiteTrackingRepository interface {
 	GetOrCreateSettings(ctx context.Context, orgID uuid.UUID, siteKey string) (*models.WebsiteTrackingSettings, error)
 	UpdateSettings(ctx context.Context, orgID, updatedBy uuid.UUID, s *models.WebsiteTrackingSettings) error
@@ -29,7 +27,10 @@ type WebsiteTrackingRepository interface {
 
 	UpsertVisitor(ctx context.Context, orgID uuid.UUID, visitorKey string, seenAt time.Time) (*models.WebsiteVisitor, error)
 	CreateVisitor(ctx context.Context, orgID uuid.UUID, visitorKey string, contactID uuid.UUID, via string, at time.Time) (*models.WebsiteVisitor, error)
-	IdentifyVisitor(ctx context.Context, visitorID, contactID uuid.UUID, via string, at time.Time) error
+	// IdentifyVisitor ties an anonymous browser to a contact. false when the
+	// row was already tied (possibly by a concurrent hit) and nothing changed.
+	IdentifyVisitor(ctx context.Context, visitorID, contactID uuid.UUID, via string, at time.Time) (bool, error)
+	GetVisitorByID(ctx context.Context, visitorID uuid.UUID) (*models.WebsiteVisitor, error)
 	InsertHit(ctx context.Context, orgID uuid.UUID, hit *models.WebsitePageHit) error
 
 	ListHitsForContact(ctx context.Context, orgID, contactID uuid.UUID, before time.Time, limit int) ([]models.WebsitePageHit, error)
@@ -59,9 +60,7 @@ func scanWebsiteSettings(row pgx.Row) (*models.WebsiteTrackingSettings, error) {
 	return &s, nil
 }
 
-// GetOrCreateSettings returns the workspace's row, creating a disabled one
-// with the given site key on first read so the dashboard always has a key to
-// show.
+// GetOrCreateSettings creates a disabled row with the given site key on first read.
 func (r *websiteTrackingRepository) GetOrCreateSettings(ctx context.Context, orgID uuid.UUID, siteKey string) (*models.WebsiteTrackingSettings, error) {
 	_, err := r.db.Exec(ctx, `
 		INSERT INTO website_tracking_settings (organization_id, site_key)
@@ -115,9 +114,7 @@ func (r *websiteTrackingRepository) GetSiteByKey(ctx context.Context, siteKey st
 	`, siteKey))
 }
 
-// SiteForCampaign is what the click redirect consults to decide whether to
-// append the identification ticket. nil, nil when the workspace never opened
-// the tracking settings.
+// SiteForCampaign backs the click redirect's identify decision. nil, nil when unset.
 func (r *websiteTrackingRepository) SiteForCampaign(ctx context.Context, campaignID uuid.UUID) (*models.WebsiteSite, error) {
 	return scanWebsiteSite(r.db.QueryRow(ctx, `
 		SELECT s.organization_id, s.enabled, s.consent_mode, s.location_precision, s.allowed_hosts
@@ -169,9 +166,8 @@ func (r *websiteTrackingRepository) UpsertVisitor(ctx context.Context, orgID uui
 		RETURNING `+websiteVisitorColumns, orgID, visitorKey, seenAt))
 }
 
-// CreateVisitor opens a fresh, already-identified browser record. Used when a
-// ticket names a different contact than the one tied to the browser: the old
-// record keeps its history and the browser moves to this one.
+// CreateVisitor opens a fresh, already-identified browser record (the split
+// path: a ticket named a different contact than the one tied to the browser).
 func (r *websiteTrackingRepository) CreateVisitor(ctx context.Context, orgID uuid.UUID, visitorKey string, contactID uuid.UUID, via string, at time.Time) (*models.WebsiteVisitor, error) {
 	return scanWebsiteVisitor(r.db.QueryRow(ctx, `
 		INSERT INTO website_visitors (organization_id, visitor_key, contact_id, identified_at, identified_via, first_seen_at, last_seen_at)
@@ -179,15 +175,21 @@ func (r *websiteTrackingRepository) CreateVisitor(ctx context.Context, orgID uui
 		RETURNING `+websiteVisitorColumns, orgID, visitorKey, contactID, at, via))
 }
 
-// IdentifyVisitor ties an anonymous browser to a contact. A browser already
-// tied to someone is left alone; the caller splits it instead.
-func (r *websiteTrackingRepository) IdentifyVisitor(ctx context.Context, visitorID, contactID uuid.UUID, via string, at time.Time) error {
-	_, err := r.db.Exec(ctx, `
+func (r *websiteTrackingRepository) IdentifyVisitor(ctx context.Context, visitorID, contactID uuid.UUID, via string, at time.Time) (bool, error) {
+	tag, err := r.db.Exec(ctx, `
 		UPDATE website_visitors
 		SET contact_id = $2, identified_at = $3, identified_via = $4
 		WHERE id = $1 AND contact_id IS NULL
 	`, visitorID, contactID, at, via)
-	return err
+	if err != nil {
+		return false, err
+	}
+	return tag.RowsAffected() == 1, nil
+}
+
+func (r *websiteTrackingRepository) GetVisitorByID(ctx context.Context, visitorID uuid.UUID) (*models.WebsiteVisitor, error) {
+	return scanWebsiteVisitor(r.db.QueryRow(ctx,
+		`SELECT `+websiteVisitorColumns+` FROM website_visitors WHERE id = $1`, visitorID))
 }
 
 func (r *websiteTrackingRepository) InsertHit(ctx context.Context, orgID uuid.UUID, h *models.WebsitePageHit) error {
@@ -221,8 +223,7 @@ func (r *websiteTrackingRepository) InsertHit(ctx context.Context, orgID uuid.UU
 	return err
 }
 
-// ListHitsForContact feeds the contact timeline: every counted view from any
-// browser tied to the contact, newest first, strictly before the cursor.
+// ListHitsForContact: every view from any browser tied to the contact, newest first.
 func (r *websiteTrackingRepository) ListHitsForContact(ctx context.Context, orgID, contactID uuid.UUID, before time.Time, limit int) ([]models.WebsitePageHit, error) {
 	rows, err := r.db.Query(ctx, `
 		SELECT h.id, h.visitor_id, h.session_key, h.occurred_at,
@@ -283,8 +284,7 @@ func (r *websiteTrackingRepository) RetentionCutoffs(ctx context.Context, now ti
 	return out, rows.Err()
 }
 
-// PruneBefore drops one workspace's page views older than the cutoff, then the
-// browser records that have nothing left and have not been seen since.
+// PruneBefore drops old page views, then browser rows with nothing left.
 func (r *websiteTrackingRepository) PruneBefore(ctx context.Context, orgID uuid.UUID, before time.Time) (int64, error) {
 	tag, err := r.db.Exec(ctx, `
 		DELETE FROM website_page_hits WHERE organization_id = $1 AND occurred_at < $2

tracking/Cargo.lock

@@ -3280,6 +3280,7 @@ dependencies = [
  "axum",
  "base64 0.21.7",
  "chrono",
+ "ipnet",
  "moka",
  "rdkafka",
  "reqwest 0.11.27",

tracking/Cargo.toml

@@ -38,6 +38,8 @@ chrono = { version = "0.4", features = ["serde"] }
 sha2 = "0.10"
 base64 = "0.21"
 moka = { version = "0.12", features = ["future"] }
+# Trusted-proxy CIDR matching for the client IP
+ipnet = "2"
 
 # Logging
 tracing = "0.1"

tracking/src/config.rs

@@ -58,6 +58,9 @@ pub struct Config {
     /// Page-view ingest budget per source per minute. Lower than the pixel
     /// budget: a person does not view a page a second, a script does.
     pub pagehit_rate_limit_per_min: u32,
+    /// CIDRs whose forwarded-IP headers are believed. Empty trusts nothing,
+    /// so the socket peer is the client, the same rule as the backend.
+    pub trusted_proxies: Vec<ipnet::IpNet>,
 }
 
 impl Config {
@@ -179,6 +182,10 @@ impl Config {
             pagehit_rate_limit_per_min
         );
 
+        let trusted_proxies =
+            parse_trusted_proxies(&env::var("TRACKING_TRUSTED_PROXIES").unwrap_or_default());
+        info!("Trusted proxies: {:?}", trusted_proxies);
+
         Ok(Self {
             env: env_name,
             host,
@@ -197,6 +204,7 @@ impl Config {
             internal_api_token,
             rate_limit_per_min,
             pagehit_rate_limit_per_min,
+            trusted_proxies,
         })
     }
 
@@ -272,6 +280,9 @@ impl Config {
             internal_api_token,
             rate_limit_per_min: 300,
             pagehit_rate_limit_per_min: 60,
+            trusted_proxies: parse_trusted_proxies(
+                &env::var("TRACKING_TRUSTED_PROXIES").unwrap_or_default(),
+            ),
         })
     }
 
@@ -366,3 +377,16 @@ impl Config {
         }
     }
 }
+
+/// Parses a comma-separated CIDR list; a bare address is a /32 or /128.
+pub fn parse_trusted_proxies(raw: &str) -> Vec<ipnet::IpNet> {
+    raw.split(',')
+        .map(str::trim)
+        .filter(|v| !v.is_empty())
+        .filter_map(|v| {
+            v.parse::<ipnet::IpNet>()
+                .ok()
+                .or_else(|| v.parse::<std::net::IpAddr>().ok().map(ipnet::IpNet::from))
+        })
+        .collect()
+}

tracking/src/handlers.rs

@@ -1,12 +1,13 @@
 use axum::{
     body::Bytes,
-    extract::{Path, State},
+    extract::{ConnectInfo, Path, State},
     http::{header, HeaderMap, StatusCode},
     response::{IntoResponse, Redirect, Response},
 };
 use chrono::Utc;
 use moka::future::Cache;
 use sha2::{Digest, Sha256};
+use std::net::{IpAddr, SocketAddr};
 use std::sync::Arc;
 use std::time::Duration;
 
@@ -42,6 +43,8 @@ pub struct AppState {
     pub hits: Arc<HitForwarder>,
     /// Tighter per-source budget for page views than for pixels
     pub hit_rate_limiter: Arc<RateLimiter>,
+    /// Proxies whose forwarded-IP headers are believed
+    pub trusted_proxies: Arc<Vec<ipnet::IpNet>>,
 }
 
 impl AppState {
@@ -69,6 +72,7 @@ impl AppState {
                 config.internal_api_token.clone(),
             )),
             hit_rate_limiter: Arc::new(RateLimiter::new(config.pagehit_rate_limit_per_min)),
+            trusted_proxies: Arc::new(config.trusted_proxies.clone()),
         }
     }
 
@@ -106,6 +110,7 @@ pub async fn health() -> impl IntoResponse {
 /// GET /t/o/{task_id}.png
 pub async fn track_open(
     State(state): State<AppState>,
+    ConnectInfo(peer): ConnectInfo<SocketAddr>,
     Path(task_id): Path<String>,
     headers: HeaderMap,
 ) -> Response {
@@ -118,7 +123,7 @@ pub async fn track_open(
     }
 
     // Extract IP hash for deduplication + rate limiting
-    let ip_hash = extract_ip_hash(&headers);
+    let ip_hash = Some(hash_ip(&client_ip(peer, &headers, &state.trusted_proxies)));
 
     // Anti-flood: over-budget sources still get the pixel (real mail clients
     // must never see a broken image), but nothing is published.
@@ -170,6 +175,7 @@ pub async fn track_open(
 /// Unknown tickets 404.
 pub async fn track_click(
     State(state): State<AppState>,
+    ConnectInfo(peer): ConnectInfo<SocketAddr>,
     Path(link_id): Path<String>,
     headers: HeaderMap,
 ) -> Response {
@@ -179,7 +185,7 @@ pub async fn track_click(
     }
 
     // Anti-flood: cap total request rate per source
-    let ip_hash = extract_ip_hash(&headers);
+    let ip_hash = Some(hash_ip(&client_ip(peer, &headers, &state.trusted_proxies)));
     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();
@@ -289,12 +295,12 @@ const TRACKING_JS: &str = include_str!("../static/tracking.js");
 /// the URL, and nothing in the body names a contact.
 pub async fn track_page_hit(
     State(state): State<AppState>,
+    ConnectInfo(peer): ConnectInfo<SocketAddr>,
     headers: HeaderMap,
     body: Bytes,
 ) -> Response {
-    let ip = extract_ip(&headers);
-    let ip_hash = ip.as_deref().map(hash_ip);
-    let source = ip_hash.clone().unwrap_or_else(|| "unknown".to_string());
+    let ip = client_ip(peer, &headers, &state.trusted_proxies);
+    let source = hash_ip(&ip);
 
     // Anti-flood: page views have their own, tighter budget on top of the
     // shared one, and the shared one counts too so a flood here also
@@ -349,10 +355,12 @@ pub async fn track_page_hit(
         screen_height: payload.sh,
         landing: payload.ld,
         user_agent: user_agent.unwrap_or_default(),
-        ip: ip.unwrap_or_default(),
+        ip,
         origin_host,
     };
 
+    let dedupe_visitor = hit.visitor_key.clone();
+    let dedupe_url = hit.url.clone();
     match state.hits.forward(hit, &source).await {
         Outcome::Accepted(None) => StatusCode::NO_CONTENT.into_response(),
         Outcome::Accepted(Some(vid)) => (
@@ -366,6 +374,8 @@ pub async fn track_page_hit(
         Outcome::UnknownSite => StatusCode::NO_CONTENT.into_response(),
         Outcome::Malformed => (StatusCode::BAD_REQUEST, "Invalid payload").into_response(),
         Outcome::Unavailable => {
+            // Nothing was stored, so the next attempt must not be deduped away.
+            state.hits.forget(&dedupe_visitor, &dedupe_url).await;
             (StatusCode::SERVICE_UNAVAILABLE, "Try again shortly").into_response()
         }
     }
@@ -405,32 +415,30 @@ fn pixel_response() -> Response {
         .into_response()
 }
 
-/// Extract and hash IP address for privacy
-fn extract_ip_hash(headers: &HeaderMap) -> Option<String> {
-    extract_ip(headers).map(|ip| hash_ip(&ip))
-}
-
-/// The client IP as the proxy in front reported it. Used raw only for the
-/// page-view forward, where the backend turns it into a location and drops it.
-fn extract_ip(headers: &HeaderMap) -> Option<String> {
-    // Try various headers for the real IP
-    headers
-        .get("x-forwarded-for")
-        .and_then(|h| h.to_str().ok())
-        .and_then(|s| s.split(',').next())
-        .map(|s| s.trim().to_string())
-        .or_else(|| {
-            headers
-                .get("x-real-ip")
-                .and_then(|h| h.to_str().ok())
-                .map(|s| s.to_string())
-        })
-        .or_else(|| {
-            headers
-                .get("cf-connecting-ip")
-                .and_then(|h| h.to_str().ok())
-                .map(|s| s.to_string())
-        })
+/// The client address. Forwarded headers are believed only when the socket
+/// peer is a trusted proxy; otherwise anyone could pick their own rate-limit
+/// bucket and the location stored with a page view.
+fn client_ip(peer: SocketAddr, headers: &HeaderMap, trusted: &[ipnet::IpNet]) -> String {
+    let peer_ip = peer.ip();
+    if !trusted.iter().any(|net| net.contains(&peer_ip)) {
+        return peer_ip.to_string();
+    }
+    let header = |name: &str| {
+        headers
+            .get(name)
+            .and_then(|h| h.to_str().ok())
+            .map(str::trim)
+            .filter(|v| !v.is_empty())
+    };
+    // The proxy appends the address it saw as the LAST X-Forwarded-For entry;
+    // the first is whatever the client claimed.
+    let forwarded = header("cf-connecting-ip")
+        .or_else(|| header("x-forwarded-for").and_then(|v| v.rsplit(',').next().map(str::trim)))
+        .or_else(|| header("x-real-ip"));
+    match forwarded.and_then(|v| v.parse::<IpAddr>().ok()) {
+        Some(ip) => ip.to_string(),
+        None => peer_ip.to_string(),
+    }
 }
 
 fn hash_ip(ip: &str) -> String {
@@ -457,6 +465,33 @@ mod tests {
         );
     }
 
+    fn hdr(pairs: &[(&'static str, &'static str)]) -> HeaderMap {
+        let mut h = HeaderMap::new();
+        for (k, v) in pairs {
+            h.insert(*k, v.parse().unwrap());
+        }
+        h
+    }
+
+    #[test]
+    fn client_ip_ignores_forwarded_headers_from_untrusted_peers() {
+        let peer: SocketAddr = "203.0.113.9:4000".parse().unwrap();
+        let h = hdr(&[("x-forwarded-for", "1.1.1.1, 2.2.2.2")]);
+        assert_eq!(client_ip(peer, &h, &[]), "203.0.113.9");
+    }
+
+    #[test]
+    fn client_ip_takes_the_proxy_appended_entry_from_trusted_peers() {
+        let trusted = vec!["10.0.0.0/8".parse::<ipnet::IpNet>().unwrap()];
+        let peer: SocketAddr = "10.1.2.3:4000".parse().unwrap();
+        let h = hdr(&[("x-forwarded-for", "1.1.1.1, 198.51.100.7")]);
+        assert_eq!(client_ip(peer, &h, &trusted), "198.51.100.7");
+        let h = hdr(&[("cf-connecting-ip", "198.51.100.8")]);
+        assert_eq!(client_ip(peer, &h, &trusted), "198.51.100.8");
+        let h = hdr(&[("x-forwarded-for", "not an ip")]);
+        assert_eq!(client_ip(peer, &h, &trusted), "10.1.2.3");
+    }
+
     #[test]
     fn host_of_strips_scheme_and_path() {
         assert_eq!(host_of("https://WWW.Example.com/a?b#c"), "www.example.com");

tracking/src/hits.rs

@@ -1,21 +1,9 @@
-//! Website page-view ingest.
-//!
-//! The snippet POSTs a small JSON document; this module validates it, keeps
-//! everything a browser could lie about out of the forwarded payload, and
-//! hands the hit to the backend's internal API, where the user agent and IP
-//! are turned into device and location and the row is stored. The same
-//! layered defenses as the click resolver keep a key-spray away from the
-//! backend:
-//!
-//! 1. negative cache: site keys confirmed unknown are dropped from memory
-//! 2. per-source miss budget: real snippets never miss, so a source
-//!    accumulating unknown keys is probing and gets cut off
-//! 3. circuit breaker: when the backend errors, forwarding pauses for a
-//!    cooldown instead of piling on
-//!
-//! Nothing in the request can name a contact. The only link between a
-//! browser and a person is the click ticket the redirect appended, and the
-//! backend checks that ticket against the workspace that owns the site key.
+//! Website page-view ingest: validate the snippet's payload, keep anything a
+//! browser could lie about out of it, and forward to the backend, which turns
+//! the user agent and IP into device and location and stores the row. Same
+//! layered defenses as the click resolver: negative cache for unknown keys,
+//! per-source miss budget, circuit breaker. Nothing in a request can name a
+//! contact; only the click ticket does, and the backend verifies it.
 
 use moka::future::Cache;
 use serde::{Deserialize, Serialize};
@@ -196,6 +184,13 @@ impl HitForwarder {
         false
     }
 
+    /// Drops a dedupe entry after a failed forward so the retry is counted.
+    pub async fn forget(&self, visitor: &str, url: &str) {
+        self.dedupe
+            .invalidate(&format!("{}:{}", visitor, url))
+            .await;
+    }
+
     pub async fn forward(&self, hit: ForwardedHit, source: &str) -> Outcome {
         if self.unknown_sites.contains_key(&hit.site_key) {
             self.count_miss(source).await;

tracking/src/main.rs

@@ -135,7 +135,14 @@ async fn main() {
         }
     };
 
-    if let Err(e) = axum::serve(listener, app).await {
+    // Connect info carries the socket peer, which is the client unless it is a
+    // trusted proxy (TRACKING_TRUSTED_PROXIES).
+    if let Err(e) = axum::serve(
+        listener,
+        app.into_make_service_with_connect_info::<SocketAddr>(),
+    )
+    .await
+    {
         observability::report_issue("Tracking server terminated unexpectedly", &e.to_string());
         std::process::exit(1);
     }