diff --git a/api/api.go b/api/api.go index fef528b8..c3dc801d 100644 --- a/api/api.go +++ b/api/api.go @@ -51,6 +51,7 @@ 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/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..b843c05b --- /dev/null +++ b/api/handlers/provider_egress_location_handlers.go @@ -0,0 +1,93 @@ +package handlers + +import ( + "crypto/hmac" + "encoding/json" + "io" + "net/http" + "sync" + + "github.com/urnetwork/glog" + + "github.com/urnetwork/server" + "github.com/urnetwork/server/controller" +) + +// 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) + } +} 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..4af00971 --- /dev/null +++ b/api/handlers/provider_egress_location_handlers_test.go @@ -0,0 +1,150 @@ +package handlers + +import ( + "bytes" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/urnetwork/server" + "github.com/urnetwork/server/controller" +) + +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) + } +} 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..2c28c494 --- /dev/null +++ b/controller/provider_egress_location_controller.go @@ -0,0 +1,161 @@ +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 +} 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..dfc27b62 100644 --- a/db_migrations.go +++ b/db_migrations.go @@ -4385,4 +4385,33 @@ 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) + `), } 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..240bb8b4 --- /dev/null +++ b/model/provider_egress_location_model.go @@ -0,0 +1,287 @@ +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 + +// 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(), + )) + }) +} + +// 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 +} + +// 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(), + )) + }) +} diff --git a/model/provider_egress_location_model_test.go b/model/provider_egress_location_model_test.go new file mode 100644 index 00000000..f92bb2a7 --- /dev/null +++ b/model/provider_egress_location_model_test.go @@ -0,0 +1,214 @@ +package model + +import ( + "context" + "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") + } + }) +} 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..94901c34 --- /dev/null +++ b/taskworker/work/provider_egress_location_work.go @@ -0,0 +1,49 @@ +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) { + minObservedAt := server.NowUtc().Add(-4 * model.ProviderEgressLocationMaxAge) + model.RemoveExpiredProviderEgressLocations(clientSession.Ctx, minObservedAt) + return &RemoveExpiredProviderEgressLocationsResult{}, nil +} + +func RemoveExpiredProviderEgressLocationsPost( + _ *RemoveExpiredProviderEgressLocationsArgs, + _ *RemoveExpiredProviderEgressLocationsResult, + clientSession *session.ClientSession, + tx server.PgTx, +) error { + ScheduleRemoveExpiredProviderEgressLocations(clientSession, tx) + return nil +}