diff --git a/TODO.md b/TODO.md index da0ced6..27f0147 100644 --- a/TODO.md +++ b/TODO.md @@ -31,6 +31,21 @@ P2: security: per-IP rate limiting on the image routes # Completed Steps +- 2026-10-04 an integration test of the image proxy flow (closes #80): + `TestImageProxyFlow` in `internal/server` starts the database, handlers and + middleware from the constructors `pixad` uses, with a fresh state directory, + and replaces only the upstream origin with a local test server. For a resize + with a change to JPEG and for `orig`, the first request goes through the + router, the real fetcher, libvips, the disk cache and SQLite and answers 200 + with the right content type and size and `X-Pixa-Cache: MISS`; the second + answers `HIT` with the same image and the upstream has had one request; the + source and the converted image are then in `cache/sources` and + `cache/variants`, with their rows in `source_content`, `source_metadata` and + `variant_content`. Two optional fields make this possible, which `pixad` does + not set and the config file and environment cannot: + `httpfetcher.Config.DialContext` connects in place of the dialer that refuses + internal addresses, the URL and redirect checks still running, and + `handlers.Params.Fetcher` replaces the fetcher the handlers build. - 2026-10-04 a URL made on the generator page with a `ttl` is tested to expire (closes #199): a new test in `internal/handlers` makes a URL on the generator page with a `ttl` of one second, checks that `/v1/e/` serves it at @@ -591,5 +606,4 @@ P2: security: per-IP rate limiting on the image routes - optional Sentry error reporting - comprehensive request logging - Prometheus performance metrics - - integration tests for the image proxy flow - load tests to verify the 1k to 5k req/s target diff --git a/internal/handlers/fetcher_internal_test.go b/internal/handlers/fetcher_internal_test.go new file mode 100644 index 0000000..d78091a --- /dev/null +++ b/internal/handlers/fetcher_internal_test.go @@ -0,0 +1,61 @@ +package handlers + +import ( + "net/http" + "net/netip" + "path/filepath" + "testing" + "time" + + "github.com/go-chi/chi/v5" + "go.uber.org/fx" + "go.uber.org/fx/fxtest" + + "sneak.berlin/go/pixa/internal/config" + "sneak.berlin/go/pixa/internal/database" + "sneak.berlin/go/pixa/internal/globals" + "sneak.berlin/go/pixa/internal/healthcheck" + "sneak.berlin/go/pixa/internal/logger" +) + +// TestHandlersBuildTheirOwnFetcherWhenNoneIsProvided builds the handlers as +// pixad does, in an fx app that provides no fetcher, and requests an image +// from 192.0.2.10, which is on the allowlist and in blocked_networks. The URL +// check accepts that address; only the dialer that refuses internal +// addresses checks blocked_networks, so the answer is 403 only if the +// fetcher the handlers build from the config connects with that dialer. Any +// other dialer would try to connect until the upstream fetch timeout, which +// is short so that the test then fails quickly. +func TestHandlersBuildTheirOwnFetcherWhenNoneIsProvided(t *testing.T) { + t.Parallel() + + const host = "192.0.2.10" + + stateDir := t.TempDir() + cfg := &config.Config{ + SigningKey: testSigningKey, + StateDir: stateDir, + DBURL: "file:" + filepath.Join(stateDir, "state.sqlite3"), + AllowlistHosts: []string{host}, + BlockedNetworks: []netip.Prefix{netip.MustParsePrefix("192.0.2.0/24")}, + UpstreamFetchTimeout: 2 * time.Second, + // With no connection slots, the fetch would fail before dialing. + UpstreamConnections: config.DefaultUpstreamConnections, + } + + var h *Handlers + + app := fxtest.New(t, + fx.Supply(cfg), + fx.Provide(globals.New, logger.New, database.New, healthcheck.New, New), + fx.Populate(&h), + ) + app.RequireStart() + t.Cleanup(app.RequireStop) + + r := chi.NewRouter() + r.Get("/v1/image/*", h.HandleImage()) + + rec := sendGet(t, r, photoURL(host)) + checkErrorBody(t, rec, http.StatusForbidden, "forbidden") +} diff --git a/internal/handlers/handlers.go b/internal/handlers/handlers.go index a65f3b5..0414026 100644 --- a/internal/handlers/handlers.go +++ b/internal/handlers/handlers.go @@ -28,6 +28,11 @@ type Params struct { Healthcheck *healthcheck.Healthcheck Database *database.Database Config *config.Config + + // Fetcher, when provided, fetches upstream images in place of the + // fetcher the handlers build from the config. Only tests provide one; + // pixad does not. + Fetcher httpfetcher.Fetcher `optional:"true"` } // Handlers provides HTTP request handlers. @@ -36,6 +41,7 @@ type Handlers struct { hc *healthcheck.Healthcheck db *database.Database config *config.Config + fetcher httpfetcher.Fetcher imgSvc *imgcache.Service imgCache *imgcache.Cache sessMgr *session.Manager @@ -59,6 +65,7 @@ func New(lc fx.Lifecycle, params Params) (*Handlers, error) { hc: params.Healthcheck, db: params.Database, config: params.Config, + fetcher: params.Fetcher, csrfProtect: csrfProtect, refererBlocklist: allowlist.New(params.Config.RefererBlocklist), } @@ -128,10 +135,12 @@ func (s *Handlers) initImageService() error { fetcherCfg.MaxConnections = s.config.UpstreamConnections fetcherCfg.BlockedNetworks = s.config.BlockedNetworks - // Create the service + // Create the service. With no fetcher provided, it builds its own from + // fetcherCfg. svc, err := imgcache.NewService(&imgcache.ServiceConfig{ Cache: cache, FetcherConfig: fetcherCfg, + Fetcher: s.fetcher, SigningKey: s.config.SigningKey, Allowlist: s.config.AllowlistHosts, MaxConcurrentProcessing: s.config.MaxConcurrentProcessing, diff --git a/internal/httpfetcher/dial_context_internal_test.go b/internal/httpfetcher/dial_context_internal_test.go new file mode 100644 index 0000000..4553799 --- /dev/null +++ b/internal/httpfetcher/dial_context_internal_test.go @@ -0,0 +1,66 @@ +package httpfetcher + +import ( + "errors" + "net" + "testing" +) + +// TestNewUsesCheckedDialerWithoutDialContext checks that a fetcher built +// without DialContext, as pixa builds it, refuses to connect to a local +// server. +func TestNewUsesCheckedDialerWithoutDialContext(t *testing.T) { + t.Parallel() + + srv := startUpstream(t) + transport := transportOf(t, New(DefaultConfig())) + + addr := srv.Listener.Addr().String() + + _, err := transport.DialContext(testContext(t), "tcp", addr) + if !errors.Is(err, ErrSSRFBlocked) { + t.Fatalf("DialContext(%s) error = %v, want ErrSSRFBlocked", addr, err) + } +} + +// TestDialContextReplacesOnlyTheDialer checks that a fetcher built with +// DialContext connects through it, while the URL check still refuses a +// loopback URL and the redirect check a redirect to a link-local address. +func TestDialContextReplacesOnlyTheDialer(t *testing.T) { + t.Parallel() + + srv := startUpstream(t) + dialer := &recordingDialer{target: srv.Listener.Addr().String()} + + cfg := DefaultConfig() + cfg.AllowHTTP = true + cfg.DialContext = dialer.dialContext + f := New(cfg) + + if body := fetchBody(t, f, "/image"); body != imagePayload { + t.Errorf("body = %q, want %q", body, imagePayload) + } + + _, err := f.Fetch(testContext(t), "http://127.0.0.1/image") + if !errors.Is(err, ErrSSRFBlocked) { + t.Errorf("Fetch(loopback URL) error = %v, want ErrSSRFBlocked", err) + } + + _, err = f.Fetch(testContext(t), upstreamURL("/redirect/private")) + if !errors.Is(err, ErrSSRFBlocked) { + t.Errorf("Fetch(/redirect/private) error = %v, want ErrSSRFBlocked", err) + } + + // The upstream server is reached through DialContext, and nothing else + // is asked of it. + dialed := dialer.dialedAddrs() + if len(dialed) == 0 { + t.Error("DialContext was never called") + } + + for _, addr := range dialed { + if addr != net.JoinHostPort(testPublicHost, "80") { + t.Errorf("DialContext was asked to connect to %s", addr) + } + } +} diff --git a/internal/httpfetcher/httpfetcher.go b/internal/httpfetcher/httpfetcher.go index f76e8b1..5640cf0 100644 --- a/internal/httpfetcher/httpfetcher.go +++ b/internal/httpfetcher/httpfetcher.go @@ -137,6 +137,11 @@ type Config struct { // BlockedNetworks are operator-supplied CIDR ranges refused by the // dialer, in addition to the always-enforced built-in ranges. BlockedNetworks []netip.Prefix + // DialContext, when set, makes the fetcher's connections in place of + // the dialer that refuses internal addresses; the URL and redirect + // checks still run. Only tests set it, to reach a local server; the + // config file and the environment cannot. + DialContext func(ctx context.Context, network, addr string) (net.Conn, error) } // DefaultConfig returns a Config with sensible defaults. @@ -190,13 +195,19 @@ func New(config *Config) *HTTPFetcher { config = DefaultConfig() } - // Create transport with SSRF-safe dialer. The dialer re-resolves and - // re-checks at connect time (closing the DNS-rebinding window) against - // both the built-in ranges and the operator-supplied blocklist. - transport := &http.Transport{ - DialContext: func(ctx context.Context, network, addr string) (net.Conn, error) { + // Unless config.DialContext replaces it, the transport connects with + // the SSRF-safe dialer, which re-resolves and re-checks at connect time + // (closing the DNS-rebinding window) against both the built-in ranges + // and the operator-supplied blocklist. + dialContext := config.DialContext + if dialContext == nil { + dialContext = func(ctx context.Context, network, addr string) (net.Conn, error) { return dialSSRFSafe(ctx, network, addr, config.BlockedNetworks) - }, + } + } + + transport := &http.Transport{ + DialContext: dialContext, TLSHandshakeTimeout: DefaultTLSTimeout, MaxIdleConns: DefaultMaxIdleConns, IdleConnTimeout: DefaultIdleConnTimeout, diff --git a/internal/imgcache/service.go b/internal/imgcache/service.go index 562595c..2a40c8c 100644 --- a/internal/imgcache/service.go +++ b/internal/imgcache/service.go @@ -42,7 +42,8 @@ type Service struct { type ServiceConfig struct { // Cache is the cache instance Cache *Cache - // FetcherConfig configures the upstream fetcher (ignored if Fetcher is set) + // FetcherConfig configures the upstream fetcher built when Fetcher is + // not set. Its AllowHTTP and MaxResponseSize are used either way. FetcherConfig *httpfetcher.Config // Fetcher is an optional custom fetcher (for testing) Fetcher httpfetcher.Fetcher diff --git a/internal/server/image_proxy_internal_test.go b/internal/server/image_proxy_internal_test.go new file mode 100644 index 0000000..24b7961 --- /dev/null +++ b/internal/server/image_proxy_internal_test.go @@ -0,0 +1,302 @@ +package server + +import ( + "bytes" + "context" + "crypto/sha256" + "database/sql" + "encoding/hex" + "image" + "image/color" + "image/jpeg" + "image/png" + "io" + "net" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "sync/atomic" + "testing" + + "go.uber.org/fx" + "go.uber.org/fx/fxtest" + + "sneak.berlin/go/pixa/internal/config" + "sneak.berlin/go/pixa/internal/database" + "sneak.berlin/go/pixa/internal/globals" + "sneak.berlin/go/pixa/internal/handlers" + "sneak.berlin/go/pixa/internal/healthcheck" + "sneak.berlin/go/pixa/internal/httpfetcher" + "sneak.berlin/go/pixa/internal/logger" + "sneak.berlin/go/pixa/internal/middleware" +) + +// upstreamHost is the upstream host of the image URLs below. It is a +// documentation address (RFC 5737), which the fetcher's URL check accepts as +// public; the fetcher's dial function connects it to the test upstream server. +const upstreamHost = "192.0.2.10" + +// TestImageProxyFlow requests images through pixa's router, handlers, +// upstream fetcher, image processor, disk cache and database, with only the +// upstream origin replaced by a local test server. The first request for a URL +// is fetched and converted; the second is served from the cache without +// another upstream request. The source and the converted image are then on +// disk, with their rows in the database. +func TestImageProxyFlow(t *testing.T) { + t.Parallel() + + source := encodeTestPNG(t, 64, 48) + + tests := []struct { + name string + sizeFormat string // the . part of the image URL + contentType string + decodeConfig func(io.Reader) (image.Config, error) + width, height int + }{ + {"resize and convert to JPEG", "32x24.jpeg", "image/jpeg", + jpeg.DecodeConfig, 32, 24}, + {"orig", "orig.orig", "image/png", png.DecodeConfig, 64, 48}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + var upstreamRequests atomic.Int32 + + upstream := httptest.NewServer(http.HandlerFunc( + func(w http.ResponseWriter, _ *http.Request) { + upstreamRequests.Add(1) + w.Header().Set("Content-Type", "image/png") + _, _ = w.Write(source) + })) + t.Cleanup(upstream.Close) + + s, db, stateDir := startImageProxy(t, upstream) + target := "/v1/image/" + upstreamHost + "/photo.png/" + tt.sizeFormat + + first := getImage(t, s, target) + + if got := first.Header().Get("X-Pixa-Cache"); got != "MISS" { + t.Errorf("first X-Pixa-Cache = %q, want MISS", got) + } + + if got := first.Header().Get("Content-Type"); got != tt.contentType { + t.Errorf("Content-Type = %q, want %q", got, tt.contentType) + } + + decoded, err := tt.decodeConfig(bytes.NewReader(first.Body.Bytes())) + if err != nil { + t.Fatalf("decoding the image: %v", err) + } + + if decoded.Width != tt.width || decoded.Height != tt.height { + t.Errorf("image is %dx%d, want %dx%d", + decoded.Width, decoded.Height, tt.width, tt.height) + } + + second := getImage(t, s, target) + + if got := second.Header().Get("X-Pixa-Cache"); got != "HIT" { + t.Errorf("second X-Pixa-Cache = %q, want HIT", got) + } + + if !bytes.Equal(second.Body.Bytes(), first.Body.Bytes()) { + t.Error("the second response is not the image the first served") + } + + if got := upstreamRequests.Load(); got != 1 { + t.Errorf("upstream received %d requests, want 1", got) + } + + checkSourceCached(t, db, stateDir, source) + checkVariantCached(t, db, stateDir, first.Body.Bytes(), tt.contentType) + }) + } +} + +// startImageProxy starts the components pixad's fx app builds, from a config +// with a fresh state directory and upstreamHost on the allowlist, and with an +// upstream fetcher that connects every upstream address to upstream. It +// returns the server with its routes, the database and the state directory. +func startImageProxy( + t *testing.T, upstream *httptest.Server, +) (*Server, *sql.DB, string) { + t.Helper() + + stateDir := t.TempDir() + cfg := &config.Config{ + SigningKey: testSigningKey, + StateDir: stateDir, + DBURL: "file:" + filepath.Join(stateDir, "state.sqlite3"), + AllowlistHosts: []string{upstreamHost}, + // The test upstream server has no TLS. + AllowHTTP: true, + // A limit of its own, so the cache does not size itself from the + // host's free disk space. + CacheMaxBytes: 64 << 20, + CacheMaxBytesExplicit: true, + UpstreamMaxResponseSize: config.DefaultUpstreamMaxResponseSize, + DownstreamTimeout: config.DefaultDownstreamTimeout, + } + + fetcherCfg := httpfetcher.DefaultConfig() + fetcherCfg.AllowHTTP = true + fetcherCfg.DialContext = func( + ctx context.Context, network, _ string, + ) (net.Conn, error) { + var dialer net.Dialer + + return dialer.DialContext(ctx, network, upstream.Listener.Addr().String()) + } + fetcher := httpfetcher.New(fetcherCfg) + + var ( + h *handlers.Handlers + mw *middleware.Middleware + db *database.Database + ) + + app := fxtest.New(t, + fx.Supply(cfg), + fx.Provide( + globals.New, + logger.New, + database.New, + healthcheck.New, + handlers.New, + middleware.New, + func() httpfetcher.Fetcher { return fetcher }, + ), + fx.Populate(&h, &mw, &db), + ) + app.RequireStart() + t.Cleanup(app.RequireStop) + + // Requests go straight to the router, as in newTestServer; the server's + // own start hook, which listens on a port, is left out. + s := &Server{config: cfg, mw: mw, h: h} + s.SetupRoutes() + + return s, db.DB(), stateDir +} + +// getImage sends a GET for target to s and fails unless it answers 200. +func getImage(t *testing.T, s *Server, target string) *httptest.ResponseRecorder { + t.Helper() + + rec := httptest.NewRecorder() + s.ServeHTTP(rec, httptest.NewRequestWithContext( + t.Context(), http.MethodGet, target, nil)) + t.Logf("GET %s: %d, X-Pixa-Cache %s", + target, rec.Code, rec.Header().Get("X-Pixa-Cache")) + + if rec.Code != http.StatusOK { + t.Fatalf("GET %s status = %d, want %d; body %s", + target, rec.Code, http.StatusOK, rec.Body.String()) + } + + return rec +} + +// checkSourceCached checks that source is stored under its SHA-256 in +// cache/sources, recorded in source_content, and that the source URL's row in +// source_metadata points at it. +func checkSourceCached(t *testing.T, db *sql.DB, stateDir string, source []byte) { + t.Helper() + + sum := sha256.Sum256(source) + hash := hex.EncodeToString(sum[:]) + + checkFile(t, filepath.Join(stateDir, "cache", "sources", hash[0:2], hash[2:4], hash), + source) + + var rows int + + err := db.QueryRowContext(t.Context(), + "SELECT COUNT(*) FROM source_content WHERE content_hash = ?", hash, + ).Scan(&rows) + if err != nil || rows != 1 { + t.Errorf("source_content rows for the source = %d (error %v), want 1", + rows, err) + } + + var metadataHash string + + err = db.QueryRowContext(t.Context(), + `SELECT content_hash FROM source_metadata + WHERE source_host = ? AND source_path = ?`, + upstreamHost, "/photo.png", + ).Scan(&metadataHash) + if err != nil || metadataHash != hash { + t.Errorf("source_metadata content_hash = %q (error %v), want %q", + metadataHash, err, hash) + } +} + +// checkVariantCached checks that the converted image served is recorded in +// variant_content with contentType, and stored under its cache key in +// cache/variants. +func checkVariantCached( + t *testing.T, db *sql.DB, stateDir string, served []byte, contentType string, +) { + t.Helper() + + var cacheKey, storedType string + + err := db.QueryRowContext(t.Context(), + "SELECT cache_key, content_type FROM variant_content", + ).Scan(&cacheKey, &storedType) + if err != nil { + t.Fatalf("variant_content row: %v", err) + } + + if storedType != contentType { + t.Errorf("variant_content content_type = %q, want %q", + storedType, contentType) + } + + checkFile(t, filepath.Join(stateDir, "cache", "variants", + cacheKey[0:2], cacheKey[2:4], cacheKey), served) +} + +// checkFile checks that the file at path holds want. +func checkFile(t *testing.T, path string, want []byte) { + t.Helper() + + //nolint:gosec // G304: a path under the test's state directory + got, err := os.ReadFile(path) + if err != nil { + t.Errorf("reading %s: %v", path, err) + + return + } + + if !bytes.Equal(got, want) { + t.Errorf("%s holds %d bytes that are not the %d expected", + path, len(got), len(want)) + } +} + +// encodeTestPNG returns an opaque width x height PNG of one color. +func encodeTestPNG(t *testing.T, width, height int) []byte { + t.Helper() + + img := image.NewRGBA(image.Rect(0, 0, width, height)) + for y := range height { + for x := range width { + img.Set(x, y, color.RGBA{R: 200, G: 40, B: 40, A: 255}) + } + } + + var buf bytes.Buffer + + err := png.Encode(&buf, img) + if err != nil { + t.Fatalf("encoding the test PNG: %v", err) + } + + return buf.Bytes() +}