diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 2062153..065f51c 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -34,16 +34,11 @@ jobs: lint: runs-on: ubuntu-latest timeout-minutes: 10 - env: - GOLANGCI_LINT_VERSION: v2.11.4 steps: - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 - uses: actions/setup-go@924ae3a1cded613372ab5595356fb5720e22ba16 with: go-version-file: go.mod - - name: Install golangci-lint - run: | - curl -sSfL "https://raw.githubusercontent.com/golangci/golangci-lint/${GOLANGCI_LINT_VERSION}/install.sh" | - sh -s -- -b "$(go env GOPATH)/bin" "${GOLANGCI_LINT_VERSION}" - - name: Lint - run: golangci-lint run ./... + - uses: golangci/golangci-lint-action@ba0d7d2ec06a0ea1cb5fa41b2e4a3ab91d21278a + with: + version: v2.11.4 diff --git a/.github/workflows/image.yml b/.github/workflows/image.yml index 6f6c0af..7d75fc3 100644 --- a/.github/workflows/image.yml +++ b/.github/workflows/image.yml @@ -1,11 +1,11 @@ name: Image -# Builds and publishes the relay image to ghcr.io/perfloop/relay. A pull -# request build proves the Dockerfile still builds; a push to main publishes -# the image tagged with the commit it was built from (the short and the full -# SHA), signs the digest with keyless cosign, and attaches an SBOM. There is -# no scheduled or manual publish: one main-branch build is the sole publisher -# for a tag. +# Builds the relay image on every pull request and publishes it on every push +# to main, to ghcr.io/perfloop/relay, tagged with the commit it was built from +# (the short and the full SHA), signed with keyless cosign, with an SBOM +# attached. The two are separate jobs so a pull request build runs with +# read-only permissions: only the main-branch job can write a package or +# obtain an OIDC token. There is no scheduled or manual publish. # # The registry credential is the workflow's own GITHUB_TOKEN; the signing # identity is the workflow's OIDC token. No long-lived key is held anywhere. @@ -27,15 +27,32 @@ permissions: jobs: build: + # A pull request proves the Dockerfile still builds. Nothing is pushed. + if: github.event_name == 'pull_request' runs-on: ubuntu-latest timeout-minutes: 30 permissions: contents: read - packages: write # publish to ghcr.io on main; a pull request never pushes - id-token: write # keyless cosign on main + steps: + - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 + - uses: docker/setup-buildx-action@d7f5e7f509e45cec5c76c4d5afdd7de93d0b3df5 + - uses: docker/build-push-action@f9f3042f7e2789586610d6e8b85c8f03e5195baf + with: + context: . + file: Dockerfile + platforms: linux/amd64 + push: false + + publish: + if: github.event_name == 'push' && github.ref == 'refs/heads/main' + runs-on: ubuntu-latest + timeout-minutes: 30 + permissions: + contents: read + packages: write # ghcr.io/perfloop/relay + id-token: write # keyless cosign env: IMAGE: ghcr.io/perfloop/relay - PUBLISH: ${{ github.event_name == 'push' && github.ref == 'refs/heads/main' }} steps: - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 @@ -49,14 +66,11 @@ jobs: { echo "tags<>"$GITHUB_OUTPUT" - - if: env.PUBLISH == 'true' - uses: docker/login-action@650006c6eb7dba73a995cc03b0b2d7f5ca915bee + - uses: docker/login-action@650006c6eb7dba73a995cc03b0b2d7f5ca915bee with: registry: ghcr.io username: ${{ github.actor }} @@ -70,16 +84,14 @@ jobs: context: . file: Dockerfile platforms: linux/amd64 - push: ${{ env.PUBLISH == 'true' }} + push: true tags: ${{ steps.tags.outputs.tags }} # A published image carries a pinned Syft SPDX package inventory. - sbom: ${{ env.PUBLISH == 'true' && 'generator=docker/buildkit-syft-scanner@sha256:79e7b013cbec16bbb436f312819a49a4a57752b2270c1a9332ae1a10fcc82a68' || 'false' }} + sbom: generator=docker/buildkit-syft-scanner@sha256:79e7b013cbec16bbb436f312819a49a4a57752b2270c1a9332ae1a10fcc82a68 - - if: env.PUBLISH == 'true' - uses: sigstore/cosign-installer@6f9f17788090df1f26f669e9d70d6ae9567deba6 + - uses: sigstore/cosign-installer@6f9f17788090df1f26f669e9d70d6ae9567deba6 - - if: env.PUBLISH == 'true' - name: Sign the published digest + - name: Sign the published digest # Keyless: the signing identity is this workflow's OIDC token, so a # verifier can require exactly this repository, workflow, and branch # (docs/image.md). No signing key exists. @@ -87,8 +99,7 @@ jobs: DIGEST: ${{ steps.build.outputs.digest }} run: cosign sign --yes "${IMAGE}@${DIGEST}" - - if: env.PUBLISH == 'true' - name: Report digest + - name: Report digest env: DIGEST: ${{ steps.build.outputs.digest }} TAGS: ${{ steps.tags.outputs.tags }} diff --git a/README.md b/README.md index 8b12cb2..0cbe4ff 100644 --- a/README.md +++ b/README.md @@ -6,7 +6,7 @@ This repository is the complete source of what runs there. The image Perfloop publishes is built from it, by the workflow in this repository, and is tagged with the commit it was built from. -710 lines of Go outside tests, three runtime dependencies (`coder/websocket`, +756 lines of Go outside tests, three runtime dependencies (`coder/websocket`, `go.yaml.in/yaml/v3`, `golang.org/x/net`): read it in an afternoon. ## What it does diff --git a/docs/operations.md b/docs/operations.md index 4dfca10..b3f4a50 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -17,14 +17,15 @@ perfloop-relay -config relay.yaml -print-routes ``` loads and validates the configuration and prints, per upstream, its kind, -its URL, the names of its configured headers, and every request the relay -will forward, then exits. An invalid configuration exits non-zero with the +its URL, the request headers it passes from Perfloop, the names of its +configured headers, and every request the relay will forward, then exits. An invalid configuration exits non-zero with the validation error. Header values and the token are not printed. Example: ``` upstream vm kind victoriametrics url http://vm.internal:8428/select/0/prometheus + passes Accept, Accept-Encoding, User-Agent, X-Scope-OrgID from Perfloop; every other request header is dropped headers Authorization, X-Scope-OrgID (values not shown) forwards GET, with the query string unchanged, to: http://vm.internal:8428/select/0/prometheus/api/v1/query diff --git a/docs/security.md b/docs/security.md index 1744457..59ffea3 100644 --- a/docs/security.md +++ b/docs/security.md @@ -105,13 +105,18 @@ are not in the table and are refused before any upstream request; the tests in `admin/tsdb/snapshot`, `write`, `push`, and a `..` in the path. A forwarded request is built as: the upstream `url` plus the read's path, -the read's query string unchanged, the end-to-end headers Perfloop sent -(hop-by-hop headers are removed), then the upstream's configured `headers`, -which replace any header of the same name. The `Perfloop-Request-Id` header -is removed before the upstream request. The upstream response's status, -headers, and body are returned to Perfloop as they are. A redirect from the -upstream is not followed: the read fails with `502 upstream request failed`, -like any other upstream error. +the read's query string unchanged, and, of the request headers Perfloop +sent, exactly `Accept`, `Accept-Encoding`, `User-Agent`, and `X-Scope-OrgID` +(`link.ForwardedHeaders`; `-print-routes` prints them). Every other request +header, the `Perfloop-Request-Id` included, stops at the relay. Then the +upstream's configured `headers` are set and replace any header of the same +name: a customer who wants to pin the Loki tenant sets `X-Scope-OrgID` in +`headers:`, and the value Perfloop sent never reaches the upstream. The +upstream response's status, headers, and body are returned to Perfloop as +they are, except that hop-by-hop headers and every `Perfloop-*` header the +upstream sent are removed, so an upstream cannot speak as Perfloop. A +redirect from the upstream is not followed: the read fails with `502 +upstream request failed`, like any other upstream error. Bounds on one read: two minutes end to end, and a response body of at most 32 MiB. A larger body aborts the read so that Perfloop sees a broken read, diff --git a/examples/kubernetes/relay.yaml b/examples/kubernetes/relay.yaml index 805badb..f2609fc 100644 --- a/examples/kubernetes/relay.yaml +++ b/examples/kubernetes/relay.yaml @@ -140,6 +140,7 @@ spec: - 172.16.0.0/12 - 192.168.0.0/16 - 169.254.0.0/16 + - 100.64.0.0/10 ports: - protocol: TCP port: 443 diff --git a/link/link.go b/link/link.go index ab856f2..1cc7b60 100644 --- a/link/link.go +++ b/link/link.go @@ -15,6 +15,7 @@ package link import ( + "errors" "fmt" "net/http" "os" @@ -38,8 +39,17 @@ const ( RequestIDHeader = "Perfloop-Request-Id" ) -// hopHeaders never cross a tunnel in either direction (RFC 9110 § 7.6.1). -var hopHeaders = []string{"Connection", "Keep-Alive", "Proxy-Connection", "Transfer-Encoding", "Upgrade", "Te", "Trailer"} +// hopHeaders never cross a tunnel in either direction (RFC 9110 § 7.6.1), +// nor do proxy credentials, which belong to the hop that used them. +var hopHeaders = []string{"Connection", "Keep-Alive", "Proxy-Connection", "Proxy-Authorization", "Proxy-Authenticate", "Transfer-Encoding", "Upgrade", "Te", "Trailer"} + +// ForwardedHeaders are the request headers a relay copies from a read onto +// the upstream request. They are exactly what Perfloop sends on a read: +// Accept and X-Scope-OrgID (the Loki tenant) from its provider adapters, +// Accept-Encoding and User-Agent from its HTTP client. Every other request +// header stops at the relay; the customer's configured headers are set after +// these and replace any of the same name. +var ForwardedHeaders = []string{"Accept", "Accept-Encoding", "User-Agent", "X-Scope-OrgID"} // StripHop removes the hop-by-hop headers from a request or response header // that is about to cross a tunnel. Both ends call it on the way in and on the @@ -72,6 +82,18 @@ func ValidUpstreamName(name string) bool { // Every referenced variable must be set; the error names the missing ones. A // literal `$` cannot appear in these files; put such a value in a variable. func ExpandEnv(data []byte) ([]byte, error) { + // os.Expand drops an unterminated `${` and everything after it; a file + // with one is a mistake to report, never a shorter file to run. + for rest := string(data); ; { + i := strings.Index(rest, "${") + if i < 0 { + break + } + rest = rest[i+2:] + if !strings.Contains(rest, "}") { + return nil, errors.New("config has an unterminated ${ reference") + } + } missing := map[string]bool{} out := os.Expand(string(data), func(name string) string { value, ok := os.LookupEnv(name) diff --git a/link/link_test.go b/link/link_test.go index 89a57e6..b761b1a 100644 --- a/link/link_test.go +++ b/link/link_test.go @@ -37,4 +37,16 @@ func TestExpandEnvNamesEveryUnsetVariable(t *testing.T) { if err == nil || !strings.HasSuffix(err.Error(), "LINK_TEST_MISSING_A, LINK_TEST_MISSING_B") { t.Fatalf("unset variables: %v", err) } + // An unterminated reference is an error, never a silently shorter file. + if out, err := ExpandEnv([]byte("token: ${LINK_TEST_TOKEN\nupstreams: []\n")); err == nil || !strings.Contains(err.Error(), "unterminated") { + t.Fatalf("unterminated reference: %q, %v", out, err) + } +} + +func TestStripHopRemovesProxyCredentials(t *testing.T) { + header := http.Header{"Proxy-Authorization": {"Basic x"}, "Proxy-Authenticate": {"Basic"}, "Accept": {"*/*"}} + StripHop(header) + if len(header) != 1 || header.Get("Accept") != "*/*" { + t.Fatalf("stripped header = %v", header) + } } diff --git a/relay.go b/relay.go index 9af747a..98862e9 100644 --- a/relay.go +++ b/relay.go @@ -20,6 +20,8 @@ import ( "net/url" "os" "path" + "slices" + "strings" "sync" "time" @@ -252,18 +254,29 @@ func (r *Relay) forward(w http.ResponseWriter, req *http.Request) (int, string, if !ok { return refuse(w, http.StatusForbidden, "route is not a documented read"), "route", nil } + // One read is bounded end to end on both sides: the upstream client's + // timeout, and the same deadline on the response write, so a tunnel peer + // that withholds flow-control credit cannot hold the stream forever. + ctx, cancel := context.WithTimeout(req.Context(), readTimeout) + defer cancel() + if err := http.NewResponseController(w).SetWriteDeadline(time.Now().Add(readTimeout)); err != nil { + return refuse(w, http.StatusInternalServerError, "write deadline is not supported"), "path", err + } target := *up.base target.Path = path.Join(up.base.Path, req.URL.Path) target.RawQuery = req.URL.RawQuery - out, err := http.NewRequestWithContext(req.Context(), http.MethodGet, target.String(), http.NoBody) + out, err := http.NewRequestWithContext(ctx, http.MethodGet, target.String(), http.NoBody) if err != nil { return refuse(w, http.StatusBadRequest, "request could not be built"), "path", err } - maps.Copy(out.Header, req.Header) - link.StripHop(out.Header) - // The read id is for this log, not for the upstream: Perfloop's - // identifiers stop here. - out.Header.Del(link.RequestIDHeader) + // Only the headers Perfloop is known to send cross to the upstream; the + // read id and anything else stop here. The customer's headers come last + // and win. + for _, name := range link.ForwardedHeaders { + if values := req.Header.Values(name); len(values) > 0 { + out.Header[http.CanonicalHeaderKey(name)] = slices.Clone(values) + } + } for header, value := range up.Headers { out.Header.Set(header, value) } @@ -280,6 +293,13 @@ func (r *Relay) forward(w http.ResponseWriter, req *http.Request) (int, string, defer resp.Body.Close() maps.Copy(w.Header(), resp.Header) link.StripHop(w.Header()) + // An upstream cannot speak as Perfloop: no Perfloop-* header of its + // making reaches the tunnel, whatever the API strips on its side. + for name := range w.Header() { + if strings.HasPrefix(name, "Perfloop-") { + w.Header().Del(name) + } + } w.WriteHeader(resp.StatusCode) n, err := io.Copy(w, io.LimitReader(resp.Body, maxResponseBytes+1)) if err != nil { diff --git a/relay_headers_test.go b/relay_headers_test.go new file mode 100644 index 0000000..1057c9f --- /dev/null +++ b/relay_headers_test.go @@ -0,0 +1,154 @@ +package relay + +import ( + "context" + "io" + "net/http" + "net/http/httptest" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/perfloop/relay/link" +) + +// Only the headers Perfloop is known to send reach the upstream; a hostile +// extra header never does, and the customer's configured header wins over +// the one Perfloop sent. +func TestOnlyAllowlistedRequestHeadersReachTheUpstream(t *testing.T) { + var mu sync.Mutex + var seen http.Header + provider := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { + mu.Lock() + seen = req.Header.Clone() + mu.Unlock() + io.WriteString(w, "ok") + })) + defer provider.Close() + term := newTerminator(t, Upstream{Name: "logs", Kind: "loki", URL: provider.URL, Headers: map[string]string{"X-Scope-OrgID": "team-a"}}) + term.run(t) + term.waitTunnels(t, tunnels) + resp := term.read(t, "logs", "/loki/api/v1/labels", + "Accept", "application/json", "Accept-Encoding", "gzip", "User-Agent", "perfloop", "X-Scope-OrgID", "from-perfloop", + "X-Forwarded-Host", "evil", "Forwarded", "for=evil", "Cookie", "a=b", "Authorization", "Bearer from-perfloop", link.RequestIDHeader, "ses/call") + if resp.status != http.StatusOK { + t.Fatalf("read: %+v", resp) + } + mu.Lock() + defer mu.Unlock() + for name, want := range map[string]string{"Accept": "application/json", "Accept-Encoding": "gzip", "User-Agent": "perfloop", "X-Scope-Orgid": "team-a"} { + if got := seen.Get(name); got != want { + t.Errorf("upstream saw %s=%q, want %q", name, got, want) + } + } + for _, name := range []string{"X-Forwarded-Host", "Forwarded", "Cookie", "Authorization", link.RequestIDHeader} { + if _, ok := seen[http.CanonicalHeaderKey(name)]; ok { + t.Errorf("upstream saw %s=%q; want it dropped", name, seen.Get(name)) + } + } +} + +// An upstream's Perfloop-* response headers are stripped before the answer +// crosses the tunnel, so an upstream cannot speak as Perfloop. +func TestUpstreamCannotForgePerfloopResponseHeaders(t *testing.T) { + provider := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Perfloop-Relay-Error", "1") + w.Header().Set("Perfloop-Request-Id", "forged/id") + w.Header().Set("Perfloop-Tenant", "1") + w.Header().Set("Content-Type", "application/json") + io.WriteString(w, `{}`) + })) + defer provider.Close() + term := newTerminator(t, Upstream{Name: "vm", Kind: "prometheus", URL: provider.URL}) + term.run(t) + term.waitTunnels(t, tunnels) + resp := term.read(t, "vm", "/api/v1/query") + if resp.status != http.StatusOK || resp.header.Get("Content-Type") != "application/json" { + t.Fatalf("read: %+v", resp) + } + for name := range resp.header { + if len(name) >= 9 && name[:9] == "Perfloop-" { + t.Errorf("forged %s=%q crossed the tunnel", name, resp.header.Get(name)) + } + } +} + +// A redirect from the upstream is not followed: the read fails as an +// upstream error, and the redirect's target sees no request. +func TestUpstreamRedirectIsRefusedNotFollowed(t *testing.T) { + var followed atomic.Int64 + mux := http.NewServeMux() + mux.HandleFunc("/elsewhere/", func(w http.ResponseWriter, _ *http.Request) { + followed.Add(1) + io.WriteString(w, "moved") + }) + mux.HandleFunc("/", func(w http.ResponseWriter, req *http.Request) { + http.Redirect(w, req, "/elsewhere"+req.URL.Path, http.StatusFound) + }) + provider := httptest.NewServer(mux) + defer provider.Close() + term := newTerminator(t, Upstream{Name: "vm", Kind: "prometheus", URL: provider.URL}) + term.run(t) + term.waitTunnels(t, tunnels) + if resp := term.read(t, "vm", "/api/v1/query"); resp.status != http.StatusBadGateway || followed.Load() != 0 { + t.Fatalf("redirect: status=%d followed=%d; want 502 and no follow", resp.status, followed.Load()) + } +} + +// One tunnel carries at most eight reads at a time; the HTTP/2 settings the +// relay announces make the ninth wait, and the upstream never sees it while +// eight are in flight. +func TestOneTunnelCarriesAtMostEightReadsAtOnce(t *testing.T) { + var inFlight atomic.Int64 + release := make(chan struct{}) + provider := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { + inFlight.Add(1) + select { + case <-release: + case <-req.Context().Done(): + } + io.WriteString(w, "ok") + })) + defer provider.Close() + term := newTerminator(t, Upstream{Name: "vm", Kind: "prometheus", URL: provider.URL}) + term.run(t) + term.waitTunnels(t, tunnels) + cc := term.live()[0] + ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second) + defer cancel() + // The client refuses to open a ninth stream while eight are active, or + // waits for a slot; either way the upstream never sees a ninth read in + // flight. + var refused atomic.Int64 + var wg sync.WaitGroup + for range maxStreams + 1 { + wg.Go(func() { + req, _ := http.NewRequestWithContext(ctx, http.MethodGet, "http://vm/api/v1/query", http.NoBody) + req.Host = "vm" + resp, err := cc.RoundTrip(req) + if err != nil { + refused.Add(1) + return + } + io.Copy(io.Discard, resp.Body) + resp.Body.Close() + }) + } + deadline := time.Now().Add(5 * time.Second) + for inFlight.Load() < int64(maxStreams) && time.Now().Before(deadline) { + time.Sleep(10 * time.Millisecond) + } + time.Sleep(200 * time.Millisecond) + if n := inFlight.Load(); n != int64(maxStreams) { + t.Fatalf("upstream saw %d reads in flight, want exactly %d", n, maxStreams) + } + if active := cc.State().StreamsActive; active != maxStreams { + t.Fatalf("tunnel has %d active streams, want %d", active, maxStreams) + } + close(release) + wg.Wait() + if n := inFlight.Load(); n+refused.Load() != int64(maxStreams)+1 || refused.Load() > 1 { + t.Fatalf("upstream saw %d reads in total and %d were refused; want %d in all", n, refused.Load(), maxStreams+1) + } +} diff --git a/routes.go b/routes.go index fad0e0f..0d2a561 100644 --- a/routes.go +++ b/routes.go @@ -6,6 +6,8 @@ import ( "maps" "slices" "strings" + + "github.com/perfloop/relay/link" ) // base returns the provider API base for a supported kind. It is the one @@ -65,8 +67,9 @@ func route(kind, path string) (string, bool) { } // Routes writes what a validated configuration permits, one upstream at a -// time: the kind, the URL the relay forwards to, the names of the headers it -// adds, and every request the relay will forward. The relay forwards the +// time: the kind, the URL the relay forwards to, the request headers it passes +// from Perfloop, the names of the headers it adds, and every request the relay +// will forward. The relay forwards the // query string of a permitted route unchanged; the proxy validated it. // Nothing else is forwarded. Secrets are never written: header values and the // token stay out. The listing is written whole, in one write, so a failed @@ -83,6 +86,7 @@ func Routes(w io.Writer, cfg Config) error { fmt.Fprintf(&out, "upstream %s\n", up.Name) fmt.Fprintf(&out, " kind %s\n", up.Kind) fmt.Fprintf(&out, " url %s\n", up.URL) + fmt.Fprintf(&out, " passes %s from Perfloop; every other request header is dropped\n", strings.Join(link.ForwardedHeaders, ", ")) headers := slices.Sorted(maps.Keys(up.Headers)) if len(headers) > 0 { fmt.Fprintf(&out, " headers %s (values not shown)\n", strings.Join(headers, ", ")) diff --git a/testdata/routes.golden b/testdata/routes.golden index 6c5f2a7..21bc10b 100644 --- a/testdata/routes.golden +++ b/testdata/routes.golden @@ -1,6 +1,7 @@ upstream vm kind victoriametrics url http://vm.internal:8428/select/0/prometheus + passes Accept, Accept-Encoding, User-Agent, X-Scope-OrgID from Perfloop; every other request header is dropped headers Authorization, X-Scope-OrgID (values not shown) forwards GET, with the query string unchanged, to: http://vm.internal:8428/select/0/prometheus/api/v1/query @@ -13,6 +14,7 @@ upstream vm upstream logs kind loki url https://loki.internal + passes Accept, Accept-Encoding, User-Agent, X-Scope-OrgID from Perfloop; every other request header is dropped forwards GET, with the query string unchanged, to: https://loki.internal/loki/api/v1/query https://loki.internal/loki/api/v1/query_range