Code review fixes: app 2.9.0, server 2.2.0

Telemetry no longer loses queued batches. The server drops a bad value,
or a sample without a plausible time, instead of rejecting the batch,
and answers 200 with what it dropped. The app sends only values inside
the server's ranges (speed as a magnitude, so reversing is fine) and no
samples before the head unit's clock is set.

Server:
- Pairing hands out and accepts a token only after the device is saved;
  a failed save leaves the request pending.
- A paired token serves only the installation id it was paired with
  (403 otherwise); the shared INGEST_TOKEN still serves any.
- Wrong-password lockouts and open pairing requests are limited per
  client address (last X-Forwarded-For entry), so a stranger cannot
  lock the owner out; a car starting again replaces its own request.
- ABRP forwarding skips samples stamped ahead of the server's clock,
  which used to stop it until a restart.

App:
- Settings written to plaintext during a keystore failure move back
  into encrypted storage; the stored installation id wins.
- The installation id is derived from ANDROID_ID on first start, so a
  reinstall finds its history; a stored or imported id still wins.
- The scanned config file is deleted after import (it holds the token).
- GPS stops while live telemetry is off or the server unconfigured, and
  pauses while parked and idle, keeping the fix the car parked with.
- Range estimates use the monotonic clock.
- One HTTP helper (ServerHttp) for telemetry, sync and pairing.
- Last Turkish strings translated; mise logs shows the NilsMG4 tags.

Grafana reads InfluxDB with its own read-only token (new grafana-influx
Secret, created by hand as described in k8s/README.md).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01QLJfZQZeAKAwSqViTmLzki

server/README.md

@@ -31,18 +31,22 @@ The ABRP credentials never reach the car.
 ## Endpoints
 
 The ingest endpoints accept `Authorization: Bearer <token>` with either the shared `INGEST_TOKEN`
-or a paired car's own token.
+or a paired car's own token. A paired token only works for the installation id it was paired
+with (403 otherwise); the shared token works for any.
 
 - `POST /v1/telemetry` — `{ installationId, samples: [...] }`, up to 500 samples in ABRP's `tlm`
   field names and units (`utc` in epoch seconds; flags as 0/1 or booleans). Unknown fields are
-  dropped, known ones bounded. Stored, then the newest sample is forwarded to ABRP if it is under
-  5 minutes old.
+  dropped, known ones bounded: a value out of range is dropped on its own, and a sample without
+  a plausible time (before 2001, or more than an hour ahead) is dropped whole. The rest is stored
+  and the answer is `200 {stored, droppedCount, dropped}`, even when nothing was usable, since
+  the car would only send it again. The newest sample is forwarded to ABRP if it is under
+  5 minutes old and not ahead of the server's clock.
 - `PUT /v1/consumption/daily` — one delta chunk of a day's consumption.
 - `PUT /v1/charging-sessions/{sessionId}` — a finished DC session with its curve.
 - `GET /v1/consumption/history/{installationId}` — the car reads its own totals back.
 - `GET /v1/health` — `{"status":"ok"}`.
 
-A bad request is a 400 naming the field; a database failure is a 500 without detail.
+A malformed request is a 400 naming the field; a database failure is a 500 without detail.
 
 ## Pairing a car by QR code
 
@@ -52,13 +56,15 @@ on only when `APPROVAL_PASSWORD` is set.
 1. The car calls `POST /v1/pair/start` `{installationId}` and shows the returned
    `verificationUriComplete` as a QR code, plus the short `userCode` (e.g. `K7F-3QX`).
 2. The phone opens `/pair?code=…`, the owner checks the code and approves with the approval
-   password. Five wrong passwords lock approvals for a minute.
+   password. Five wrong passwords from one address lock that address out for a minute; other
+   addresses are unaffected (the address is the last `X-Forwarded-For` entry, which Traefik sets).
+   An address may hold three open requests; a car starting again replaces its own.
 3. The car polls `POST /v1/pair/poll` `{deviceCode}`: 202 pending, 200 with its own token
    (handed out once), 403 denied, 410 expired (after 10 minutes) or unknown.
 
 Each car gets its own 43-character token; only its SHA-256 is stored (measurement `device` in
-the same bucket), and the device is persisted before the car can collect the token, so a
-restart never loses a paired car. Manage cars with the approval password:
+the same bucket), and the token is only handed out, and only accepted, once the device is
+persisted, so a restart never loses a paired car. Manage cars with the approval password:
 
 ```sh
 curl -H "Authorization: Bearer $APPROVAL_PASSWORD" https://car.k8s.stinnesbeck.com/v1/devices

server/abrp.go

@@ -37,6 +37,8 @@ const (
 	abrpEndpoint = "https://api.iternio.com/1/tlm/send"
 	// Older than this, a sample is history rather than live state, and ABRP gets nothing.
 	abrpMaxAge = 5 * time.Minute
+	// Newer than now by more than this, a sample's clock is wrong, and ABRP gets nothing.
+	abrpMaxAhead = 30 * time.Second
 )
 
 func (f *Forwarder) configured() bool {
@@ -46,16 +48,17 @@ func (f *Forwarder) configured() bool {
 // pick chooses the sample to forward, or false when there is nothing new or it is too
 // soon. It records the choice, so two batches arriving together cannot both forward.
 func (f *Forwarder) pick(samples []TelemetrySample, now time.Time) (TelemetrySample, bool) {
-	if len(samples) == 0 {
-		return TelemetrySample{}, false
-	}
-	newest := samples[0]
-	for _, s := range samples[1:] {
-		if s.UTC > newest.UTC {
-			newest = s
+	// A sample stamped ahead of now has a wrong clock. Taking it would also set lastUTC into
+	// the future, and every real sample after it would look older and never be sent.
+	latest := now.Add(abrpMaxAhead).Unix()
+	var newest TelemetrySample
+	found := false
+	for _, s := range samples {
+		if s.UTC <= latest && (!found || s.UTC > newest.UTC) {
+			newest, found = s, true
 		}
 	}
-	if now.Sub(time.Unix(newest.UTC, 0)) > abrpMaxAge {
+	if !found || now.Sub(time.Unix(newest.UTC, 0)) > abrpMaxAge {
 		return TelemetrySample{}, false
 	}
 	f.mu.Lock()

server/api.go

@@ -8,6 +8,7 @@ import (
 	"log/slog"
 	"net/http"
 	"strings"
+	"time"
 )
 
 // API serves the car. Every endpoint but /v1/health needs the ingest token.
@@ -45,30 +46,49 @@ func (a *API) Routes() http.Handler {
 	return mux
 }
 
-// hasIngestToken accepts the shared token (compared in constant time; main refuses to start
+// caller is who sent an ingest request. A paired car's token is bound to the installation
+// id it was paired with; the shared INGEST_TOKEN may write and read for any.
+type caller struct {
+	installationID string // empty for the shared token
+}
+
+// mayUse reports whether the caller may write or read data stored under id.
+func (c caller) mayUse(id string) bool {
+	return c.installationID == "" || strings.EqualFold(c.installationID, id)
+}
+
+// errOtherInstallation is a 403: a paired car's token used for another car's data.
+var errOtherInstallation = errors.New("this token belongs to another installation; pair again after changing the installation id")
+
+// authenticate accepts the shared token (compared in constant time; main refuses to start
 // with one under 24 characters, so an empty or short one never matches) or a paired car's
 // own, unrevoked token.
-func (a *API) hasIngestToken(r *http.Request) bool {
+func (a *API) authenticate(r *http.Request) (caller, bool) {
 	header := r.Header.Get("Authorization")
 	supplied, ok := strings.CutPrefix(header, "Bearer ")
 	if !ok || supplied == "" {
-		return false
+		return caller{}, false
 	}
 	if len(a.IngestToken) >= minIngestTokenLength &&
 		subtle.ConstantTimeCompare([]byte(supplied), []byte(a.IngestToken)) == 1 {
-		return true
+		return caller{}, true
 	}
-	return a.Pairing.Authorized(supplied)
+	d, ok := a.Pairing.Authorized(supplied)
+	if !ok {
+		return caller{}, false
+	}
+	return caller{installationID: d.InstallationID}, true
 }
 
-type handler func(w http.ResponseWriter, r *http.Request, body []byte) error
+type handler func(w http.ResponseWriter, r *http.Request, c caller, body []byte) error
 
 // ingest wraps a handler with the token check, the body limit and error mapping: a
-// validation failure is a 400 naming the field; anything else is a 500 whose detail
-// stays in the server log.
+// validation failure is a 400 naming the field, another car's data a 403; anything else
+// is a 500 whose detail stays in the server log.
 func (a *API) ingest(h handler) http.HandlerFunc {
 	return func(w http.ResponseWriter, r *http.Request) {
-		if !a.hasIngestToken(r) {
+		c, ok := a.authenticate(r)
+		if !ok {
 			writeJSON(w, http.StatusUnauthorized, map[string]string{"error": "unauthorized"})
 			return
 		}
@@ -81,7 +101,11 @@ func (a *API) ingest(h handler) http.HandlerFunc {
 			writeJSON(w, http.StatusBadRequest, map[string]string{"error": "body is too large"})
 			return
 		}
-		if err := h(w, r, body); err != nil {
+		if err := h(w, r, c, body); err != nil {
+			if errors.Is(err, errOtherInstallation) {
+				writeJSON(w, http.StatusForbidden, map[string]string{"error": err.Error()})
+				return
+			}
 			var verr *ValidationError
 			if errors.As(err, &verr) {
 				writeJSON(w, http.StatusBadRequest, map[string]string{"error": verr.Error()})
@@ -93,29 +117,53 @@ func (a *API) ingest(h handler) http.HandlerFunc {
 	}
 }
 
+// maxDroppedListed caps the dropped values named in a telemetry response and log line.
+const maxDroppedListed = 20
+
 // telemetry stores the batch first; only once the write succeeded is the newest sample
 // handed to ABRP, so the car's retry of a failed request cannot leave ABRP with data the
-// database never got.
-func (a *API) telemetry(w http.ResponseWriter, r *http.Request, body []byte) error {
-	batch, err := validateTelemetryBatch(body)
+// database never got. The answer says how many samples were stored and what was dropped;
+// a batch with nothing usable is still a 200, since sending it again would not help.
+func (a *API) telemetry(w http.ResponseWriter, r *http.Request, c caller, body []byte) error {
+	batch, err := validateTelemetryBatch(body, time.Now())
 	if err != nil {
 		return err
 	}
-	if err := a.Store.WriteTelemetry(r.Context(), batch.InstallationID, batch.Samples); err != nil {
-		return err
+	if !c.mayUse(batch.InstallationID) {
+		return errOtherInstallation
 	}
-	a.Forwarder.Forward(batch.Samples)
-	w.WriteHeader(http.StatusNoContent)
+	if len(batch.Samples) > 0 {
+		if err := a.Store.WriteTelemetry(r.Context(), batch.InstallationID, batch.Samples); err != nil {
+			return err
+		}
+		a.Forwarder.Forward(batch.Samples)
+	}
+	dropped := batch.Dropped
+	if len(dropped) > maxDroppedListed {
+		dropped = dropped[:maxDroppedListed]
+	}
+	if len(batch.Dropped) > 0 {
+		slog.Warn("telemetry values dropped", "installation", batch.InstallationID,
+			"count", len(batch.Dropped), "first", dropped)
+	}
+	writeJSON(w, http.StatusOK, map[string]any{
+		"stored":       len(batch.Samples),
+		"droppedCount": len(batch.Dropped),
+		"dropped":      append([]string{}, dropped...),
+	})
 	return nil
 }
 
 // consumption stores one delta chunk at its recordedAt. Reads add a day's chunks up, so a
 // chunk the car re-sends from its offline cache replaces itself instead of counting twice.
-func (a *API) consumption(w http.ResponseWriter, r *http.Request, body []byte) error {
+func (a *API) consumption(w http.ResponseWriter, r *http.Request, c caller, body []byte) error {
 	chunk, err := validateConsumption(body)
 	if err != nil {
 		return err
 	}
+	if !c.mayUse(chunk.InstallationID) {
+		return errOtherInstallation
+	}
 	if err := a.Store.WriteConsumptionChunk(r.Context(), chunk); err != nil {
 		return err
 	}
@@ -123,11 +171,14 @@ func (a *API) consumption(w http.ResponseWriter, r *http.Request, body []byte) e
 	return nil
 }
 
-func (a *API) chargingSession(w http.ResponseWriter, r *http.Request, body []byte) error {
+func (a *API) chargingSession(w http.ResponseWriter, r *http.Request, c caller, body []byte) error {
 	session, err := validateChargingSession(body, r.PathValue("sessionId"))
 	if err != nil {
 		return err
 	}
+	if !c.mayUse(session.InstallationID) {
+		return errOtherInstallation
+	}
 	if err := a.Store.WriteChargingSession(r.Context(), session); err != nil {
 		return err
 	}
@@ -137,11 +188,14 @@ func (a *API) chargingSession(w http.ResponseWriter, r *http.Request, body []byt
 
 // history is the car's read-back of its own uploaded totals, so an app reinstall does not
 // blank its Consumption tab. Up to 400 days, newest first, plus the lifetime sum of all.
-func (a *API) history(w http.ResponseWriter, r *http.Request, _ []byte) error {
+func (a *API) history(w http.ResponseWriter, r *http.Request, c caller, _ []byte) error {
 	id, err := validUUID(r.PathValue("installationId"), "installationId")
 	if err != nil {
 		return err
 	}
+	if !c.mayUse(id) {
+		return errOtherInstallation
+	}
 	chunks, err := a.Store.ReadConsumptionChunks(r.Context(), id)
 	if err != nil {
 		return err

server/api_test.go

@@ -116,17 +116,32 @@ func TestEveryOtherEndpointNeedsTheIngestToken(t *testing.T) {
 	}
 }
 
-func TestTelemetryIsStoredAndBadInputIsA400(t *testing.T) {
+func TestTelemetryIsStoredAndBadValuesAreDropped(t *testing.T) {
 	api, store := newTestAPI()
 	h := api.Routes()
 	rec := do(t, h, "POST", "/v1/telemetry", testToken, string(telemetryBody(`{"utc":1790000000,"soc":80}`)))
-	if rec.Code != 204 || len(store.telemetry) != 1 {
-		t.Fatalf("%d, stored %d", rec.Code, len(store.telemetry))
+	if rec.Code != 200 || len(store.telemetry) != 1 || !strings.Contains(rec.Body.String(), `"stored":1`) {
+		t.Fatalf("%d %s, stored %d", rec.Code, rec.Body, len(store.telemetry))
 	}
-	rec = do(t, h, "POST", "/v1/telemetry", testToken, string(telemetryBody(`{"utc":1790000000,"soc":120}`)))
-	if rec.Code != 400 || !strings.Contains(rec.Body.String(), "samples[0].soc is invalid") {
+	// One bad value costs that value, not the batch: the car drops what the server rejects.
+	rec = do(t, h, "POST", "/v1/telemetry", testToken,
+		string(telemetryBody(`{"utc":1790000000,"soc":120,"speed":-3},{"utc":1790000015,"soc":79}`)))
+	if rec.Code != 200 || len(store.telemetry) != 3 || !strings.Contains(rec.Body.String(), "samples[0].soc") ||
+		!strings.Contains(rec.Body.String(), "samples[0].speed") || !strings.Contains(rec.Body.String(), `"stored":2`) {
+		t.Fatalf("%d %s, stored %d", rec.Code, rec.Body, len(store.telemetry))
+	}
+	if _, ok := store.telemetry[1].Numbers["soc"]; ok {
+		t.Fatal("out-of-range soc stored")
+	}
+	// Nothing usable is still a 200: sending it again would not help.
+	rec = do(t, h, "POST", "/v1/telemetry", testToken, string(telemetryBody(`{"utc":5}`)))
+	if rec.Code != 200 || len(store.telemetry) != 3 || !strings.Contains(rec.Body.String(), `"stored":0`) {
 		t.Fatalf("%d %s", rec.Code, rec.Body)
 	}
+	// A malformed request is still a 400.
+	if rec := do(t, h, "POST", "/v1/telemetry", testToken, `{"installationId":"nope","samples":[]}`); rec.Code != 400 {
+		t.Fatalf("malformed: %d", rec.Code)
+	}
 	if rec := do(t, h, "GET", "/v1/telemetry", testToken, ""); rec.Code != 405 {
 		t.Fatalf("GET on a POST route: want 405, got %d", rec.Code)
 	}

server/logic_test.go

@@ -75,6 +75,18 @@ func TestForwarderPicksOnlyTheNewestFreshSampleOncePerInterval(t *testing.T) {
 	if _, ok := fresh.pick([]TelemetrySample{{UTC: testNowS - 600}}, now); ok {
 		t.Fatal("forwarded a stale sample")
 	}
+	// A sample from a clock a day ahead is skipped, and does not block the real ones after it.
+	ahead := &Forwarder{MinInterval: 5 * time.Second}
+	got, ok = ahead.pick([]TelemetrySample{{UTC: testNowS + 86400}, {UTC: testNowS - 1}}, now)
+	if !ok || got.UTC != testNowS-1 {
+		t.Fatalf("picked %+v %v", got, ok)
+	}
+	if _, ok := ahead.pick([]TelemetrySample{{UTC: testNowS + 86400}}, now.Add(time.Minute)); ok {
+		t.Fatal("forwarded a sample from the future")
+	}
+	if _, ok := ahead.pick([]TelemetrySample{{UTC: testNowS + 60}}, now.Add(time.Minute)); !ok {
+		t.Fatal("a real sample after a future one was not forwarded")
+	}
 }
 
 func TestForwarderPutsCredentialsInTheBodyNotTheURL(t *testing.T) {

server/model.go

@@ -105,4 +105,6 @@ type TelemetrySample struct {
 type TelemetryBatch struct {
 	InstallationID string
 	Samples        []TelemetrySample
+	// Dropped names the values and samples validation left out, e.g. "samples[3].speed".
+	Dropped []string
 }

server/pairing.go

@@ -21,12 +21,19 @@ import (
 // replica. Approved devices are persisted (see Device) so their tokens survive a restart.
 
 const (
-	pairingTTL        = 10 * time.Minute
-	pairingInterval   = 5 * time.Second
-	maxPendingPairing = 50
-	// After this many wrong approval passwords in a window, approvals pause for the window.
-	maxFailedLogins    = 5
-	failedLoginsWindow = time.Minute
+	pairingTTL      = 10 * time.Minute
+	pairingInterval = 5 * time.Second
+	// Pending requests per client address, and in all. A car needs one; a client that opens
+	// more only crowds out its own, and the total bounds memory.
+	maxPendingPerClient = 3
+	maxPendingPairing   = 1000
+	// After this many wrong approval passwords from one client address in a window, that
+	// address waits for the window. Other addresses are not affected, so a stranger guessing
+	// cannot lock the owner out. The global cap only bounds guessing spread over many
+	// addresses; it is far above anything a person types.
+	maxFailedLogins       = 5
+	maxFailedLoginsGlobal = 100
+	failedLoginsWindow    = time.Minute
 	// Unambiguous characters only: no 0/O, 1/I/L, so a code reads right off a car screen.
 	userCodeAlphabet = "ABCDEFGHJKMNPQRSTUVWXYZ23456789"
 )
@@ -52,6 +59,9 @@ type pairingState int
 
 const (
 	pairingPending pairingState = iota
+	// pairingSaving: approved, and the device is being persisted. The car still sees
+	// pending, so it cannot collect a token whose device might not be saved.
+	pairingSaving
 	pairingApproved
 	pairingDenied
 )
@@ -60,6 +70,7 @@ type pairingRequest struct {
 	deviceCode     string
 	userCode       string
 	installationID string
+	client         string // address that opened the request
 	expires        time.Time
 	state          pairingState
 	token          string // set on approval, handed out once
@@ -75,7 +86,9 @@ type Pairing struct {
 	byDevice map[string]*pairingRequest
 	byUser   map[string]*pairingRequest
 	devices  map[string]Device // by token hash
-	failures []time.Time
+	// Wrong passwords: per client address, and all of them together.
+	failures       map[string][]time.Time
+	globalFailures []time.Time
 }
 
 func NewPairing(password string, devices []Device) *Pairing {
@@ -85,6 +98,7 @@ func NewPairing(password string, devices []Device) *Pairing {
 		byDevice:         map[string]*pairingRequest{},
 		byUser:           map[string]*pairingRequest{},
 		devices:          map[string]Device{},
+		failures:         map[string][]time.Time{},
 	}
 	for _, d := range devices {
 		p.devices[d.TokenHash] = d
@@ -137,25 +151,43 @@ func hashToken(token string) string {
 
 func (p *Pairing) expireLocked(now time.Time) {
 	for code, r := range p.byDevice {
-		if now.After(r.expires) {
+		// A request being saved is finished by Commit or Forget, not expired under them.
+		if now.After(r.expires) && r.state != pairingSaving {
 			delete(p.byDevice, code)
 			delete(p.byUser, r.userCode)
 		}
 	}
 }
 
-// Start opens a pairing request for a car.
-func (p *Pairing) Start(installationID string) (*pairingRequest, error) {
+func (p *Pairing) dropLocked(r *pairingRequest) {
+	delete(p.byDevice, r.deviceCode)
+	delete(p.byUser, r.userCode)
+}
+
+// Start opens a pairing request for a car. A new request replaces the same installation's
+// undecided one; a client already holding maxPendingPerClient requests gets errTooMany.
+func (p *Pairing) Start(installationID, client string) (*pairingRequest, error) {
 	p.mu.Lock()
 	defer p.mu.Unlock()
 	now := p.Now()
 	p.expireLocked(now)
-	if len(p.byDevice) >= maxPendingPairing {
+	perClient := 0
+	for _, r := range p.byDevice {
+		if r.state == pairingPending && strings.EqualFold(r.installationID, installationID) {
+			p.dropLocked(r)
+			continue
+		}
+		if r.client == client {
+			perClient++
+		}
+	}
+	if perClient >= maxPendingPerClient || len(p.byDevice) >= maxPendingPairing {
 		return nil, errTooMany
 	}
 	r := &pairingRequest{
 		deviceCode:     randomString(32),
 		installationID: installationID,
+		client:         client,
 		expires:        now.Add(pairingTTL),
 	}
 	for {
@@ -182,32 +214,46 @@ func (p *Pairing) Lookup(userCode string) (pairingRequest, error) {
 	return *r, nil
 }
 
-// checkPassword counts failures and refuses everything while too many are recent.
-func (p *Pairing) checkPasswordLocked(password string, now time.Time) error {
-	recent := p.failures[:0]
-	for _, t := range p.failures {
+func recentOnly(times []time.Time, now time.Time) []time.Time {
+	recent := times[:0]
+	for _, t := range times {
 		if now.Sub(t) < failedLoginsWindow {
 			recent = append(recent, t)
 		}
 	}
-	p.failures = recent
-	if len(p.failures) >= maxFailedLogins {
+	return recent
+}
+
+// checkPasswordLocked counts failures per client address and refuses a client while it
+// has too many recent ones (or while too many arrive from everywhere at once).
+func (p *Pairing) checkPasswordLocked(password, client string, now time.Time) error {
+	for c, times := range p.failures {
+		if recent := recentOnly(times, now); len(recent) == 0 {
+			delete(p.failures, c)
+		} else {
+			p.failures[c] = recent
+		}
+	}
+	p.globalFailures = recentOnly(p.globalFailures, now)
+	if len(p.failures[client]) >= maxFailedLogins || len(p.globalFailures) >= maxFailedLoginsGlobal {
 		return errTooManyFailed
 	}
 	if subtle.ConstantTimeCompare([]byte(password), []byte(p.ApprovalPassword)) != 1 {
-		p.failures = append(p.failures, now)
+		p.failures[client] = append(p.failures[client], now)
+		p.globalFailures = append(p.globalFailures, now)
 		return errWrongPassword
 	}
 	return nil
 }
 
 // Decide approves or denies a pending request. On approval it returns the new device, which
-// the caller must persist before the car can collect the token.
-func (p *Pairing) Decide(userCode, password string, approve bool) (Device, error) {
+// the caller must persist and then Commit (or Forget, if saving failed): until Commit the
+// car keeps seeing pending and the token is not accepted anywhere.
+func (p *Pairing) Decide(userCode, password, client string, approve bool) (Device, error) {
 	p.mu.Lock()
 	defer p.mu.Unlock()
 	now := p.Now()
-	if err := p.checkPasswordLocked(password, now); err != nil {
+	if err := p.checkPasswordLocked(password, client, now); err != nil {
 		return Device{}, err
 	}
 	p.expireLocked(now)
@@ -229,20 +275,29 @@ func (p *Pairing) Decide(userCode, password string, approve bool) (Device, error
 		TokenHash:      hashToken(token),
 		CreatedAt:      now,
 	}
-	r.state = pairingApproved
+	r.state = pairingSaving
 	r.token = token
 	r.deviceID = d.ID
-	p.devices[d.TokenHash] = d
 	return d, nil
 }
 
+// Commit finishes an approval once its device is persisted: the token now works, and the
+// car collects it at its next poll.
+func (p *Pairing) Commit(userCode string, d Device) {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	p.devices[d.TokenHash] = d
+	if r, ok := p.byUser[normalizeUserCode(userCode)]; ok && r.deviceID == d.ID {
+		r.state = pairingApproved
+	}
+}
+
 // Forget undoes an approval whose device could not be persisted, so no token is handed out
-// that would stop working at the next restart.
+// that would stop working at the next restart. The request is pending again.
 func (p *Pairing) Forget(userCode string, d Device) {
 	p.mu.Lock()
 	defer p.mu.Unlock()
-	delete(p.devices, d.TokenHash)
-	if r, ok := p.byUser[normalizeUserCode(userCode)]; ok {
+	if r, ok := p.byUser[normalizeUserCode(userCode)]; ok && r.deviceID == d.ID {
 		r.state = pairingPending
 		r.token = ""
 		r.deviceID = ""
@@ -259,23 +314,27 @@ func (p *Pairing) Poll(deviceCode string) (state pairingState, token, deviceID s
 		return 0, "", "", errUnknownCode
 	}
 	switch r.state {
+	case pairingSaving:
+		return pairingPending, "", "", nil
 	case pairingApproved, pairingDenied:
 		// Decided: hand the outcome out once, then forget the request.
-		delete(p.byDevice, r.deviceCode)
-		delete(p.byUser, r.userCode)
+		p.dropLocked(r)
 	}
 	return r.state, r.token, r.deviceID, nil
 }
 
-// Authorized reports whether a token belongs to a paired, unrevoked device.
-func (p *Pairing) Authorized(token string) bool {
+// Authorized returns the paired, unrevoked device a token belongs to.
+func (p *Pairing) Authorized(token string) (Device, bool) {
 	if p == nil || token == "" {
-		return false
+		return Device{}, false
 	}
 	p.mu.Lock()
 	defer p.mu.Unlock()
 	d, ok := p.devices[hashToken(token)]
-	return ok && !d.Revoked
+	if !ok || d.Revoked {
+		return Device{}, false
+	}
+	return d, true
 }
 
 func (p *Pairing) Devices() []Device {
@@ -304,8 +363,8 @@ func (p *Pairing) Revoke(id string) (Device, bool) {
 
 // CheckAdmin verifies the approval password for the device-management API, with the same
 // failure counting as the approval page.
-func (p *Pairing) CheckAdmin(password string) error {
+func (p *Pairing) CheckAdmin(password, client string) error {
 	p.mu.Lock()
 	defer p.mu.Unlock()
-	return p.checkPasswordLocked(password, p.Now())
+	return p.checkPasswordLocked(password, client, p.Now())
 }

server/pairing_http.go

@@ -5,6 +5,7 @@ import (
 	"html/template"
 	"io"
 	"log/slog"
+	"net"
 	"net/http"
 	"net/url"
 	"strings"
@@ -79,7 +80,7 @@ func (a *API) pairStart(w http.ResponseWriter, r *http.Request) {
 		writeJSON(w, http.StatusBadRequest, map[string]string{"error": err.Error()})
 		return
 	}
-	req, err := a.Pairing.Start(id)
+	req, err := a.Pairing.Start(id, clientAddress(r))
 	if err != nil {
 		writeJSON(w, http.StatusTooManyRequests, map[string]string{"error": err.Error()})
 		return
@@ -118,6 +119,25 @@ func (a *API) pairPoll(w http.ResponseWriter, r *http.Request) {
 	}
 }
 
+// clientAddress is who is asking, for the per-client limits: the last X-Forwarded-For entry,
+// which is the one Traefik adds (anything left of it comes from the client and can be
+// made up), or the connection's own address without a proxy.
+func clientAddress(r *http.Request) string {
+	if forwarded := r.Header.Values("X-Forwarded-For"); len(forwarded) > 0 {
+		last := forwarded[len(forwarded)-1]
+		if i := strings.LastIndex(last, ","); i >= 0 {
+			last = last[i+1:]
+		}
+		if addr := strings.TrimSpace(last); addr != "" {
+			return addr
+		}
+	}
+	if host, _, err := net.SplitHostPort(r.RemoteAddr); err == nil {
+		return host
+	}
+	return r.RemoteAddr
+}
+
 // ---------- Approval page ----------
 
 type pageData struct {
@@ -204,7 +224,7 @@ func (a *API) pairDecide(w http.ResponseWriter, r *http.Request) {
 	}
 	code := normalizeUserCode(r.PostForm.Get("code"))
 	approve := r.PostForm.Get("action") != "deny"
-	device, err := a.Pairing.Decide(code, r.PostForm.Get("password"), approve)
+	device, err := a.Pairing.Decide(code, r.PostForm.Get("password"), clientAddress(r), approve)
 	switch {
 	case errors.Is(err, errWrongPassword):
 		renderPage(w, http.StatusUnauthorized, pageData{Code: code, Error: "Wrong password."})
@@ -231,6 +251,7 @@ func (a *API) pairDecide(w http.ResponseWriter, r *http.Request) {
 		renderPage(w, http.StatusInternalServerError, pageData{Code: code, Error: "The car could not be saved. Try again."})
 		return
 	}
+	a.Pairing.Commit(code, device)
 	slog.Info("car paired", "device", device.ID, "installation", device.InstallationID)
 	renderPage(w, http.StatusOK, pageData{Done: true, Message: "Approved. The car finishes pairing on its own within a few seconds — you can close this page."})
 }
@@ -246,7 +267,7 @@ func (a *API) adminAuthorized(w http.ResponseWriter, r *http.Request) bool {
 		writeJSON(w, http.StatusUnauthorized, map[string]string{"error": "unauthorized"})
 		return false
 	}
-	switch err := a.Pairing.CheckAdmin(supplied); {
+	switch err := a.Pairing.CheckAdmin(supplied, clientAddress(r)); {
 	case errors.Is(err, errTooManyFailed):
 		writeJSON(w, http.StatusTooManyRequests, map[string]string{"error": err.Error()})
 		return false

server/pairing_test.go

@@ -34,10 +34,19 @@ func poll(t *testing.T, h http.Handler, deviceCode string) (int, map[string]stri
 }
 
 func submit(t *testing.T, h http.Handler, code, password, action string) *httptest.ResponseRecorder {
+	t.Helper()
+	return submitFrom(t, h, "", code, password, action)
+}
+
+// submitFrom submits the approval form as if from forwardedFor, the address Traefik adds.
+func submitFrom(t *testing.T, h http.Handler, forwardedFor, code, password, action string) *httptest.ResponseRecorder {
 	t.Helper()
 	form := url.Values{"code": {code}, "password": {password}, "action": {action}}
 	req := httptest.NewRequest("POST", "/pair", strings.NewReader(form.Encode()))
 	req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
+	if forwardedFor != "" {
+		req.Header.Set("X-Forwarded-For", forwardedFor)
+	}
 	rec := httptest.NewRecorder()
 	h.ServeHTTP(rec, req)
 	return rec
@@ -86,7 +95,7 @@ func TestPairingHappyPath(t *testing.T) {
 	}
 	// And it works on the ingest endpoints.
 	rec := do(t, h, "POST", "/v1/telemetry", got["token"], string(telemetryBody(`{"utc":1790000000,"soc":80}`)))
-	if rec.Code != 204 {
+	if rec.Code != 200 {
 		t.Fatalf("paired token on telemetry: %d %s", rec.Code, rec.Body)
 	}
 }
@@ -107,6 +116,11 @@ func TestPairingWrongPasswordAndLockout(t *testing.T) {
 	if status, _ := poll(t, h, deviceCode); status != 202 || len(store.devices) != 0 {
 		t.Fatal("a locked-out attempt changed the request")
 	}
+	// The lockout is per address: someone guessing elsewhere does not lock the owner out.
+	if rec := submitFrom(t, h, "198.51.100.7", userCode, testApprovalPassword, "deny"); rec.Code != 200 {
+		t.Fatalf("other address locked out too: %d", rec.Code)
+	}
+	deviceCode, userCode, _ = startPairing(t, h)
 	// After the window the right password works again.
 	api.Pairing.Now = func() time.Time { return time.Now().Add(failedLoginsWindow + time.Second) }
 	if rec := submit(t, h, userCode, testApprovalPassword, "approve"); rec.Code != 200 {
@@ -186,7 +200,7 @@ func TestRevokedTokenIsRejected(t *testing.T) {
 		t.Fatalf("revoked token accepted: %d", rec.Code)
 	}
 	// The shared ingest token is unaffected.
-	if rec := do(t, h, "POST", "/v1/telemetry", testToken, string(telemetryBody(`{"utc":1790000000,"soc":1}`))); rec.Code != 204 {
+	if rec := do(t, h, "POST", "/v1/telemetry", testToken, string(telemetryBody(`{"utc":1790000000,"soc":1}`))); rec.Code != 200 {
 		t.Fatalf("shared token: %d", rec.Code)
 	}
 }
@@ -195,7 +209,9 @@ func TestPairedDevicesSurviveARestart(t *testing.T) {
 	devices := []Device{{ID: "d1", TokenHash: hashToken("restored-token-0123456789abcdef"), CreatedAt: time.Now()},
 		{ID: "d2", TokenHash: hashToken("revoked-token-0123456789abcdef"), CreatedAt: time.Now(), Revoked: true}}
 	p := NewPairing(testApprovalPassword, devices)
-	if !p.Authorized("restored-token-0123456789abcdef") || p.Authorized("revoked-token-0123456789abcdef") {
+	_, restored := p.Authorized("restored-token-0123456789abcdef")
+	_, revoked := p.Authorized("revoked-token-0123456789abcdef")
+	if !restored || revoked {
 		t.Fatal("restored devices not honoured")
 	}
 }
@@ -232,3 +248,84 @@ func TestApprovalPasswordMustBeLongEnough(t *testing.T) {
 }
 
 var regexpUserCode = regexp.MustCompile(`^[` + userCodeAlphabet + `]{3}-[` + userCodeAlphabet + `]{3}$`)
+
+func TestPairedTokenOnlyServesItsOwnInstallation(t *testing.T) {
+	api, _ := newTestAPI()
+	h := api.Routes()
+	deviceCode, userCode, _ := startPairing(t, h)
+	submit(t, h, userCode, testApprovalPassword, "approve")
+	_, got := poll(t, h, deviceCode)
+	token := got["token"]
+	other := "7e2b3c4d-5e6f-4a70-9b8c-1d2e3f4a5b6c"
+	body := `{"installationId":"` + other + `","samples":[{"utc":1790000000,"soc":5}]}`
+	if rec := do(t, h, "POST", "/v1/telemetry", token, body); rec.Code != 403 {
+		t.Fatalf("telemetry for another car: %d", rec.Code)
+	}
+	if rec := do(t, h, "GET", "/v1/consumption/history/"+other, token, ""); rec.Code != 403 {
+		t.Fatalf("another car's history: %d", rec.Code)
+	}
+	if rec := do(t, h, "GET", "/v1/consumption/history/"+testInstallation, token, ""); rec.Code != 200 {
+		t.Fatalf("own history: %d", rec.Code)
+	}
+	// The shared token serves any installation.
+	if rec := do(t, h, "POST", "/v1/telemetry", testToken, body); rec.Code != 200 {
+		t.Fatalf("shared token: %d", rec.Code)
+	}
+}
+
+func TestATokenIsCollectableOnlyOnceSaved(t *testing.T) {
+	p := NewPairing(testApprovalPassword, nil)
+	r, err := p.Start(testInstallation, "192.0.2.1")
+	if err != nil {
+		t.Fatal(err)
+	}
+	d, err := p.Decide(r.userCode, testApprovalPassword, "192.0.2.1", true)
+	if err != nil {
+		t.Fatal(err)
+	}
+	// Saving is in flight: the car still waits, and the token is not accepted yet.
+	if state, token, _, _ := p.Poll(r.deviceCode); state != pairingPending || token != "" {
+		t.Fatalf("collectable while saving: %v %q", state, token)
+	}
+	p.Forget(r.userCode, d)
+	if _, err := p.Decide(r.userCode, testApprovalPassword, "192.0.2.1", true); err != nil {
+		t.Fatalf("not approvable again after a failed save: %v", err)
+	}
+}
+
+func TestPairingStartsAreLimitedPerAddress(t *testing.T) {
+	p := NewPairing(testApprovalPassword, nil)
+	ids := []string{"6f1a2b3c-4d5e-4f60-8a9b-0c1d2e3f4a51", "6f1a2b3c-4d5e-4f60-8a9b-0c1d2e3f4a52",
+		"6f1a2b3c-4d5e-4f60-8a9b-0c1d2e3f4a53", "6f1a2b3c-4d5e-4f60-8a9b-0c1d2e3f4a54"}
+	for _, id := range ids[:maxPendingPerClient] {
+		if _, err := p.Start(id, "203.0.113.9"); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if _, err := p.Start(ids[3], "203.0.113.9"); !errors.Is(err, errTooMany) {
+		t.Fatalf("over the per-address limit: %v", err)
+	}
+	// The car on its own address is unaffected, and starting again replaces its request.
+	first, err := p.Start(testInstallation, "192.0.2.1")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := p.Start(testInstallation, "192.0.2.1"); err != nil {
+		t.Fatal(err)
+	}
+	if _, _, _, err := p.Poll(first.deviceCode); !errors.Is(err, errUnknownCode) {
+		t.Fatal("the replaced request is still open")
+	}
+}
+
+func TestClientAddressTakesTheProxysEntry(t *testing.T) {
+	req := httptest.NewRequest("GET", "/", nil)
+	req.Header.Set("X-Forwarded-For", "10.9.9.9, 198.51.100.7")
+	if got := clientAddress(req); got != "198.51.100.7" {
+		t.Fatalf("got %q", got)
+	}
+	req.Header.Del("X-Forwarded-For")
+	if got := clientAddress(req); got != "192.0.2.1" {
+		t.Fatalf("without a proxy: %q", got)
+	}
+}

server/validate.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"math"
 	"regexp"
+	"sort"
 	"time"
 )
 
@@ -252,8 +253,13 @@ func validateChargingSession(body []byte, pathSessionID string) (ChargingSession
 
 // validateTelemetryBatch checks a batch of live samples in ABRP's tlm field names and
 // units. Unknown fields are dropped rather than rejected, so a newer app can add one
-// without a server release; every known field is bounded; a flag may be 0/1 or a boolean.
-func validateTelemetryBatch(body []byte) (TelemetryBatch, error) {
+// without a server release; a flag may be 0/1 or a boolean.
+//
+// Only a malformed request is a 400. A bad value inside a sample costs that value, and a
+// sample without a usable time costs that sample: the car drops a batch the server
+// rejects, and one sensor reading out of range must not take hours of queued driving with
+// it. What was dropped is listed in Dropped, for the response and the log.
+func validateTelemetryBatch(body []byte, now time.Time) (TelemetryBatch, error) {
 	o, err := decodeObject(body)
 	if err != nil {
 		return TelemetryBatch{}, err
@@ -266,34 +272,45 @@ func validateTelemetryBatch(body []byte) (TelemetryBatch, error) {
 	if !ok || len(raw) == 0 || len(raw) > maxSamples {
 		return TelemetryBatch{}, invalid("samples is invalid")
 	}
-	samples := make([]TelemetrySample, 0, len(raw))
+	batch := TelemetryBatch{InstallationID: id, Samples: make([]TelemetrySample, 0, len(raw))}
+	latest := now.Add(maxClockAhead).Unix()
 	for i, entry := range raw {
-		sample, err := validateSample(entry, fmt.Sprintf("samples[%d]", i))
-		if err != nil {
-			return TelemetryBatch{}, err
+		name := fmt.Sprintf("samples[%d]", i)
+		sample, dropped, ok := validateSample(entry, name, latest)
+		batch.Dropped = append(batch.Dropped, dropped...)
+		if ok {
+			batch.Samples = append(batch.Samples, sample)
 		}
-		samples = append(samples, sample)
 	}
-	return TelemetryBatch{InstallationID: id, Samples: samples}, nil
+	return batch, nil
 }
 
-func validateSample(entry any, name string) (TelemetrySample, error) {
+// maxClockAhead is how far in the future a sample may be stamped. The head unit sets its
+// clock from GPS; a sample from further ahead has a wrong time, and stored there it would
+// pose as the car's latest state until that time came.
+const maxClockAhead = time.Hour
+
+// validateSample keeps every valid field of one sample. It returns false when the sample
+// as a whole is unusable: not an object, or no plausible time. dropped names what was left
+// out.
+func validateSample(entry any, name string, latestUTC int64) (sample TelemetrySample, dropped []string, ok bool) {
 	o, err := asObject(entry, name)
 	if err != nil {
-		return TelemetrySample{}, err
+		return TelemetrySample{}, []string{name}, false
 	}
 	utc, err := integer(o["utc"], name+".utc")
-	if err != nil || utc < utcMin || utc > utcMax {
-		return TelemetrySample{}, invalid("%s.utc is invalid", name)
+	if err != nil || utc < utcMin || utc > utcMax || utc > latestUTC {
+		return TelemetrySample{}, []string{name + ".utc"}, false
 	}
-	sample := TelemetrySample{UTC: utc, Numbers: map[string]float64{}, Flags: map[string]bool{}}
+	sample = TelemetrySample{UTC: utc, Numbers: map[string]float64{}, Flags: map[string]bool{}}
 	for field, limits := range telemetryNumbers {
 		if !o.has(field) {
 			continue
 		}
 		f, err := bounded(o[field], limits[0], limits[1], name+"."+field)
 		if err != nil {
-			return TelemetrySample{}, err
+			dropped = append(dropped, name+"."+field)
+			continue
 		}
 		sample.Numbers[field] = f
 	}
@@ -311,17 +328,20 @@ func validateSample(entry any, name string) (TelemetrySample, error) {
 			case "0":
 				sample.Flags[field] = false
 			default:
-				return TelemetrySample{}, invalid("%s.%s is invalid", name, field)
+				dropped = append(dropped, name+"."+field)
 			}
 		default:
-			return TelemetrySample{}, invalid("%s.%s is invalid", name, field)
+			dropped = append(dropped, name+"."+field)
 		}
 	}
 	// A position is only a position as a pair.
 	_, hasLat := sample.Numbers["lat"]
 	_, hasLon := sample.Numbers["lon"]
 	if hasLat != hasLon {
-		return TelemetrySample{}, invalid("%s.lat and lon must be sent together", name)
+		delete(sample.Numbers, "lat")
+		delete(sample.Numbers, "lon")
+		dropped = append(dropped, name+".lat/lon")
 	}
-	return sample, nil
+	sort.Strings(dropped)
+	return sample, dropped, true
 }

server/validate_test.go

@@ -4,6 +4,7 @@ import (
 	"errors"
 	"strings"
 	"testing"
+	"time"
 )
 
 const (
@@ -12,6 +13,9 @@ const (
 	testNowS         = 1_790_000_000
 )
 
+// testNow is the sample time the telemetry tests send, as "now" for the clock-ahead check.
+var testNow = time.Unix(testNowS, 0)
+
 func isValidationError(err error) bool {
 	var v *ValidationError
 	return errors.As(err, &v)
@@ -125,7 +129,7 @@ func telemetryBody(samples string) []byte {
 
 func TestTelemetryKeepsOnlyWhatWasSent(t *testing.T) {
 	b, err := validateTelemetryBatch(telemetryBody(
-		`{"utc":1790000000,"soc":81,"power":-12.5,"speed":0,"lat":52.1,"lon":7.6,"is_charging":1,"is_dcfc":false}`))
+		`{"utc":1790000000,"soc":81,"power":-12.5,"speed":0,"lat":52.1,"lon":7.6,"is_charging":1,"is_dcfc":false}`), testNow)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -146,7 +150,7 @@ func TestTelemetryKeepsOnlyWhatWasSent(t *testing.T) {
 }
 
 func TestTelemetryDropsUnknownFields(t *testing.T) {
-	b, err := validateTelemetryBatch(telemetryBody(`{"utc":1790000000,"future_field":3}`))
+	b, err := validateTelemetryBatch(telemetryBody(`{"utc":1790000000,"future_field":3}`), testNow)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -155,31 +159,49 @@ func TestTelemetryDropsUnknownFields(t *testing.T) {
 	}
 }
 
-func TestTelemetryRejectsBadSamples(t *testing.T) {
-	for name, sample := range map[string]string{
-		"milliseconds":   `{"utc":1790000000000}`,
-		"fractional utc": `{"utc":1790000000.5}`,
-		"soc above 100":  `{"utc":1790000000,"soc":101}`,
-		"string speed":   `{"utc":1790000000,"speed":"fast"}`,
-		"flag 2":         `{"utc":1790000000,"is_parked":2}`,
-		"half position":  `{"utc":1790000000,"lat":52.1}`,
-		"not an object":  `5`,
+func TestTelemetryDropsBadValuesAndKeepsTheRest(t *testing.T) {
+	for name, c := range map[string]struct {
+		sample  string
+		dropped string
+		kept    bool // the sample itself survives
+	}{
+		"milliseconds":      {`{"utc":1790000000000,"soc":5}`, "samples[0].utc", false},
+		"fractional utc":    {`{"utc":1790000000.5,"soc":5}`, "samples[0].utc", false},
+		"clock unset":       {`{"utc":0,"soc":5}`, "samples[0].utc", false},
+		"clock a day ahead": {`{"utc":1790086400,"soc":5}`, "samples[0].utc", false},
+		"not an object":     {`5`, "samples[0]", false},
+		"soc above 100":     {`{"utc":1790000000,"soc":101,"speed":20}`, "samples[0].soc", true},
+		"reversing":         {`{"utc":1790000000,"soc":5,"speed":-4}`, "samples[0].speed", true},
+		"string speed":      {`{"utc":1790000000,"soc":5,"speed":"fast"}`, "samples[0].speed", true},
+		"flag 2":            {`{"utc":1790000000,"soc":5,"is_parked":2}`, "samples[0].is_parked", true},
+		"half position":     {`{"utc":1790000000,"soc":5,"lat":52.1}`, "samples[0].lat/lon", true},
 	} {
-		if _, err := validateTelemetryBatch(telemetryBody(sample)); !isValidationError(err) {
-			t.Errorf("%s: want ValidationError, got %v", name, err)
+		b, err := validateTelemetryBatch(telemetryBody(c.sample), testNow)
+		if err != nil {
+			t.Errorf("%s: the batch was rejected: %v", name, err)
+			continue
+		}
+		if len(b.Dropped) != 1 || b.Dropped[0] != c.dropped {
+			t.Errorf("%s: dropped %v, want %s", name, b.Dropped, c.dropped)
+		}
+		if (len(b.Samples) == 1) != c.kept {
+			t.Errorf("%s: %d samples kept", name, len(b.Samples))
+		}
+		if c.kept && b.Samples[0].Numbers["soc"] != 5 && name != "soc above 100" {
+			t.Errorf("%s: the valid soc was lost", name)
 		}
 	}
 }
 
 func TestTelemetryRejectsEmptyAndOversizedBatches(t *testing.T) {
-	if _, err := validateTelemetryBatch(telemetryBody("")); !isValidationError(err) {
+	if _, err := validateTelemetryBatch(telemetryBody(""), testNow); !isValidationError(err) {
 		t.Error("empty batch accepted")
 	}
 	many := strings.TrimSuffix(strings.Repeat(`{"utc":1790000000},`, 501), ",")
-	if _, err := validateTelemetryBatch(telemetryBody(many)); !isValidationError(err) {
+	if _, err := validateTelemetryBatch(telemetryBody(many), testNow); !isValidationError(err) {
 		t.Error("501 samples accepted")
 	}
-	if _, err := validateTelemetryBatch([]byte(`{"installationId":"nope","samples":[{"utc":1790000000}]}`)); !isValidationError(err) {
+	if _, err := validateTelemetryBatch([]byte(`{"installationId":"nope","samples":[{"utc":1790000000}]}`), testNow); !isValidationError(err) {
 		t.Error("bad installation id accepted")
 	}
 }