| 1 | package edge |
| 2 | |
| 3 | import ( |
| 4 | "bufio" |
| 5 | "context" |
| 6 | "encoding/json" |
| 7 | "io" |
| 8 | "net" |
| 9 | "net/http" |
| 10 | "net/http/httptest" |
| 11 | "net/http/httputil" |
| 12 | "path/filepath" |
| 13 | "strings" |
| 14 | "sync/atomic" |
| 15 | "testing" |
| 16 | "time" |
| 17 | |
| 18 | "github.com/carlosframework/platform/internal/geo" |
| 19 | "github.com/carlosframework/platform/internal/geoip" |
| 20 | "github.com/carlosframework/platform/internal/registry" |
| 21 | ) |
| 22 | |
| 23 | var geoHeaderNames = []string{geoip.HeaderCountry, geoip.HeaderSubdivision, geoip.HeaderLat, geoip.HeaderLon, geoip.HeaderSource, geoip.HeaderHop} |
| 24 | |
| 25 | // geoEchoBackend answers with the geo headers it received, as JSON. |
| 26 | func geoEchoBackend(t *testing.T) *httptest.Server { |
| 27 | t.Helper() |
| 28 | ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { |
| 29 | got := map[string]string{} |
| 30 | for _, k := range geoHeaderNames { |
| 31 | if v := r.Header.Get(k); v != "" { |
| 32 | got[k] = v |
| 33 | } |
| 34 | } |
| 35 | w.Header().Set("Content-Type", "application/json") |
| 36 | json.NewEncoder(w).Encode(got) |
| 37 | })) |
| 38 | t.Cleanup(ts.Close) |
| 39 | return ts |
| 40 | } |
| 41 | |
| 42 | // geoEdge serves host from a geo-echo backend. |
| 43 | func geoEdge(t *testing.T, host string) *Edge { |
| 44 | t.Helper() |
| 45 | backend := geoEchoBackend(t) |
| 46 | reg := openReg(t) |
| 47 | if err := reg.Upsert(registry.Route{Host: host, Kind: "service", Addr: backend.Listener.Addr().String(), Status: "ready"}); err != nil { |
| 48 | t.Fatal(err) |
| 49 | } |
| 50 | e, err := New(reg, nil) |
| 51 | if err != nil { |
| 52 | t.Fatal(err) |
| 53 | } |
| 54 | t.Cleanup(e.probers.Stop) |
| 55 | return e |
| 56 | } |
| 57 | |
| 58 | // spyGeo makes every lookup a hit for country and counts the calls. |
| 59 | func spyGeo(e *Edge, country string) *atomic.Int64 { |
| 60 | var n atomic.Int64 |
| 61 | e.geoLookup = func(net.IP) geo.Snapshot { |
| 62 | n.Add(1) |
| 63 | return geo.Snapshot{Loaded: true, Found: true, Source: "TestDB-" + country, |
| 64 | Result: geo.Result{Country: country, Point: geo.Point{Lat: 1.234, Lon: 5.678}, HasPoint: true}} |
| 65 | } |
| 66 | return &n |
| 67 | } |
| 68 | |
| 69 | // serveGeo drives one request through ServeHTTP and returns the headers the |
| 70 | // app saw. |
| 71 | func serveGeo(t *testing.T, e *Edge, r *http.Request) (int, map[string]string) { |
| 72 | t.Helper() |
| 73 | w := httptest.NewRecorder() |
| 74 | e.ServeHTTP(w, r) |
| 75 | got := map[string]string{} |
| 76 | if w.Code == http.StatusOK { |
| 77 | if err := json.Unmarshal(w.Body.Bytes(), &got); err != nil { |
| 78 | t.Fatalf("backend body %q: %v", w.Body.String(), err) |
| 79 | } |
| 80 | } |
| 81 | return w.Code, got |
| 82 | } |
| 83 | |
| 84 | func geoReq(host, from string) *http.Request { |
| 85 | r := httptest.NewRequest("GET", "http://"+host+"/", nil) |
| 86 | r.RemoteAddr = net.JoinHostPort(from, "4444") |
| 87 | return r |
| 88 | } |
| 89 | |
| 90 | func TestDirectRequestGetsGeoHeadersFromTheRealReader(t *testing.T) { |
| 91 | e := geoEdge(t, "app.test") |
| 92 | db, err := geo.OpenMMDB("../geo/testdata/GeoIP2-City-Test.mmdb") |
| 93 | if err != nil { |
| 94 | t.Fatal(err) |
| 95 | } |
| 96 | e.geo.Swap(db) |
| 97 | code, h := serveGeo(t, e, geoReq("app.test", "81.2.69.142")) |
| 98 | if code != 200 || h[geoip.HeaderCountry] != "GB" || h[geoip.HeaderSubdivision] != "GB-ENG" || |
| 99 | h[geoip.HeaderSource] != "GeoIP2-City" || h[geoip.HeaderLat] != "51.51" || h[geoip.HeaderLon] != "-0.09" { |
| 100 | t.Fatalf("app saw %v", h) |
| 101 | } |
| 102 | if _, leaked := h[geoip.HeaderHop]; leaked { |
| 103 | t.Fatal("X-Carlos-Hop-Geo reached the app") |
| 104 | } |
| 105 | } |
| 106 | |
| 107 | func TestNoDatabaseMeansNoHeaders(t *testing.T) { |
| 108 | e := geoEdge(t, "app.test") |
| 109 | r := geoReq("app.test", "81.2.69.142") |
| 110 | r.Header.Set(geoip.HeaderCountry, "KP") |
| 111 | if _, h := serveGeo(t, e, r); len(h) != 0 { |
| 112 | t.Fatalf("app saw %v with no database loaded", h) |
| 113 | } |
| 114 | } |
| 115 | |
| 116 | // Review Focus 1. |
| 117 | func TestForgedGeoHeadersNeverReachTheApp(t *testing.T) { |
| 118 | e := geoEdge(t, "app.test") |
| 119 | spyGeo(e, "GB") |
| 120 | for name, mutate := range map[string]func(*http.Request){ |
| 121 | "plain forgery": func(r *http.Request) { |
| 122 | r.Header.Set(geoip.HeaderCountry, "KP") |
| 123 | r.Header.Set(geoip.HeaderSource, "forged") |
| 124 | r.Header.Set(geoip.HeaderHop, "hit;src=forged") |
| 125 | }, |
| 126 | "connection nomination": func(r *http.Request) { |
| 127 | r.Header.Set("Connection", "X-Carlos-Geo-Country, X-Carlos-Geo-Source") |
| 128 | }, |
| 129 | } { |
| 130 | r := geoReq("app.test", "81.2.69.142") |
| 131 | mutate(r) |
| 132 | _, h := serveGeo(t, e, r) |
| 133 | if h[geoip.HeaderCountry] != "GB" || h[geoip.HeaderSource] != "TestDB-GB" { |
| 134 | t.Errorf("%s: app saw %v, want the edge's own GB answer", name, h) |
| 135 | } |
| 136 | if _, leaked := h[geoip.HeaderHop]; leaked { |
| 137 | t.Errorf("%s: hop header reached the app", name) |
| 138 | } |
| 139 | } |
| 140 | } |
| 141 | |
| 142 | // meshGeoEdge is geoEdge plus a minted mesh key, so hop requests can be |
| 143 | // signed (homemap_test.go's hopEdge, with a geo-echo backend). |
| 144 | func meshGeoEdge(t *testing.T, host string) (*Edge, []byte) { |
| 145 | t.Helper() |
| 146 | e := geoEdge(t, host) |
| 147 | s := meshTestStore(t) |
| 148 | if err := mintMeshKey(context.Background(), s); err != nil { |
| 149 | t.Fatal(err) |
| 150 | } |
| 151 | m := newHomeMap(s, "home") |
| 152 | refreshHomeMap(t, m) |
| 153 | e.homes.Store(m) |
| 154 | key, _ := m.meshKeys() |
| 155 | return e, key |
| 156 | } |
| 157 | |
| 158 | func TestVerifiedHopAdoptsTheSendersAnswerWithoutLookingUp(t *testing.T) { |
| 159 | e, key := meshGeoEdge(t, "app.test") |
| 160 | spy := spyGeo(e, "XX") // a lookup here would show as XX |
| 161 | r := geoReq("app.test", "198.51.100.7") // the sending edge's address |
| 162 | r.Header.Set(hopHeader, "1") |
| 163 | r.Header.Set(hopAuthHeader, signHop(key, "app.test", time.Now())) |
| 164 | r.Header.Set("X-Forwarded-For", "81.2.69.142") |
| 165 | r.Header.Set(geoip.HeaderHop, "hit;src=GeoLite2-City") |
| 166 | r.Header.Set(geoip.HeaderCountry, "FR") |
| 167 | code, h := serveGeo(t, e, r) |
| 168 | if code != 200 || h[geoip.HeaderCountry] != "FR" || h[geoip.HeaderSource] != "GeoLite2-City" { |
| 169 | t.Fatalf("app saw %d %v, want the adopted FR", code, h) |
| 170 | } |
| 171 | if spy.Load() != 0 { |
| 172 | t.Fatal("the home edge looked the visitor up instead of adopting") |
| 173 | } |
| 174 | } |
| 175 | |
| 176 | // Review Focus 2, codex R2-1. |
| 177 | func TestRejectedHopIsServedWithNoGeo(t *testing.T) { |
| 178 | e, _ := meshGeoEdge(t, "app.test") |
| 179 | spy := spyGeo(e, "XX") |
| 180 | r := geoReq("app.test", "81.2.69.142") |
| 181 | r.Header.Set(hopHeader, "1") |
| 182 | r.Header.Set(hopAuthHeader, signHop([]byte("not-the-mesh-key-not-the-mesh-key"), "app.test", time.Now())) |
| 183 | r.Header.Set(geoip.HeaderHop, "hit;src=GeoLite2-City") |
| 184 | r.Header.Set(geoip.HeaderCountry, "FR") |
| 185 | code, h := serveGeo(t, e, r) |
| 186 | if code != 200 { |
| 187 | t.Fatalf("a rejected hop must still be served (as today), got %d", code) |
| 188 | } |
| 189 | if len(h) != 0 || spy.Load() != 0 { |
| 190 | t.Fatalf("rejected hop: app saw %v, lookups %d; want nothing and no lookup", h, spy.Load()) |
| 191 | } |
| 192 | } |
| 193 | |
| 194 | func TestDecideGeoDoorIsOffWithoutALookup(t *testing.T) { |
| 195 | e := geoEdge(t, "app.test") |
| 196 | spy := spyGeo(e, "GB") |
| 197 | a := e.decideGeo(geoReq("app.test", "81.2.69.142"), geoip.Captured{}, false, false, true) |
| 198 | if a.State != geoip.StateOff || spy.Load() != 0 { |
| 199 | t.Fatalf("door = %+v, lookups %d", a, spy.Load()) |
| 200 | } |
| 201 | } |
| 202 | |
| 203 | func TestDoorCallReachesTheAppWithNoGeo(t *testing.T) { |
| 204 | s := newDoorScene(t) |
| 205 | spy := spyGeo(s.e, "GB") |
| 206 | rec := s.do(t, "GET", doorOwnHost, "/", s.key, map[string]string{geoip.HeaderCountry: "KP"}, nil) |
| 207 | if rec.Code != http.StatusOK { |
| 208 | t.Fatalf("door call = %d %s", rec.Code, rec.Body.String()) |
| 209 | } |
| 210 | if spy.Load() != 0 { |
| 211 | t.Fatal("a door call was looked up: it would locate the box itself") |
| 212 | } |
| 213 | s.own.mu.Lock() |
| 214 | hdr := s.own.hdr |
| 215 | s.own.mu.Unlock() |
| 216 | for _, k := range geoHeaderNames { |
| 217 | if v := hdr.Get(k); v != "" { |
| 218 | t.Errorf("door call reached the app with %s: %q", k, v) |
| 219 | } |
| 220 | } |
| 221 | } |
| 222 | |
| 223 | func TestSetGeoForBackhaulCarriesStateAndAppsNeverSeeIt(t *testing.T) { |
| 224 | in := httptest.NewRequest("GET", "http://app.test/", nil) |
| 225 | for _, c := range []struct { |
| 226 | a geoip.Answer |
| 227 | forward bool |
| 228 | hop string |
| 229 | }{ |
| 230 | {geoip.Answer{State: geoip.StateHit, Country: "GB", Source: "S"}, true, "hit;src=S"}, |
| 231 | {geoip.Off(), true, "off"}, // a door call being backhauled |
| 232 | {geoip.Answer{State: geoip.StateHit, Country: "GB", Source: "S"}, false, ""}, |
| 233 | } { |
| 234 | pr := &httputil.ProxyRequest{In: in.WithContext(context.WithValue(in.Context(), geoResultKey{}, c.a)), Out: in.Clone(context.Background())} |
| 235 | pr.Out.Header.Set(geoip.HeaderHop, "hit;src=forged") |
| 236 | pr.Out.Header.Set(geoip.HeaderCountry, "KP") |
| 237 | setGeo(pr, c.forward) |
| 238 | if got := pr.Out.Header.Get(geoip.HeaderHop); got != c.hop { |
| 239 | t.Errorf("%+v forward=%v: hop = %q, want %q", c.a, c.forward, got, c.hop) |
| 240 | } |
| 241 | if got := pr.Out.Header.Get(geoip.HeaderCountry); got == "KP" { |
| 242 | t.Errorf("%+v: forged country survived setGeo", c.a) |
| 243 | } |
| 244 | } |
| 245 | // A request that never passed ServeHTTP (the scheduler's UnixProxy) gets nothing. |
| 246 | pr := &httputil.ProxyRequest{In: in, Out: in.Clone(context.Background())} |
| 247 | pr.Out.Header.Set(geoip.HeaderCountry, "KP") |
| 248 | setGeo(pr, false) |
| 249 | if pr.Out.Header.Get(geoip.HeaderCountry) != "" { |
| 250 | t.Fatal("no context answer must mean no geo headers") |
| 251 | } |
| 252 | } |
| 253 | |
| 254 | func TestBackhaulDeliversTheSendersAnswer(t *testing.T) { |
| 255 | s := newBackhaulScene(t) |
| 256 | s.converge(t) |
| 257 | homeSpy, senderSpy := spyGeo(s.a.edge, "XX"), spyGeo(s.b.edge, "GB") |
| 258 | _, body := s.get(t, s.b, "https://"+e2eHomedHost+"/", nil, true) |
| 259 | seen := e2eParseApp(t, body) |
| 260 | if seen.geoCountry != "GB" || seen.hopGeo != "" { |
| 261 | t.Fatalf("home app saw country=%q hopGeo=%q, want GB and no hop header", seen.geoCountry, seen.hopGeo) |
| 262 | } |
| 263 | if homeSpy.Load() != 0 || senderSpy.Load() != 1 { |
| 264 | t.Fatalf("lookups home=%d sender=%d, want 0 and 1 (ingress computes, spec §8.2)", homeSpy.Load(), senderSpy.Load()) |
| 265 | } |
| 266 | } |
| 267 | |
| 268 | // Review Focus 1: Vary on the geo header picks the right variant. |
| 269 | func TestCacheVariesOnTheEdgesCountry(t *testing.T) { |
| 270 | var hits atomic.Int64 |
| 271 | backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { |
| 272 | hits.Add(1) |
| 273 | w.Header().Set("Cache-Control", "max-age=60") |
| 274 | w.Header().Set("Vary", geoip.HeaderCountry) |
| 275 | w.Write([]byte("country=" + r.Header.Get(geoip.HeaderCountry))) |
| 276 | })) |
| 277 | t.Cleanup(backend.Close) |
| 278 | reg := openReg(t) |
| 279 | reg.Upsert(registry.Route{Host: "app.test", Kind: "instance", Addr: backend.Listener.Addr().String(), Status: "ready"}) |
| 280 | e, err := New(reg, nil) |
| 281 | if err != nil { |
| 282 | t.Fatal(err) |
| 283 | } |
| 284 | t.Cleanup(e.probers.Stop) |
| 285 | e.EnableCache(NewCache(DefaultCacheMaxEntries, DefaultCacheMaxBodyBytes)) |
| 286 | db, _ := geo.OpenMMDB("../geo/testdata/GeoIP2-City-Test.mmdb") |
| 287 | e.geo.Swap(db) |
| 288 | get := func(from string, forged string) string { |
| 289 | r := geoReq("app.test", from) |
| 290 | if forged != "" { |
| 291 | r.Header.Set(geoip.HeaderCountry, forged) |
| 292 | } |
| 293 | w := httptest.NewRecorder() |
| 294 | e.ServeHTTP(w, r) |
| 295 | return w.Body.String() |
| 296 | } |
| 297 | if b := get("81.2.69.142", ""); b != "country=GB" { |
| 298 | t.Fatalf("GB = %q", b) |
| 299 | } |
| 300 | if b := get("216.160.83.56", ""); b != "country=US" { |
| 301 | t.Fatalf("US = %q", b) |
| 302 | } |
| 303 | if b := get("81.2.69.142", "US"); b != "country=GB" { |
| 304 | t.Fatalf("a forged header picked another country's variant: %q", b) |
| 305 | } |
| 306 | if hits.Load() != 2 || cachedVariants(e, "app.test", "/") != 2 { |
| 307 | t.Fatalf("hits=%d variants=%d, want 2 and 2", hits.Load(), cachedVariants(e, "app.test", "/")) |
| 308 | } |
| 309 | } |
| 310 | |
| 311 | func TestUpgradeRequestCarriesGeo(t *testing.T) { |
| 312 | // The backend answers the upgrade with the country it was sent, in a |
| 313 | // 101 response header (ReverseProxy copies those back to the client). |
| 314 | backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { |
| 315 | hj, ok := w.(http.Hijacker) |
| 316 | if !ok { |
| 317 | http.Error(w, "not hijackable", http.StatusInternalServerError) |
| 318 | return |
| 319 | } |
| 320 | conn, buf, err := hj.Hijack() |
| 321 | if err != nil { |
| 322 | return |
| 323 | } |
| 324 | defer conn.Close() |
| 325 | buf.WriteString("HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nX-Seen-Country: " + |
| 326 | r.Header.Get(geoip.HeaderCountry) + "\r\n\r\n") |
| 327 | buf.Flush() |
| 328 | })) |
| 329 | t.Cleanup(backend.Close) |
| 330 | reg := openReg(t) |
| 331 | if err := reg.Upsert(registry.Route{Host: "app.test", Kind: "service", Addr: backend.Listener.Addr().String(), Status: "ready"}); err != nil { |
| 332 | t.Fatal(err) |
| 333 | } |
| 334 | e, err := New(reg, nil) |
| 335 | if err != nil { |
| 336 | t.Fatal(err) |
| 337 | } |
| 338 | t.Cleanup(e.probers.Stop) |
| 339 | spyGeo(e, "GB") |
| 340 | srv := httptest.NewServer(e) |
| 341 | t.Cleanup(srv.Close) |
| 342 | conn, err := net.Dial("tcp", srv.Listener.Addr().String()) |
| 343 | if err != nil { |
| 344 | t.Fatal(err) |
| 345 | } |
| 346 | defer conn.Close() |
| 347 | io.WriteString(conn, "GET / HTTP/1.1\r\nHost: app.test\r\nUpgrade: websocket\r\nConnection: Upgrade\r\n"+ |
| 348 | "X-Carlos-Geo-Country: KP\r\n\r\n") |
| 349 | resp, err := http.ReadResponse(bufio.NewReader(conn), nil) |
| 350 | if err != nil { |
| 351 | t.Fatal(err) |
| 352 | } |
| 353 | if resp.StatusCode != http.StatusSwitchingProtocols || resp.Header.Get("X-Seen-Country") != "GB" { |
| 354 | t.Fatalf("upgrade = %d, backend saw country %q, want 101 and GB", resp.StatusCode, resp.Header.Get("X-Seen-Country")) |
| 355 | } |
| 356 | } |
| 357 | |
| 358 | // Plan review R1-18 / spec §8.4: forgery, Connection nomination and an |
| 359 | // unknown X-Carlos-Geo-* suffix, through every proxy type a local route can |
| 360 | // use (unix, tcp, multi). Backhaul is covered by |
| 361 | // TestBackhaulDeliversTheSendersAnswer and the tunnel by Task 10's e2e. |
| 362 | func TestForgeryMatrixAcrossLocalProxyTypes(t *testing.T) { |
| 363 | echo := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { |
| 364 | got := map[string]string{} |
| 365 | for k := range r.Header { |
| 366 | if n := strings.ToLower(strings.ReplaceAll(k, "_", "-")); strings.HasPrefix(n, "x-carlos-geo-") || n == strings.ToLower(geoip.HeaderHop) { |
| 367 | got[k] = r.Header.Get(k) |
| 368 | } |
| 369 | } |
| 370 | w.Header().Set("Content-Type", "application/json") |
| 371 | json.NewEncoder(w).Encode(got) |
| 372 | }) |
| 373 | tcp := httptest.NewServer(echo) |
| 374 | t.Cleanup(tcp.Close) |
| 375 | routes := map[string]registry.Route{ |
| 376 | "tcp": {Host: "tcp.test", Kind: "service", Addr: tcp.Listener.Addr().String(), Status: "ready"}, |
| 377 | "multi": {Host: "multi.test", Kind: "service", Addrs: tcp.Listener.Addr().String(), Health: healthJSON(t, frozenHealth()), Status: "ready"}, |
| 378 | } |
| 379 | sock := filepath.Join(t.TempDir(), "g.sock") |
| 380 | if ln, err := net.Listen("unix", sock); err == nil { |
| 381 | us := &httptest.Server{Listener: ln, Config: &http.Server{Handler: echo}} |
| 382 | us.Start() |
| 383 | t.Cleanup(us.Close) |
| 384 | routes["unix"] = registry.Route{Host: "unix.test", Kind: "instance", Socket: sock, Status: "ready"} |
| 385 | } else { |
| 386 | t.Logf("unix listener unavailable here (%v); unix leg skipped", err) |
| 387 | } |
| 388 | reg := openReg(t) |
| 389 | for _, rt := range routes { |
| 390 | if err := reg.Upsert(rt); err != nil { |
| 391 | t.Fatal(err) |
| 392 | } |
| 393 | } |
| 394 | e, err := New(reg, nil) |
| 395 | if err != nil { |
| 396 | t.Fatal(err) |
| 397 | } |
| 398 | t.Cleanup(e.probers.Stop) |
| 399 | spyGeo(e, "GB") |
| 400 | for name, rt := range routes { |
| 401 | r := geoReq(rt.Host, "81.2.69.142") |
| 402 | r.Header.Set(geoip.HeaderCountry, "KP") |
| 403 | r.Header.Set("X-Carlos-Geo-City", "Atlantis") |
| 404 | r.Header.Set(geoip.HeaderHop, "hit;src=forged") |
| 405 | r.Header.Set("Connection", "X-Carlos-Geo-Country, X-Carlos-Geo-Source") |
| 406 | r.Header["X_Carlos_Geo_Country"] = []string{"KP"} // WSGI reads this as HTTP_X_CARLOS_GEO_COUNTRY |
| 407 | r.Header["X_Carlos_Hop_Geo"] = []string{"hit;src=forged"} |
| 408 | code, h := serveGeo(t, e, r) |
| 409 | if code != 200 { |
| 410 | t.Errorf("%s: status %d", name, code) |
| 411 | continue |
| 412 | } |
| 413 | if h[geoip.HeaderCountry] != "GB" || h[geoip.HeaderSource] != "TestDB-GB" { |
| 414 | t.Errorf("%s: app saw %v, want the edge's GB answer", name, h) |
| 415 | } |
| 416 | // Only the edge's own five, under their exact names, may arrive: |
| 417 | // any other spelling of the namespace (a city, the hop header, an |
| 418 | // underscore alias WSGI would read as the real one) is a forgery. |
| 419 | allowed := map[string]bool{geoip.HeaderCountry: true, geoip.HeaderSubdivision: true, |
| 420 | geoip.HeaderLat: true, geoip.HeaderLon: true, geoip.HeaderSource: true} |
| 421 | for k := range h { |
| 422 | if !allowed[k] { |
| 423 | t.Errorf("%s: %s reached the app", name, k) |
| 424 | } |
| 425 | } |
| 426 | } |
| 427 | } |
| 428 | |
amadan