diff --git a/api/api.go b/api/api.go index fef528b8..bb9d6c21 100644 --- a/api/api.go +++ b/api/api.go @@ -51,6 +51,9 @@ func Routes() []*router.Route { router.NewRoute("POST", "/auth/upgrade-guest-existing", handlers.UpgradeGuestExisting), router.NewRoute("POST", "/network/auth-client", handlers.AuthNetworkClient), router.NewRoute("POST", "/network/remove-client", handlers.RemoveNetworkClient), + router.NewRoute("POST", "/network/provider-egress-location", handlers.ProviderEgressLocationSubmit), + router.NewRoute("GET", "/network/provider-egress-due", handlers.ProviderEgressLocationDue), + router.NewRoute("POST", "/network/provider-egress-attempt", handlers.ProviderEgressLocationAttempt), router.NewRoute("GET", "/network/clients", handlers.NetworkClients), router.NewRoute("GET", "/network/peers", handlers.NetworkPeers), router.NewRoute("GET", "/network/provider-locations", handlers.NetworkGetProviderLocations), diff --git a/api/handlers/provider_egress_location_handlers.go b/api/handlers/provider_egress_location_handlers.go new file mode 100644 index 00000000..264da746 --- /dev/null +++ b/api/handlers/provider_egress_location_handlers.go @@ -0,0 +1,229 @@ +package handlers + +import ( + "crypto/hmac" + "encoding/json" + "io" + "net/http" + "strconv" + "sync" + + "github.com/urnetwork/glog" + + "github.com/urnetwork/server" + "github.com/urnetwork/server/controller" + "github.com/urnetwork/server/model" +) + +// operatorSecretHeader carries the operator ingest secret. This endpoint is +// operator-to-server, not a client route: it is authenticated by a shared +// secret from the vault, not by a network jwt. +const operatorSecretHeader = "X-UR-Operator-Secret" + +// maxProviderEgressLocationBody bounds the request body. +const maxProviderEgressLocationBody = 16 * 1024 + +// operatorIngestSecret memoizes readOperatorIngestSecret for the life of the +// process. It is a package-level var (not a plain sync.OnceValue call site) +// so tests can swap it for a stub and restore it with defer; production code +// never reassigns it. +var operatorIngestSecret func() string = sync.OnceValue(readOperatorIngestSecret) + +// readOperatorIngestSecret reads the operator ingest secret from the vault +// resource "provider_egress.yml", key "ingest_secret". It returns "" +// when the vault resource is absent, the key is absent, or the key is empty, +// which makes the endpoint fail closed (every request is rejected) rather +// than open. `SimpleResource`/`String` are the non-panicking lookups (unlike +// `RequireSimpleResource`/`RequireString`), so a missing vault resource +// disables the endpoint instead of panicking the api process at startup or +// per-request. +func readOperatorIngestSecret() string { + res, err := server.Vault.SimpleResource("provider_egress.yml") + if err != nil { + glog.Infof("[pegl]no provider_egress.yml in the vault; ingest endpoint disabled\n") + return "" + } + values := res.String("ingest_secret") + if len(values) != 1 || values[0] == "" { + glog.Infof("[pegl]no ingest_secret in provider_egress.yml; ingest endpoint disabled\n") + return "" + } + return values[0] +} + +// ProviderEgressLocationSubmit ingests a probed provider egress location from +// the operator's prober. The prober routes geolocation lookups through a +// provider's own egress -- rather than relying on a lookup against the +// provider's control-connection ip -- and submits the result here so the +// server can prefer it over the built-in mmdb lookup. The route is +// operator-to-server, authenticated by the shared secret above rather than a +// network jwt. +func ProviderEgressLocationSubmit(w http.ResponseWriter, r *http.Request) { + secret := operatorIngestSecret() + provided := r.Header.Get(operatorSecretHeader) + if secret == "" || provided == "" || !hmac.Equal([]byte(secret), []byte(provided)) { + http.Error(w, "Unauthorized", http.StatusUnauthorized) + return + } + + body, err := io.ReadAll(io.LimitReader(r.Body, maxProviderEgressLocationBody+1)) + if err != nil { + http.Error(w, "Bad request", http.StatusBadRequest) + return + } + if len(body) > maxProviderEgressLocationBody { + http.Error(w, "Request too large", http.StatusRequestEntityTooLarge) + return + } + + var args controller.SubmitProviderEgressLocationArgs + if err := json.Unmarshal(body, &args); err != nil { + http.Error(w, "Bad request", http.StatusBadRequest) + return + } + + result, err := controller.SubmitProviderEgressLocation(r.Context(), &args) + if err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + + w.Header().Set("Content-Type", "application/json") + if err := json.NewEncoder(w).Encode(result); err != nil { + glog.Infof("[pegl]could not write response. err = %s\n", err) + } +} + +// ProviderEgressLocationAttempt records that the operator's prober tried to +// probe a provider, whether or not the try produced a location. +// +// The prober reports a *failure* here; a success is reported by +// ProviderEgressLocationSubmit above, whose provider_egress_location row +// already defers the provider for the full staleness window. Reporting a +// success here as well is harmless -- the attempt backoff is far shorter than +// that window -- but redundant. +// +// This exists because ProviderEgressLocationDue would otherwise be starved by +// providers that can never be probed successfully: they never get an egress +// row, so they sort to the head of the queue on every poll forever. See +// model.GetProviderEgressLocationDue. +// +// Same auth as the two endpoints around it: operator-to-server, the shared +// secret header rather than a network jwt, fail-closed when the vault resource +// is missing. +func ProviderEgressLocationAttempt(w http.ResponseWriter, r *http.Request) { + secret := operatorIngestSecret() + provided := r.Header.Get(operatorSecretHeader) + if secret == "" || provided == "" || !hmac.Equal([]byte(secret), []byte(provided)) { + http.Error(w, "Unauthorized", http.StatusUnauthorized) + return + } + + body, err := io.ReadAll(io.LimitReader(r.Body, maxProviderEgressLocationBody+1)) + if err != nil { + http.Error(w, "Bad request", http.StatusBadRequest) + return + } + if len(body) > maxProviderEgressLocationBody { + http.Error(w, "Request too large", http.StatusRequestEntityTooLarge) + return + } + + var args controller.RecordProviderEgressProbeAttemptArgs + if err := json.Unmarshal(body, &args); err != nil { + http.Error(w, "Bad request", http.StatusBadRequest) + return + } + + result, err := controller.RecordProviderEgressProbeAttempt(r.Context(), &args) + if err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + + w.Header().Set("Content-Type", "application/json") + if err := json.NewEncoder(w).Encode(result); err != nil { + glog.Infof("[pegl]could not write response. err = %s\n", err) + } +} + +const ( + // defaultProviderEgressDueLimit is the batch size when the caller does not + // ask for one. + defaultProviderEgressDueLimit = 100 + // maxProviderEgressDueLimit bounds the batch size regardless of what the + // caller asks for, so one request cannot ask the database for the entire + // provider population. + maxProviderEgressDueLimit = 500 +) + +// providerEgressDueAge is how stale a stored probe must be before its provider +// is offered up for re-probing. It is deliberately shorter than +// model.ProviderEgressLocationMaxAge -- the age past which a stored location +// stops being trusted at all. If the two were equal, every location would lapse +// to the mmdb fallback at the exact moment it became due and stay lapsed until +// the prober worked its way around to it; at half the max age the prober has a +// full max-age/2 window to refresh a location before it expires. +const providerEgressDueAge = model.ProviderEgressLocationMaxAge / 2 + +// ProviderEgressLocationDueResult is the response body of +// ProviderEgressLocationDue. +type ProviderEgressLocationDueResult struct { + ClientIds []server.Id `json:"client_ids"` +} + +// ProviderEgressLocationDue tells the operator's prober which providers to +// probe next: those whose egress location has gone stale, and those that have +// never been probed at all, oldest first. +// +// This moves the probe schedule from the prober's memory into the database. +// The prober used to decide what to probe from an in-memory ttl cache, so a +// restart re-probed the whole population and nothing durable recorded what was +// actually due; observed_at already carries that information server-side, and +// this exposes it. +// +// A provider is skipped if it has a fresh success *or* a recent attempt. The +// second cutoff, ProviderEgressProbeAttemptBackoff, is much shorter than the +// first: a provider that failed to probe should be retried within hours, but +// must not be handed back on every poll, which is what would starve the rest of +// the queue (see ProviderEgressLocationAttempt above). +// +// Same auth as ProviderEgressLocationSubmit above: operator-to-server, the +// shared secret header rather than a network jwt, fail-closed when the vault +// resource is missing. +func ProviderEgressLocationDue(w http.ResponseWriter, r *http.Request) { + secret := operatorIngestSecret() + provided := r.Header.Get(operatorSecretHeader) + if secret == "" || provided == "" || !hmac.Equal([]byte(secret), []byte(provided)) { + http.Error(w, "Unauthorized", http.StatusUnauthorized) + return + } + + limit := defaultProviderEgressDueLimit + if raw := r.URL.Query().Get("limit"); raw != "" { + parsed, err := strconv.Atoi(raw) + if err != nil || parsed < 1 { + // not clamped up to 1: `limit=0` would come back as an empty list, + // which the prober cannot distinguish from "nothing is due" + http.Error(w, "Bad request", http.StatusBadRequest) + return + } + limit = min(parsed, maxProviderEgressDueLimit) + } + + // both cutoffs are computed here and passed as arguments; observed_at and + // attempt_at are naive timestamps holding utc, so comparing them to sql + // now() in the query would cast through the session timezone + now := server.NowUtc() + minObservedAt := now.Add(-providerEgressDueAge) + minAttemptAt := now.Add(-model.ProviderEgressProbeAttemptBackoff) + + result := &ProviderEgressLocationDueResult{ + ClientIds: model.GetProviderEgressLocationDue(r.Context(), minObservedAt, minAttemptAt, limit), + } + + w.Header().Set("Content-Type", "application/json") + if err := json.NewEncoder(w).Encode(result); err != nil { + glog.Infof("[pegl]could not write response. err = %s\n", err) + } +} diff --git a/api/handlers/provider_egress_location_handlers_test.go b/api/handlers/provider_egress_location_handlers_test.go new file mode 100644 index 00000000..0f094ca0 --- /dev/null +++ b/api/handlers/provider_egress_location_handlers_test.go @@ -0,0 +1,552 @@ +package handlers + +import ( + "bytes" + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "slices" + "strings" + "testing" + "time" + + "github.com/urnetwork/server" + "github.com/urnetwork/server/controller" + "github.com/urnetwork/server/model" +) + +func TestProviderEgressLocationSubmitRejectsMissingSecret(t *testing.T) { + body, _ := json.Marshal(map[string]any{ + "client_id": "019f8835-158d-6fd8-e9dd-fd0e4c6d6792", + }) + req := httptest.NewRequest(http.MethodPost, "/network/provider-egress-location", bytes.NewReader(body)) + w := httptest.NewRecorder() + + ProviderEgressLocationSubmit(w, req) + + if w.Code != http.StatusUnauthorized { + t.Fatalf("status = %d, want 401 when the operator secret header is absent", w.Code) + } +} + +func TestProviderEgressLocationSubmitRejectsWrongSecret(t *testing.T) { + body, _ := json.Marshal(map[string]any{ + "client_id": "019f8835-158d-6fd8-e9dd-fd0e4c6d6792", + }) + req := httptest.NewRequest(http.MethodPost, "/network/provider-egress-location", bytes.NewReader(body)) + req.Header.Set(operatorSecretHeader, "definitely-not-the-secret") + w := httptest.NewRecorder() + + ProviderEgressLocationSubmit(w, req) + + if w.Code != http.StatusUnauthorized { + t.Fatalf("status = %d, want 401 on a wrong operator secret", w.Code) + } +} + +// withStubOperatorIngestSecret swaps the package-level operatorIngestSecret +// memo for a stub that always returns secret, and returns a func to restore +// the original (real, still-memoized) reader. This lets a test exercise the +// "configured vault" path without the sync.OnceValue in the real reader ever +// touching the vault, and without the reject tests above (which rely on an +// unconfigured vault) observing any change. +func withStubOperatorIngestSecret(secret string) (restore func()) { + prev := operatorIngestSecret + operatorIngestSecret = func() string { return secret } + return func() { operatorIngestSecret = prev } +} + +// TestProviderEgressLocationSubmitAcceptsCorrectSecret proves the auth gate +// can ACCEPT a correct secret and hand off to the controller. Without this +// test, the two reject tests above (which both run with the vault +// unconfigured and take the secret=="" short-circuit) would pass unchanged +// even if the handler's entire body were replaced with an unconditional 401 - +// hmac.Equal would never be proven to run on a real match. +// +// Clearing auth hands the request to controller.SubmitProviderEgressLocation, +// which looks up the client in the database before it can return "Unknown +// client.", so this test needs a real (throwaway) test database - see +// server.DefaultTestEnv, the same harness +// controller/provider_egress_location_controller_test.go uses. t.Setenv makes +// it self-sufficient under a plain `go test`, matching the pattern in +// router/warp_handlers_status_test.go. +func TestProviderEgressLocationSubmitAcceptsCorrectSecret(t *testing.T) { + t.Setenv("WARP_ENV", "local") + server.DefaultTestEnv().Run(t, func(t testing.TB) { + const secret = "correct-operator-secret-0123456789" + defer withStubOperatorIngestSecret(secret)() + + // A syntactically valid, semantically unregistered submission: once + // auth clears, the controller looks up the client and (since it does + // not exist in the fresh test database) returns "Unknown client.", + // surfaced by the handler as 400. That 400 is proof the request + // reached the controller, i.e. proof auth passed. + args := controller.SubmitProviderEgressLocationArgs{ + ClientId: server.NewId(), + CountryCode: "US", + Country: "United States", + CountryConfident: true, + ObservedAt: server.NowUtc(), + } + body, err := json.Marshal(args) + if err != nil { + t.Fatalf("marshal args: %s", err) + } + + req := httptest.NewRequest(http.MethodPost, "/network/provider-egress-location", bytes.NewReader(body)) + req.Header.Set(operatorSecretHeader, secret) + w := httptest.NewRecorder() + + ProviderEgressLocationSubmit(w, req) + + if w.Code == http.StatusUnauthorized { + t.Fatalf("status = %d, want the correct secret to clear auth (not 401)", w.Code) + } + if w.Code != http.StatusBadRequest { + t.Fatalf("status = %d, want 400 (Unknown client.) once auth clears for an unregistered client id; body = %s", w.Code, w.Body.String()) + } + if !strings.Contains(w.Body.String(), "Unknown client.") { + t.Fatalf("body = %q, want it to report the unknown client", w.Body.String()) + } + }) +} + +// TestProviderEgressLocationSubmitRejectsAlteredSecret proves that once the +// vault is configured, hmac.Equal is actually consulted rather than the +// endpoint accepting any request once secret != "". Same configured secret as +// the accept test above, but the request carries a one-character-altered +// value. +func TestProviderEgressLocationSubmitRejectsAlteredSecret(t *testing.T) { + const secret = "correct-operator-secret-0123456789" + const wrongSecret = "correct-operator-secret-0123456780" // last char changed + defer withStubOperatorIngestSecret(secret)() + + body, _ := json.Marshal(map[string]any{ + "client_id": "019f8835-158d-6fd8-e9dd-fd0e4c6d6792", + }) + req := httptest.NewRequest(http.MethodPost, "/network/provider-egress-location", bytes.NewReader(body)) + req.Header.Set(operatorSecretHeader, wrongSecret) + w := httptest.NewRecorder() + + ProviderEgressLocationSubmit(w, req) + + if w.Code != http.StatusUnauthorized { + t.Fatalf("status = %d, want 401 when the operator secret is configured but the request's secret is wrong", w.Code) + } +} + +// TestProviderEgressLocationSubmitReadsSecretFromVault proves the vault +// plumbing itself - not the test stub - returns the configured secret: +// readOperatorIngestSecret (the un-memoized reader) reads a +// PushSimpleResource-injected provider_egress.yml. +func TestProviderEgressLocationSubmitReadsSecretFromVault(t *testing.T) { + const secret = "vault-provisioned-secret-abcdef" + pop := server.Vault.PushSimpleResource( + "provider_egress.yml", + []byte(`ingest_secret: "`+secret+`"`), + ) + defer pop() + + if got := readOperatorIngestSecret(); got != secret { + t.Fatalf("readOperatorIngestSecret() = %q, want %q", got, secret) + } +} + +func TestProviderEgressLocationDueRejectsMissingSecret(t *testing.T) { + req := httptest.NewRequest(http.MethodGet, "/network/provider-egress-due", nil) + w := httptest.NewRecorder() + + ProviderEgressLocationDue(w, req) + + if w.Code != http.StatusUnauthorized { + t.Fatalf("status = %d, want 401 when the operator secret header is absent", w.Code) + } +} + +func TestProviderEgressLocationDueRejectsWrongSecret(t *testing.T) { + req := httptest.NewRequest(http.MethodGet, "/network/provider-egress-due", nil) + req.Header.Set(operatorSecretHeader, "definitely-not-the-secret") + w := httptest.NewRecorder() + + ProviderEgressLocationDue(w, req) + + if w.Code != http.StatusUnauthorized { + t.Fatalf("status = %d, want 401 on a wrong operator secret", w.Code) + } +} + +// TestProviderEgressLocationDueRejectsAlteredSecret is the reject case with +// the vault *configured*, so the request gets past the secret == "" fail-closed +// short-circuit and hmac.Equal is what does the rejecting. +func TestProviderEgressLocationDueRejectsAlteredSecret(t *testing.T) { + const secret = "correct-operator-secret-0123456789" + const wrongSecret = "correct-operator-secret-0123456780" // last char changed + defer withStubOperatorIngestSecret(secret)() + + req := httptest.NewRequest(http.MethodGet, "/network/provider-egress-due", nil) + req.Header.Set(operatorSecretHeader, wrongSecret) + w := httptest.NewRecorder() + + ProviderEgressLocationDue(w, req) + + if w.Code != http.StatusUnauthorized { + t.Fatalf("status = %d, want 401 when the operator secret is configured but the request's secret is wrong", w.Code) + } +} + +// testing_connectDueProvider stands up a connected + valid provider holding a +// Public provide key and no probe result, i.e. a provider the due query must +// return. The caller runs model.UpdateClientLocationReliabilities afterward. +func testing_connectDueProvider( + t testing.TB, + ctx context.Context, + clientId server.Id, + locationId server.Id, + clientAddress string, +) { + model.Testing_CreateDevice(ctx, server.NewId(), server.NewId(), clientId, "", "") + + handlerId := model.CreateNetworkClientHandler(ctx) + connectionId, _, _, _, err := model.ConnectNetworkClient(ctx, clientId, clientAddress, handlerId) + if err != nil { + t.Fatalf("connect client: %s", err) + } + if err := model.SetConnectionLocation(ctx, connectionId, locationId, &model.ConnectionLocationScores{}); err != nil { + t.Fatalf("set connection location: %s", err) + } + model.SetProvide(ctx, clientId, map[model.ProvideMode][]byte{ + model.ProvideModePublic: []byte("provide-secret"), + }) +} + +// TestProviderEgressLocationDueAcceptsCorrectSecret proves the auth gate can +// ACCEPT. This is the test that gives the three reject tests above their +// meaning: without it, a handler whose entire body was replaced with an +// unconditional `http.Error(w, "Unauthorized", 401)` would still pass all +// three, because a suite that only ever asserts rejections cannot tell a +// working auth check from a broken-shut one. +// +// It deliberately asserts more than "not 401": a real, never-probed provider +// is stood up in the test database and must come back in the response body, so +// the test also fails if the handler clears auth but never reaches the model +// query or writes the wrong json shape. +func TestProviderEgressLocationDueAcceptsCorrectSecret(t *testing.T) { + t.Setenv("WARP_ENV", "local") + server.DefaultTestEnv().Run(t, func(t testing.TB) { + const secret = "correct-operator-secret-0123456789" + defer withStubOperatorIngestSecret(secret)() + + ctx := context.Background() + + city := &model.Location{ + LocationType: model.LocationTypeCity, + City: "Palo Alto", + Region: "California", + Country: "United States", + CountryCode: "us", + } + model.CreateLocation(ctx, city) + + due := server.NewId() + testing_connectDueProvider(t, ctx, due, city.LocationId, "0.0.0.1:0") + model.UpdateClientLocationReliabilities(ctx, server.NowUtc().Add(-time.Hour), server.NowUtc()) + + req := httptest.NewRequest(http.MethodGet, "/network/provider-egress-due?limit=10", nil) + req.Header.Set(operatorSecretHeader, secret) + w := httptest.NewRecorder() + + ProviderEgressLocationDue(w, req) + + if w.Code == http.StatusUnauthorized { + t.Fatalf("status = %d, want the correct secret to clear auth (not 401)", w.Code) + } + if w.Code != http.StatusOK { + t.Fatalf("status = %d, want 200; body = %s", w.Code, w.Body.String()) + } + + var result ProviderEgressLocationDueResult + if err := json.Unmarshal(w.Body.Bytes(), &result); err != nil { + t.Fatalf("decode body %q: %s", w.Body.String(), err) + } + if !slices.Contains(result.ClientIds, due) { + t.Fatalf("client_ids = %v, want it to contain the never-probed provider %s", result.ClientIds, due) + } + // the wire name the prober reads + if !strings.Contains(w.Body.String(), `"client_ids"`) { + t.Fatalf("body = %s, want a client_ids field", w.Body.String()) + } + }) +} + +// The prober asks for a batch; the server must not hand back more than asked. +func TestProviderEgressLocationDueHonoursLimit(t *testing.T) { + t.Setenv("WARP_ENV", "local") + server.DefaultTestEnv().Run(t, func(t testing.TB) { + const secret = "correct-operator-secret-0123456789" + defer withStubOperatorIngestSecret(secret)() + + ctx := context.Background() + + city := &model.Location{ + LocationType: model.LocationTypeCity, + City: "Palo Alto", + Region: "California", + Country: "United States", + CountryCode: "us", + } + model.CreateLocation(ctx, city) + + testing_connectDueProvider(t, ctx, server.NewId(), city.LocationId, "0.0.0.1:0") + testing_connectDueProvider(t, ctx, server.NewId(), city.LocationId, "0.0.0.2:0") + model.UpdateClientLocationReliabilities(ctx, server.NowUtc().Add(-time.Hour), server.NowUtc()) + + req := httptest.NewRequest(http.MethodGet, "/network/provider-egress-due?limit=1", nil) + req.Header.Set(operatorSecretHeader, secret) + w := httptest.NewRecorder() + + ProviderEgressLocationDue(w, req) + + if w.Code != http.StatusOK { + t.Fatalf("status = %d, want 200; body = %s", w.Code, w.Body.String()) + } + var result ProviderEgressLocationDueResult + if err := json.Unmarshal(w.Body.Bytes(), &result); err != nil { + t.Fatalf("decode body %q: %s", w.Body.String(), err) + } + if len(result.ClientIds) != 1 { + t.Fatalf("len(client_ids) = %d, want 1 for limit=1; body = %s", len(result.ClientIds), w.Body.String()) + } + }) +} + +// Every other due test in this file stands up never-probed providers, which +// come back regardless of what cutoff the handler computes -- so nothing here +// actually exercised providerEgressDueAge. This one does: a provider probed +// just now must be held back, and one probed past the cutoff must come through. +// Defeating the cutoff (dropping it, computing it in the wrong direction, +// comparing against sql now() through the session timezone) fails this. +func TestProviderEgressLocationDueHonoursStalenessCutoff(t *testing.T) { + t.Setenv("WARP_ENV", "local") + server.DefaultTestEnv().Run(t, func(t testing.TB) { + const secret = "correct-operator-secret-0123456789" + defer withStubOperatorIngestSecret(secret)() + + ctx := context.Background() + + city := &model.Location{ + LocationType: model.LocationTypeCity, + City: "Palo Alto", + Region: "California", + Country: "United States", + CountryCode: "us", + } + model.CreateLocation(ctx, city) + + fresh := server.NewId() + stale := server.NewId() + testing_connectDueProvider(t, ctx, fresh, city.LocationId, "0.0.0.1:0") + testing_connectDueProvider(t, ctx, stale, city.LocationId, "0.0.0.2:0") + model.UpdateClientLocationReliabilities(ctx, server.NowUtc().Add(-time.Hour), server.NowUtc()) + + now := server.NowUtc() + model.SetProviderEgressLocation(ctx, &model.ProviderEgressLocation{ + ClientId: fresh, LocationId: city.LocationId, + CountryCode: "us", ObservedAt: now, + }) + // comfortably past providerEgressDueAge, which is half + // model.ProviderEgressLocationMaxAge + model.SetProviderEgressLocation(ctx, &model.ProviderEgressLocation{ + ClientId: stale, LocationId: city.LocationId, + CountryCode: "us", ObservedAt: now.Add(-providerEgressDueAge - time.Hour), + }) + + req := httptest.NewRequest(http.MethodGet, "/network/provider-egress-due?limit=100", nil) + req.Header.Set(operatorSecretHeader, secret) + w := httptest.NewRecorder() + + ProviderEgressLocationDue(w, req) + + if w.Code != http.StatusOK { + t.Fatalf("status = %d, want 200; body = %s", w.Code, w.Body.String()) + } + var result ProviderEgressLocationDueResult + if err := json.Unmarshal(w.Body.Bytes(), &result); err != nil { + t.Fatalf("decode body %q: %s", w.Body.String(), err) + } + if slices.Contains(result.ClientIds, fresh) { + t.Fatalf("client_ids = %v, must not contain the just-probed provider %s", result.ClientIds, fresh) + } + if !slices.Contains(result.ClientIds, stale) { + t.Fatalf("client_ids = %v, must contain the provider probed past the cutoff %s", result.ClientIds, stale) + } + }) +} + +func TestProviderEgressLocationAttemptRejectsMissingSecret(t *testing.T) { + body, _ := json.Marshal(map[string]any{ + "client_id": "019f8835-158d-6fd8-e9dd-fd0e4c6d6792", + }) + req := httptest.NewRequest(http.MethodPost, "/network/provider-egress-attempt", bytes.NewReader(body)) + w := httptest.NewRecorder() + + ProviderEgressLocationAttempt(w, req) + + if w.Code != http.StatusUnauthorized { + t.Fatalf("status = %d, want 401 when the operator secret header is absent", w.Code) + } +} + +// The reject case with the vault *configured*, so the request gets past the +// secret == "" fail-closed short-circuit and hmac.Equal is what rejects. +func TestProviderEgressLocationAttemptRejectsAlteredSecret(t *testing.T) { + const secret = "correct-operator-secret-0123456789" + const wrongSecret = "correct-operator-secret-0123456780" // last char changed + defer withStubOperatorIngestSecret(secret)() + + body, _ := json.Marshal(map[string]any{ + "client_id": "019f8835-158d-6fd8-e9dd-fd0e4c6d6792", + }) + req := httptest.NewRequest(http.MethodPost, "/network/provider-egress-attempt", bytes.NewReader(body)) + req.Header.Set(operatorSecretHeader, wrongSecret) + w := httptest.NewRecorder() + + ProviderEgressLocationAttempt(w, req) + + if w.Code != http.StatusUnauthorized { + t.Fatalf("status = %d, want 401 when the operator secret is configured but the request's secret is wrong", w.Code) + } +} + +// The whole point of the attempt endpoint, end to end over http: a provider +// that has never been probed successfully is due; the prober reports that it +// tried and failed; the provider stops being due. Without that, a provider +// whose probes always fail sits at the head of the queue on every poll forever +// (observed_at IS NULL sorts first) and starves every provider behind it. +func TestProviderEgressLocationAttemptDefersProvider(t *testing.T) { + t.Setenv("WARP_ENV", "local") + server.DefaultTestEnv().Run(t, func(t testing.TB) { + const secret = "correct-operator-secret-0123456789" + defer withStubOperatorIngestSecret(secret)() + + ctx := context.Background() + + city := &model.Location{ + LocationType: model.LocationTypeCity, + City: "Palo Alto", + Region: "California", + Country: "United States", + CountryCode: "us", + } + model.CreateLocation(ctx, city) + + dead := server.NewId() + testing_connectDueProvider(t, ctx, dead, city.LocationId, "0.0.0.1:0") + model.UpdateClientLocationReliabilities(ctx, server.NowUtc().Add(-time.Hour), server.NowUtc()) + + if !slices.Contains(due(t, secret), dead) { + t.Fatalf("the never-probed provider %s must be due before any attempt is reported", dead) + } + + attemptBody, err := json.Marshal(controller.RecordProviderEgressProbeAttemptArgs{ + ClientId: dead, + ProbeFailure: "tunnel_failed", + }) + if err != nil { + t.Fatalf("marshal attempt: %s", err) + } + req := httptest.NewRequest(http.MethodPost, "/network/provider-egress-attempt", bytes.NewReader(attemptBody)) + req.Header.Set(operatorSecretHeader, secret) + w := httptest.NewRecorder() + + ProviderEgressLocationAttempt(w, req) + + if w.Code != http.StatusOK { + t.Fatalf("status = %d, want 200; body = %s", w.Code, w.Body.String()) + } + + attempt := model.GetProviderEgressProbeAttempt(ctx, dead) + if attempt == nil { + t.Fatal("expected the attempt to be recorded") + } + if attempt.ProbeFailure != "tunnel_failed" { + t.Fatalf("probe_failure = %q, want %q", attempt.ProbeFailure, "tunnel_failed") + } + + if slices.Contains(due(t, secret), dead) { + t.Fatalf("the provider %s must not be due again immediately after a failed attempt", dead) + } + }) +} + +// due drives the due endpoint over http and returns the batch. +func due(t testing.TB, secret string) []server.Id { + req := httptest.NewRequest(http.MethodGet, "/network/provider-egress-due?limit=100", nil) + req.Header.Set(operatorSecretHeader, secret) + w := httptest.NewRecorder() + + ProviderEgressLocationDue(w, req) + + if w.Code != http.StatusOK { + t.Fatalf("due: status = %d, want 200; body = %s", w.Code, w.Body.String()) + } + var result ProviderEgressLocationDueResult + if err := json.Unmarshal(w.Body.Bytes(), &result); err != nil { + t.Fatalf("due: decode body %q: %s", w.Body.String(), err) + } + return result.ClientIds +} + +// An unknown client id must be rejected rather than writing an attempt row +// keyed to a client that does not exist, which nothing would ever read. +func TestProviderEgressLocationAttemptRejectsUnknownClient(t *testing.T) { + t.Setenv("WARP_ENV", "local") + server.DefaultTestEnv().Run(t, func(t testing.TB) { + const secret = "correct-operator-secret-0123456789" + defer withStubOperatorIngestSecret(secret)() + + body, err := json.Marshal(controller.RecordProviderEgressProbeAttemptArgs{ + ClientId: server.NewId(), + ProbeFailure: "tunnel_failed", + }) + if err != nil { + t.Fatalf("marshal attempt: %s", err) + } + req := httptest.NewRequest(http.MethodPost, "/network/provider-egress-attempt", bytes.NewReader(body)) + req.Header.Set(operatorSecretHeader, secret) + w := httptest.NewRecorder() + + ProviderEgressLocationAttempt(w, req) + + if w.Code == http.StatusUnauthorized { + t.Fatalf("status = %d, want the correct secret to clear auth (not 401)", w.Code) + } + if w.Code != http.StatusBadRequest { + t.Fatalf("status = %d, want 400 for an unregistered client id; body = %s", w.Code, w.Body.String()) + } + if !strings.Contains(w.Body.String(), "Unknown client.") { + t.Fatalf("body = %q, want it to report the unknown client", w.Body.String()) + } + }) +} + +// A limit that is not a positive integer is a caller bug. Silently clamping it +// to 1 (or to the default) would answer a question the prober did not ask -- +// `limit=0` would come back as an empty list, indistinguishable from "nothing +// is due" -- so it is rejected instead. +func TestProviderEgressLocationDueRejectsBadLimit(t *testing.T) { + const secret = "correct-operator-secret-0123456789" + defer withStubOperatorIngestSecret(secret)() + + for _, raw := range []string{"0", "-1", "abc", "1.5"} { + req := httptest.NewRequest(http.MethodGet, "/network/provider-egress-due?limit="+raw, nil) + req.Header.Set(operatorSecretHeader, secret) + w := httptest.NewRecorder() + + ProviderEgressLocationDue(w, req) + + if w.Code != http.StatusBadRequest { + t.Fatalf("limit=%q: status = %d, want 400", raw, w.Code) + } + } +} diff --git a/controller/network_client_controller.go b/controller/network_client_controller.go index 5ecd7252..4a723b6e 100644 --- a/controller/network_client_controller.go +++ b/controller/network_client_controller.go @@ -2,6 +2,7 @@ package controller import ( "context" + "net/netip" // "strings" "time" @@ -56,6 +57,64 @@ func SetConnectionLocation( connectionId server.Id, clientIp string, ) error { + // a provider probed through its own egress is located from that probe, not + // from a lookup on its control-connection ip: the egress is where user + // traffic actually exits, and an operator-run prober learns it by routing + // geolocation lookups through the provider itself and cross-checking them + // across several sources, then submits the result here. When a fresh + // probed entry exists we prefer it over the built-in mmdb lookup on the + // control ip. GetFreshProviderEgressLocationForConnection is + // a single query joining network_client_connection to + // provider_egress_location: this runs for every connection (provider or + // not) on the connect-announce path and inside a retry loop, so it must + // not cost the two round trips (client lookup, then egress lookup) the + // naive version would. + if egress := model.GetFreshProviderEgressLocationForConnection( + ctx, + connectionId, + model.ProviderEgressLocationMaxAge, + ); egress != nil { + scores := &model.ConnectionLocationScores{} + if egress.Hosting { + scores.NetTypeHosting = 1 + } + if egress.Proxy { + scores.NetTypePrivacy = 1 + } + // egress.Mobile deliberately does NOT feed NetTypeVirtual: unlike + // Hosting/Proxy, Mobile has no mmdb-path equivalent (IpInfo has no + // Mobile concept; NetTypeVirtual is set from the ipinfo schema's + // is_satellite field only, see GetLocationForIp, and never from + // DB-IP). Deriving NetTypeVirtual from Mobile here would penalize a + // probed mobile provider's ranking with no equivalent penalty for an + // otherwise-identical unprobed one -- the opposite of the parity + // this feature is meant to preserve (see arinForeignScore's doc for + // the same parity reasoning applied to net_type_foreign). Mobile + // stays on the model/wire contract as metadata; it just does not + // feed the ranking score. + // keep the ARIN org-vs-country foreign check on the probed path too, + // so a probed provider is ranked on equal terms with an equivalent + // unprobed one (net_type_foreign feeds the ranking columns). Compute + // it exactly as the mmdb path does in GetLocationForIp: the ARIN org + // country of the control ip against the mmdb country of that SAME + // control ip -- not the probed country, which is a different + // question (whether probing changed the answer) and must not be + // silently folded into this ranking penalty. Any lookup failure + // just leaves NetTypeForeign at 0; it must never fail or panic this + // path. + if addr, err := netip.ParseAddr(clientIp); err == nil { + if ipInfo, err := server.GetIpInfo(addr); err == nil { + scores.NetTypeForeign = arinForeignScore(addr, ipInfo.CountryCode) + } + } + err := model.SetConnectionLocation(ctx, connectionId, egress.LocationId, scores) + if err == nil { + return nil + } + // fall through to the mmdb path on a storage error + glog.Infof("[ncc][%s]could not set probed egress location. err = %s\n", connectionId, err) + } + location, connectionLocationScores, err := GetLocationForIp(ctx, clientIp) if err != nil { // server.Logger().Printf("Get ip for location error: %s", err) diff --git a/controller/network_client_controller_test.go b/controller/network_client_controller_test.go new file mode 100644 index 00000000..f6561fb1 --- /dev/null +++ b/controller/network_client_controller_test.go @@ -0,0 +1,335 @@ +package controller + +import ( + "context" + "testing" + "time" + + "github.com/urnetwork/connect" + "github.com/urnetwork/server" + "github.com/urnetwork/server/model" +) + +// A provider with a fresh probed egress location must be located from that +// entry, not from the mmdb lookup on its control ip. +func TestSetConnectionLocationPrefersEgressLocation(t *testing.T) { + server.DefaultTestEnv().Run(t, func(t testing.TB) { + ctx := context.Background() + + // the probed egress location: japan + probed := &model.Location{ + LocationType: model.LocationTypeCountry, + Country: "Japan", + CountryCode: "jp", + } + model.CreateLocation(ctx, probed) + + networkId := server.NewId() + clientId := server.NewId() + model.Testing_CreateDevice(ctx, networkId, server.NewId(), clientId, "", "") + + handlerId := model.CreateNetworkClientHandler(ctx) + connectionId, _, _, _, err := model.ConnectNetworkClient(ctx, clientId, "8.8.8.8:0", handlerId) + connect.AssertEqual(t, err, nil) + + model.SetProviderEgressLocation(ctx, &model.ProviderEgressLocation{ + ClientId: clientId, + LocationId: probed.LocationId, + CountryCode: "jp", + ObservedAt: server.NowUtc(), + }) + + err = SetConnectionLocation(ctx, connectionId, "8.8.8.8") + connect.AssertEqual(t, err, nil) + + var countryLocationId server.Id + server.Db(ctx, func(conn server.PgConn) { + result, qerr := conn.Query( + ctx, + `SELECT country_location_id FROM network_client_location WHERE connection_id = $1`, + connectionId, + ) + server.WithPgResult(result, qerr, func() { + if result.Next() { + server.Raise(result.Scan(&countryLocationId)) + } + }) + }) + connect.AssertEqual(t, countryLocationId, probed.CountryLocationId) + }) +} + +// With no probed entry, the existing mmdb path still applies. +func TestSetConnectionLocationFallsBackToMmdb(t *testing.T) { + server.DefaultTestEnv().Run(t, func(t testing.TB) { + ctx := context.Background() + + networkId := server.NewId() + clientId := server.NewId() + model.Testing_CreateDevice(ctx, networkId, server.NewId(), clientId, "", "") + + handlerId := model.CreateNetworkClientHandler(ctx) + connectionId, _, _, _, err := model.ConnectNetworkClient(ctx, clientId, "8.8.8.8:0", handlerId) + connect.AssertEqual(t, err, nil) + + // no SetProviderEgressLocation call -> mmdb path + err = SetConnectionLocation(ctx, connectionId, "8.8.8.8") + connect.AssertEqual(t, err, nil) + + var count int + server.Db(ctx, func(conn server.PgConn) { + result, qerr := conn.Query( + ctx, + `SELECT COUNT(*) FROM network_client_location WHERE connection_id = $1`, + connectionId, + ) + server.WithPgResult(result, qerr, func() { + if result.Next() { + server.Raise(result.Scan(&count)) + } + }) + }) + connect.AssertEqual(t, count, 1) + }) +} + +// A probed egress location observed longer ago than ProviderEgressLocationMaxAge +// must be ignored, falling back to the mmdb lookup exactly as if there were no +// probed entry at all. +func TestSetConnectionLocationStaleProbedFallsBackToMmdb(t *testing.T) { + server.DefaultTestEnv().Run(t, func(t testing.TB) { + ctx := context.Background() + + clientIp := "8.8.8.8" + + // the mmdb location this ip actually resolves to; created up front so + // its canonical (deduped) location id is known for the assertion. + mmdbLocation, _, err := GetLocationForIp(ctx, clientIp) + connect.AssertEqual(t, err, nil) + model.CreateLocation(ctx, mmdbLocation) + + // a stale probed location, deliberately a different country than + // whatever the mmdb lookup returns, so a wrongly-preferred probed + // value would be caught by the assertion below. + probed := &model.Location{ + LocationType: model.LocationTypeCountry, + Country: "Japan", + CountryCode: "jp", + } + model.CreateLocation(ctx, probed) + + networkId := server.NewId() + clientId := server.NewId() + model.Testing_CreateDevice(ctx, networkId, server.NewId(), clientId, "", "") + + handlerId := model.CreateNetworkClientHandler(ctx) + connectionId, _, _, _, err := model.ConnectNetworkClient(ctx, clientId, clientIp+":0", handlerId) + connect.AssertEqual(t, err, nil) + + model.SetProviderEgressLocation(ctx, &model.ProviderEgressLocation{ + ClientId: clientId, + LocationId: probed.LocationId, + CountryCode: "jp", + ObservedAt: server.NowUtc().Add(-(model.ProviderEgressLocationMaxAge + time.Hour)), + }) + + err = SetConnectionLocation(ctx, connectionId, clientIp) + connect.AssertEqual(t, err, nil) + + var countryLocationId server.Id + server.Db(ctx, func(conn server.PgConn) { + result, qerr := conn.Query( + ctx, + `SELECT country_location_id FROM network_client_location WHERE connection_id = $1`, + connectionId, + ) + server.WithPgResult(result, qerr, func() { + if result.Next() { + server.Raise(result.Scan(&countryLocationId)) + } + }) + }) + connect.AssertEqual(t, countryLocationId, mmdbLocation.CountryLocationId) + if countryLocationId == probed.CountryLocationId { + t.Fatal("stale probed location must not be used") + } + }) +} + +// A probed egress location whose location_id points at no existing location +// row makes the storage write fail (SetConnectionLocation returns an error +// rather than panicking, see model.SetConnectionLocation). The connection +// must still end up located via the mmdb path, and SetConnectionLocation +// itself must not panic or error out on this. +func TestSetConnectionLocationProbedWriteErrorFallsBackToMmdb(t *testing.T) { + server.DefaultTestEnv().Run(t, func(t testing.TB) { + ctx := context.Background() + + clientIp := "8.8.8.8" + + mmdbLocation, _, err := GetLocationForIp(ctx, clientIp) + connect.AssertEqual(t, err, nil) + model.CreateLocation(ctx, mmdbLocation) + + networkId := server.NewId() + clientId := server.NewId() + model.Testing_CreateDevice(ctx, networkId, server.NewId(), clientId, "", "") + + handlerId := model.CreateNetworkClientHandler(ctx) + connectionId, _, _, _, err := model.ConnectNetworkClient(ctx, clientId, clientIp+":0", handlerId) + connect.AssertEqual(t, err, nil) + + // a fresh probed entry pointing at a location id that was never + // created via CreateLocation -- the storage write in + // model.SetConnectionLocation must fail cleanly on this, not panic. + model.SetProviderEgressLocation(ctx, &model.ProviderEgressLocation{ + ClientId: clientId, + LocationId: server.NewId(), + CountryCode: "jp", + ObservedAt: server.NowUtc(), + }) + + err = SetConnectionLocation(ctx, connectionId, clientIp) + connect.AssertEqual(t, err, nil) + + var countryLocationId server.Id + server.Db(ctx, func(conn server.PgConn) { + result, qerr := conn.Query( + ctx, + `SELECT country_location_id FROM network_client_location WHERE connection_id = $1`, + connectionId, + ) + server.WithPgResult(result, qerr, func() { + if result.Next() { + server.Raise(result.Scan(&countryLocationId)) + } + }) + }) + connect.AssertEqual(t, countryLocationId, mmdbLocation.CountryLocationId) + }) +} + +// net_type_foreign on the probed path must be computed the same way as the +// mmdb path: the ARIN org country of the control ip against the mmdb country +// of that SAME control ip, not against the probed country. 8.8.8.8 resolves +// (both via mmdb and ARIN org registration) to "us", so it is non-foreign by +// construction -- this is the same ip used by the parity tests above. The +// probed country here is deliberately "jp", a different country than the +// control ip's mmdb country: probing having changed the answer must not, by +// itself, flip net_type_foreign, or a probed provider is penalized a full +// ranking tier precisely for the reason this feature exists. A prior version +// of this code compared the ARIN org country against the probed country +// instead of the control ip's mmdb country, which made this exact scenario +// (egress country differs from control-ip country) foreign=1 instead of the +// correct foreign=0. +func TestSetConnectionLocationProbedNetTypeForeignMatchesMmdbParity(t *testing.T) { + server.DefaultTestEnv().Run(t, func(t testing.TB) { + ctx := context.Background() + + clientIp := "8.8.8.8" + + probed := &model.Location{ + LocationType: model.LocationTypeCountry, + Country: "Japan", + CountryCode: "jp", + } + model.CreateLocation(ctx, probed) + + networkId := server.NewId() + clientId := server.NewId() + model.Testing_CreateDevice(ctx, networkId, server.NewId(), clientId, "", "") + + handlerId := model.CreateNetworkClientHandler(ctx) + connectionId, _, _, _, err := model.ConnectNetworkClient(ctx, clientId, clientIp+":0", handlerId) + connect.AssertEqual(t, err, nil) + + // probed country ("jp") deliberately differs from the control ip's + // mmdb country ("us" for 8.8.8.8). + model.SetProviderEgressLocation(ctx, &model.ProviderEgressLocation{ + ClientId: clientId, + LocationId: probed.LocationId, + CountryCode: "jp", + ObservedAt: server.NowUtc(), + }) + + err = SetConnectionLocation(ctx, connectionId, clientIp) + connect.AssertEqual(t, err, nil) + + var netTypeForeign int + server.Db(ctx, func(conn server.PgConn) { + result, qerr := conn.Query( + ctx, + `SELECT net_type_foreign FROM network_client_location WHERE connection_id = $1`, + connectionId, + ) + server.WithPgResult(result, qerr, func() { + if result.Next() { + server.Raise(result.Scan(&netTypeForeign)) + } + }) + }) + connect.AssertEqual(t, netTypeForeign, 0) + }) +} + +// A fresh probed location's Hosting/Proxy flags must map onto the stored +// connection's net_type_hosting/net_type_privacy scores. Mobile must NOT map +// onto net_type_virtual: Hosting/Proxy have direct mmdb-path equivalents +// (ipInfo.Hosting/ipInfo.Privacy, see GetLocationForIp), but Mobile has none +// (IpInfo has no Mobile concept at all, and NetTypeVirtual is only ever set +// from the ipinfo schema's is_satellite field, never from DB-IP or from +// anything Mobile-shaped) -- deriving NetTypeVirtual from Mobile would give +// a probed mobile provider a ranking penalty an identical unprobed mobile +// provider never takes, breaking the parity this feature promises. +func TestSetConnectionLocationMapsProbedFlagsToScores(t *testing.T) { + server.DefaultTestEnv().Run(t, func(t testing.TB) { + ctx := context.Background() + + probed := &model.Location{ + LocationType: model.LocationTypeCountry, + Country: "Japan", + CountryCode: "jp", + } + model.CreateLocation(ctx, probed) + + networkId := server.NewId() + clientId := server.NewId() + model.Testing_CreateDevice(ctx, networkId, server.NewId(), clientId, "", "") + + handlerId := model.CreateNetworkClientHandler(ctx) + connectionId, _, _, _, err := model.ConnectNetworkClient(ctx, clientId, "8.8.8.8:0", handlerId) + connect.AssertEqual(t, err, nil) + + model.SetProviderEgressLocation(ctx, &model.ProviderEgressLocation{ + ClientId: clientId, + LocationId: probed.LocationId, + CountryCode: "jp", + Hosting: true, + Proxy: true, + Mobile: true, + ObservedAt: server.NowUtc(), + }) + + err = SetConnectionLocation(ctx, connectionId, "8.8.8.8") + connect.AssertEqual(t, err, nil) + + var netTypeHosting int + var netTypePrivacy int + var netTypeVirtual int + server.Db(ctx, func(conn server.PgConn) { + result, qerr := conn.Query( + ctx, + `SELECT net_type_hosting, net_type_privacy, net_type_virtual FROM network_client_location WHERE connection_id = $1`, + connectionId, + ) + server.WithPgResult(result, qerr, func() { + if result.Next() { + server.Raise(result.Scan(&netTypeHosting, &netTypePrivacy, &netTypeVirtual)) + } + }) + }) + connect.AssertEqual(t, netTypeHosting, 1) + connect.AssertEqual(t, netTypePrivacy, 1) + connect.AssertEqual(t, netTypeVirtual, 0) + }) +} diff --git a/controller/network_client_location_controller.go b/controller/network_client_location_controller.go index 3376a9af..a8796b8b 100644 --- a/controller/network_client_location_controller.go +++ b/controller/network_client_location_controller.go @@ -55,21 +55,35 @@ func GetLocationForIp(ctx context.Context, clientIp string) (*model.Location, *m connectionLocationScores.NetTypeVirtual = 1 } + connectionLocationScores.NetTypeForeign = arinForeignScore(addr, ipInfo.CountryCode) + + return location, connectionLocationScores, nil +} + +// arinForeignScore cross-checks the ARIN org registration country for addr +// against countryCode. Both the ordinary path (GetLocationForIp) and the +// provider-egress path (SetConnectionLocation, in network_client_controller.go) pass the +// mmdb-resolved country of addr here, not the probed egress country: this is +// deliberate parity, so a probed and an unprobed provider on the same +// control ip are scored on the same basis and probing does not, by itself, +// change this ranking signal. If the org's registered country differs, the +// use case is considered foreign (VPN/proxy-like), matching the heuristic +// previously inlined in GetLocationForIp. +// +// If the ARIN lookup fails, this returns 0 without error: it must never fail +// or panic a caller on the connect-announce hot path over a missing/failed +// foreign check. +func arinForeignScore(addr netip.Addr, countryCode string) int { arinInfo, err := server.GetArinInfo(addr) - if err == nil { - // if the org ownership does not match the ip country, - // we consider the use case of the ip to be virtual - foreign := false - for _, orgCountryCode := range arinInfo.OrgCountryCodes { - if orgCountryCode != ipInfo.CountryCode { - foreign = true - break - } - } - if foreign { - connectionLocationScores.NetTypeForeign = 1 + if err != nil { + return 0 + } + // if the org ownership does not match the claimed country, + // we consider the use case of the ip to be foreign + for _, orgCountryCode := range arinInfo.OrgCountryCodes { + if orgCountryCode != countryCode { + return 1 } } - - return location, connectionLocationScores, nil + return 0 } diff --git a/controller/provider_egress_location_controller.go b/controller/provider_egress_location_controller.go new file mode 100644 index 00000000..4c688150 --- /dev/null +++ b/controller/provider_egress_location_controller.go @@ -0,0 +1,213 @@ +package controller + +import ( + "context" + "fmt" + "strings" + "time" + + "github.com/urnetwork/server" + "github.com/urnetwork/server/model" +) + +// MaxProviderEgressLocationSubmissionAge rejects a submission whose probe is +// already older than this when it arrives. It bounds replay of an old probe. +const MaxProviderEgressLocationSubmissionAge = 24 * time.Hour + +// MaxProviderEgressLocationSubmissionSkew rejects a submission whose +// observed_at is further in the future than this. The prober and server +// clocks should be roughly in sync, so a few minutes of allowance covers +// ordinary clock drift without opening the door to a far-future timestamp. +// Without this bound, a future observed_at would defeat every other +// safeguard at once: it always wins the monotonic upsert in +// model.SetProviderEgressLocation (so no later, legitimate probe can ever +// overwrite it), it reads as "fresh" forever against +// ProviderEgressLocationMaxAge, and it outlives the taskworker sweep in +// RemoveExpiredProviderEgressLocations -- permanently pinning a provider to +// whatever location was submitted, with no API-side recovery. +const MaxProviderEgressLocationSubmissionSkew = 5 * time.Minute + +// maxLocationNameLen bounds country/city/region as submitted: these flow into +// model.CreateLocation, whose location_name column is varchar(128). Rejecting +// an over-long value here with a clear error is preferable to letting +// CreateLocation panic on a Postgres "value too long for type character +// varying(128)" error. +const maxLocationNameLen = 128 + +// maxOrgLen mirrors maxLocationNameLen for org, which is stored in +// provider_egress_location.org, a varchar(256) column. +const maxOrgLen = 256 + +type SubmitProviderEgressLocationArgs struct { + ClientId server.Id `json:"client_id"` + CountryCode string `json:"country_code"` + Country string `json:"country"` + Region string `json:"region,omitempty"` + City string `json:"city,omitempty"` + ASN int `json:"asn,omitempty"` + Org string `json:"org,omitempty"` + Hosting bool `json:"hosting,omitempty"` + Proxy bool `json:"proxy,omitempty"` + Mobile bool `json:"mobile,omitempty"` + CountryConfident bool `json:"country_confident"` + CityConfident bool `json:"city_confident,omitempty"` + ObservedAt time.Time `json:"observed_at"` +} + +type SubmitProviderEgressLocationResult struct { + LocationId server.Id `json:"location_id"` +} + +// SubmitProviderEgressLocation records a probed egress location for a provider. +// Only country-confident submissions are accepted; city/region are stored only +// when the probe was also city-confident (free geolocation sources disagree on +// city often enough that an unconfirmed city is worse than none). +func SubmitProviderEgressLocation( + ctx context.Context, + args *SubmitProviderEgressLocationArgs, +) (*SubmitProviderEgressLocationResult, error) { + if !args.CountryConfident { + return nil, fmt.Errorf("Submission is not country-confident.") + } + countryCode := strings.ToLower(strings.TrimSpace(args.CountryCode)) + if len(countryCode) != 2 { + return nil, fmt.Errorf("Country code must be alpha-2.") + } + if args.ObservedAt.IsZero() { + return nil, fmt.Errorf("Missing observed_at.") + } + if args.ObservedAt.Before(server.NowUtc().Add(-MaxProviderEgressLocationSubmissionAge)) { + return nil, fmt.Errorf("Submission is too old.") + } + if server.NowUtc().Add(MaxProviderEgressLocationSubmissionSkew).Before(args.ObservedAt) { + return nil, fmt.Errorf("Submission is too far in the future.") + } + if networkId := model.GetNetworkClientNetwork(ctx, args.ClientId); networkId == nil { + return nil, fmt.Errorf("Unknown client.") + } + + // country is always used to resolve/create a location row (at minimum + // the country-granular one), and model.CreateLocation dedupes country + // rows on (location_type, country_code): an empty name here would create + // a canonical row with location_name='' that every later lookup for this + // country reuses forever, even after a subsequent real mmdb lookup. Reject + // rather than silently falling back, so the prober learns it sent a bad + // payload instead of the server permanently corrupting shared data. + country := strings.TrimSpace(args.Country) + if country == "" { + return nil, fmt.Errorf("Missing country.") + } + if maxLocationNameLen < len(country) { + return nil, fmt.Errorf("Country is too long.") + } + if maxOrgLen < len(args.Org) { + return nil, fmt.Errorf("Org is too long.") + } + + // city/region are only used (and their rows only created) when the probe + // was city-confident; the same empty-name corruption applies to them, so + // require both are present and reject rather than silently dropping to + // country granularity on a bad payload. + var city, region string + if args.CityConfident { + city = strings.TrimSpace(args.City) + region = strings.TrimSpace(args.Region) + if city == "" { + return nil, fmt.Errorf("Missing city for a city-confident submission.") + } + if region == "" { + return nil, fmt.Errorf("Missing region for a city-confident submission.") + } + if maxLocationNameLen < len(city) { + return nil, fmt.Errorf("City is too long.") + } + if maxLocationNameLen < len(region) { + return nil, fmt.Errorf("Region is too long.") + } + } + + // resolve to a canonical location row. city granularity only when the + // probe agreed on a city; otherwise country. + location := &model.Location{ + LocationType: model.LocationTypeCountry, + Country: country, + CountryCode: countryCode, + } + if args.CityConfident { + location = &model.Location{ + LocationType: model.LocationTypeCity, + City: city, + Region: region, + Country: country, + CountryCode: countryCode, + } + } + model.CreateLocation(ctx, location) + + model.SetProviderEgressLocation(ctx, &model.ProviderEgressLocation{ + ClientId: args.ClientId, + LocationId: location.LocationId, + CountryCode: countryCode, + ASN: args.ASN, + Org: args.Org, + Hosting: args.Hosting, + Proxy: args.Proxy, + Mobile: args.Mobile, + CityConfident: args.CityConfident, + ObservedAt: args.ObservedAt, + }) + + return &SubmitProviderEgressLocationResult{LocationId: location.LocationId}, nil +} + +// maxProbeFailureLen bounds the failure class as submitted: +// provider_egress_probe_attempt.probe_failure is a varchar(64), and rejecting +// an over-long value with a clear error beats letting the insert panic on a +// Postgres "value too long" error and spin in the retry loop. +const maxProbeFailureLen = 64 + +type RecordProviderEgressProbeAttemptArgs struct { + ClientId server.Id `json:"client_id"` + // ProbeFailure is "" when the attempt succeeded, otherwise a short failure + // class (`contract_failed`, `tunnel_failed`, `no_consensus`, ...). + ProbeFailure string `json:"probe_failure,omitempty"` +} + +type RecordProviderEgressProbeAttemptResult struct { + AttemptAt time.Time `json:"attempt_at"` +} + +// RecordProviderEgressProbeAttempt records that the prober tried this provider. +// A failed attempt defers the provider from the due queue for +// ProviderEgressProbeAttemptBackoff, exactly as a successful probe defers it +// for the (much longer) staleness window -- without this, a provider that +// always fails to probe never gets a provider_egress_location row and so stays +// permanently at the head of the queue, starving every other provider. See +// model.GetProviderEgressLocationDue. +// +// The attempt is timestamped by the server, not the prober: the prober is +// reporting something it just did, and a prober whose clock ran fast could +// otherwise defer a provider far past the backoff window. +func RecordProviderEgressProbeAttempt( + ctx context.Context, + args *RecordProviderEgressProbeAttemptArgs, +) (*RecordProviderEgressProbeAttemptResult, error) { + if maxProbeFailureLen < len(args.ProbeFailure) { + return nil, fmt.Errorf("Probe failure class is too long.") + } + // same check as SubmitProviderEgressLocation: without it a typo'd or stale + // client id writes a row keyed to a client that does not exist, which + // nothing ever reads and only the sweep ever removes. + if networkId := model.GetNetworkClientNetwork(ctx, args.ClientId); networkId == nil { + return nil, fmt.Errorf("Unknown client.") + } + + attemptAt := server.NowUtc() + model.SetProviderEgressProbeAttempt(ctx, &model.ProviderEgressProbeAttempt{ + ClientId: args.ClientId, + AttemptAt: attemptAt, + ProbeFailure: args.ProbeFailure, + }) + + return &RecordProviderEgressProbeAttemptResult{AttemptAt: attemptAt}, nil +} diff --git a/controller/provider_egress_location_controller_test.go b/controller/provider_egress_location_controller_test.go new file mode 100644 index 00000000..d11c08c8 --- /dev/null +++ b/controller/provider_egress_location_controller_test.go @@ -0,0 +1,389 @@ +package controller + +import ( + "context" + "strings" + "testing" + "time" + + "github.com/urnetwork/connect" + "github.com/urnetwork/server" + "github.com/urnetwork/server/model" +) + +func TestSubmitProviderEgressLocationCountryOnly(t *testing.T) { + server.DefaultTestEnv().Run(t, func(t testing.TB) { + ctx := context.Background() + + networkId := server.NewId() + clientId := server.NewId() + model.Testing_CreateDevice(ctx, networkId, server.NewId(), clientId, "", "") + + res, err := SubmitProviderEgressLocation(ctx, &SubmitProviderEgressLocationArgs{ + ClientId: clientId, + CountryCode: "US", + Country: "United States", + ASN: 401486, + Org: "RAVNIX LLC", + Hosting: true, + CountryConfident: true, + ObservedAt: server.NowUtc(), + }) + connect.AssertEqual(t, err, nil) + if res.LocationId == (server.Id{}) { + t.Fatal("expected a resolved location id") + } + + stored := model.GetProviderEgressLocation(ctx, clientId) + if stored == nil { + t.Fatal("expected the submission to be stored") + } + connect.AssertEqual(t, stored.CountryCode, "us") + connect.AssertEqual(t, stored.ASN, 401486) + connect.AssertEqual(t, stored.Hosting, true) + connect.AssertEqual(t, stored.CityConfident, false) + + // the resolved location must be the country-granular row, with no + // city/region association + loc := model.GetLocation(ctx, stored.LocationId) + if loc == nil { + t.Fatal("expected the resolved location row to exist") + } + connect.AssertEqual(t, loc.LocationType, model.LocationTypeCountry) + if loc.CityLocationId != (server.Id{}) { + t.Fatal("a country-granularity row must not have a city association") + } + if loc.RegionLocationId != (server.Id{}) { + t.Fatal("a country-granularity row must not have a region association") + } + }) +} + +func TestSubmitProviderEgressLocationRejectsNotCountryConfident(t *testing.T) { + server.DefaultTestEnv().Run(t, func(t testing.TB) { + ctx := context.Background() + + networkId := server.NewId() + clientId := server.NewId() + model.Testing_CreateDevice(ctx, networkId, server.NewId(), clientId, "", "") + + _, err := SubmitProviderEgressLocation(ctx, &SubmitProviderEgressLocationArgs{ + ClientId: clientId, + CountryCode: "us", + CountryConfident: false, + ObservedAt: server.NowUtc(), + }) + if err == nil { + t.Fatal("a submission that is not country-confident must be rejected") + } + if model.GetProviderEgressLocation(ctx, clientId) != nil { + t.Fatal("rejected submission must not be stored") + } + }) +} + +func TestSubmitProviderEgressLocationRejectsUnknownClient(t *testing.T) { + server.DefaultTestEnv().Run(t, func(t testing.TB) { + ctx := context.Background() + _, err := SubmitProviderEgressLocation(ctx, &SubmitProviderEgressLocationArgs{ + ClientId: server.NewId(), // never created + CountryCode: "us", + CountryConfident: true, + ObservedAt: server.NowUtc(), + }) + if err == nil { + t.Fatal("unknown client_id must be rejected") + } + }) +} + +func TestSubmitProviderEgressLocationCityConfidentStoresCity(t *testing.T) { + server.DefaultTestEnv().Run(t, func(t testing.TB) { + ctx := context.Background() + + networkId := server.NewId() + clientId := server.NewId() + model.Testing_CreateDevice(ctx, networkId, server.NewId(), clientId, "", "") + + _, err := SubmitProviderEgressLocation(ctx, &SubmitProviderEgressLocationArgs{ + ClientId: clientId, + CountryCode: "us", + Country: "United States", + Region: "Colorado", + City: "Denver", + CountryConfident: true, + CityConfident: true, + ObservedAt: server.NowUtc(), + }) + connect.AssertEqual(t, err, nil) + + stored := model.GetProviderEgressLocation(ctx, clientId) + if stored == nil { + t.Fatal("expected the submission to be stored") + } + connect.AssertEqual(t, stored.CityConfident, true) + + // the resolved location must be the city-granular row + loc := model.GetLocation(ctx, stored.LocationId) + if loc == nil { + t.Fatal("expected the resolved location row to exist") + } + connect.AssertEqual(t, loc.LocationType, model.LocationTypeCity) + }) +} + +// An empty Country must be rejected, not silently stored: model.CreateLocation +// dedupes country rows on (location_type, country_code), so an empty name +// would create a canonical, permanently-blank row that every later lookup for +// that country reuses forever. +func TestSubmitProviderEgressLocationRejectsEmptyCountry(t *testing.T) { + server.DefaultTestEnv().Run(t, func(t testing.TB) { + ctx := context.Background() + + networkId := server.NewId() + clientId := server.NewId() + model.Testing_CreateDevice(ctx, networkId, server.NewId(), clientId, "", "") + + _, err := SubmitProviderEgressLocation(ctx, &SubmitProviderEgressLocationArgs{ + ClientId: clientId, + CountryCode: "us", + Country: " ", // blank after trimming + CountryConfident: true, + ObservedAt: server.NowUtc(), + }) + if err == nil { + t.Fatal("an empty country must be rejected") + } + if model.GetProviderEgressLocation(ctx, clientId) != nil { + t.Fatal("rejected submission must not be stored") + } + }) +} + +// A city-confident submission with an empty City must be rejected rather than +// silently falling back to country granularity: the same empty-canonical-row +// corruption applies to city/region rows. +func TestSubmitProviderEgressLocationRejectsEmptyCityWhenCityConfident(t *testing.T) { + server.DefaultTestEnv().Run(t, func(t testing.TB) { + ctx := context.Background() + + networkId := server.NewId() + clientId := server.NewId() + model.Testing_CreateDevice(ctx, networkId, server.NewId(), clientId, "", "") + + _, err := SubmitProviderEgressLocation(ctx, &SubmitProviderEgressLocationArgs{ + ClientId: clientId, + CountryCode: "us", + Country: "United States", + Region: "Colorado", + City: "", + CountryConfident: true, + CityConfident: true, + ObservedAt: server.NowUtc(), + }) + if err == nil { + t.Fatal("a city-confident submission with an empty city must be rejected") + } + if model.GetProviderEgressLocation(ctx, clientId) != nil { + t.Fatal("rejected submission must not be stored") + } + }) +} + +// A city-confident submission with an empty Region must likewise be rejected: +// the region row is created with the same dedupe-on-empty-name hazard. +func TestSubmitProviderEgressLocationRejectsEmptyRegionWhenCityConfident(t *testing.T) { + server.DefaultTestEnv().Run(t, func(t testing.TB) { + ctx := context.Background() + + networkId := server.NewId() + clientId := server.NewId() + model.Testing_CreateDevice(ctx, networkId, server.NewId(), clientId, "", "") + + _, err := SubmitProviderEgressLocation(ctx, &SubmitProviderEgressLocationArgs{ + ClientId: clientId, + CountryCode: "us", + Country: "United States", + Region: "", + City: "Denver", + CountryConfident: true, + CityConfident: true, + ObservedAt: server.NowUtc(), + }) + if err == nil { + t.Fatal("a city-confident submission with an empty region must be rejected") + } + if model.GetProviderEgressLocation(ctx, clientId) != nil { + t.Fatal("rejected submission must not be stored") + } + }) +} + +// An over-long Country must be rejected with a clear error instead of +// panicking inside model.CreateLocation on a Postgres "value too long for +// type character varying(128)" error. +func TestSubmitProviderEgressLocationRejectsOverLongCountry(t *testing.T) { + server.DefaultTestEnv().Run(t, func(t testing.TB) { + ctx := context.Background() + + networkId := server.NewId() + clientId := server.NewId() + model.Testing_CreateDevice(ctx, networkId, server.NewId(), clientId, "", "") + + _, err := SubmitProviderEgressLocation(ctx, &SubmitProviderEgressLocationArgs{ + ClientId: clientId, + CountryCode: "us", + Country: strings.Repeat("a", 129), + CountryConfident: true, + ObservedAt: server.NowUtc(), + }) + if err == nil { + t.Fatal("an over-long country must be rejected") + } + if model.GetProviderEgressLocation(ctx, clientId) != nil { + t.Fatal("rejected submission must not be stored") + } + }) +} + +// An over-long Org must likewise be rejected rather than panicking inside +// storage. +func TestSubmitProviderEgressLocationRejectsOverLongOrg(t *testing.T) { + server.DefaultTestEnv().Run(t, func(t testing.TB) { + ctx := context.Background() + + networkId := server.NewId() + clientId := server.NewId() + model.Testing_CreateDevice(ctx, networkId, server.NewId(), clientId, "", "") + + _, err := SubmitProviderEgressLocation(ctx, &SubmitProviderEgressLocationArgs{ + ClientId: clientId, + CountryCode: "us", + Country: "United States", + Org: strings.Repeat("a", 257), + CountryConfident: true, + ObservedAt: server.NowUtc(), + }) + if err == nil { + t.Fatal("an over-long org must be rejected") + } + if model.GetProviderEgressLocation(ctx, clientId) != nil { + t.Fatal("rejected submission must not be stored") + } + }) +} + +func TestSubmitProviderEgressLocationRejectsStaleObservedAt(t *testing.T) { + server.DefaultTestEnv().Run(t, func(t testing.TB) { + ctx := context.Background() + + networkId := server.NewId() + clientId := server.NewId() + model.Testing_CreateDevice(ctx, networkId, server.NewId(), clientId, "", "") + + _, err := SubmitProviderEgressLocation(ctx, &SubmitProviderEgressLocationArgs{ + ClientId: clientId, + CountryCode: "us", + CountryConfident: true, + ObservedAt: server.NowUtc().Add(-30 * 24 * time.Hour), + }) + if err == nil { + t.Fatal("a submission observed long ago must be rejected") + } + }) +} + +// A far-future observed_at must be rejected, not just an old one: unchecked, +// it would defeat the monotonic upsert (it always "wins"), read as fresh +// forever, and outlive the taskworker sweep -- permanently pinning a +// provider's location with no API-side recovery. See +// MaxProviderEgressLocationSubmissionSkew. +func TestSubmitProviderEgressLocationRejectsFutureObservedAt(t *testing.T) { + server.DefaultTestEnv().Run(t, func(t testing.TB) { + ctx := context.Background() + + networkId := server.NewId() + clientId := server.NewId() + model.Testing_CreateDevice(ctx, networkId, server.NewId(), clientId, "", "") + + _, err := SubmitProviderEgressLocation(ctx, &SubmitProviderEgressLocationArgs{ + ClientId: clientId, + CountryCode: "jp", + Country: "Japan", + CountryConfident: true, + ObservedAt: server.NowUtc().Add(10 * 365 * 24 * time.Hour), + }) + if err == nil { + t.Fatal("a far-future observed_at must be rejected") + } + if model.GetProviderEgressLocation(ctx, clientId) != nil { + t.Fatal("rejected submission must not be stored") + } + }) +} + +// A submission within the allowed clock-skew window must still be accepted: +// the future-timestamp rejection must not be so strict that ordinary clock +// drift between the prober and server breaks legitimate submissions. +func TestSubmitProviderEgressLocationAcceptsWithinSkewObservedAt(t *testing.T) { + server.DefaultTestEnv().Run(t, func(t testing.TB) { + ctx := context.Background() + + networkId := server.NewId() + clientId := server.NewId() + model.Testing_CreateDevice(ctx, networkId, server.NewId(), clientId, "", "") + + res, err := SubmitProviderEgressLocation(ctx, &SubmitProviderEgressLocationArgs{ + ClientId: clientId, + CountryCode: "us", + Country: "United States", + CountryConfident: true, + ObservedAt: server.NowUtc().Add(1 * time.Minute), + }) + connect.AssertEqual(t, err, nil) + if res.LocationId == (server.Id{}) { + t.Fatal("expected a resolved location id") + } + + stored := model.GetProviderEgressLocation(ctx, clientId) + if stored == nil { + t.Fatal("expected the submission to be stored") + } + }) +} + +// A large 32-bit ASN (e.g. from the private-use range, common on +// hosting/VPN infrastructure) must round-trip cleanly, not panic. The asn +// column used to be `int` (Postgres int4, max ~2.147e9); ASNs are 32-bit +// unsigned (max ~4.295e9), so a value above int4's range panicked deep in +// pgx's arg encoding. +func TestSubmitProviderEgressLocationAcceptsLargeAsn(t *testing.T) { + server.DefaultTestEnv().Run(t, func(t testing.TB) { + ctx := context.Background() + + networkId := server.NewId() + clientId := server.NewId() + model.Testing_CreateDevice(ctx, networkId, server.NewId(), clientId, "", "") + + const largeAsn = 4200000000 + + res, err := SubmitProviderEgressLocation(ctx, &SubmitProviderEgressLocationArgs{ + ClientId: clientId, + CountryCode: "us", + Country: "United States", + ASN: largeAsn, + CountryConfident: true, + ObservedAt: server.NowUtc(), + }) + connect.AssertEqual(t, err, nil) + if res.LocationId == (server.Id{}) { + t.Fatal("expected a resolved location id") + } + + stored := model.GetProviderEgressLocation(ctx, clientId) + if stored == nil { + t.Fatal("expected the submission to be stored") + } + connect.AssertEqual(t, stored.ASN, largeAsn) + }) +} diff --git a/db_migrations.go b/db_migrations.go index 063f091e..7bcb3344 100644 --- a/db_migrations.go +++ b/db_migrations.go @@ -4385,4 +4385,69 @@ var migrations = []any{ CREATE INDEX IF NOT EXISTS network_client_top_level_contract_time ON network_client (contract_time) WHERE (active = true AND source_client_id IS NULL AND contract_time IS NOT NULL) `), + + // provider egress locations: locations learned by an operator-run prober + // that routes geolocation lookups through a provider's own egress rather + // than trusting a lookup on the provider's control-connection ip, since + // the egress is where user traffic actually exits and can differ from + // where the provider's control connection originates (e.g. behind a VPN + // or hosting network). Keyed by client_id, one row per provider, upserted + // by the operator's prober. location_id is the canonical country (or + // city, when the probe was city-confident) location row. observed_at is + // when the probe ran, and is what freshness is judged against. + newSqlMigration(` + CREATE TABLE IF NOT EXISTS provider_egress_location ( + client_id uuid NOT NULL PRIMARY KEY, + location_id uuid NOT NULL, + country_code varchar(2) NOT NULL, + asn bigint NOT NULL DEFAULT 0, + org varchar(256) NOT NULL DEFAULT '', + hosting bool NOT NULL DEFAULT false, + proxy bool NOT NULL DEFAULT false, + mobile bool NOT NULL DEFAULT false, + city_confident bool NOT NULL DEFAULT false, + observed_at timestamp NOT NULL, + update_time timestamp NOT NULL + ) + `), + newSqlMigration(` + CREATE INDEX IF NOT EXISTS provider_egress_location_observed_at + ON provider_egress_location (observed_at) + `), + + // provider egress probe attempts: when the prober last *tried* a provider, + // successful or not, and how the try failed. + // + // This cannot live on provider_egress_location, because the case it exists + // to handle is precisely a provider that has no row there. A provider that + // connects, holds a Public provide key and fails every probe (firewalled + // egress, dead upstream) never gets an egress row, so its observed_at stays + // NULL, so it sorts to the head of the due queue forever. Enough of them and + // every batch the prober asks for is the same set of permanently-dead + // providers, and no healthy provider's location is ever refreshed -- while + // the endpoint keeps returning a full, plausible-looking batch. + // GetProviderEgressLocationDue defers on a recent attempt as well as a fresh + // success, which needs somewhere to record the attempt. + // + // Pulled forward from the P2 verdict model + // (docs/superpowers/specs/2026-07-25-enforced-provider-geo-probing-design.md, + // probe_attempt_at / probe_failure) because the P1 schedule cannot function + // without it. Deliberately only the two columns the schedule reads, not the + // rest of that model. + newSqlMigration(` + CREATE TABLE IF NOT EXISTS provider_egress_probe_attempt ( + client_id uuid NOT NULL PRIMARY KEY, + attempt_at timestamp NOT NULL, + probe_failure varchar(64) NOT NULL DEFAULT '', + update_time timestamp NOT NULL + ) + `), + + // serves the sweep in RemoveExpiredProviderEgressProbeAttempts. The due + // query reaches this table by primary key through the left join, so it + // needs no index of its own. + newSqlMigration(` + CREATE INDEX IF NOT EXISTS provider_egress_probe_attempt_attempt_at + ON provider_egress_probe_attempt (attempt_at) + `), } diff --git a/docs/operator/prober-systemd.md b/docs/operator/prober-systemd.md new file mode 100644 index 00000000..6a3442c6 --- /dev/null +++ b/docs/operator/prober-systemd.md @@ -0,0 +1,256 @@ +# Running the egress prober under systemd + +The egress prober measures each provider's real exit location by opening a tunnel +**through that provider** and asking public geolocation APIs what address they see. +It must never be able to reach those APIs any other way. A direct lookup would +record the *operator's own* location for the provider and hand the operator's +address to third-party APIs. + +A Docker deployment can confine it by attaching the prober to an +`internal: true` network -- no gateway, no NAT -- and to no other network. The +mainstream deployment uses no Docker at all, so this page gives the equivalent: +a systemd unit whose egress is denied by the kernel. + +`IPAddressDeny=` / `IPAddressAllow=` are enforced by systemd's cgroup/BPF filter. +No container, no network namespace, no capability grant, and no root — the +filter applies to the service's cgroup regardless of the user it runs as. + +## Prerequisites + +- systemd 235 or newer, cgroup v2, and BPF available. If any is missing systemd + logs `Failed to add IP address ... ignoring` and **the address rules do + nothing**. See [Verifying it is actually enforced](#verifying-it-is-actually-enforced). +- The `egress-prober` binary from + `https://github.com/Ryanmello07/urnetwork-operator-proxy` at + `/usr/local/bin/egress-prober`: + + ```bash + git clone https://github.com/Ryanmello07/urnetwork-operator-proxy.git + git clone https://github.com/urnetwork/connect.git # ../connect replace + git clone https://github.com/urnetwork/glog.git # ../glog replace + cd urnetwork-operator-proxy + CGO_ENABLED=0 go build -trimpath -o /usr/local/bin/egress-prober ./cmd/egress-prober + ``` + +## The unit + +Save as `/etc/systemd/system/urnetwork-egress-prober.service`. + +```ini +[Unit] +Description=URnetwork provider egress prober +Documentation=https://github.com/urnetwork/server/blob/main/docs/operator/prober-systemd.md +After=network-online.target nss-lookup.target +Wants=network-online.target + +[Service] +Type=simple +ExecStart=/usr/local/bin/egress-prober \ + --api-url=https://api.example.net \ + --platform-url=wss://connect.example.net \ + --interval=1h \ + --concurrency=4 \ + --probe-timeout=60s \ + --confinement-timeout=3s + +# UR_PROBER_BY_JWT and UR_OPERATOR_SECRET. systemd reads this file as PID 1, +# before dropping privileges, so it can be root-owned 0600 even though the +# service itself runs as an unprivileged dynamic user: +# chown root:root /etc/urnetwork/prober.env && chmod 600 /etc/urnetwork/prober.env +# Passing either secret on the command line would publish it in `ps` and in +# `systemctl show`. +EnvironmentFile=/etc/urnetwork/prober.env + +# --------------------------------------------------------------------------- +# THE CONFINEMENT. Everything below this line is the point of the unit. +# --------------------------------------------------------------------------- +# Deny all IP traffic, then allow back exactly the operator's own platform and +# loopback. Enforced by the kernel through systemd's cgroup BPF filter, which +# DROPS the packet rather than rejecting it: a connection to anything not +# listed hangs until its own deadline instead of failing fast. That is what a +# deny rule looks like from inside, and the self-check counts a dial timeout as +# blocked -- but it means the check costs one --confinement-timeout per address +# at startup (6 addresses x 3s = ~18s observed), not one in total. +IPAddressDeny=any + +# Loopback. Required so the process can talk to the local DNS stub resolver +# (systemd-resolved on 127.0.0.53). See "DNS" below -- this is what lets the +# self-check resolve the geolocation hostnames and then discover it cannot +# reach them, which is the evidence it needs to start. +IPAddressAllow=localhost + +# The operator's platform. IPAddressAllow TAKES ADDRESSES, NOT HOSTNAMES -- +# see "Addresses, not hostnames" below. Replace these with the addresses of +# your own api and connect hosts: +# getent ahostsv4 api.example.net connect.example.net +IPAddressAllow=198.51.100.10 # api.example.net +IPAddressAllow=198.51.100.11 # connect.example.net +# If your DNS resolver is not on loopback, add its address here too, or use +# --confinement-address (see "DNS" below): +# IPAddressAllow=192.0.2.53 + +# No privileges of any kind. The provider tunnel is a userspace gvisor +# netstack, not a kernel tun device, so the prober needs no capabilities. +DynamicUser=yes +NoNewPrivileges=yes +CapabilityBoundingSet= +AmbientCapabilities= +RestrictAddressFamilies=AF_INET AF_INET6 +PrivateDevices=yes +ProtectKernelTunables=yes +ProtectKernelModules=yes +ProtectControlGroups=yes +RestrictNamespaces=yes +RestrictSUIDSGID=yes +LockPersonality=yes +MemoryDenyWriteExecute=yes +SystemCallFilter=@system-service +SystemCallArchitectures=native + +# A failed confinement self-check is a configuration fault, not a blip. Retry +# a few times in case the platform was briefly unreachable, then stop and stay +# failed so it is visible in `systemctl status` instead of looping forever. +Restart=on-failure +RestartSec=60s +StartLimitIntervalSec=600 +StartLimitBurst=5 + +[Install] +WantedBy=multi-user.target +``` + +```bash +sudo systemctl daemon-reload +sudo systemctl enable --now urnetwork-egress-prober +journalctl -u urnetwork-egress-prober -f +``` + +A healthy start logs: + +``` +egress-prober: confinement self-check: 3 geolocation host(s) -> 4 address(es): ... +egress-prober: confinement self-check passed: 4 address(es) tested, none directly reachable +``` + +## Never set `--skip-confinement-check` + +The flag exists for a one-shot manual probe from a host you know is not the +operator's. A check disabled in a unit file is not a check: it makes the +process start in exactly the situation the unit is meant to prevent, and the +only trace is two `WARNING` lines in the journal. It is not in the unit above +and it must not be added to it. + +## DNS: why loopback is allowed + +The prober's self-check refuses to start when it cannot obtain evidence either +way. If it cannot resolve the geolocation hostnames it exits with +`ErrNoEvidence` — being unable to verify is not evidence of confinement. So a +unit with `IPAddressDeny=any` and nothing allowed back does **not** produce a +confined prober; it produces one that will not start. + +There are two ways to resolve this, and this unit takes the first: + +1. **Permit DNS (this unit).** `IPAddressAllow=localhost` lets the prober reach + the local stub resolver. `systemd-resolved` is a separate, unrestricted + service, so it performs the upstream query on the prober's behalf and the + prober's own egress stays denied. The self-check then resolves + `ip.pn`, `free.freeipapi.com` and `ipinfo.io`, tries to connect to each + address, has every packet dropped by the BPF filter until the dial + deadline expires, and starts. That is the healthy case, and it needs no + hand-maintained list of geolocation addresses: + + ``` + egress-prober: confinement self-check: 3 geolocation host(s) -> 6 address(es): 134.119.216.174:443 [2606:4700:3037::6815:5e88]:443 ... 34.117.59.81:443 + egress-prober: confinement self-check passed: 6 address(es) tested, none directly reachable + ``` + + If your resolver is *not* on loopback (`/etc/resolv.conf` points at + `192.0.2.53`, say), add that address to `IPAddressAllow` as well — a DNS + server address is not a geolocation API, so allowing it does not weaken the + confinement. + +2. **Skip resolution.** Where DNS is genuinely unavailable, pass the addresses + to dial directly and the check stays real: + + ```ini + ExecStart=/usr/local/bin/egress-prober \ + ... \ + --confinement-address=134.119.216.174:443 \ + --confinement-address=104.21.94.136:443 \ + --confinement-address=172.67.168.79:443 \ + --confinement-address=34.117.59.81:443 + ``` + + The flag is repeatable and rejects hostnames — a name could not be resolved + at dial time either, so accepting one would mean dialling nothing. This is a + second copy of the endpoint list and it can go stale; the prober prints the + hostnames it expects (`geolocation hosts: ...`) on every start so drift is + visible in the journal. A confined docker deployment has to take this + route, because an `internal: true` docker network cannot resolve external + names at all -- the embedded resolver answers SERVFAIL, having no route to + forward the query. + +## Addresses, not hostnames + +`IPAddressAllow=` takes IP addresses and CIDR ranges (plus the aliases +`localhost`, `link-local`, `multicast`, `any`). **It does not take hostnames.** +systemd resolves nothing here — there is no name to resolve at packet-filter +time. + +That has two consequences an operator has to live with: + +- You must list your platform's addresses literally, and **update the unit when + they change**. A platform address that moves — a new load balancer, a new + region, a provider migration — breaks the prober silently in the *safe* + direction: it can no longer reach `api` or `connect`, every pass logs a + failure, and no probe is submitted. Nothing leaks. +- The tempting fix is to widen the rule: `IPAddressAllow=any` "just to get it + running", or a whole `/8` because the platform "is somewhere in there". + **That is the dangerous direction, and the prober's own self-check is what + catches it.** On the next start the check finds a geolocation address + directly reachable and refuses to run: + + ``` + egress-prober: confinement self-check failed: confinement: a direct connection + to a geolocation address succeeded; this process is not confined: 134.119.216.174:443 + egress-prober: this process must not be able to reach a geolocation api except + through a provider tunnel. ... + ``` + + Exit code 1, and with the `Restart=on-failure` limits above the unit lands in + `failed`. The same is true if the address rules are not being enforced at all + (no cgroup v2, no BPF, an old systemd): the prober discovers it can reach the + APIs and stops. The confinement and the check are two halves of one + mechanism — the unit restricts, the check proves the restriction is real. + +## Verifying it is actually enforced + +Confirm systemd accepted the rules, rather than trusting that it did: + +```bash +systemctl show urnetwork-egress-prober -p IPAddressDeny -p IPAddressAllow +journalctl -u urnetwork-egress-prober | grep -i 'ip address' # expect no "ignoring" +``` + +Then confirm from the outside that the deny actually bites, using the service's +own settings but a harmless command: + +```bash +sudo systemd-run --uid=nobody \ + -p IPAddressDeny=any -p IPAddressAllow=localhost \ + --wait --pty /bin/sh -c 'curl -sS --max-time 5 https://ipinfo.io/ip; echo rc=$?' +# expect a connect failure, not an address: +# curl: (28) Connection timed out after 5002 milliseconds +# rc=28 +``` + +If that prints an address, the filter is not being applied on this host and the +prober must not be run here until it is. + +## Alternative: one pass per timer + +`--interval=0` runs a single pass and exits, reporting success or failure in the +exit code, for operators who prefer a timer to a long-running service. Use +`Type=oneshot`, drop `Restart=`, and add a companion +`urnetwork-egress-prober.timer`. Everything in the confinement block above stays +exactly the same — including the self-check, which then runs on every firing. diff --git a/model/network_client_location_model.go b/model/network_client_location_model.go index a2650fc4..811cc80e 100644 --- a/model/network_client_location_model.go +++ b/model/network_client_location_model.go @@ -1242,6 +1242,36 @@ func SetConnectionLocation( } }) + // geo databases vary in how deep their coverage goes: many IPs -- + // datacenter, mobile, and VPN egress especially -- only resolve to + // country or region granularity, with no city. network_client_location + // requires city_location_id and region_location_id NOT NULL, so a + // country-only location's NULL city/region made this INSERT panic + // inside server.Tx. That panic propagated out of the connection + // announce goroutine (connect/transport_announce.go), whose + // HandleError wrapper then cancelled the whole connection context -- + // tearing down every country-only client's connect connection right + // after auth (the app itself included, whichever egress it resolved + // to), and, because the panic hit before the disconnect-cleanup + // defer was registered, orphaning the connection row as + // connected=true forever. Fall back to the coarsest available + // granularity so the columns are always non-null: a country-only + // location stores its country id for city/region too, which keeps + // the provider locatable at country level instead of crashing the + // connection. If even the country id is missing (location row absent + // or malformed), return a clean error so the caller's existing + // graceful retry path handles it -- never panic here. + if countryLocationId == nil { + returnErr = fmt.Errorf("Location %s has no country granularity.", locationId) + return + } + if cityLocationId == nil { + cityLocationId = countryLocationId + } + if regionLocationId == nil { + regionLocationId = countryLocationId + } + server.RaisePgResult(tx.Exec( ctx, ` diff --git a/model/network_client_location_model_test.go b/model/network_client_location_model_test.go index ab91c7ee..888a1ce2 100644 --- a/model/network_client_location_model_test.go +++ b/model/network_client_location_model_test.go @@ -1198,3 +1198,62 @@ func TestFindProviders2ReliabilityDeployGap(t *testing.T) { connect.AssertEqual(t, len(res.Providers), n) }) } + +// SetConnectionLocation must not panic when the resolved location has no city +// (country-only). network_client_location requires city_location_id and +// region_location_id NOT NULL, but a country-granularity location row has them +// NULL, so the insert panicked inside server.Tx. That panic propagated out of +// the connection announce goroutine, whose HandleError wrapper cancelled the +// connection context -- tearing down the client's connection right after auth, +// and (because the panic hit before the disconnect-cleanup defer was +// registered) orphaning the connection row as connected=true. This asserts a +// country-only location is stored, falling back to country granularity, with +// no panic and no error. +func TestSetConnectionLocationToleratesCountryOnlyLocation(t *testing.T) { + server.DefaultTestEnv().Run(t, func(t testing.TB) { + ctx := context.Background() + + // country-only location -- its row has NULL city_location_id and + // NULL region_location_id, exactly what crashed the insert before + country := &Location{ + LocationType: LocationTypeCountry, + Country: "United States", + CountryCode: "us", + } + CreateLocation(ctx, country) + + networkId := server.NewId() + clientId := server.NewId() + Testing_CreateDevice(ctx, networkId, server.NewId(), clientId, "", "") + + handlerId := CreateNetworkClientHandler(ctx) + connectionId, _, _, _, err := ConnectNetworkClient(ctx, clientId, "0.0.0.1:0", handlerId) + connect.AssertEqual(t, err, nil) + + // this call panicked before the fix; now it must succeed and store + // the connection at country granularity + err = SetConnectionLocation(ctx, connectionId, country.LocationId, &ConnectionLocationScores{}) + connect.AssertEqual(t, err, nil) + + var city, region, cty *server.Id + server.Db(ctx, func(conn server.PgConn) { + result, qerr := conn.Query( + ctx, + `SELECT city_location_id, region_location_id, country_location_id FROM network_client_location WHERE connection_id = $1`, + connectionId, + ) + server.WithPgResult(result, qerr, func() { + if result.Next() { + server.Raise(result.Scan(&city, ®ion, &cty)) + } + }) + }) + if city == nil || region == nil || cty == nil { + t.Fatal("expected all three location ids to be set (falling back to country), got a nil") + } + // city and region fall back to the country id for a country-only location + connect.AssertEqual(t, *city, country.CountryLocationId) + connect.AssertEqual(t, *region, country.CountryLocationId) + connect.AssertEqual(t, *cty, country.CountryLocationId) + }) +} diff --git a/model/provider_egress_location_model.go b/model/provider_egress_location_model.go new file mode 100644 index 00000000..55d0aaf3 --- /dev/null +++ b/model/provider_egress_location_model.go @@ -0,0 +1,500 @@ +package model + +import ( + "context" + "strings" + "time" + + "github.com/urnetwork/server" +) + +// ProviderEgressLocationMaxAge bounds how long a probed egress location is +// trusted. Past this, the location is ignored and the caller falls back to the +// mmdb lookup on the observed control ip. +const ProviderEgressLocationMaxAge = 7 * 24 * time.Hour + +// ProviderEgressProbeAttemptBackoff is how long a probe *attempt* defers a +// provider from being offered up again, whether or not the attempt succeeded. +// +// It is much shorter than the staleness window a successful probe buys +// (providerEgressDueAge in api/handlers, half ProviderEgressLocationMaxAge): a +// provider that fails to probe should be retried periodically -- the fault may +// be transient -- just not on every single poll, which is what starves the rest +// of the queue. +const ProviderEgressProbeAttemptBackoff = 6 * time.Hour + +// ProviderEgressLocation is a provider location learned by probing the +// provider's own egress, rather than by looking up its control-connection ip. +type ProviderEgressLocation struct { + ClientId server.Id + LocationId server.Id + CountryCode string + ASN int + Org string + Hosting bool + Proxy bool + Mobile bool + CityConfident bool + ObservedAt time.Time + UpdateTime time.Time +} + +// SetProviderEgressLocation upserts the probed location for a provider. The +// upsert is monotonic in observed_at: a replayed or out-of-order submission +// older than what is already stored is silently dropped rather than +// clobbering a newer probe result. +func SetProviderEgressLocation(ctx context.Context, e *ProviderEgressLocation) { + // country codes are stored/compared lowercased (see CreateLocation in + // network_client_location_model.go); the geolocation APIs that feed this + // return uppercase codes (e.g. "US"), so normalize before writing. + countryCode := strings.ToLower(e.CountryCode) + + server.Tx(ctx, func(tx server.PgTx) { + server.RaisePgResult(tx.Exec( + ctx, + ` + INSERT INTO provider_egress_location ( + client_id, + location_id, + country_code, + asn, + org, + hosting, + proxy, + mobile, + city_confident, + observed_at, + update_time + ) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11) + ON CONFLICT (client_id) DO UPDATE + SET + location_id = $2, + country_code = $3, + asn = $4, + org = $5, + hosting = $6, + proxy = $7, + mobile = $8, + city_confident = $9, + observed_at = $10, + update_time = $11 + WHERE provider_egress_location.observed_at < EXCLUDED.observed_at + `, + e.ClientId, + e.LocationId, + countryCode, + e.ASN, + e.Org, + e.Hosting, + e.Proxy, + e.Mobile, + e.CityConfident, + e.ObservedAt.UTC(), + server.NowUtc(), + )) + }) +} + +// ProviderEgressProbeAttempt records that the prober tried a provider, whether +// or not the try produced a location. +// +// A provider that has never been probed successfully has no +// ProviderEgressLocation row at all, so an attempt cannot be recorded there -- +// see the provider_egress_probe_attempt migration for why that matters. +// ProbeFailure is "" for a successful attempt, otherwise a short failure class +// (`tunnel_failed`, `no_consensus`, ...). +type ProviderEgressProbeAttempt struct { + ClientId server.Id + AttemptAt time.Time + ProbeFailure string + UpdateTime time.Time +} + +// SetProviderEgressProbeAttempt upserts the last probe attempt for a provider. +// +// Like SetProviderEgressLocation the upsert is monotonic in its timestamp: a +// replayed or out-of-order report older than what is already stored is dropped +// rather than moving the provider's last-attempt time backwards, which would +// hand it back to the prober early. +func SetProviderEgressProbeAttempt(ctx context.Context, a *ProviderEgressProbeAttempt) { + server.Tx(ctx, func(tx server.PgTx) { + server.RaisePgResult(tx.Exec( + ctx, + ` + INSERT INTO provider_egress_probe_attempt ( + client_id, + attempt_at, + probe_failure, + update_time + ) + VALUES ($1, $2, $3, $4) + ON CONFLICT (client_id) DO UPDATE + SET + attempt_at = $2, + probe_failure = $3, + update_time = $4 + WHERE provider_egress_probe_attempt.attempt_at < EXCLUDED.attempt_at + `, + a.ClientId, + a.AttemptAt.UTC(), + a.ProbeFailure, + server.NowUtc(), + )) + }) +} + +// GetProviderEgressProbeAttempt returns the last recorded probe attempt for a +// provider, or nil. +func GetProviderEgressProbeAttempt(ctx context.Context, clientId server.Id) *ProviderEgressProbeAttempt { + var a *ProviderEgressProbeAttempt + server.Db(ctx, func(conn server.PgConn) { + result, err := conn.Query( + ctx, + ` + SELECT + client_id, + attempt_at, + probe_failure, + update_time + FROM provider_egress_probe_attempt + WHERE client_id = $1 + `, + clientId, + ) + server.WithPgResult(result, err, func() { + if result.Next() { + a = &ProviderEgressProbeAttempt{} + server.Raise(result.Scan( + &a.ClientId, + &a.AttemptAt, + &a.ProbeFailure, + &a.UpdateTime, + )) + } + }) + }) + return a +} + +// GetProviderEgressLocation returns the stored location for a provider, or nil. +func GetProviderEgressLocation(ctx context.Context, clientId server.Id) *ProviderEgressLocation { + var e *ProviderEgressLocation + server.Db(ctx, func(conn server.PgConn) { + result, err := conn.Query( + ctx, + ` + SELECT + client_id, + location_id, + country_code, + asn, + org, + hosting, + proxy, + mobile, + city_confident, + observed_at, + update_time + FROM provider_egress_location + WHERE client_id = $1 + `, + clientId, + ) + server.WithPgResult(result, err, func() { + if result.Next() { + e = &ProviderEgressLocation{} + server.Raise(result.Scan( + &e.ClientId, + &e.LocationId, + &e.CountryCode, + &e.ASN, + &e.Org, + &e.Hosting, + &e.Proxy, + &e.Mobile, + &e.CityConfident, + &e.ObservedAt, + &e.UpdateTime, + )) + } + }) + }) + return e +} + +// GetFreshProviderEgressLocation is GetProviderEgressLocation, filtered to +// entries probed within maxAge. The cutoff is computed in Go and bound as a +// parameter: observed_at is a naive timestamp holding utc, and comparing it +// against sql now() would cast through the session timezone. +func GetFreshProviderEgressLocation( + ctx context.Context, + clientId server.Id, + maxAge time.Duration, +) *ProviderEgressLocation { + e := GetProviderEgressLocation(ctx, clientId) + if e == nil { + return nil + } + if e.ObservedAt.Before(server.NowUtc().Add(-maxAge)) { + return nil + } + return e +} + +// GetFreshProviderEgressLocationForConnection resolves the probed provider +// egress location for a connection in a single query, joining +// network_client_connection to provider_egress_location on client_id. This +// exists for the connect-announce hot path (SetConnectionLocation), which +// previously spent two round trips per connection -- resolving the client id +// for the connection, then fetching its fresh egress location -- before ever +// reaching the mmdb fallback; collapsing to one query matters on a path that +// runs for every connection and inside a retry loop. +// +// As with GetFreshProviderEgressLocation, the maxAge cutoff is computed in Go +// and compared in Go: observed_at is a naive timestamp holding utc, and +// comparing it against sql now() would cast through the session timezone. +func GetFreshProviderEgressLocationForConnection( + ctx context.Context, + connectionId server.Id, + maxAge time.Duration, +) *ProviderEgressLocation { + var e *ProviderEgressLocation + server.Db(ctx, func(conn server.PgConn) { + result, err := conn.Query( + ctx, + ` + SELECT + pel.client_id, + pel.location_id, + pel.country_code, + pel.asn, + pel.org, + pel.hosting, + pel.proxy, + pel.mobile, + pel.city_confident, + pel.observed_at, + pel.update_time + FROM network_client_connection ncc + INNER JOIN provider_egress_location pel ON pel.client_id = ncc.client_id + WHERE ncc.connection_id = $1 + `, + connectionId, + ) + server.WithPgResult(result, err, func() { + if result.Next() { + e = &ProviderEgressLocation{} + server.Raise(result.Scan( + &e.ClientId, + &e.LocationId, + &e.CountryCode, + &e.ASN, + &e.Org, + &e.Hosting, + &e.Proxy, + &e.Mobile, + &e.CityConfident, + &e.ObservedAt, + &e.UpdateTime, + )) + } + }) + }) + if e == nil { + return nil + } + if e.ObservedAt.Before(server.NowUtc().Add(-maxAge)) { + return nil + } + return e +} + +// GetLocation returns the canonical location row, or nil. +// +// Note: the location table also has a location_name column, but it holds the +// name for whichever granularity that specific row represents (e.g. a city +// row's own name), not a single name field on the Location struct. Location +// instead splits City/Region/Country by joining sibling rows (see +// IndexSearchLocationsInTx in network_client_location_model.go). This helper +// only needs to resolve identity/type, so it selects the columns that map +// directly onto Location's fields and leaves City/Region/Country empty. +func GetLocation(ctx context.Context, locationId server.Id) *Location { + var loc *Location + server.Db(ctx, func(conn server.PgConn) { + result, err := conn.Query( + ctx, + ` + SELECT location_id, location_type, city_location_id, region_location_id, country_location_id, country_code + FROM location + WHERE location_id = $1 + `, + locationId, + ) + server.WithPgResult(result, err, func() { + if result.Next() { + loc = &Location{} + // city_location_id/region_location_id are only set once the + // row's hierarchy reaches that granularity (e.g. a country + // row has both NULL); server.Id.Scan errors on a nil source, + // so scan through nullable pointers as in + // IndexSearchLocationsInTx (network_client_location_model.go). + var cityLocationId *server.Id + var regionLocationId *server.Id + var countryLocationId *server.Id + server.Raise(result.Scan( + &loc.LocationId, + &loc.LocationType, + &cityLocationId, + ®ionLocationId, + &countryLocationId, + &loc.CountryCode, + )) + if cityLocationId != nil { + loc.CityLocationId = *cityLocationId + } + if regionLocationId != nil { + loc.RegionLocationId = *regionLocationId + } + if countryLocationId != nil { + loc.CountryLocationId = *countryLocationId + } + } + }) + }) + return loc +} + +// GetProviderEgressLocationDue returns the client ids of providers whose +// egress location is due for a probe: no fresh success (newest probe older than +// minObservedAt, or never probed) *and* no recent attempt (last attempt older +// than minAttemptAt, or never attempted). Oldest first, so the +// longest-unprobed are handed out first, capped at limit. +// +// This is the durable replacement for the prober's in-memory ttl cache: the +// schedule lives in the database, so a prober restart resumes where it left +// off instead of re-probing everything. +// +// Three things about the shape of this query matter. +// +// First, candidates are sourced from the live provider population +// (network_client_location_reliability, connected + valid) and the egress row +// is LEFT JOINed on. The dominant case by far is a provider that has *never* +// been probed and therefore has no provider_egress_location row at all; +// selecting from provider_egress_location would return exactly the providers +// that least need probing and none of the ones that most do. +// +// Second, only providers holding a Public provide key are returned. Probing +// tunnels through the provider itself, which means opening a contract from +// outside the provider's own network -- something a provider without a Public +// key refuses. Offering one to the prober would burn a probe slot on a +// guaranteed failure. This is the same filter UpdateClientLocations and +// UpdateClientScores apply (network_client_location_model.go). +// +// Third, a recent *attempt* defers a provider the same way a recent success +// does. Without that, a provider that connects and holds a Public provide key +// but always fails to probe -- for any reason other than the missing Public key +// screened for above -- never gets an egress row, so its observed_at stays +// NULL, so it sorts ahead of every stale-but-refreshable provider on every +// single poll, forever. Enough such providers to fill a batch and no healthy +// provider is ever refreshed again, while this endpoint goes on returning a +// full, plausible-looking batch. The in-memory ttl cache this replaced was +// incidentally immune, because it marked a provider probed whether or not the +// probe worked; moving the schedule server-side dropped that protection, and +// provider_egress_probe_attempt is what restores it. +// +// Both cutoffs are computed by the caller in Go and bound as parameters: +// observed_at and attempt_at are naive `timestamp` columns holding utc, and +// comparing them against sql now() would cast through the session timezone and +// silently skip a window. +func GetProviderEgressLocationDue( + ctx context.Context, + minObservedAt time.Time, + minAttemptAt time.Time, + limit int, +) []server.Id { + clientIds := []server.Id{} + server.Db(ctx, func(conn server.PgConn) { + result, err := conn.Query( + ctx, + ` + SELECT + network_client_location_reliability.client_id + FROM network_client_location_reliability + + LEFT JOIN provider_egress_location ON + provider_egress_location.client_id = network_client_location_reliability.client_id + + LEFT JOIN provider_egress_probe_attempt ON + provider_egress_probe_attempt.client_id = network_client_location_reliability.client_id + + WHERE + network_client_location_reliability.connected = true AND + network_client_location_reliability.valid = true AND + EXISTS ( + SELECT 1 FROM provide_key + WHERE + provide_key.client_id = network_client_location_reliability.client_id AND + provide_key.provide_mode = $1 + ) AND + ( + provider_egress_location.observed_at IS NULL OR + provider_egress_location.observed_at < $2 + ) AND + ( + provider_egress_probe_attempt.attempt_at IS NULL OR + provider_egress_probe_attempt.attempt_at < $3 + ) + + -- never-probed sorts ahead of merely stale: a missing observed_at + -- is infinitely old. client_id breaks the tie so batch composition + -- is deterministic instead of plan-dependent -- otherwise the whole + -- never-probed population ties on NULL and which slice of it the + -- prober gets back under a limit is whatever order the executor + -- happened to produce. + ORDER BY + provider_egress_location.observed_at ASC NULLS FIRST, + network_client_location_reliability.client_id ASC + LIMIT $4 + `, + ProvideModePublic, + minObservedAt.UTC(), + minAttemptAt.UTC(), + limit, + ) + server.WithPgResult(result, err, func() { + for result.Next() { + var clientId server.Id + server.Raise(result.Scan(&clientId)) + clientIds = append(clientIds, clientId) + } + }) + }) + return clientIds +} + +// RemoveExpiredProviderEgressLocations drops entries probed before +// minObservedAt. +func RemoveExpiredProviderEgressLocations(ctx context.Context, minObservedAt time.Time) { + server.MaintenanceTx(ctx, func(tx server.PgTx) { + server.RaisePgResult(tx.Exec( + ctx, + `DELETE FROM provider_egress_location WHERE observed_at < $1`, + minObservedAt.UTC(), + )) + }) +} + +// RemoveExpiredProviderEgressProbeAttempts drops attempts older than +// minAttemptAt. An attempt only carries information for as long as it defers +// the provider (ProviderEgressProbeAttemptBackoff); past that the row is just +// storage held for a client id that may no longer exist. +func RemoveExpiredProviderEgressProbeAttempts(ctx context.Context, minAttemptAt time.Time) { + server.MaintenanceTx(ctx, func(tx server.PgTx) { + server.RaisePgResult(tx.Exec( + ctx, + `DELETE FROM provider_egress_probe_attempt WHERE attempt_at < $1`, + minAttemptAt.UTC(), + )) + }) +} diff --git a/model/provider_egress_location_model_test.go b/model/provider_egress_location_model_test.go new file mode 100644 index 00000000..892f01b3 --- /dev/null +++ b/model/provider_egress_location_model_test.go @@ -0,0 +1,545 @@ +package model + +import ( + "context" + "slices" + "testing" + "time" + + "github.com/urnetwork/connect" + "github.com/urnetwork/server" +) + +func TestProviderEgressLocationUpsertAndGet(t *testing.T) { + server.DefaultTestEnv().Run(t, func(t testing.TB) { + ctx := context.Background() + + country := &Location{ + LocationType: LocationTypeCountry, + Country: "United States", + CountryCode: "us", + } + CreateLocation(ctx, country) + + clientId := server.NewId() + now := server.NowUtc() + SetProviderEgressLocation(ctx, &ProviderEgressLocation{ + ClientId: clientId, + LocationId: country.LocationId, + CountryCode: "us", + ASN: 401486, + Org: "RAVNIX LLC", + Hosting: true, + ObservedAt: now, + }) + + got := GetProviderEgressLocation(ctx, clientId) + if got == nil { + t.Fatal("expected a stored egress location") + } + connect.AssertEqual(t, got.LocationId, country.LocationId) + connect.AssertEqual(t, got.CountryCode, "us") + connect.AssertEqual(t, got.ASN, 401486) + connect.AssertEqual(t, got.Hosting, true) + connect.AssertEqual(t, got.Proxy, false) + + // upsert replaces, given a strictly newer observed_at: the upsert is + // monotonic (see TestProviderEgressLocationUpsertIgnoresOlderReplay below), + // so a second submission at the same observed_at would not win. + SetProviderEgressLocation(ctx, &ProviderEgressLocation{ + ClientId: clientId, + LocationId: country.LocationId, + CountryCode: "us", + ASN: 999, + Hosting: false, + Proxy: true, + ObservedAt: now.Add(time.Minute), + }) + got = GetProviderEgressLocation(ctx, clientId) + connect.AssertEqual(t, got.ASN, 999) + connect.AssertEqual(t, got.Hosting, false) + connect.AssertEqual(t, got.Proxy, true) + }) +} + +// The upsert is monotonic in observed_at: a replayed submission older than +// what is already stored must not clobber the newer row. +func TestProviderEgressLocationUpsertIgnoresOlderReplay(t *testing.T) { + server.DefaultTestEnv().Run(t, func(t testing.TB) { + ctx := context.Background() + + usCountry := &Location{ + LocationType: LocationTypeCountry, + Country: "United States", + CountryCode: "us", + } + CreateLocation(ctx, usCountry) + + jpCountry := &Location{ + LocationType: LocationTypeCountry, + Country: "Japan", + CountryCode: "jp", + } + CreateLocation(ctx, jpCountry) + + clientId := server.NewId() + newer := server.NowUtc() + older := newer.Add(-1 * time.Hour) + + // the newer probe lands first + SetProviderEgressLocation(ctx, &ProviderEgressLocation{ + ClientId: clientId, + LocationId: jpCountry.LocationId, + CountryCode: "jp", + ASN: 111, + ObservedAt: newer, + }) + + // a stale/replayed older probe arrives afterward and must not win + SetProviderEgressLocation(ctx, &ProviderEgressLocation{ + ClientId: clientId, + LocationId: usCountry.LocationId, + CountryCode: "us", + ASN: 222, + ObservedAt: older, + }) + + got := GetProviderEgressLocation(ctx, clientId) + if got == nil { + t.Fatal("expected a stored egress location") + } + connect.AssertEqual(t, got.CountryCode, "jp") + connect.AssertEqual(t, got.ASN, 111) + connect.AssertEqual(t, got.LocationId, jpCountry.LocationId) + }) +} + +func TestProviderEgressLocationCountryCodeLowercased(t *testing.T) { + server.DefaultTestEnv().Run(t, func(t testing.TB) { + ctx := context.Background() + + country := &Location{ + LocationType: LocationTypeCountry, + Country: "United States", + CountryCode: "us", + } + CreateLocation(ctx, country) + + clientId := server.NewId() + // geolocation APIs return uppercase codes (e.g. "US"); the model must + // normalize to lowercase before storing, matching CreateLocation's + // established invariant that country codes are stored/compared lowercased. + SetProviderEgressLocation(ctx, &ProviderEgressLocation{ + ClientId: clientId, + LocationId: country.LocationId, + CountryCode: "US", + ASN: 12345, + Org: "TEST ORG", + ObservedAt: server.NowUtc(), + }) + + got := GetProviderEgressLocation(ctx, clientId) + if got == nil { + t.Fatal("expected a stored egress location") + } + connect.AssertEqual(t, got.CountryCode, "us") + }) +} + +func TestProviderEgressLocationFreshness(t *testing.T) { + server.DefaultTestEnv().Run(t, func(t testing.TB) { + ctx := context.Background() + + country := &Location{ + LocationType: LocationTypeCountry, + Country: "United States", + CountryCode: "us", + } + CreateLocation(ctx, country) + + fresh := server.NewId() + SetProviderEgressLocation(ctx, &ProviderEgressLocation{ + ClientId: fresh, LocationId: country.LocationId, CountryCode: "us", + ObservedAt: server.NowUtc(), + }) + stale := server.NewId() + SetProviderEgressLocation(ctx, &ProviderEgressLocation{ + ClientId: stale, LocationId: country.LocationId, CountryCode: "us", + ObservedAt: server.NowUtc().Add(-8 * 24 * time.Hour), + }) + + if GetFreshProviderEgressLocation(ctx, fresh, ProviderEgressLocationMaxAge) == nil { + t.Fatal("fresh entry must be returned") + } + if GetFreshProviderEgressLocation(ctx, stale, ProviderEgressLocationMaxAge) != nil { + t.Fatal("stale entry must not be returned") + } + // absent + if GetFreshProviderEgressLocation(ctx, server.NewId(), ProviderEgressLocationMaxAge) != nil { + t.Fatal("absent entry must return nil") + } + }) +} + +func TestRemoveExpiredProviderEgressLocations(t *testing.T) { + server.DefaultTestEnv().Run(t, func(t testing.TB) { + ctx := context.Background() + + country := &Location{ + LocationType: LocationTypeCountry, + Country: "United States", + CountryCode: "us", + } + CreateLocation(ctx, country) + + keep := server.NewId() + drop := server.NewId() + SetProviderEgressLocation(ctx, &ProviderEgressLocation{ + ClientId: keep, LocationId: country.LocationId, CountryCode: "us", + ObservedAt: server.NowUtc(), + }) + SetProviderEgressLocation(ctx, &ProviderEgressLocation{ + ClientId: drop, LocationId: country.LocationId, CountryCode: "us", + ObservedAt: server.NowUtc().Add(-30 * 24 * time.Hour), + }) + + RemoveExpiredProviderEgressLocations(ctx, server.NowUtc().Add(-14*24*time.Hour)) + + if GetProviderEgressLocation(ctx, keep) == nil { + t.Fatal("recent entry must survive the sweep") + } + if GetProviderEgressLocation(ctx, drop) != nil { + t.Fatal("old entry must be swept") + } + }) +} + +// testing_connectProbeableProvider stands up the minimum a client needs to +// look like a live provider to the due-selection query: a device, a live +// connection with a resolved location, and a provide key of the given mode. +// The caller must run UpdateClientLocationReliabilities afterward -- that is +// what rolls the live connection tables up into the +// network_client_location_reliability row (connected + valid) the query reads. +// It returns the connection id, so a caller can disconnect the provider again. +func testing_connectProbeableProvider( + t testing.TB, + ctx context.Context, + clientId server.Id, + locationId server.Id, + clientAddress string, + provideMode ProvideMode, +) server.Id { + Testing_CreateDevice(ctx, server.NewId(), server.NewId(), clientId, "", "") + + handlerId := CreateNetworkClientHandler(ctx) + connectionId, _, _, _, err := ConnectNetworkClient(ctx, clientId, clientAddress, handlerId) + if err != nil { + t.Fatalf("connect client: %s", err) + } + + if err := SetConnectionLocation(ctx, connectionId, locationId, &ConnectionLocationScores{}); err != nil { + t.Fatalf("set connection location: %s", err) + } + + SetProvide(ctx, clientId, map[ProvideMode][]byte{ + provideMode: []byte("provide-secret"), + }) + + return connectionId +} + +// The prober asks the server what to probe next. The answer must be sourced +// from the live provider population and not from provider_egress_location, +// because the dominant case -- a provider that has never been probed at all -- +// has no row there. +func TestGetProviderEgressLocationDue(t *testing.T) { + server.DefaultTestEnv().Run(t, func(t testing.TB) { + ctx := context.Background() + now := server.NowUtc() + + city := &Location{ + LocationType: LocationTypeCity, + City: "Palo Alto", + Region: "California", + Country: "United States", + CountryCode: "us", + } + CreateLocation(ctx, city) + + fresh := server.NewId() + stale := server.NewId() + never := server.NewId() + // a provider that cannot serve a stranger is unprobeable: the tunnel + // contract would be refused, so it must never be offered to the prober + nonPublic := server.NewId() + + testing_connectProbeableProvider(t, ctx, fresh, city.LocationId, "0.0.0.1:0", ProvideModePublic) + testing_connectProbeableProvider(t, ctx, stale, city.LocationId, "0.0.0.2:0", ProvideModePublic) + testing_connectProbeableProvider(t, ctx, never, city.LocationId, "0.0.0.3:0", ProvideModePublic) + testing_connectProbeableProvider(t, ctx, nonPublic, city.LocationId, "0.0.0.4:0", ProvideModeNetwork) + + UpdateClientLocationReliabilities(ctx, now.Add(-time.Hour), now) + + SetProviderEgressLocation(ctx, &ProviderEgressLocation{ + ClientId: fresh, LocationId: city.LocationId, + CountryCode: "us", ObservedAt: now.Add(-1 * time.Hour), + }) + SetProviderEgressLocation(ctx, &ProviderEgressLocation{ + ClientId: stale, LocationId: city.LocationId, + CountryCode: "us", ObservedAt: now.Add(-72 * time.Hour), + }) + // `never` and `nonPublic` deliberately get no row at all + + // no attempt rows exist in this test, so the attempt cutoff never + // excludes anything; freshness is the only variable + due := GetProviderEgressLocationDue(ctx, now.Add(-24*time.Hour), now, 100) + + // a provider probed an hour ago must not be re-probed; one probed three + // days ago must be; one never probed must be + if slices.Contains(due, fresh) { + t.Fatalf("due = %v, must not contain the provider probed an hour ago (%s)", due, fresh) + } + if !slices.Contains(due, stale) { + t.Fatalf("due = %v, must contain the provider probed three days ago (%s)", due, stale) + } + if !slices.Contains(due, never) { + t.Fatalf("due = %v, must contain the never-probed provider (%s)", due, never) + } + // unprobeable regardless of freshness + if slices.Contains(due, nonPublic) { + t.Fatalf("due = %v, must not contain the provider without a Public provide key (%s)", due, nonPublic) + } + + // oldest first, so the longest-unprobed are probed first: the + // never-probed provider sorts ahead of the three-days-stale one + neverIndex := slices.Index(due, never) + staleIndex := slices.Index(due, stale) + if staleIndex < neverIndex { + t.Fatalf("never-probed provider at %d must sort before the stale one at %d", neverIndex, staleIndex) + } + + // limit is honoured + limited := GetProviderEgressLocationDue(ctx, now.Add(-24*time.Hour), now, 1) + if len(limited) != 1 { + t.Fatalf("len(due) = %d for limit 1, want 1", len(limited)) + } + if limited[0] != never { + t.Fatalf("due[0] = %s for limit 1, want the never-probed provider %s", limited[0], never) + } + }) +} + +// A provider that connects, holds a Public provide key and fails every probe +// never gets a provider_egress_location row, so its observed_at stays NULL, so +// it sorts ahead of every stale-but-refreshable provider -- forever, on every +// poll. Enough of them to fill a batch and no healthy provider's location is +// ever refreshed again, silently: the endpoint keeps returning a full, +// plausible-looking batch of the same dead providers. +// +// A recent attempt must therefore defer a provider exactly as a fresh success +// does. Deleting the attempt predicate from GetProviderEgressLocationDue must +// fail this test. +func TestGetProviderEgressLocationDueDefersRecentlyAttempted(t *testing.T) { + server.DefaultTestEnv().Run(t, func(t testing.TB) { + ctx := context.Background() + now := server.NowUtc() + + city := &Location{ + LocationType: LocationTypeCity, + City: "Palo Alto", + Region: "California", + Country: "United States", + CountryCode: "us", + } + CreateLocation(ctx, city) + + // never probed successfully, and the prober just tried it and failed + dead := server.NewId() + // probed successfully three days ago, never attempted since: the + // provider that actually needs the next probe slot + healthyStale := server.NewId() + + testing_connectProbeableProvider(t, ctx, dead, city.LocationId, "0.0.0.1:0", ProvideModePublic) + testing_connectProbeableProvider(t, ctx, healthyStale, city.LocationId, "0.0.0.2:0", ProvideModePublic) + + UpdateClientLocationReliabilities(ctx, now.Add(-time.Hour), now) + + SetProviderEgressLocation(ctx, &ProviderEgressLocation{ + ClientId: healthyStale, LocationId: city.LocationId, + CountryCode: "us", ObservedAt: now.Add(-72 * time.Hour), + }) + // `dead` deliberately gets no location row -- it has never succeeded -- + // only a failed attempt seconds ago + SetProviderEgressProbeAttempt(ctx, &ProviderEgressProbeAttempt{ + ClientId: dead, + AttemptAt: now.Add(-5 * time.Second), + ProbeFailure: "tunnel_failed", + }) + + minObservedAt := now.Add(-24 * time.Hour) + minAttemptAt := now.Add(-ProviderEgressProbeAttemptBackoff) + + due := GetProviderEgressLocationDue(ctx, minObservedAt, minAttemptAt, 100) + + if slices.Contains(due, dead) { + t.Fatalf("due = %v, must not contain the provider attempted seconds ago (%s)", due, dead) + } + if !slices.Contains(due, healthyStale) { + t.Fatalf("due = %v, must contain the stale-but-refreshable provider (%s)", due, healthyStale) + } + + // the starvation itself: with a batch big enough for exactly one + // provider, the slot must go to the one that can actually be refreshed, + // not to the never-probed one that just failed. Without the attempt + // predicate `dead` wins this on observed_at IS NULL every single poll. + limited := GetProviderEgressLocationDue(ctx, minObservedAt, minAttemptAt, 1) + if len(limited) != 1 { + t.Fatalf("len(due) = %d for limit 1, want 1", len(limited)) + } + if limited[0] != healthyStale { + t.Fatalf("due[0] = %s for limit 1, want the refreshable provider %s, not the just-failed one %s", limited[0], healthyStale, dead) + } + + // ... and the deferral is a backoff, not a ban: once the backoff has + // elapsed the same provider is offered again. The caller computes the + // cutoff as (wall clock - backoff), so a poll exactly one backoff period + // after the attempt computes `now`. + afterBackoff := GetProviderEgressLocationDue(ctx, minObservedAt, now, 100) + if !slices.Contains(afterBackoff, dead) { + t.Fatalf("due = %v, must contain the failed provider (%s) again once the attempt backoff has elapsed", afterBackoff, dead) + } + }) +} + +// Only live, routable providers are probeable. A provider that has gone offline +// (connected = false) or that looks messed up from a routing perspective +// (valid = false, a generated column: more than one address hash or location on +// its live connections) must not be handed to the prober. Deleting either +// predicate from GetProviderEgressLocationDue must fail this test. +func TestGetProviderEgressLocationDueRequiresConnectedAndValid(t *testing.T) { + server.DefaultTestEnv().Run(t, func(t testing.TB) { + ctx := context.Background() + now := server.NowUtc() + + city := &Location{ + LocationType: LocationTypeCity, + City: "Palo Alto", + Region: "California", + Country: "United States", + CountryCode: "us", + } + CreateLocation(ctx, city) + + good := server.NewId() + disconnected := server.NewId() + invalid := server.NewId() + + testing_connectProbeableProvider(t, ctx, good, city.LocationId, "0.0.0.1:0", ProvideModePublic) + disconnectedConnectionId := testing_connectProbeableProvider(t, ctx, disconnected, city.LocationId, "0.0.0.2:0", ProvideModePublic) + + // `invalid` holds two simultaneous connections from two different + // addresses, which makes client_address_hash_count = 2 and so the + // generated `valid` column false. The two addresses must be in + // different /29s: server.ClientIpHash buckets ipv4 to the /29 network, + // so e.g. 0.0.0.3 and 0.0.0.4 would hash the same and count as one. + testing_connectProbeableProvider(t, ctx, invalid, city.LocationId, "0.0.0.3:0", ProvideModePublic) + secondHandlerId := CreateNetworkClientHandler(ctx) + secondConnectionId, _, _, _, err := ConnectNetworkClient(ctx, invalid, "0.0.8.3:0", secondHandlerId) + if err != nil { + t.Fatalf("connect second address: %s", err) + } + if err := SetConnectionLocation(ctx, secondConnectionId, city.LocationId, &ConnectionLocationScores{}); err != nil { + t.Fatalf("set second connection location: %s", err) + } + + // first roll-up: everything above is connected + UpdateClientLocationReliabilities(ctx, now.Add(-time.Hour), now) + + // `disconnected` drops off, and a second roll-up flips its reliability + // row's connected to false (the row itself survives) + if err := DisconnectNetworkClient(ctx, disconnectedConnectionId); err != nil { + t.Fatalf("disconnect client: %s", err) + } + UpdateClientLocationReliabilities(ctx, now.Add(-time.Hour), server.NowUtc()) + + due := GetProviderEgressLocationDue(ctx, now.Add(-24*time.Hour), now, 100) + + if !slices.Contains(due, good) { + t.Fatalf("due = %v, must contain the connected, valid provider (%s)", due, good) + } + if slices.Contains(due, disconnected) { + t.Fatalf("due = %v, must not contain the disconnected provider (%s)", due, disconnected) + } + if slices.Contains(due, invalid) { + t.Fatalf("due = %v, must not contain the provider whose reliability row is not valid (%s)", due, invalid) + } + }) +} + +// The attempt upsert is monotonic in attempt_at, for the same reason the +// location upsert is: a replayed or out-of-order report must not move the last +// attempt backwards and hand the provider back to the prober early. +func TestProviderEgressProbeAttemptUpsertIgnoresOlderReplay(t *testing.T) { + server.DefaultTestEnv().Run(t, func(t testing.TB) { + ctx := context.Background() + + clientId := server.NewId() + newer := server.NowUtc() + older := newer.Add(-time.Hour) + + SetProviderEgressProbeAttempt(ctx, &ProviderEgressProbeAttempt{ + ClientId: clientId, AttemptAt: newer, ProbeFailure: "no_consensus", + }) + SetProviderEgressProbeAttempt(ctx, &ProviderEgressProbeAttempt{ + ClientId: clientId, AttemptAt: older, ProbeFailure: "tunnel_failed", + }) + + got := GetProviderEgressProbeAttempt(ctx, clientId) + if got == nil { + t.Fatal("expected a stored probe attempt") + } + connect.AssertEqual(t, got.ProbeFailure, "no_consensus") + // postgres `timestamp` keeps microseconds, Go keeps nanoseconds, so + // compare with a tolerance rather than for equality + if delta := got.AttemptAt.Sub(newer); delta < -time.Millisecond || time.Millisecond < delta { + t.Fatalf("attempt_at = %s, want the newer attempt %s", got.AttemptAt, newer) + } + + // a strictly newer report does win + newest := newer.Add(time.Minute) + SetProviderEgressProbeAttempt(ctx, &ProviderEgressProbeAttempt{ + ClientId: clientId, AttemptAt: newest, ProbeFailure: "", + }) + got = GetProviderEgressProbeAttempt(ctx, clientId) + connect.AssertEqual(t, got.ProbeFailure, "") + + // absent + if GetProviderEgressProbeAttempt(ctx, server.NewId()) != nil { + t.Fatal("absent attempt must return nil") + } + }) +} + +func TestRemoveExpiredProviderEgressProbeAttempts(t *testing.T) { + server.DefaultTestEnv().Run(t, func(t testing.TB) { + ctx := context.Background() + + keep := server.NewId() + drop := server.NewId() + SetProviderEgressProbeAttempt(ctx, &ProviderEgressProbeAttempt{ + ClientId: keep, AttemptAt: server.NowUtc(), + }) + SetProviderEgressProbeAttempt(ctx, &ProviderEgressProbeAttempt{ + ClientId: drop, AttemptAt: server.NowUtc().Add(-30 * 24 * time.Hour), + }) + + RemoveExpiredProviderEgressProbeAttempts(ctx, server.NowUtc().Add(-24*time.Hour)) + + if GetProviderEgressProbeAttempt(ctx, keep) == nil { + t.Fatal("recent attempt must survive the sweep") + } + if GetProviderEgressProbeAttempt(ctx, drop) != nil { + t.Fatal("old attempt must be swept") + } + }) +} diff --git a/taskworker/taskworker.go b/taskworker/taskworker.go index 3364dce0..9cbc1c05 100644 --- a/taskworker/taskworker.go +++ b/taskworker/taskworker.go @@ -52,6 +52,7 @@ func InitTasks(ctx context.Context) { work.ScheduleRemoveExpiredAuthAttempts(clientSession, tx) work.ScheduleRemoveExpiredWalletAuthChallenges(clientSession, tx) work.ScheduleRemoveExpiredWalletNonces(clientSession, tx) + work.ScheduleRemoveExpiredProviderEgressLocations(clientSession, tx) work.ScheduleRemoveOldAuditNetworkEvents(clientSession, tx) work.ScheduleRemoveOldAuditEvents(clientSession, tx) work.ScheduleRemoveOldClientReliabilityStats(clientSession, tx) @@ -236,6 +237,10 @@ func InitTaskWorkerWithSettings(ctx context.Context, settings *task.TaskWorkerSe work.RemoveExpiredWalletNoncesPost, "github.com/urnetwork/server/taskworker/work.RemoveExpiredWalletNonces", ), + task.NewTaskTargetWithPost( + work.RemoveExpiredProviderEgressLocations, + work.RemoveExpiredProviderEgressLocationsPost, + ), task.NewTaskTargetWithPost( work.RemoveOldAuditNetworkEvents, work.RemoveOldAuditNetworkEventsPost, diff --git a/taskworker/work/provider_egress_location_work.go b/taskworker/work/provider_egress_location_work.go new file mode 100644 index 00000000..533d9609 --- /dev/null +++ b/taskworker/work/provider_egress_location_work.go @@ -0,0 +1,55 @@ +package work + +import ( + "time" + + "github.com/urnetwork/server" + "github.com/urnetwork/server/model" + "github.com/urnetwork/server/session" + "github.com/urnetwork/server/task" +) + +type RemoveExpiredProviderEgressLocationsArgs struct{} + +type RemoveExpiredProviderEgressLocationsResult struct{} + +func ScheduleRemoveExpiredProviderEgressLocations(clientSession *session.ClientSession, tx server.PgTx) { + task.ScheduleTaskInTx( + tx, + RemoveExpiredProviderEgressLocations, + &RemoveExpiredProviderEgressLocationsArgs{}, + clientSession, + task.RunOnce("remove_expired_provider_egress_locations"), + task.RunAt(server.NowUtc().Add(6*time.Hour)), + ) +} + +// RemoveExpiredProviderEgressLocations drops probed locations well past their +// trust window, so a provider that stops being probed eventually falls back to +// the mmdb path instead of being pinned to a stale location forever. The cutoff +// is deliberately looser than ProviderEgressLocationMaxAge: reads already +// ignore stale rows, so this is only reclaiming storage. +func RemoveExpiredProviderEgressLocations( + _ *RemoveExpiredProviderEgressLocationsArgs, + clientSession *session.ClientSession, +) (*RemoveExpiredProviderEgressLocationsResult, error) { + now := server.NowUtc() + minObservedAt := now.Add(-4 * model.ProviderEgressLocationMaxAge) + model.RemoveExpiredProviderEgressLocations(clientSession.Ctx, minObservedAt) + // probe attempts stop meaning anything once they no longer defer the + // provider; same reasoning as above, a looser multiple of the window that + // actually matters. + minAttemptAt := now.Add(-4 * model.ProviderEgressProbeAttemptBackoff) + model.RemoveExpiredProviderEgressProbeAttempts(clientSession.Ctx, minAttemptAt) + return &RemoveExpiredProviderEgressLocationsResult{}, nil +} + +func RemoveExpiredProviderEgressLocationsPost( + _ *RemoveExpiredProviderEgressLocationsArgs, + _ *RemoveExpiredProviderEgressLocationsResult, + clientSession *session.ClientSession, + tx server.PgTx, +) error { + ScheduleRemoveExpiredProviderEgressLocations(clientSession, tx) + return nil +}