From 8daefbe8c49be8233299fa6b6d014ff63e769385 Mon Sep 17 00:00:00 2001 From: Matthew Meszaros Date: Sat, 29 Aug 2026 04:16:28 -0700 Subject: [PATCH] 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 | 3 + docker-compose.yml | 2 + .../docs/development/configuration.mdx | 1 + docs/content/docs/guides/website-tracking.mdx | 2 +- internal/api/handler/website_tracking.go | 17 +-- internal/app/websitetracking/service.go | 103 +++++++++--------- internal/jobs/website_tracking_retention.go | 7 +- internal/repository/pg_website_tracking.go | 46 ++++---- tracking/Cargo.lock | 1 + tracking/Cargo.toml | 2 + tracking/src/config.rs | 24 ++++ tracking/src/handlers.rs | 101 +++++++++++------ tracking/src/hits.rs | 31 +++--- tracking/src/main.rs | 9 +- 14 files changed, 203 insertions(+), 146 deletions(-) diff --git a/deploy/config/env.example b/deploy/config/env.example index 9eb015b9..81edb6a3 100644 --- a/deploy/config/env.example +++ b/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 diff --git a/docker-compose.yml b/docker-compose.yml index f0c43365..50186477 100644 --- a/docker-compose.yml +++ b/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. diff --git a/docs/content/docs/development/configuration.mdx b/docs/content/docs/development/configuration.mdx index 88ca0d68..459beee6 100644 --- a/docs/content/docs/development/configuration.mdx +++ b/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 `.` | `nats://localhost:4222`, `warmbly` | | `KAFKA_TRACKING_TOPIC` | Event topic, read by the Rust publisher **and** the Go subscriber | `tracking-events` | diff --git a/docs/content/docs/guides/website-tracking.mdx b/docs/content/docs/guides/website-tracking.mdx index 24f08c59..345c64de 100644 --- a/docs/content/docs/guides/website-tracking.mdx +++ b/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. diff --git a/internal/api/handler/website_tracking.go b/internal/api/handler/website_tracking.go index 91a048b2..f050f806 100644 --- a/internal/api/handler/website_tracking.go +++ b/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 diff --git a/internal/app/websitetracking/service.go b/internal/app/websitetracking/service.go index c9e99be2..42b50fec 100644 --- a/internal/app/websitetracking/service.go +++ b/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 diff --git a/internal/jobs/website_tracking_retention.go b/internal/jobs/website_tracking_retention.go index 1d0247ab..1f7e25a5 100644 --- a/internal/jobs/website_tracking_retention.go +++ b/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 diff --git a/internal/repository/pg_website_tracking.go b/internal/repository/pg_website_tracking.go index a6d84ce6..d0d91a57 100644 --- a/internal/repository/pg_website_tracking.go +++ b/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 diff --git a/tracking/Cargo.lock b/tracking/Cargo.lock index 99ef7672..0e18dfed 100644 --- a/tracking/Cargo.lock +++ b/tracking/Cargo.lock @@ -3280,6 +3280,7 @@ dependencies = [ "axum", "base64 0.21.7", "chrono", + "ipnet", "moka", "rdkafka", "reqwest 0.11.27", diff --git a/tracking/Cargo.toml b/tracking/Cargo.toml index a260ec6e..bedb20cd 100644 --- a/tracking/Cargo.toml +++ b/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" diff --git a/tracking/src/config.rs b/tracking/src/config.rs index 52b70a26..076e8b73 100644 --- a/tracking/src/config.rs +++ b/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, } 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 { + raw.split(',') + .map(str::trim) + .filter(|v| !v.is_empty()) + .filter_map(|v| { + v.parse::() + .ok() + .or_else(|| v.parse::().ok().map(ipnet::IpNet::from)) + }) + .collect() +} diff --git a/tracking/src/handlers.rs b/tracking/src/handlers.rs index be2b7731..1d37b7f1 100644 --- a/tracking/src/handlers.rs +++ b/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, /// Tighter per-source budget for page views than for pixels pub hit_rate_limiter: Arc, + /// Proxies whose forwarded-IP headers are believed + pub trusted_proxies: Arc>, } 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, + ConnectInfo(peer): ConnectInfo, Path(task_id): Path, 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, + ConnectInfo(peer): ConnectInfo, Path(link_id): Path, 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, + ConnectInfo(peer): ConnectInfo, 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 { - 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 { - // 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::().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::().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"); diff --git a/tracking/src/hits.rs b/tracking/src/hits.rs index eafe8fa4..fcbf7e48 100644 --- a/tracking/src/hits.rs +++ b/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; diff --git a/tracking/src/main.rs b/tracking/src/main.rs index b2d7be35..7f0ebead 100644 --- a/tracking/src/main.rs +++ b/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::(), + ) + .await + { observability::report_issue("Tracking server terminated unexpectedly", &e.to_string()); std::process::exit(1); }