Compare commits
2
Commits
next
..
85161353ea
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
85161353ea | ||
|
|
13c06f5ad1 |
+17
-16
@@ -3,14 +3,13 @@
|
||||
# golangci/golangci-lint:v2.12.2-alpine, 2026-08-07
|
||||
FROM golangci/golangci-lint:v2.12.2-alpine@sha256:91b27804074a0bacea298707f016911e60cf0cdbc6c7bf5ccacb5f0606d18d60 AS lint
|
||||
|
||||
RUN apk add --no-cache make build-base vips-dev libheif-dev pkgconfig
|
||||
|
||||
WORKDIR /src
|
||||
|
||||
# script/bootstrap installs the build dependencies and downloads the Go
|
||||
# modules. Only script/, go.mod and go.sum are copied first, so this
|
||||
# layer is reused until one of them changes.
|
||||
COPY script/ ./script/
|
||||
# Copy go mod files first for better layer caching
|
||||
COPY go.mod go.sum ./
|
||||
RUN script/bootstrap
|
||||
RUN go mod download
|
||||
|
||||
# Copy source code
|
||||
COPY . .
|
||||
@@ -29,12 +28,20 @@ FROM golang:1.25.4-alpine@sha256:d3f0cf7723f3429e3f9ed846243970b20a2de7bae6a5b66
|
||||
# Depend on lint stage passing
|
||||
COPY --from=lint /src/go.sum /dev/null
|
||||
|
||||
ARG VERSION=dev
|
||||
|
||||
# Install build dependencies for CGO image libraries
|
||||
RUN apk add --no-cache \
|
||||
build-base \
|
||||
vips-dev \
|
||||
libheif-dev \
|
||||
pkgconfig
|
||||
|
||||
WORKDIR /src
|
||||
|
||||
# Build dependencies and Go modules, as in the lint stage
|
||||
COPY script/ ./script/
|
||||
# Copy go mod files first for better layer caching
|
||||
COPY go.mod go.sum ./
|
||||
RUN script/bootstrap
|
||||
RUN GOTOOLCHAIN=auto go mod download
|
||||
|
||||
# Copy source code
|
||||
COPY . .
|
||||
@@ -42,14 +49,8 @@ COPY . .
|
||||
# Run tests
|
||||
RUN make test
|
||||
|
||||
# VERSION is declared here, not earlier: a new value reruns only the
|
||||
# build, not script/bootstrap or the tests. CGO stays enabled for
|
||||
# govips; -trimpath keeps build paths out of the binary, and -s -w
|
||||
# leave out the symbol table and debug information.
|
||||
ARG VERSION=dev
|
||||
RUN CGO_ENABLED=1 GOTOOLCHAIN=auto go build -trimpath \
|
||||
-ldflags "-s -w -X main.Version=${VERSION}" \
|
||||
-o /pixad ./cmd/pixad
|
||||
# Build with CGO enabled
|
||||
RUN CGO_ENABLED=1 GOTOOLCHAIN=auto go build -ldflags "-X main.Version=${VERSION}" -o /pixad ./cmd/pixad
|
||||
|
||||
# Runtime stage
|
||||
# alpine:3.21, 2026-02-25
|
||||
|
||||
+7
-7
@@ -6,16 +6,16 @@
|
||||
# golangci/golangci-lint:v2.12.2-alpine, 2026-08-07
|
||||
FROM golangci/golangci-lint:v2.12.2-alpine@sha256:91b27804074a0bacea298707f016911e60cf0cdbc6c7bf5ccacb5f0606d18d60
|
||||
|
||||
# pixa is CGO/libvips: the type-aware linters compile every package, so
|
||||
# this image needs the same C libraries the build does.
|
||||
RUN apk add --no-cache build-base vips-dev libheif-dev pkgconfig
|
||||
|
||||
WORKDIR /src
|
||||
|
||||
# pixa is CGO/libvips: the type-aware linters compile every package, so
|
||||
# this image needs the same C libraries the build does. script/bootstrap
|
||||
# installs them and downloads the Go modules. Only script/, go.mod and
|
||||
# go.sum are copied first; they settle this layer's result, so it may
|
||||
# safely be reused between runs.
|
||||
COPY script/ ./script/
|
||||
# Modules first for layer caching; go.mod/go.sum settle this layer's
|
||||
# result, so it may safely be reused between runs.
|
||||
COPY go.mod go.sum ./
|
||||
RUN script/bootstrap
|
||||
RUN go mod download
|
||||
|
||||
COPY . .
|
||||
|
||||
|
||||
@@ -40,8 +40,9 @@ What the [upaas](https://git.eeqj.de/sneak/upaas) app for pixa needs:
|
||||
|
||||
- **Port:** pixa listens on container port `8080`.
|
||||
- **Volume:** container path `/var/lib/pixa`, where pixa keeps its
|
||||
database and cache. Creating the host directory when it is missing is
|
||||
upaas's job, tracked in https://git.eeqj.de/sneak/upaas/issues/235.
|
||||
database and cache. upaas bind-mounts the host path it is given and
|
||||
does not create it, so the host directory must exist before the first
|
||||
deploy.
|
||||
- **Environment variables:**
|
||||
- `PIXA_SIGNING_KEY` (required): secret for signed and encrypted URLs
|
||||
and login, 32+ characters, for example from
|
||||
@@ -57,6 +58,10 @@ What the [upaas](https://git.eeqj.de/sneak/upaas) app for pixa needs:
|
||||
`healthy`. The probe uses the port from `PORT` (default `8080`), so a
|
||||
port changed only in a mounted config file is not seen by it: change
|
||||
the port with `PORT`.
|
||||
- **First run:** create the host directory, owned by root or by uid
|
||||
`65532` and gid `65532`. The server runs as the container's `pixad`
|
||||
user, which has that uid and gid, and the container gives the
|
||||
directory to `pixad` when it starts.
|
||||
|
||||
## Rationale
|
||||
|
||||
@@ -233,28 +238,15 @@ variables set by the file's `env:` section are checked the same way.
|
||||
| `PIXA_TRUSTED_PROXIES` | `trusted_proxies` | CIDR ranges of proxies whose `X-Forwarded-For` is believed; default RFC 1918 |
|
||||
| `PIXA_ALLOW_HTTP` | `allow_http` | Allow plain-HTTP upstreams, for testing only; default `false` |
|
||||
| `PIXA_UPSTREAM_CONNECTIONS_PER_HOST` | `upstream_connections_per_host` | Concurrent connections per upstream host; default `20` |
|
||||
| `PIXA_UPSTREAM_CONNECTIONS` | `upstream_connections` | Concurrent connections to all upstream hosts together; default `64` |
|
||||
| `PIXA_MAX_CONCURRENT_PROCESSING` | `max_concurrent_processing` | Images processed at once; default the number of CPUs |
|
||||
| `PIXA_UPSTREAM_FETCH_TIMEOUT` | `upstream_fetch_timeout` | Time allowed for one fetch from an upstream host; default `30s` |
|
||||
| `PIXA_UPSTREAM_MAX_RESPONSE_SIZE` | `upstream_max_response_size` | Largest upstream response accepted, in bytes; default 50 MiB |
|
||||
| `PIXA_DOWNSTREAM_TIMEOUT` | `downstream_timeout` | Time allowed for answering one client request; default `60s` |
|
||||
| `PIXA_ACCESS_CONTROL_ALLOW_ORIGIN` | `access_control_allow_origin` | CORS origin allowed to read responses: `*` or one origin; default `*` |
|
||||
| `PIXA_METRICS_USERNAME` | `metrics.username` | Username for `/metrics`, which is served only when both are set |
|
||||
| `PIXA_METRICS_PASSWORD` | `metrics.password` | Password for `/metrics`; set together with the username |
|
||||
| `PIXA_SENTRY_DSN` | `sentry_dsn` | Sentry DSN for error reporting; empty disables it |
|
||||
| `PIXA_DEBUG` | `debug` | Debug logging and plain-HTTP local development; default `false` |
|
||||
| `PIXA_MAINTENANCE_MODE` | `maintenance_mode` | Answer image requests with 503; the health check stays 200; default `false` |
|
||||
| `PIXA_MAINTENANCE_MODE` | `maintenance_mode` | Maintenance flag reported by the health check; default `false` |
|
||||
|
||||
Key settings in more detail:
|
||||
|
||||
- `access_control_allow_origin` — the origin a browser lets read pixa's
|
||||
responses, sent as the CORS `Access-Control-Allow-Origin` header: `*`, the
|
||||
default, is any site; otherwise one `http` or `https` origin such as
|
||||
`https://example.com`, whose host is a lowercase host name (letters,
|
||||
digits, hyphens and dots, with a letter in its last part) or an IP address
|
||||
(IPv6 in brackets, in its shortest form), with an optional port 1-65535
|
||||
that has no leading zero and is not the scheme's default. Any other value,
|
||||
including another scheme such as a browser extension's, aborts startup
|
||||
- `access_control_allow_origin` — CORS origin
|
||||
- `allowlist_hosts` — list of allowed upstream hosts
|
||||
- `blocked_networks` — list of CIDR ranges to refuse for SSRF protection,
|
||||
added to the always-enforced built-in ranges (loopback, private,
|
||||
@@ -280,37 +272,13 @@ Key settings in more detail:
|
||||
the host's addresses is seen with that address. To be sure which address it
|
||||
is, set this to `[]` (or `PIXA_TRUSTED_PROXIES` to empty), send a request
|
||||
through the proxy, and read `remoteIP` in pixa's request log line for it
|
||||
- `upstream_fetch_timeout` — time allowed for one fetch from an upstream
|
||||
host, as a duration such as `30s` (the default) or `2m`
|
||||
- `upstream_max_response_size` — largest upstream response accepted, in
|
||||
bytes; default `52428800` (50 MiB). It also limits the image data pixa
|
||||
decodes
|
||||
- `downstream_timeout` — time allowed for answering one client request, as a
|
||||
duration; default `60s`. The upstream fetch counts toward it, and so do the
|
||||
waits for an upstream connection and for a processing slot (up to 10 seconds
|
||||
each), so keep it longer than `upstream_fetch_timeout` plus 20 seconds
|
||||
- `upstream_fetch_timeout` — timeout for origin requests
|
||||
- `upstream_max_response_size` — max origin response size
|
||||
- `downstream_timeout` — client response timeout
|
||||
- `signing_key` — HMAC secret for URL signatures
|
||||
- `cache_max_bytes` — disk cache size limit in bytes; `0` disables the
|
||||
disk cache entirely; omitted defaults to 75% of the free space on
|
||||
the filesystem containing `<state_dir>/cache/` (minimum 500 MiB)
|
||||
- `upstream_connections` — the most connections to upstream hosts at once, all
|
||||
hosts together, on top of `upstream_connections_per_host`; default `64`. A
|
||||
fetch holds its connection until its image has been processed. A fetch that
|
||||
finds all of them in use waits up to 10 seconds for one to free up; if none
|
||||
does, and `downstream_timeout` has not ended first, the request is answered
|
||||
503 with the error `server busy, try again later`
|
||||
- `max_concurrent_processing` — the most images decoded and encoded at once;
|
||||
default the number of CPUs pixa can use (`GOMAXPROCS`), which follows a
|
||||
container's CPU limit. A request that finds all of them in use waits up to 10
|
||||
seconds for one to free up; if none does, and `downstream_timeout` has not
|
||||
ended first, it is answered 503 the same way
|
||||
- `maintenance_mode` — while `true`, the image routes (`/v1/image/` and
|
||||
`/v1/e/`) answer every request with 503, a `Retry-After` header and a JSON
|
||||
error body. The health check (`/.well-known/healthcheck.json`) still answers
|
||||
200 and reports `"maintenance_mode": true`. It stays 200 because the image's
|
||||
Docker `HEALTHCHECK` requests it: a 503 there would make the container
|
||||
unhealthy, and upaas marks a deploy failed when its container is unhealthy.
|
||||
The login and URL generator pages and `/metrics` keep working
|
||||
|
||||
See `config.example.yml` for all options with defaults.
|
||||
|
||||
|
||||
@@ -25,54 +25,20 @@ The disk cache is now size-bounded with LRU eviction
|
||||
|
||||
# Next Step
|
||||
|
||||
P2: security: referer blacklist
|
||||
P1: rate limit global concurrent upstream fetches to prevent resource
|
||||
exhaustion
|
||||
|
||||
# Completed Steps
|
||||
|
||||
- 2026-09-29 the container makes `/var/lib/pixa` usable by itself (closes
|
||||
#159): `deploy/docker-entrypoint.sh` creates the directory if it is missing,
|
||||
gives the directory and everything in it to `pixad` when the directory or one
|
||||
of its top-level entries belongs to another user or group, sets its mode to
|
||||
`750`, then runs the server as `pixad`; data left by an earlier run under
|
||||
another uid is taken over this way; "Running under upaas" in `README.md` no
|
||||
longer tells the operator to create or chown the host directory.
|
||||
- 2026-09-29 variant content types kept in memory (closes #70):
|
||||
`Cache.metaCache` holds the content types of up to 10,000 variants in an LRU
|
||||
(`github.com/hashicorp/golang-lru/v2`), filled by `StoreVariant` and by
|
||||
`GetVariant` after it reads a `.meta` file, where a type `StoreVariant` added
|
||||
meanwhile is kept over the one read, and never with the
|
||||
`application/octet-stream` served for a variant without one; for a variant it
|
||||
holds, `GetVariant` skips the `.meta` read, still opening the variant file and
|
||||
taking the size from it; eviction removes the entry before deleting the files,
|
||||
and `GetVariant` removes it when the file will not open; the cap is a
|
||||
constant, not a setting; the unused `variantMeta` type is gone; `README.md`
|
||||
describes it.
|
||||
- 2026-09-29 maintenance mode refuses image requests (closes #71): while
|
||||
`maintenance_mode` is on, `/v1/image/` and `/v1/e/` answer 503 with a
|
||||
`Retry-After` header and the JSON error body, from one middleware in
|
||||
`internal/server/routes.go`; the health check stays 200 and reports
|
||||
`maintenance_mode`, as the image's Docker `HEALTHCHECK` requests it and upaas
|
||||
marks a deploy failed when its container is unhealthy; the login and URL
|
||||
generator pages and `/metrics` keep working; documented in `README.md` and
|
||||
`config.example.yml`.
|
||||
- 2026-09-29 bound concurrent image processing and upstream fetches (closes
|
||||
#64): `max_concurrent_processing` (default the number of CPUs pixa can use)
|
||||
limits the images decoded and encoded at once, and `upstream_connections`
|
||||
(default 64) the connections to all upstream hosts together, on top of
|
||||
`upstream_connections_per_host`; a fetch holds its connection until its image
|
||||
has been processed, and a request whose source is cached reads it only once it
|
||||
has a processing slot; a request that finds either limit reached waits up to
|
||||
10 seconds for a free one, then gets 503 `server busy, try again later`;
|
||||
libvips runs one worker thread per image with its operation cache off;
|
||||
documented in `README.md` and `config.example.yml`.
|
||||
- 2026-09-29 Dockerfiles install through `script/bootstrap` (closes #95): the
|
||||
`Dockerfile` lint and build stages and `Dockerfile.lint` copy `script/`,
|
||||
`go.mod` and `go.sum`, then run `script/bootstrap` in place of their own
|
||||
`apk add` lines, so the build dependencies are listed in one place;
|
||||
`script/bootstrap` now also installs a C compiler when `gcc` is missing; the
|
||||
build uses `-trimpath` and `-s -w` and keeps `CGO_ENABLED=1` for govips;
|
||||
`ARG VERSION` sits just above the build, so a new version reruns neither
|
||||
`script/bootstrap` nor the tests.
|
||||
`GetVariant` after it reads a `.meta` file; for a variant it holds, `Lookup`
|
||||
skips the check of the disk and `GetVariant` skips the `.meta` read, still
|
||||
opening the variant file and taking the size from it; eviction removes the
|
||||
entry before deleting the files, and `GetVariant` removes it when the file
|
||||
will not open; the cap is a constant, not a setting; the unused `variantMeta`
|
||||
type is gone; `README.md` describes it.
|
||||
- 2026-09-29 migrations at the path `REPO_POLICIES.md` sets (closes #96): the
|
||||
migration files moved, contents unchanged, from `internal/database/schema/`
|
||||
to `internal/db/migrations/` as `000_migration.sql` and `001_schema.sql`; the
|
||||
@@ -100,17 +66,6 @@ P2: security: referer blacklist
|
||||
that is sooner, never negative; an allowlisted host's URL that has an `exp`
|
||||
follows it too; `immutable` stays, as freshness now ends at the expiry;
|
||||
documented in `README.md`.
|
||||
- 2026-09-28 add the four settings `README.md` documented but pixa did not
|
||||
have, which aborted startup as unknown keys (closes #61):
|
||||
`access_control_allow_origin` (default `*`, the CORS origin),
|
||||
`upstream_fetch_timeout` (default `30s`), `upstream_max_response_size`
|
||||
(default 50 MiB) and `downstream_timeout` (default `60s`, both the
|
||||
server's write timeout and the per-request timeout); each has a
|
||||
`PIXA_` variable; durations are positive Go duration strings, the size a
|
||||
whole number of bytes up to 1 GiB, the origin `*` or one `http` or
|
||||
`https` origin as `README.md` describes it; an invalid value
|
||||
aborts startup naming the key and the value; documented in
|
||||
`config.example.yml` and `README.md`.
|
||||
- 2026-09-28 cache stats report real numbers (closes #56): `Cache.Stats`
|
||||
counts the cached source images and processed variants (`source_content`
|
||||
plus `variant_content`) and takes their size from `Cache.UsageBytes`,
|
||||
@@ -351,6 +306,7 @@ P2: security: referer blacklist
|
||||
# Future Steps
|
||||
|
||||
- P2: security
|
||||
- referer blacklist
|
||||
- per-IP rate limiting on the image routes
|
||||
- per-origin rate limiting
|
||||
- P2: HTTP response handling
|
||||
@@ -359,6 +315,7 @@ P2: security: referer blacklist
|
||||
- X-Request-ID propagation
|
||||
- P2: auto format selection (format=auto based on Accept header)
|
||||
- P2: configuration
|
||||
- add all configuration options from README
|
||||
- YAML config file support
|
||||
- P2: operational
|
||||
- optional Sentry error reporting
|
||||
|
||||
@@ -8,20 +8,10 @@
|
||||
# this file's env: section is set while the file loads, so it overrides
|
||||
# both the environment the process was started with and this file's own
|
||||
# key.
|
||||
#
|
||||
# Durations are Go duration strings such as 30s or 2m and must be
|
||||
# positive; a bare number has no unit and aborts startup. Sizes are a
|
||||
# whole number of bytes.
|
||||
|
||||
# Server settings
|
||||
port: 8080
|
||||
debug: false
|
||||
|
||||
# While true, the image routes (/v1/image/ and /v1/e/) answer every request
|
||||
# with 503 and a Retry-After header. The health check keeps answering 200 and
|
||||
# reports maintenance_mode as true. It stays 200 because the image's Docker
|
||||
# HEALTHCHECK requests it: a 503 there would make the container unhealthy, and
|
||||
# upaas marks a deploy failed when its container is unhealthy.
|
||||
maintenance_mode: false
|
||||
|
||||
# Data directory for SQLite database and cache files
|
||||
@@ -77,42 +67,6 @@ allow_http: false
|
||||
# Maximum concurrent connections per upstream host (default: 20)
|
||||
upstream_connections_per_host: 20
|
||||
|
||||
# Maximum concurrent connections to all upstream hosts together, on top of
|
||||
# the per-host limit (default: 64). A fetch holds its connection until its
|
||||
# image has been processed. A fetch that finds none free waits up to 10
|
||||
# seconds for one, and if none frees up the request is answered 503, unless
|
||||
# downstream_timeout has ended first.
|
||||
upstream_connections: 64
|
||||
|
||||
# Maximum number of images decoded and encoded at once (default: the
|
||||
# number of CPUs pixa can use, which follows a container's CPU limit). A
|
||||
# request that finds none free waits up to 10 seconds for one, and if none
|
||||
# frees up it is answered 503, unless downstream_timeout has ended first.
|
||||
# max_concurrent_processing: 4
|
||||
|
||||
# Time allowed for one fetch from an upstream host (default: 30s)
|
||||
upstream_fetch_timeout: 30s
|
||||
|
||||
# Largest upstream response accepted, in bytes, at most 1073741824
|
||||
# (1 GiB) (default: 52428800, 50 MiB)
|
||||
upstream_max_response_size: 52428800
|
||||
|
||||
# Time allowed for answering one client request (default: 60s). The
|
||||
# upstream fetch counts toward it, and so do the waits for an upstream
|
||||
# connection and for a processing slot (up to 10 seconds each), so keep it
|
||||
# longer than upstream_fetch_timeout plus 20 seconds.
|
||||
downstream_timeout: 60s
|
||||
|
||||
# The origin a browser lets read pixa's responses, sent as the CORS
|
||||
# Access-Control-Allow-Origin header: "*" (the default) is any site;
|
||||
# otherwise one http or https origin such as https://example.com, whose
|
||||
# host is a lowercase host name (letters, digits, hyphens and dots, with a
|
||||
# letter in its last part) or an IP address (IPv6 in brackets, in its
|
||||
# shortest form), with an optional port 1-65535 that has no leading zero
|
||||
# and is not the scheme's default. Any other value, including another
|
||||
# scheme such as a browser extension's, aborts startup.
|
||||
access_control_allow_origin: "*"
|
||||
|
||||
# Maximum disk cache size in bytes. Explicit values are used exactly as
|
||||
# given; 0 disables the disk cache entirely (every request fetches and
|
||||
# processes uncached). When omitted, the default is 75% of the free
|
||||
|
||||
@@ -1,22 +1,14 @@
|
||||
#!/bin/sh
|
||||
# deploy/docker-entrypoint.sh: the Docker image's ENTRYPOINT. It runs as
|
||||
# root only to make /var/lib/pixa usable by pixad: a host directory
|
||||
# bind-mounted there keeps its host owner, often root, and data from an
|
||||
# earlier run may belong to another uid. The server itself always runs
|
||||
# as pixad.
|
||||
# root only to give /var/lib/pixa to pixad: a host directory
|
||||
# bind-mounted there keeps its host owner, often root, and pixad could
|
||||
# not write to it. The server itself always runs as pixad.
|
||||
set -eu
|
||||
|
||||
main() {
|
||||
mkdir -p /var/lib/pixa
|
||||
# Only the directory and its top-level entries are checked, so a
|
||||
# normal start does not walk the cache. -depth gives each directory
|
||||
# to pixad after its contents, so a start stopped part way leaves
|
||||
# something at the top for the next start to find; -h changes a
|
||||
# symlink itself, never the file it points to.
|
||||
if [ -n "$(find /var/lib/pixa -maxdepth 1 \( ! -user pixad -o ! -group pixad \))" ]; then
|
||||
find /var/lib/pixa -depth -exec chown -h pixad:pixad {} +
|
||||
if [ "$(stat -c %U /var/lib/pixa)" != pixad ]; then
|
||||
chown pixad:pixad /var/lib/pixa
|
||||
fi
|
||||
chmod 750 /var/lib/pixa
|
||||
exec su-exec pixad /usr/local/bin/pixad "$@"
|
||||
}
|
||||
|
||||
|
||||
@@ -1,162 +0,0 @@
|
||||
package config
|
||||
|
||||
import (
|
||||
"runtime"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// The variables that set the two concurrency limits.
|
||||
const (
|
||||
testMaxConcurrentProcessingVar = "PIXA_MAX_CONCURRENT_PROCESSING"
|
||||
testUpstreamConnectionsVar = "PIXA_UPSTREAM_CONNECTIONS"
|
||||
)
|
||||
|
||||
// TestOmittedConcurrencyLimitsUseDefaults checks that an omitted
|
||||
// max_concurrent_processing is the number of CPUs Go uses and an omitted
|
||||
// upstream_connections is 64.
|
||||
func TestOmittedConcurrencyLimitsUseDefaults(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
c, err := configFromYAML(t, signingKeyLine)
|
||||
if err != nil {
|
||||
t.Fatalf("minimal config should be valid, got error: %v", err)
|
||||
}
|
||||
|
||||
if c.MaxConcurrentProcessing != runtime.GOMAXPROCS(0) {
|
||||
t.Errorf("MaxConcurrentProcessing = %d, want %d, one per CPU",
|
||||
c.MaxConcurrentProcessing, runtime.GOMAXPROCS(0))
|
||||
}
|
||||
|
||||
if c.UpstreamConnections != 64 {
|
||||
t.Errorf("UpstreamConnections = %d, want 64", c.UpstreamConnections)
|
||||
}
|
||||
}
|
||||
|
||||
// TestExplicitConcurrencyLimitsAreUsed checks that valid values for the
|
||||
// two limits are used as given.
|
||||
func TestExplicitConcurrencyLimitsAreUsed(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
c, err := configFromYAML(t, signingKeyLine+
|
||||
"max_concurrent_processing: 3\nupstream_connections: 10\n")
|
||||
if err != nil {
|
||||
t.Fatalf("valid config should load, got error: %v", err)
|
||||
}
|
||||
|
||||
if c.MaxConcurrentProcessing != 3 {
|
||||
t.Errorf("MaxConcurrentProcessing = %d, want 3", c.MaxConcurrentProcessing)
|
||||
}
|
||||
|
||||
if c.UpstreamConnections != 10 {
|
||||
t.Errorf("UpstreamConnections = %d, want 10", c.UpstreamConnections)
|
||||
}
|
||||
}
|
||||
|
||||
// TestInvalidConcurrencyLimitAbortsStartup checks that a limit that is
|
||||
// not a whole number of at least 1, or is null, aborts startup naming the
|
||||
// key and the value, and the variable too where the value could have come
|
||||
// from it.
|
||||
func TestInvalidConcurrencyLimitAbortsStartup(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
processing := keyMaxConcurrentProcessing
|
||||
connections := keyUpstreamConnections
|
||||
|
||||
runAbortCases(t, []abortCase{
|
||||
{
|
||||
name: "max_concurrent_processing zero",
|
||||
yaml: signingKeyLine + processing + ": 0\n",
|
||||
wantErrSubstrings: []string{
|
||||
processing, testMaxConcurrentProcessingVar, "value 0",
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "max_concurrent_processing negative",
|
||||
yaml: signingKeyLine + processing + ": -2\n",
|
||||
wantErrSubstrings: []string{
|
||||
processing, testMaxConcurrentProcessingVar, "value -2",
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "max_concurrent_processing not a number",
|
||||
yaml: signingKeyLine + processing + ": lots\n",
|
||||
wantErrSubstrings: []string{
|
||||
processing, testMaxConcurrentProcessingVar, "lots",
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "max_concurrent_processing fractional",
|
||||
yaml: signingKeyLine + processing + ": 1.5\n",
|
||||
wantErrSubstrings: []string{processing, "1.5"},
|
||||
},
|
||||
{
|
||||
name: "max_concurrent_processing null",
|
||||
yaml: signingKeyLine + processing + ": null\n",
|
||||
wantErrSubstrings: []string{processing, nullValueText},
|
||||
},
|
||||
{
|
||||
name: "upstream_connections zero",
|
||||
yaml: signingKeyLine + connections + ": 0\n",
|
||||
wantErrSubstrings: []string{
|
||||
connections, testUpstreamConnectionsVar, "value 0",
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "upstream_connections negative",
|
||||
yaml: signingKeyLine + connections + ": -5\n",
|
||||
wantErrSubstrings: []string{
|
||||
connections, testUpstreamConnectionsVar, "value -5",
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "upstream_connections not a number",
|
||||
yaml: signingKeyLine + connections + ": many\n",
|
||||
wantErrSubstrings: []string{
|
||||
connections, testUpstreamConnectionsVar, "many",
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "upstream_connections null",
|
||||
yaml: signingKeyLine + connections + ": null\n",
|
||||
wantErrSubstrings: []string{connections, nullValueText},
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
// TestConcurrencyLimitsFromEnvironment checks that the two variables set
|
||||
// the limits over the config file, and that an invalid value in either
|
||||
// aborts startup naming the variable and the value.
|
||||
func TestConcurrencyLimitsFromEnvironment(t *testing.T) {
|
||||
t.Setenv(testMaxConcurrentProcessingVar, "3")
|
||||
t.Setenv(testUpstreamConnectionsVar, "10")
|
||||
|
||||
c, err := configFromYAML(t, signingKeyLine+
|
||||
"max_concurrent_processing: 5\nupstream_connections: 50\n")
|
||||
if err != nil {
|
||||
t.Fatalf("limits from the environment should load: %v", err)
|
||||
}
|
||||
|
||||
if c.MaxConcurrentProcessing != 3 || c.UpstreamConnections != 10 {
|
||||
t.Errorf("limits = %d and %d, want 3 and 10 from the environment",
|
||||
c.MaxConcurrentProcessing, c.UpstreamConnections)
|
||||
}
|
||||
|
||||
cases := []struct {
|
||||
variable string
|
||||
value string
|
||||
}{
|
||||
{testMaxConcurrentProcessingVar, "lots"},
|
||||
{testMaxConcurrentProcessingVar, "0"},
|
||||
{testUpstreamConnectionsVar, "-1"},
|
||||
{testUpstreamConnectionsVar, "ten"},
|
||||
}
|
||||
|
||||
for _, tc := range cases {
|
||||
t.Run(tc.variable+"="+tc.value, func(t *testing.T) {
|
||||
t.Setenv(tc.variable, tc.value)
|
||||
|
||||
_, err := configFromYAML(t, signingKeyLine)
|
||||
wantStartupError(t, err, tc.variable, tc.value)
|
||||
})
|
||||
}
|
||||
}
|
||||
+11
-230
@@ -10,11 +10,9 @@ import (
|
||||
"net/url"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"runtime"
|
||||
"sort"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"git.eeqj.de/sneak/smartconfig"
|
||||
"go.uber.org/fx"
|
||||
@@ -26,11 +24,6 @@ const (
|
||||
DefaultPort = 8080
|
||||
DefaultStateDir = "/var/lib/pixa"
|
||||
DefaultUpstreamConnectionsPerHost = 20
|
||||
DefaultUpstreamConnections = 64
|
||||
DefaultAccessControlAllowOrigin = "*"
|
||||
DefaultUpstreamFetchTimeout = 30 * time.Second
|
||||
DefaultUpstreamMaxResponseSize = 50 << 20 // 50 MiB
|
||||
DefaultDownstreamTimeout = 60 * time.Second
|
||||
)
|
||||
|
||||
// Configuration key names.
|
||||
@@ -48,15 +41,9 @@ const (
|
||||
keyAllowlistHosts = "allowlist_hosts"
|
||||
keyAllowHTTP = "allow_http"
|
||||
keyUpstreamConnectionsPerHost = "upstream_connections_per_host"
|
||||
keyUpstreamConnections = "upstream_connections"
|
||||
keyMaxConcurrentProcessing = "max_concurrent_processing"
|
||||
keyCacheMaxBytes = "cache_max_bytes"
|
||||
keyBlockedNetworks = "blocked_networks"
|
||||
keyTrustedProxies = "trusted_proxies"
|
||||
keyAccessControlAllowOrigin = "access_control_allow_origin"
|
||||
keyUpstreamFetchTimeout = "upstream_fetch_timeout"
|
||||
keyUpstreamMaxResponseSize = "upstream_max_response_size"
|
||||
keyDownstreamTimeout = "downstream_timeout"
|
||||
)
|
||||
|
||||
// placeholderSigningKey is the dummy signing_key shipped in
|
||||
@@ -82,8 +69,7 @@ var (
|
||||
errEmptyEntry = errors.New("contains an empty entry")
|
||||
errNotAValidURL = errors.New("not a valid URL")
|
||||
errPortOutOfRange = errors.New("outside the valid port range")
|
||||
errSizeOutOfRange = errors.New("outside the accepted range")
|
||||
errMustBeAtLeastOne = errors.New("must be at least 1")
|
||||
errTooFewConnections = errors.New("must be at least 1")
|
||||
errValueTooShort = errors.New("value too short")
|
||||
errPlaceholderKey = errors.New(
|
||||
"is the placeholder from config.example.yml; " +
|
||||
@@ -100,10 +86,6 @@ var (
|
||||
errNotBareHostname = errors.New(
|
||||
"must be a bare hostname without scheme, path, or whitespace")
|
||||
errNoHostnameLabels = errors.New("contains no hostname labels")
|
||||
errNotADuration = errors.New("not a duration such as 30s or 2m")
|
||||
errMustBePositive = errors.New("must be positive")
|
||||
errNotAnOrigin = errors.New(
|
||||
`not "*" or an origin such as https://example.com`)
|
||||
)
|
||||
|
||||
// Params defines dependencies for Config.
|
||||
@@ -130,29 +112,6 @@ type Config struct {
|
||||
AllowHTTP bool // Allow non-TLS upstream (testing only)
|
||||
UpstreamConnectionsPerHost int // Max concurrent connections per upstream host
|
||||
|
||||
// UpstreamConnections is the most concurrent connections to all
|
||||
// upstream hosts together, on top of the per-host limit.
|
||||
// MaxConcurrentProcessing is the most images processed at once.
|
||||
UpstreamConnections int
|
||||
MaxConcurrentProcessing int
|
||||
|
||||
// UpstreamFetchTimeout is the time allowed for one fetch from an
|
||||
// upstream host. UpstreamMaxResponseSize is the largest upstream
|
||||
// response accepted, in bytes, and also the image processor's input
|
||||
// limit.
|
||||
UpstreamFetchTimeout time.Duration
|
||||
UpstreamMaxResponseSize int64
|
||||
|
||||
// AccessControlAllowOrigin is the origin the CORS middleware allows
|
||||
// to read responses: "*" for any, or one origin such as
|
||||
// https://example.com.
|
||||
AccessControlAllowOrigin string
|
||||
|
||||
// DownstreamTimeout bounds how long answering one client request may
|
||||
// take. It is both the HTTP server's write timeout and the deadline
|
||||
// of the per-request timeout middleware.
|
||||
DownstreamTimeout time.Duration
|
||||
|
||||
// BlockedNetworks are operator-supplied CIDR ranges to refuse in
|
||||
// addition to the built-in SSRF blocklist. Enforced by the upstream
|
||||
// fetcher's dialer; the built-in ranges always apply.
|
||||
@@ -236,12 +195,14 @@ func New(_ fx.Lifecycle, params Params) (*Config, error) {
|
||||
// unparseable or invalid is an error: defaults apply only to omitted
|
||||
// keys, never to invalid explicit values.
|
||||
func newFromSmartConfig(sc *smartconfig.Config) (*Config, error) {
|
||||
if sc != nil {
|
||||
err := validateKnownKeys(sc)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
|
||||
err = validateAllowlistHostsValue(sc)
|
||||
err := validateAllowlistHostsValue(sc)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -279,20 +240,6 @@ func newFromSmartConfig(sc *smartconfig.Config) (*Config, error) {
|
||||
AllowHTTP: loader.boolVal(keyAllowHTTP, false),
|
||||
UpstreamConnectionsPerHost: loader.intVal(
|
||||
keyUpstreamConnectionsPerHost, DefaultUpstreamConnectionsPerHost),
|
||||
UpstreamConnections: loader.intVal(
|
||||
keyUpstreamConnections, DefaultUpstreamConnections),
|
||||
// Decoding and encoding are CPU-bound, so the default is one image
|
||||
// per CPU Go uses, which follows a container's CPU limit.
|
||||
MaxConcurrentProcessing: loader.intVal(
|
||||
keyMaxConcurrentProcessing, runtime.GOMAXPROCS(0)),
|
||||
UpstreamFetchTimeout: loader.durationVal(
|
||||
keyUpstreamFetchTimeout, DefaultUpstreamFetchTimeout),
|
||||
UpstreamMaxResponseSize: loader.int64Val(
|
||||
keyUpstreamMaxResponseSize, DefaultUpstreamMaxResponseSize),
|
||||
AccessControlAllowOrigin: loader.stringVal(
|
||||
keyAccessControlAllowOrigin, DefaultAccessControlAllowOrigin),
|
||||
DownstreamTimeout: loader.durationVal(
|
||||
keyDownstreamTimeout, DefaultDownstreamTimeout),
|
||||
CacheMaxBytes: loader.int64Val(keyCacheMaxBytes, 0),
|
||||
BlockedNetworks: blockedNetworks,
|
||||
TrustedProxies: trustedProxies,
|
||||
@@ -335,13 +282,8 @@ func newFromSmartConfig(sc *smartconfig.Config) (*Config, error) {
|
||||
// being silently ignored, and rejects keys that are explicitly set to
|
||||
// null: a null is a SET value, never an omission, so it must not
|
||||
// silently take the default. The env section is permitted because
|
||||
// smartconfig consumes it for environment variable injection. A nil sc
|
||||
// means no config file, which has no keys to check.
|
||||
// smartconfig consumes it for environment variable injection.
|
||||
func validateKnownKeys(sc *smartconfig.Config) error {
|
||||
if sc == nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
var unknown, nullKeys []string
|
||||
|
||||
for key, value := range sc.Data() {
|
||||
@@ -411,10 +353,8 @@ func isKnownConfigKey(key string) bool {
|
||||
switch key {
|
||||
case keyDebug, keyMaintenanceMode, keyPort, keyStateDir, keySentryDSN,
|
||||
keyDBURL, keyMetrics, keySigningKey, keyAllowlistHosts, keyAllowHTTP,
|
||||
keyUpstreamConnectionsPerHost, keyUpstreamConnections,
|
||||
keyMaxConcurrentProcessing, keyCacheMaxBytes, keyBlockedNetworks,
|
||||
keyTrustedProxies, keyAccessControlAllowOrigin, keyUpstreamFetchTimeout,
|
||||
keyUpstreamMaxResponseSize, keyDownstreamTimeout, "env":
|
||||
keyUpstreamConnectionsPerHost, keyCacheMaxBytes, keyBlockedNetworks,
|
||||
keyTrustedProxies, "env":
|
||||
return true
|
||||
}
|
||||
|
||||
@@ -439,15 +379,9 @@ func envVarNames() map[string]string {
|
||||
keyAllowlistHosts: "PIXA_ALLOWLIST_HOSTS",
|
||||
keyAllowHTTP: "PIXA_ALLOW_HTTP",
|
||||
keyUpstreamConnectionsPerHost: "PIXA_UPSTREAM_CONNECTIONS_PER_HOST",
|
||||
keyUpstreamConnections: "PIXA_UPSTREAM_CONNECTIONS",
|
||||
keyMaxConcurrentProcessing: "PIXA_MAX_CONCURRENT_PROCESSING",
|
||||
keyCacheMaxBytes: "PIXA_CACHE_MAX_BYTES",
|
||||
keyBlockedNetworks: "PIXA_BLOCKED_NETWORKS",
|
||||
keyTrustedProxies: "PIXA_TRUSTED_PROXIES",
|
||||
keyAccessControlAllowOrigin: "PIXA_ACCESS_CONTROL_ALLOW_ORIGIN",
|
||||
keyUpstreamFetchTimeout: "PIXA_UPSTREAM_FETCH_TIMEOUT",
|
||||
keyUpstreamMaxResponseSize: "PIXA_UPSTREAM_MAX_RESPONSE_SIZE",
|
||||
keyDownstreamTimeout: "PIXA_DOWNSTREAM_TIMEOUT",
|
||||
}
|
||||
}
|
||||
|
||||
@@ -584,9 +518,10 @@ func (c *Config) validate() error {
|
||||
settingName(keyPort), c.Port, errPortOutOfRange, maxPort)
|
||||
}
|
||||
|
||||
err = c.validateConcurrencyLimits()
|
||||
if err != nil {
|
||||
return err
|
||||
if c.UpstreamConnectionsPerHost < 1 {
|
||||
return fmt.Errorf("%s: value %d %w",
|
||||
settingName(keyUpstreamConnectionsPerHost),
|
||||
c.UpstreamConnectionsPerHost, errTooFewConnections)
|
||||
}
|
||||
|
||||
if c.StateDir == "" {
|
||||
@@ -600,11 +535,6 @@ func (c *Config) validate() error {
|
||||
settingName(keyCacheMaxBytes), c.CacheMaxBytes, errMustNotBeNegative)
|
||||
}
|
||||
|
||||
err = c.validateUpstreamMaxResponseSize()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
for _, host := range c.AllowlistHosts {
|
||||
err := validateAllowlistHost(host)
|
||||
if err != nil {
|
||||
@@ -626,106 +556,6 @@ func (c *Config) validate() error {
|
||||
errMustBeSetTogether)
|
||||
}
|
||||
|
||||
return c.validateAccessControlAllowOrigin()
|
||||
}
|
||||
|
||||
// validateUpstreamMaxResponseSize checks that upstream_max_response_size
|
||||
// is from 1 byte to 1 GiB. An upstream response is read whole into
|
||||
// memory, and the image processor reads one byte past this limit, which
|
||||
// must not overflow.
|
||||
func (c *Config) validateUpstreamMaxResponseSize() error {
|
||||
const maxUpstreamMaxResponseSize = 1 << 30 // 1 GiB
|
||||
if c.UpstreamMaxResponseSize < 1 ||
|
||||
c.UpstreamMaxResponseSize > maxUpstreamMaxResponseSize {
|
||||
return fmt.Errorf("%s: value %d is %w 1-%d",
|
||||
settingName(keyUpstreamMaxResponseSize), c.UpstreamMaxResponseSize,
|
||||
errSizeOutOfRange, maxUpstreamMaxResponseSize)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// validateAccessControlAllowOrigin accepts "*" or an origin exactly as a browser
|
||||
// sends it: http or https, an IP address as netip writes it or a lowercase name
|
||||
// with a letter in its last part, and an optional port 1-65535, not the default.
|
||||
func (c *Config) validateAccessControlAllowOrigin() error {
|
||||
origin := c.AccessControlAllowOrigin
|
||||
if origin == "*" {
|
||||
return nil
|
||||
}
|
||||
|
||||
errOrigin := fmt.Errorf("%s: value %q is %w",
|
||||
settingName(keyAccessControlAllowOrigin), origin, errNotAnOrigin)
|
||||
|
||||
parsed, err := url.Parse(origin)
|
||||
if err != nil {
|
||||
return errOrigin
|
||||
}
|
||||
|
||||
defaultPort := map[string]string{"http": "80", "https": "443"}[parsed.Scheme]
|
||||
if defaultPort == "" {
|
||||
return errOrigin
|
||||
}
|
||||
|
||||
const letters = "abcdefghijklmnopqrstuvwxyz"
|
||||
|
||||
host := parsed.Hostname()
|
||||
lastPart := host[strings.LastIndex(host, ".")+1:]
|
||||
|
||||
addr, err := netip.ParseAddr(host)
|
||||
|
||||
switch {
|
||||
case err == nil && addr.Is6():
|
||||
host = "[" + addr.String() + "]"
|
||||
case err == nil:
|
||||
host = addr.String()
|
||||
case strings.Trim(host, letters+"0123456789-.") != "": // a character other than these
|
||||
return errOrigin
|
||||
case !strings.ContainsAny(lastPart, letters):
|
||||
return errOrigin
|
||||
}
|
||||
|
||||
// The value must be exactly the origin rebuilt from its parts.
|
||||
rebuilt := parsed.Scheme + "://" + host
|
||||
|
||||
port := parsed.Port()
|
||||
if port != "" {
|
||||
_, err := strconv.ParseUint(port, 10, 16)
|
||||
if err != nil || port[0] == '0' || port == defaultPort {
|
||||
return errOrigin
|
||||
}
|
||||
|
||||
rebuilt += ":" + port
|
||||
}
|
||||
|
||||
if rebuilt != origin {
|
||||
return errOrigin
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// validateConcurrencyLimits checks that the two upstream connection limits
|
||||
// and the image processing limit are at least 1.
|
||||
func (c *Config) validateConcurrencyLimits() error {
|
||||
if c.UpstreamConnectionsPerHost < 1 {
|
||||
return fmt.Errorf("%s: value %d %w",
|
||||
settingName(keyUpstreamConnectionsPerHost),
|
||||
c.UpstreamConnectionsPerHost, errMustBeAtLeastOne)
|
||||
}
|
||||
|
||||
if c.UpstreamConnections < 1 {
|
||||
return fmt.Errorf("%s: value %d %w",
|
||||
settingName(keyUpstreamConnections),
|
||||
c.UpstreamConnections, errMustBeAtLeastOne)
|
||||
}
|
||||
|
||||
if c.MaxConcurrentProcessing < 1 {
|
||||
return fmt.Errorf("%s: value %d %w",
|
||||
settingName(keyMaxConcurrentProcessing),
|
||||
c.MaxConcurrentProcessing, errMustBeAtLeastOne)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -843,19 +673,6 @@ func (l *strictLoader) int64Val(key string, defaultVal int64) int64 {
|
||||
return val
|
||||
}
|
||||
|
||||
func (l *strictLoader) durationVal(key string, defaultVal time.Duration) time.Duration {
|
||||
if l.err != nil {
|
||||
return 0
|
||||
}
|
||||
|
||||
val, err := getDuration(l.sc, key, defaultVal)
|
||||
if err != nil {
|
||||
l.err = err
|
||||
}
|
||||
|
||||
return val
|
||||
}
|
||||
|
||||
func (l *strictLoader) boolVal(key string, defaultVal bool) bool {
|
||||
if l.err != nil {
|
||||
return false
|
||||
@@ -977,42 +794,6 @@ func getInt64(sc *smartconfig.Config, key string, defaultVal int64) (int64, erro
|
||||
}
|
||||
}
|
||||
|
||||
// getDuration returns the duration value for key, or defaultVal if the
|
||||
// key is omitted. A present value must be a positive Go duration string
|
||||
// such as "30s" or "2m", read with time.ParseDuration; a bare number has
|
||||
// no unit and is an error, as is an explicit null.
|
||||
func getDuration(
|
||||
sc *smartconfig.Config, key string, defaultVal time.Duration,
|
||||
) (time.Duration, error) {
|
||||
raw, ok := lookupValue(sc, key)
|
||||
if !ok {
|
||||
return defaultVal, nil
|
||||
}
|
||||
|
||||
if raw == nil {
|
||||
return 0, errNullConfigValue(key)
|
||||
}
|
||||
|
||||
str, ok := raw.(string)
|
||||
if !ok {
|
||||
return 0, fmt.Errorf("config key %q: value %v (%T) is %w",
|
||||
key, raw, raw, errNotADuration)
|
||||
}
|
||||
|
||||
parsed, err := time.ParseDuration(strings.TrimSpace(str))
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("%s: value %q is %w",
|
||||
settingName(key), str, errNotADuration)
|
||||
}
|
||||
|
||||
if parsed <= 0 {
|
||||
return 0, fmt.Errorf("%s: value %q %w",
|
||||
settingName(key), str, errMustBePositive)
|
||||
}
|
||||
|
||||
return parsed, nil
|
||||
}
|
||||
|
||||
// getBool returns the boolean value for key, or defaultVal if the key
|
||||
// is omitted. A present value that is not a boolean (or a ParseBool-able
|
||||
// string), or is explicitly null, is an error; numbers are not accepted
|
||||
|
||||
@@ -6,7 +6,6 @@ import (
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.eeqj.de/sneak/smartconfig"
|
||||
)
|
||||
@@ -600,244 +599,3 @@ func TestEnsureStateDirFailsOnUncreatablePath(t *testing.T) {
|
||||
t.Errorf("error %q does not name the offending key state_dir", err.Error())
|
||||
}
|
||||
}
|
||||
|
||||
// TestOmittedOriginTimeoutsAndSizeUseDefaults checks that the CORS
|
||||
// origin, the upstream fetch timeout, the upstream response size limit
|
||||
// and the downstream timeout default to the values pixa used before they
|
||||
// could be configured.
|
||||
func TestOmittedOriginTimeoutsAndSizeUseDefaults(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
c, err := configFromYAML(t, signingKeyLine)
|
||||
if err != nil {
|
||||
t.Fatalf("minimal config should be valid, got error: %v", err)
|
||||
}
|
||||
|
||||
if c.AccessControlAllowOrigin != "*" {
|
||||
t.Errorf("AccessControlAllowOrigin = %q, want *", c.AccessControlAllowOrigin)
|
||||
}
|
||||
|
||||
if c.UpstreamFetchTimeout != 30*time.Second {
|
||||
t.Errorf("UpstreamFetchTimeout = %v, want 30s", c.UpstreamFetchTimeout)
|
||||
}
|
||||
|
||||
if c.UpstreamMaxResponseSize != 50<<20 {
|
||||
t.Errorf("UpstreamMaxResponseSize = %d, want %d (50 MiB)",
|
||||
c.UpstreamMaxResponseSize, 50<<20)
|
||||
}
|
||||
|
||||
if c.DownstreamTimeout != 60*time.Second {
|
||||
t.Errorf("DownstreamTimeout = %v, want 60s", c.DownstreamTimeout)
|
||||
}
|
||||
}
|
||||
|
||||
// TestExplicitOriginTimeoutsAndSizeAreUsed checks that valid values for
|
||||
// the CORS origin, the two timeouts and the response size limit are used
|
||||
// as given. The size is the largest accepted, 1 GiB.
|
||||
func TestExplicitOriginTimeoutsAndSizeAreUsed(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
c, err := configFromYAML(t, signingKeyLine+`
|
||||
access_control_allow_origin: https://app.example.com
|
||||
upstream_fetch_timeout: 10s
|
||||
upstream_max_response_size: 1073741824
|
||||
downstream_timeout: 2m
|
||||
`)
|
||||
if err != nil {
|
||||
t.Fatalf("valid config should load, got error: %v", err)
|
||||
}
|
||||
|
||||
if c.AccessControlAllowOrigin != "https://app.example.com" {
|
||||
t.Errorf("AccessControlAllowOrigin = %q, want https://app.example.com",
|
||||
c.AccessControlAllowOrigin)
|
||||
}
|
||||
|
||||
if c.UpstreamFetchTimeout != 10*time.Second {
|
||||
t.Errorf("UpstreamFetchTimeout = %v, want 10s", c.UpstreamFetchTimeout)
|
||||
}
|
||||
|
||||
if c.UpstreamMaxResponseSize != 1073741824 {
|
||||
t.Errorf("UpstreamMaxResponseSize = %d, want 1073741824",
|
||||
c.UpstreamMaxResponseSize)
|
||||
}
|
||||
|
||||
if c.DownstreamTimeout != 2*time.Minute {
|
||||
t.Errorf("DownstreamTimeout = %v, want 2m", c.DownstreamTimeout)
|
||||
}
|
||||
}
|
||||
|
||||
// TestOriginWithPortOrAnyOriginIsAccepted checks the other accepted forms
|
||||
// of access_control_allow_origin: "*", an origin with a port, and origins
|
||||
// whose host is an IPv4 or IPv6 address.
|
||||
func TestOriginWithPortOrAnyOriginIsAccepted(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
for _, origin := range []string{
|
||||
"*", "http://localhost:3000", "http://192.0.2.1", "http://[2001:db8::1]:8080",
|
||||
} {
|
||||
c, err := configFromYAML(t, signingKeyLine+
|
||||
"access_control_allow_origin: \""+origin+"\"\n")
|
||||
if err != nil {
|
||||
t.Fatalf("origin %q should be accepted, got error: %v", origin, err)
|
||||
}
|
||||
|
||||
if c.AccessControlAllowOrigin != origin {
|
||||
t.Errorf("AccessControlAllowOrigin = %q, want %q",
|
||||
c.AccessControlAllowOrigin, origin)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// invalidTimeoutCases are configs where upstream_fetch_timeout or
|
||||
// downstream_timeout is not a positive Go duration string; each must
|
||||
// abort startup naming the key and the value.
|
||||
func invalidTimeoutCases() []abortCase {
|
||||
return []abortCase{
|
||||
{
|
||||
name: "upstream_fetch_timeout not a duration",
|
||||
yaml: signingKeyLine + "upstream_fetch_timeout: soon\n",
|
||||
wantErrSubstrings: []string{keyUpstreamFetchTimeout, "soon"},
|
||||
},
|
||||
{
|
||||
name: "upstream_fetch_timeout number without a unit",
|
||||
yaml: signingKeyLine + "upstream_fetch_timeout: 45\n",
|
||||
wantErrSubstrings: []string{keyUpstreamFetchTimeout, "45"},
|
||||
},
|
||||
{
|
||||
name: "upstream_fetch_timeout zero",
|
||||
yaml: signingKeyLine + "upstream_fetch_timeout: 0s\n",
|
||||
wantErrSubstrings: []string{keyUpstreamFetchTimeout, "0s"},
|
||||
},
|
||||
{
|
||||
name: "upstream_fetch_timeout negative",
|
||||
yaml: signingKeyLine + "upstream_fetch_timeout: -5s\n",
|
||||
wantErrSubstrings: []string{keyUpstreamFetchTimeout, "-5s"},
|
||||
},
|
||||
{
|
||||
name: "upstream_fetch_timeout null",
|
||||
yaml: signingKeyLine + "upstream_fetch_timeout: null\n",
|
||||
wantErrSubstrings: []string{keyUpstreamFetchTimeout, nullValueText},
|
||||
},
|
||||
{
|
||||
name: "downstream_timeout not a duration",
|
||||
yaml: signingKeyLine + "downstream_timeout: 1 minute\n",
|
||||
wantErrSubstrings: []string{keyDownstreamTimeout, "1 minute"},
|
||||
},
|
||||
{
|
||||
name: "downstream_timeout zero",
|
||||
yaml: signingKeyLine + "downstream_timeout: 0s\n",
|
||||
wantErrSubstrings: []string{keyDownstreamTimeout, "0s"},
|
||||
},
|
||||
{
|
||||
name: "downstream_timeout negative",
|
||||
yaml: signingKeyLine + "downstream_timeout: -1m\n",
|
||||
wantErrSubstrings: []string{keyDownstreamTimeout, "-1m"},
|
||||
},
|
||||
{
|
||||
name: "downstream_timeout null",
|
||||
yaml: signingKeyLine + "downstream_timeout:\n",
|
||||
wantErrSubstrings: []string{keyDownstreamTimeout, nullValueText},
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
// invalidSizeAndOriginCases are configs where upstream_max_response_size
|
||||
// is not a whole number of bytes from 1 to 1 GiB, or
|
||||
// access_control_allow_origin is neither "*" nor an origin; each must
|
||||
// abort startup naming the key and the value.
|
||||
func invalidSizeAndOriginCases() []abortCase {
|
||||
badOrigins := []string{
|
||||
"", // empty
|
||||
"example.com", // no scheme
|
||||
"https://example.com/images", // a path
|
||||
"https://example.com/", // a trailing slash
|
||||
// The CORS middleware reads a * inside an origin as a pattern
|
||||
// that lets other sites read responses.
|
||||
"https://*",
|
||||
"https://*.example.com",
|
||||
"https://*example.com",
|
||||
"https://a.com,b.com", // two hosts
|
||||
"https://example.com:", // an empty port
|
||||
"https://:8443", // no host
|
||||
"https://example.com:0", // a port below 1
|
||||
"https://example.com:99999", // a port above 65535
|
||||
"https://exämple.com", // a host name that is not ASCII
|
||||
"https://example.com:443", // the default port for https
|
||||
"http://example.com:80", // the default port for http
|
||||
"https://example.com:08080", // a port with a leading zero
|
||||
"https://01.2.3.4", // an IPv4 address with a leading zero
|
||||
"https://10.0.0", // an IPv4 address with three parts
|
||||
"https://192.168.1.256", // an IPv4 address part above 255
|
||||
"https://example.123", // a host name whose last part is a number
|
||||
"https://[0:0:0:0:0:0:0:1]", // an IPv6 address not in its shortest form
|
||||
"file://example.com", // a scheme other than http or https
|
||||
"HTTPS://example.com", // a scheme in upper case
|
||||
"https://Example.com", // a host name in upper case
|
||||
}
|
||||
|
||||
cases := make([]abortCase, 0, len(badOrigins))
|
||||
for _, origin := range badOrigins {
|
||||
cases = append(cases, abortCase{
|
||||
name: "access_control_allow_origin " + origin,
|
||||
yaml: signingKeyLine +
|
||||
"access_control_allow_origin: \"" + origin + "\"\n",
|
||||
wantErrSubstrings: []string{keyAccessControlAllowOrigin, origin},
|
||||
})
|
||||
}
|
||||
|
||||
return append(cases, []abortCase{
|
||||
{
|
||||
name: "access_control_allow_origin null",
|
||||
yaml: signingKeyLine + "access_control_allow_origin: null\n",
|
||||
wantErrSubstrings: []string{keyAccessControlAllowOrigin, nullValueText},
|
||||
},
|
||||
{
|
||||
name: "upstream_max_response_size with a unit",
|
||||
yaml: signingKeyLine + "upstream_max_response_size: 50MB\n",
|
||||
wantErrSubstrings: []string{keyUpstreamMaxResponseSize, "50MB"},
|
||||
},
|
||||
{
|
||||
name: "upstream_max_response_size fractional",
|
||||
yaml: signingKeyLine + "upstream_max_response_size: 1.5\n",
|
||||
wantErrSubstrings: []string{keyUpstreamMaxResponseSize, "1.5"},
|
||||
},
|
||||
{
|
||||
name: "upstream_max_response_size zero",
|
||||
yaml: signingKeyLine + "upstream_max_response_size: 0\n",
|
||||
wantErrSubstrings: []string{keyUpstreamMaxResponseSize, "0"},
|
||||
},
|
||||
{
|
||||
name: "upstream_max_response_size negative",
|
||||
yaml: signingKeyLine + "upstream_max_response_size: -1\n",
|
||||
wantErrSubstrings: []string{keyUpstreamMaxResponseSize, "-1"},
|
||||
},
|
||||
{
|
||||
name: "upstream_max_response_size null",
|
||||
yaml: signingKeyLine + "upstream_max_response_size: null\n",
|
||||
wantErrSubstrings: []string{keyUpstreamMaxResponseSize, nullValueText},
|
||||
},
|
||||
{
|
||||
name: "upstream_max_response_size above 1 GiB",
|
||||
yaml: signingKeyLine + "upstream_max_response_size: 1073741825\n",
|
||||
wantErrSubstrings: []string{keyUpstreamMaxResponseSize, "1073741825"},
|
||||
},
|
||||
{
|
||||
name: "upstream_max_response_size largest 64-bit integer",
|
||||
yaml: signingKeyLine +
|
||||
"upstream_max_response_size: 9223372036854775807\n",
|
||||
wantErrSubstrings: []string{
|
||||
keyUpstreamMaxResponseSize, "9223372036854775807",
|
||||
},
|
||||
},
|
||||
}...)
|
||||
}
|
||||
|
||||
// TestInvalidOriginTimeoutOrSizeAbortsStartup verifies the
|
||||
// no-silent-fallback rule for the CORS origin, the two timeouts and the
|
||||
// response size limit: a value that does not parse or is out of range
|
||||
// aborts startup naming the key and the value.
|
||||
func TestInvalidOriginTimeoutOrSizeAbortsStartup(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
runAbortCases(t, append(invalidTimeoutCases(), invalidSizeAndOriginCases()...))
|
||||
}
|
||||
|
||||
@@ -8,7 +8,6 @@ import (
|
||||
"slices"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"sneak.berlin/go/pixa/internal/globals"
|
||||
"sneak.berlin/go/pixa/internal/logger"
|
||||
@@ -67,15 +66,9 @@ func TestEnvironmentSetsEveryKey(t *testing.T) {
|
||||
t.Setenv("PIXA_ALLOWLIST_HOSTS", "s3.sneak.cloud,.example.com")
|
||||
t.Setenv("PIXA_ALLOW_HTTP", "true")
|
||||
t.Setenv("PIXA_UPSTREAM_CONNECTIONS_PER_HOST", "5")
|
||||
t.Setenv("PIXA_UPSTREAM_CONNECTIONS", "10")
|
||||
t.Setenv("PIXA_MAX_CONCURRENT_PROCESSING", "3")
|
||||
t.Setenv("PIXA_CACHE_MAX_BYTES", "1024")
|
||||
t.Setenv("PIXA_BLOCKED_NETWORKS", "203.0.113.0/24")
|
||||
t.Setenv("PIXA_TRUSTED_PROXIES", "192.0.2.0/24")
|
||||
t.Setenv("PIXA_ACCESS_CONTROL_ALLOW_ORIGIN", "https://app.example.com")
|
||||
t.Setenv("PIXA_UPSTREAM_FETCH_TIMEOUT", "10s")
|
||||
t.Setenv("PIXA_UPSTREAM_MAX_RESPONSE_SIZE", "1048576")
|
||||
t.Setenv("PIXA_DOWNSTREAM_TIMEOUT", "2m")
|
||||
|
||||
c, err := newFromSmartConfig(nil)
|
||||
if err != nil {
|
||||
@@ -95,16 +88,10 @@ func TestEnvironmentSetsEveryKey(t *testing.T) {
|
||||
AllowlistHosts: []string{testHostS3, ".example.com"},
|
||||
AllowHTTP: true,
|
||||
UpstreamConnectionsPerHost: 5,
|
||||
UpstreamConnections: 10,
|
||||
MaxConcurrentProcessing: 3,
|
||||
CacheMaxBytes: 1024,
|
||||
cacheMaxBytesExplicit: true,
|
||||
BlockedNetworks: []netip.Prefix{netip.MustParsePrefix("203.0.113.0/24")},
|
||||
TrustedProxies: []netip.Prefix{netip.MustParsePrefix("192.0.2.0/24")},
|
||||
AccessControlAllowOrigin: "https://app.example.com",
|
||||
UpstreamFetchTimeout: 10 * time.Second,
|
||||
UpstreamMaxResponseSize: 1048576,
|
||||
DownstreamTimeout: 2 * time.Minute,
|
||||
}
|
||||
|
||||
if !reflect.DeepEqual(*c, want) {
|
||||
@@ -293,30 +280,6 @@ func TestInvalidDebugFromEnvironmentAbortsStartup(t *testing.T) {
|
||||
wantStartupError(t, err, "PIXA_DEBUG", "maybe")
|
||||
}
|
||||
|
||||
// TestInvalidOriginTimeoutOrSizeFromEnvironmentAbortsStartup checks that
|
||||
// an invalid CORS origin, timeout or response size limit in its variable
|
||||
// aborts startup naming the variable and the value.
|
||||
func TestInvalidOriginTimeoutOrSizeFromEnvironmentAbortsStartup(t *testing.T) {
|
||||
cases := []struct {
|
||||
variable string
|
||||
value string
|
||||
}{
|
||||
{"PIXA_ACCESS_CONTROL_ALLOW_ORIGIN", "example.com"},
|
||||
{"PIXA_UPSTREAM_FETCH_TIMEOUT", "soon"},
|
||||
{"PIXA_UPSTREAM_MAX_RESPONSE_SIZE", "50MB"},
|
||||
{"PIXA_DOWNSTREAM_TIMEOUT", "0s"},
|
||||
}
|
||||
|
||||
for _, tc := range cases {
|
||||
t.Run(tc.variable, func(t *testing.T) {
|
||||
t.Setenv(tc.variable, tc.value)
|
||||
|
||||
_, err := configFromYAML(t, signingKeyLine)
|
||||
wantStartupError(t, err, tc.variable, tc.value)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// TestConfigFileAloneBehavesAsBefore checks that with no variables set
|
||||
// (TestMain unsets them) the config file's values are used and omitted
|
||||
// keys take their defaults.
|
||||
|
||||
@@ -106,14 +106,11 @@ func (s *Handlers) initImageService() error {
|
||||
// Create the fetcher config
|
||||
fetcherCfg := httpfetcher.DefaultConfig()
|
||||
fetcherCfg.AllowHTTP = s.config.AllowHTTP
|
||||
fetcherCfg.Timeout = s.config.UpstreamFetchTimeout
|
||||
fetcherCfg.MaxResponseSize = s.config.UpstreamMaxResponseSize
|
||||
|
||||
if s.config.UpstreamConnectionsPerHost > 0 {
|
||||
fetcherCfg.MaxConnectionsPerHost = s.config.UpstreamConnectionsPerHost
|
||||
}
|
||||
|
||||
fetcherCfg.MaxConnections = s.config.UpstreamConnections
|
||||
fetcherCfg.BlockedNetworks = s.config.BlockedNetworks
|
||||
|
||||
// Create the service
|
||||
@@ -122,7 +119,6 @@ func (s *Handlers) initImageService() error {
|
||||
FetcherConfig: fetcherCfg,
|
||||
SigningKey: s.config.SigningKey,
|
||||
Allowlist: s.config.AllowlistHosts,
|
||||
MaxConcurrentProcessing: s.config.MaxConcurrentProcessing,
|
||||
Logger: s.log,
|
||||
})
|
||||
if err != nil {
|
||||
|
||||
@@ -12,7 +12,6 @@ import (
|
||||
"github.com/go-chi/chi/v5"
|
||||
"sneak.berlin/go/pixa/internal/encurl"
|
||||
"sneak.berlin/go/pixa/internal/httpfetcher"
|
||||
"sneak.berlin/go/pixa/internal/imageprocessor"
|
||||
"sneak.berlin/go/pixa/internal/imgcache"
|
||||
)
|
||||
|
||||
@@ -218,14 +217,6 @@ func (s *Handlers) respondImageError(
|
||||
return
|
||||
}
|
||||
|
||||
if errors.Is(err, httpfetcher.ErrTooManyConnections) ||
|
||||
errors.Is(err, imageprocessor.ErrTooManyImages) {
|
||||
s.respondError(w, "server busy, try again later",
|
||||
http.StatusServiceUnavailable)
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
s.respondError(w, "internal error", http.StatusInternalServerError)
|
||||
}
|
||||
|
||||
|
||||
@@ -12,7 +12,6 @@ import (
|
||||
|
||||
"sneak.berlin/go/pixa/internal/encurl"
|
||||
"sneak.berlin/go/pixa/internal/httpfetcher"
|
||||
"sneak.berlin/go/pixa/internal/imageprocessor"
|
||||
"sneak.berlin/go/pixa/internal/imgcache"
|
||||
)
|
||||
|
||||
@@ -125,10 +124,6 @@ func (s *Handlers) handleImageError(w http.ResponseWriter, err error) {
|
||||
s.respondError(w, "upstream error", http.StatusBadGateway)
|
||||
case errors.Is(err, httpfetcher.ErrUpstreamTimeout):
|
||||
s.respondError(w, "upstream timeout", http.StatusGatewayTimeout)
|
||||
case errors.Is(err, httpfetcher.ErrTooManyConnections),
|
||||
errors.Is(err, imageprocessor.ErrTooManyImages):
|
||||
s.respondError(w, "server busy, try again later",
|
||||
http.StatusServiceUnavailable)
|
||||
default:
|
||||
s.log.Error("image request failed", "error", err)
|
||||
s.respondError(w, "internal error", http.StatusInternalServerError)
|
||||
|
||||
@@ -1,49 +0,0 @@
|
||||
package handlers
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"sneak.berlin/go/pixa/internal/httpfetcher"
|
||||
"sneak.berlin/go/pixa/internal/imageprocessor"
|
||||
"sneak.berlin/go/pixa/internal/imgcache"
|
||||
)
|
||||
|
||||
// TestServerBusyAnswers503 checks that both image routes answer 503 with
|
||||
// a clear error when the image service gives up waiting for a free upstream
|
||||
// connection or processing slot, wrapped as the service wraps them.
|
||||
func TestServerBusyAnswers503(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
h := &Handlers{log: slog.New(slog.DiscardHandler)}
|
||||
req := &imgcache.ImageRequest{SourceHost: "img.example.com", SourcePath: "/a.jpg"}
|
||||
|
||||
for _, err := range []error{
|
||||
fmt.Errorf("upstream fetch failed: %w", httpfetcher.ErrTooManyConnections),
|
||||
fmt.Errorf("image processing failed: %w", imageprocessor.ErrTooManyImages),
|
||||
} {
|
||||
plain := httptest.NewRecorder()
|
||||
h.respondImageError(plain, req, err)
|
||||
|
||||
encrypted := httptest.NewRecorder()
|
||||
h.handleImageError(encrypted, err)
|
||||
|
||||
for route, rec := range map[string]*httptest.ResponseRecorder{
|
||||
"/v1/image/": plain, "/v1/e/": encrypted,
|
||||
} {
|
||||
if rec.Code != http.StatusServiceUnavailable {
|
||||
t.Errorf("%s for %v: status = %d, want %d",
|
||||
route, err, rec.Code, http.StatusServiceUnavailable)
|
||||
}
|
||||
|
||||
if !strings.Contains(rec.Body.String(), "server busy, try again later") {
|
||||
t.Errorf("%s for %v: body = %q, want the server busy error",
|
||||
route, err, rec.Body.String())
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,6 +1,5 @@
|
||||
// Package httpfetcher fetches content from upstream HTTP origins with SSRF
|
||||
// protection, connection limits per host and for all hosts together, and
|
||||
// content-type validation.
|
||||
// protection, per-host connection limits, and content-type validation.
|
||||
package httpfetcher
|
||||
|
||||
import (
|
||||
@@ -29,13 +28,8 @@ const (
|
||||
DefaultIdleConnTimeout = 90 * time.Second
|
||||
DefaultMaxRedirects = 10
|
||||
DefaultMaxConnectionsPerHost = 20
|
||||
DefaultMaxConnections = 64
|
||||
)
|
||||
|
||||
// ConnectionWaitTimeout is how long Fetch waits for a free connection when
|
||||
// MaxConnections fetches are already in progress.
|
||||
const ConnectionWaitTimeout = 10 * time.Second
|
||||
|
||||
// MIME content types.
|
||||
const (
|
||||
contentTypeJPEG = "image/jpeg"
|
||||
@@ -76,7 +70,6 @@ var (
|
||||
ErrInvalidContentType = errors.New("invalid or unsupported content type")
|
||||
ErrUpstreamError = errors.New("upstream server error")
|
||||
ErrUpstreamTimeout = errors.New("upstream request timeout")
|
||||
ErrTooManyConnections = errors.New("too many concurrent upstream connections")
|
||||
)
|
||||
|
||||
// Internal fetcher errors.
|
||||
@@ -129,9 +122,6 @@ type Config struct {
|
||||
AllowHTTP bool
|
||||
// MaxConnectionsPerHost limits concurrent connections to each upstream host.
|
||||
MaxConnectionsPerHost int
|
||||
// MaxConnections limits concurrent connections to all upstream hosts
|
||||
// together.
|
||||
MaxConnections int
|
||||
// BlockedNetworks are operator-supplied CIDR ranges refused by the
|
||||
// dialer, in addition to the always-enforced built-in ranges.
|
||||
BlockedNetworks []netip.Prefix
|
||||
@@ -153,22 +143,15 @@ func DefaultConfig() *Config {
|
||||
},
|
||||
AllowHTTP: false,
|
||||
MaxConnectionsPerHost: DefaultMaxConnectionsPerHost,
|
||||
MaxConnections: DefaultMaxConnections,
|
||||
}
|
||||
}
|
||||
|
||||
// HTTPFetcher implements Fetcher with SSRF protection and connection limits
|
||||
// per host and for all hosts together.
|
||||
// HTTPFetcher implements Fetcher with SSRF protection and per-host connection limits.
|
||||
type HTTPFetcher struct {
|
||||
client *http.Client
|
||||
config *Config
|
||||
hostSems map[string]chan struct{} // per-host semaphores
|
||||
hostSemMu sync.Mutex // protects hostSems map
|
||||
// allHostsSemaphore has one slot per connection allowed to all hosts
|
||||
// together (config.MaxConnections).
|
||||
allHostsSemaphore chan struct{}
|
||||
// connectionWaitTimeout is ConnectionWaitTimeout; tests shorten it.
|
||||
connectionWaitTimeout time.Duration
|
||||
}
|
||||
|
||||
// New creates a new HTTPFetcher with SSRF protection.
|
||||
@@ -212,15 +195,10 @@ func New(config *Config) *HTTPFetcher {
|
||||
client: client,
|
||||
config: config,
|
||||
hostSems: make(map[string]chan struct{}),
|
||||
allHostsSemaphore: make(chan struct{}, config.MaxConnections),
|
||||
connectionWaitTimeout: ConnectionWaitTimeout,
|
||||
}
|
||||
}
|
||||
|
||||
// Fetch retrieves content from the given URL with SSRF protection. When
|
||||
// MaxConnections fetches are already in progress, it waits up to
|
||||
// ConnectionWaitTimeout for one to finish, then fails with
|
||||
// ErrTooManyConnections.
|
||||
// Fetch retrieves content from the given URL with SSRF protection.
|
||||
func (f *HTTPFetcher) Fetch(ctx context.Context, url string) (*FetchResult, error) {
|
||||
// Validate URL before making request
|
||||
err := validateURL(ctx, url, f.config.AllowHTTP)
|
||||
@@ -228,17 +206,24 @@ func (f *HTTPFetcher) Fetch(ctx context.Context, url string) (*FetchResult, erro
|
||||
return nil, err
|
||||
}
|
||||
|
||||
release, err := f.acquireConnection(ctx, extractHost(url))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
// Extract host for rate limiting
|
||||
host := extractHost(url)
|
||||
|
||||
// Acquire semaphore slot for this host
|
||||
sem := f.getHostSemaphore(host)
|
||||
select {
|
||||
case sem <- struct{}{}:
|
||||
// Acquired slot
|
||||
case <-ctx.Done():
|
||||
return nil, ctx.Err()
|
||||
}
|
||||
|
||||
// If we fail before returning a result, release the connection
|
||||
// If we fail before returning a result, release the slot
|
||||
success := false
|
||||
|
||||
defer func() {
|
||||
if !success {
|
||||
release()
|
||||
<-sem
|
||||
}
|
||||
}()
|
||||
|
||||
@@ -282,52 +267,17 @@ func (f *HTTPFetcher) Fetch(ctx context.Context, url string) (*FetchResult, erro
|
||||
return nil, fmt.Errorf("upstream request failed: %w", err)
|
||||
}
|
||||
|
||||
result, err := f.buildResult(resp, remoteAddr, fetchDuration, release)
|
||||
result, err := f.buildResult(resp, remoteAddr, fetchDuration, sem)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// Mark success so defer doesn't release the connection; closing the
|
||||
// result's Content does
|
||||
// Mark success so defer doesn't release the semaphore
|
||||
success = true
|
||||
|
||||
return result, nil
|
||||
}
|
||||
|
||||
// acquireConnection takes a slot for host, then one of the slots shared by
|
||||
// all hosts, and returns the func that gives both back. The host's slot
|
||||
// comes first, so fetches queued for one busy host hold no shared slot.
|
||||
// Only the wait for a shared slot is bounded: after connectionWaitTimeout
|
||||
// it fails with ErrTooManyConnections.
|
||||
func (f *HTTPFetcher) acquireConnection(
|
||||
ctx context.Context, host string,
|
||||
) (func(), error) {
|
||||
hostSem := f.getHostSemaphore(host)
|
||||
|
||||
select {
|
||||
case hostSem <- struct{}{}:
|
||||
case <-ctx.Done():
|
||||
return nil, ctx.Err()
|
||||
}
|
||||
|
||||
select {
|
||||
case f.allHostsSemaphore <- struct{}{}:
|
||||
case <-time.After(f.connectionWaitTimeout):
|
||||
<-hostSem
|
||||
|
||||
return nil, ErrTooManyConnections
|
||||
case <-ctx.Done():
|
||||
<-hostSem
|
||||
|
||||
return nil, ctx.Err()
|
||||
}
|
||||
|
||||
return func() {
|
||||
<-hostSem
|
||||
<-f.allHostsSemaphore
|
||||
}, nil
|
||||
}
|
||||
|
||||
// getHostSemaphore returns the semaphore for a host, creating it if necessary.
|
||||
func (f *HTTPFetcher) getHostSemaphore(host string) chan struct{} {
|
||||
f.hostSemMu.Lock()
|
||||
@@ -343,12 +293,12 @@ func (f *HTTPFetcher) getHostSemaphore(host string) chan struct{} {
|
||||
}
|
||||
|
||||
// buildResult validates the upstream response and assembles a FetchResult
|
||||
// whose Content calls release when closed.
|
||||
// whose Content releases the host semaphore slot when closed.
|
||||
func (f *HTTPFetcher) buildResult(
|
||||
resp *http.Response,
|
||||
remoteAddr string,
|
||||
fetchDuration time.Duration,
|
||||
release func(),
|
||||
sem chan struct{},
|
||||
) (*FetchResult, error) {
|
||||
// Extract HTTP version (strip "HTTP/" prefix)
|
||||
httpVersion := strings.TrimPrefix(resp.Proto, "HTTP/")
|
||||
@@ -383,7 +333,7 @@ func (f *HTTPFetcher) buildResult(
|
||||
}
|
||||
|
||||
return &FetchResult{
|
||||
Content: &semaphoreReleasingReadCloser{limitedBody, resp.Body, release},
|
||||
Content: &semaphoreReleasingReadCloser{limitedBody, resp.Body, sem},
|
||||
ContentLength: resp.ContentLength,
|
||||
ContentType: contentType,
|
||||
Headers: resp.Header,
|
||||
@@ -624,18 +574,17 @@ func (r *limitedReader) Read(p []byte) (int, error) {
|
||||
return n, err
|
||||
}
|
||||
|
||||
// semaphoreReleasingReadCloser releases the fetch's connection slots when
|
||||
// closed.
|
||||
// semaphoreReleasingReadCloser releases a semaphore slot when closed.
|
||||
type semaphoreReleasingReadCloser struct {
|
||||
*limitedReader
|
||||
|
||||
closer io.Closer
|
||||
release func()
|
||||
sem chan struct{}
|
||||
}
|
||||
|
||||
func (r *semaphoreReleasingReadCloser) Close() error {
|
||||
err := r.closer.Close()
|
||||
r.release()
|
||||
<-r.sem // Release semaphore slot
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -1,165 +0,0 @@
|
||||
package httpfetcher
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"net"
|
||||
"strconv"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// imageURLOnPort is the fake upstream's image route on testPublicHost at
|
||||
// port. Each port is a different host to the per-host limit, while the test
|
||||
// dialer sends every port to the one test server.
|
||||
func imageURLOnPort(port int) string {
|
||||
return "http://" + net.JoinHostPort(testPublicHost, strconv.Itoa(port)) +
|
||||
"/image"
|
||||
}
|
||||
|
||||
func TestDefaultConfigMaxConnections(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
if got := DefaultConfig().MaxConnections; got != DefaultMaxConnections {
|
||||
t.Errorf("MaxConnections = %d, want %d", got, DefaultMaxConnections)
|
||||
}
|
||||
}
|
||||
|
||||
// TestFetchLimitsConnectionsToAllHostsTogether checks that MaxConnections
|
||||
// counts the fetches to every host together, apart from the per-host
|
||||
// limit: with MaxConnections at 2 and two responses open from two hosts, a
|
||||
// fetch from a third host, which has nothing open, waits the whole wait
|
||||
// timeout and fails with ErrTooManyConnections. Closing one response lets
|
||||
// it through.
|
||||
func TestFetchLimitsConnectionsToAllHostsTogether(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
srv := startUpstream(t)
|
||||
|
||||
cfg := DefaultConfig()
|
||||
cfg.MaxConnections = 2
|
||||
|
||||
f, _ := newServerFetcher(t, srv, cfg)
|
||||
f.connectionWaitTimeout = 100 * time.Millisecond
|
||||
|
||||
first, err := f.Fetch(testContext(t), imageURLOnPort(81))
|
||||
if err != nil {
|
||||
t.Fatalf("first Fetch() error = %v", err)
|
||||
}
|
||||
|
||||
second, err := f.Fetch(testContext(t), imageURLOnPort(82))
|
||||
if err != nil {
|
||||
t.Fatalf("second Fetch() error = %v", err)
|
||||
}
|
||||
|
||||
defer func() { _ = second.Content.Close() }()
|
||||
|
||||
start := time.Now()
|
||||
|
||||
_, err = f.Fetch(testContext(t), imageURLOnPort(83))
|
||||
if !errors.Is(err, ErrTooManyConnections) {
|
||||
t.Fatalf("third Fetch() error = %v, want ErrTooManyConnections", err)
|
||||
}
|
||||
|
||||
if waited := time.Since(start); waited < f.connectionWaitTimeout {
|
||||
t.Errorf("third Fetch() failed after %v, before waiting %v",
|
||||
waited, f.connectionWaitTimeout)
|
||||
}
|
||||
|
||||
if held := semLen(f, testPublicHost+":83"); held != 0 {
|
||||
t.Errorf("the refused fetch kept its host's slot: %d held", held)
|
||||
}
|
||||
|
||||
err = first.Content.Close()
|
||||
if err != nil {
|
||||
t.Fatalf("close first body: %v", err)
|
||||
}
|
||||
|
||||
third, err := f.Fetch(testContext(t), imageURLOnPort(83))
|
||||
if err != nil {
|
||||
t.Fatalf("Fetch() after a response was closed: error = %v", err)
|
||||
}
|
||||
|
||||
_ = third.Content.Close()
|
||||
}
|
||||
|
||||
// TestFetchFreesHostSlotWhenContextEndsWaitingForConnection checks that a
|
||||
// fetch whose request context ends while it waits for a connection shared
|
||||
// by all hosts gives its host's slot back. With MaxConnections at 1 and one
|
||||
// response open, a fetch from another host takes that host's slot and waits;
|
||||
// its context ends long before the 10 second wait timeout.
|
||||
func TestFetchFreesHostSlotWhenContextEndsWaitingForConnection(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
srv := startUpstream(t)
|
||||
|
||||
cfg := DefaultConfig()
|
||||
cfg.MaxConnections = 1
|
||||
|
||||
f, _ := newServerFetcher(t, srv, cfg)
|
||||
|
||||
first, err := f.Fetch(testContext(t), imageURLOnPort(81))
|
||||
if err != nil {
|
||||
t.Fatalf("first Fetch() error = %v", err)
|
||||
}
|
||||
|
||||
defer func() { _ = first.Content.Close() }()
|
||||
|
||||
ctx, cancel := context.WithTimeout(t.Context(), 100*time.Millisecond)
|
||||
defer cancel()
|
||||
|
||||
_, err = f.Fetch(ctx, imageURLOnPort(82))
|
||||
if !errors.Is(err, context.DeadlineExceeded) {
|
||||
t.Fatalf("second Fetch() error = %v, want context.DeadlineExceeded", err)
|
||||
}
|
||||
|
||||
if held := semLen(f, testPublicHost+":82"); held != 0 {
|
||||
t.Errorf("the fetch kept its host's slot after its context ended: "+
|
||||
"%d held", held)
|
||||
}
|
||||
}
|
||||
|
||||
// TestFetchReleasesConnectionOnError checks that a fetch that fails after
|
||||
// taking its connection gives it back: with MaxConnections at 1, the slot
|
||||
// must be free after the failure and the next fetch must succeed.
|
||||
func TestFetchReleasesConnectionOnError(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
cases := []struct {
|
||||
name string
|
||||
url string
|
||||
want error
|
||||
}{
|
||||
{"upstream answers 500", upstreamURL("/status/500"), ErrUpstreamError},
|
||||
{"upstream sends HTML", upstreamURL("/html"), ErrInvalidContentType},
|
||||
// 198.51.100.7 (TEST-NET-2) passes the SSRF checks, and the test
|
||||
// dialer refuses every host but testPublicHost.
|
||||
{"connecting fails", "http://198.51.100.7/image", errUnexpectedDial},
|
||||
}
|
||||
|
||||
for _, tc := range cases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
srv := startUpstream(t)
|
||||
|
||||
cfg := DefaultConfig()
|
||||
cfg.MaxConnections = 1
|
||||
|
||||
f, _ := newServerFetcher(t, srv, cfg)
|
||||
f.connectionWaitTimeout = 100 * time.Millisecond
|
||||
|
||||
_, err := f.Fetch(testContext(t), tc.url)
|
||||
if !errors.Is(err, tc.want) {
|
||||
t.Fatalf("Fetch() error = %v, want %v", err, tc.want)
|
||||
}
|
||||
|
||||
if held := len(f.allHostsSemaphore); held != 0 {
|
||||
t.Fatalf("connection still held after the error: %d held", held)
|
||||
}
|
||||
|
||||
res := fetchImage(t, f, "/image")
|
||||
_ = res.Content.Close()
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -7,9 +7,7 @@ import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"runtime"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/davidbyttow/govips/v2/vips"
|
||||
)
|
||||
@@ -19,21 +17,11 @@ import (
|
||||
//nolint:gochecknoglobals // package-level sync.Once for one-time vips init
|
||||
var vipsOnce sync.Once
|
||||
|
||||
// initVips initializes libvips with quiet logging, one worker thread per
|
||||
// image and no operation cache. Process already works on one image per CPU
|
||||
// by default, so more threads per image would only compete for the CPUs.
|
||||
// Each request decodes different source bytes, so the operation cache
|
||||
// would rarely be hit and would hold memory outside MaxConcurrentProcessing;
|
||||
// repeated requests are served from pixa's disk cache instead.
|
||||
// initVips initializes libvips with quiet logging.
|
||||
func initVips() {
|
||||
vipsOnce.Do(func() {
|
||||
vips.LoggingSettings(nil, vips.LogLevelError)
|
||||
vips.Startup(&vips.Config{
|
||||
ConcurrencyLevel: 1,
|
||||
MaxCacheSize: 0,
|
||||
MaxCacheMem: 0,
|
||||
MaxCacheFiles: 0,
|
||||
})
|
||||
vips.Startup(nil)
|
||||
})
|
||||
}
|
||||
|
||||
@@ -118,23 +106,9 @@ var ErrInputDataTooLarge = errors.New("input data exceeds maximum allowed size")
|
||||
// not supported.
|
||||
var ErrUnsupportedOutputFormat = errors.New("unsupported output format")
|
||||
|
||||
// ErrTooManyImages is returned when MaxConcurrentProcessing images are being
|
||||
// processed and none finishes within ProcessingWaitTimeout.
|
||||
var ErrTooManyImages = errors.New("too many images being processed at once")
|
||||
|
||||
// ProcessingWaitTimeout is how long Process waits for a free slot when
|
||||
// MaxConcurrentProcessing images are already being processed.
|
||||
const ProcessingWaitTimeout = 10 * time.Second
|
||||
|
||||
// ImageProcessor implements image transformation using libvips via govips.
|
||||
type ImageProcessor struct {
|
||||
maxInputBytes int64
|
||||
// processingSemaphore has one slot per image that may be processed at
|
||||
// once. Process holds a slot from before it reads its input until it
|
||||
// returns, so the input, the decoded image and the output all count.
|
||||
processingSemaphore chan struct{}
|
||||
// processingWaitTimeout is ProcessingWaitTimeout; tests shorten it.
|
||||
processingWaitTimeout time.Duration
|
||||
}
|
||||
|
||||
// Params holds configuration for creating an ImageProcessor.
|
||||
@@ -143,9 +117,6 @@ type Params struct {
|
||||
// MaxInputBytes is the maximum allowed input size in bytes.
|
||||
// If <= 0, DefaultMaxInputBytes is used.
|
||||
MaxInputBytes int64
|
||||
// MaxConcurrentProcessing is the most images processed at once.
|
||||
// If <= 0, the number of CPUs Go uses (runtime.GOMAXPROCS(0)) is used.
|
||||
MaxConcurrentProcessing int
|
||||
}
|
||||
|
||||
// New creates a new image processor with the given parameters.
|
||||
@@ -158,34 +129,17 @@ func New(params Params) *ImageProcessor {
|
||||
maxInputBytes = DefaultMaxInputBytes
|
||||
}
|
||||
|
||||
maxConcurrentProcessing := params.MaxConcurrentProcessing
|
||||
if maxConcurrentProcessing <= 0 {
|
||||
maxConcurrentProcessing = runtime.GOMAXPROCS(0)
|
||||
}
|
||||
|
||||
return &ImageProcessor{
|
||||
maxInputBytes: maxInputBytes,
|
||||
processingSemaphore: make(chan struct{}, maxConcurrentProcessing),
|
||||
processingWaitTimeout: ProcessingWaitTimeout,
|
||||
}
|
||||
}
|
||||
|
||||
// Process transforms an image according to the request. When
|
||||
// MaxConcurrentProcessing images are already being processed, it waits up
|
||||
// to ProcessingWaitTimeout for one to finish, then fails with
|
||||
// ErrTooManyImages.
|
||||
// Process transforms an image according to the request.
|
||||
func (p *ImageProcessor) Process(
|
||||
ctx context.Context,
|
||||
_ context.Context,
|
||||
input io.Reader,
|
||||
req *Request,
|
||||
) (*Result, error) {
|
||||
release, err := p.acquireSlot(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
defer release()
|
||||
|
||||
// Read input with a size limit to prevent unbounded memory consumption.
|
||||
// We read at most maxInputBytes+1 so we can detect if the input exceeds
|
||||
// the limit without consuming additional memory.
|
||||
@@ -331,29 +285,6 @@ func FormatToMIME(format Format) string {
|
||||
}
|
||||
}
|
||||
|
||||
// acquireSlot takes a slot in processingSemaphore, waiting at most
|
||||
// processingWaitTimeout for one to free up, and returns the func that gives
|
||||
// it back. A free slot is taken even when ctx has ended; only the wait for
|
||||
// one stops when ctx ends, as the rest of Process does not check ctx.
|
||||
func (p *ImageProcessor) acquireSlot(ctx context.Context) (func(), error) {
|
||||
release := func() { <-p.processingSemaphore }
|
||||
|
||||
select {
|
||||
case p.processingSemaphore <- struct{}{}:
|
||||
return release, nil
|
||||
default:
|
||||
}
|
||||
|
||||
select {
|
||||
case p.processingSemaphore <- struct{}{}:
|
||||
return release, nil
|
||||
case <-time.After(p.processingWaitTimeout):
|
||||
return nil, ErrTooManyImages
|
||||
case <-ctx.Done():
|
||||
return nil, ctx.Err()
|
||||
}
|
||||
}
|
||||
|
||||
// detectFormat returns the format string from a vips image.
|
||||
func (p *ImageProcessor) detectFormat(img *vips.ImageRef) string {
|
||||
format := img.Format()
|
||||
|
||||
@@ -1,302 +0,0 @@
|
||||
package imageprocessor
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"errors"
|
||||
"io"
|
||||
"runtime"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"testing/iotest"
|
||||
"time"
|
||||
)
|
||||
|
||||
// errTestReadFailed is the error the unreadable test input returns.
|
||||
var errTestReadFailed = errors.New("test input cannot be read")
|
||||
|
||||
// readingCounter counts the Process calls reading their input at the same
|
||||
// time and remembers the most there ever were.
|
||||
type readingCounter struct {
|
||||
mu sync.Mutex
|
||||
reading int
|
||||
most int
|
||||
}
|
||||
|
||||
func (c *readingCounter) start() {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
|
||||
c.reading++
|
||||
c.most = max(c.most, c.reading)
|
||||
}
|
||||
|
||||
func (c *readingCounter) stop() {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
|
||||
c.reading--
|
||||
}
|
||||
|
||||
func (c *readingCounter) mostReading() int {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
|
||||
return c.most
|
||||
}
|
||||
|
||||
// gatedReader is a Process input. Its first Read counts the call in,
|
||||
// reports it on entered and blocks until gate is closed; it counts the call
|
||||
// out when it returns io.EOF. Process reads its input only while it holds a
|
||||
// processing slot, so the count never goes above MaxConcurrentProcessing.
|
||||
type gatedReader struct {
|
||||
data *bytes.Reader
|
||||
gate <-chan struct{}
|
||||
entered chan<- struct{}
|
||||
counter *readingCounter
|
||||
started bool
|
||||
}
|
||||
|
||||
func (r *gatedReader) Read(p []byte) (int, error) {
|
||||
if !r.started {
|
||||
r.started = true
|
||||
r.counter.start()
|
||||
|
||||
r.entered <- struct{}{}
|
||||
|
||||
<-r.gate
|
||||
}
|
||||
|
||||
n, err := r.data.Read(p)
|
||||
if errors.Is(err, io.EOF) {
|
||||
r.counter.stop()
|
||||
}
|
||||
|
||||
return n, err
|
||||
}
|
||||
|
||||
// smallJPEGRequest asks for a 5x5 JPEG.
|
||||
func smallJPEGRequest() *Request {
|
||||
return &Request{
|
||||
Size: Size{Width: 5, Height: 5},
|
||||
Format: FormatJPEG,
|
||||
Quality: 85,
|
||||
FitMode: FitCover,
|
||||
}
|
||||
}
|
||||
|
||||
// processInBackground runs Process on reader in a new goroutine and sends
|
||||
// its error on results.
|
||||
func processInBackground(
|
||||
proc *ImageProcessor, reader *gatedReader, results chan<- error,
|
||||
) {
|
||||
go func() {
|
||||
result, err := proc.Process(context.Background(), reader, smallJPEGRequest())
|
||||
if err == nil {
|
||||
_ = result.Content.Close()
|
||||
}
|
||||
|
||||
results <- err
|
||||
}()
|
||||
}
|
||||
|
||||
// waitForEntries fails the test unless count Process calls report on
|
||||
// entered within a few seconds.
|
||||
func waitForEntries(t *testing.T, entered <-chan struct{}, count int) {
|
||||
t.Helper()
|
||||
|
||||
for range count {
|
||||
select {
|
||||
case <-entered:
|
||||
case <-time.After(5 * time.Second):
|
||||
t.Fatal("Process calls did not start reading their input")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestNewDefaultsMaxConcurrentProcessingToCPUs(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
for _, limit := range []int{0, -1} {
|
||||
proc := New(Params{MaxConcurrentProcessing: limit})
|
||||
if got := cap(proc.processingSemaphore); got != runtime.GOMAXPROCS(0) {
|
||||
t.Errorf("MaxConcurrentProcessing %d: %d slots, want %d, one per CPU",
|
||||
limit, got, runtime.GOMAXPROCS(0))
|
||||
}
|
||||
}
|
||||
|
||||
proc := New(Params{MaxConcurrentProcessing: 3})
|
||||
if got := cap(proc.processingSemaphore); got != 3 {
|
||||
t.Errorf("MaxConcurrentProcessing 3: %d slots, want 3", got)
|
||||
}
|
||||
}
|
||||
|
||||
// TestProcessNeverExceedsMaxConcurrentProcessing starts more Process calls
|
||||
// than MaxConcurrentProcessing allows and holds the first ones inside
|
||||
// Process until the test lets them go. No more than the limit may be
|
||||
// working at once, and the calls held back must wait for a slot and then
|
||||
// succeed.
|
||||
func TestProcessNeverExceedsMaxConcurrentProcessing(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
const (
|
||||
limit = 2
|
||||
calls = 6
|
||||
)
|
||||
|
||||
proc := New(Params{MaxConcurrentProcessing: limit})
|
||||
input := createTestJPEG(t, 50, 50)
|
||||
|
||||
counter := &readingCounter{}
|
||||
gate := make(chan struct{})
|
||||
entered := make(chan struct{}, calls)
|
||||
results := make(chan error, calls)
|
||||
|
||||
openGate := sync.OnceFunc(func() { close(gate) })
|
||||
t.Cleanup(openGate)
|
||||
|
||||
for range calls {
|
||||
processInBackground(proc, &gatedReader{
|
||||
data: bytes.NewReader(input), gate: gate, entered: entered,
|
||||
counter: counter,
|
||||
}, results)
|
||||
}
|
||||
|
||||
waitForEntries(t, entered, limit)
|
||||
|
||||
// A call beyond the limit would start reading its input now.
|
||||
select {
|
||||
case <-entered:
|
||||
t.Fatalf("a Process call started while %d were already working", limit)
|
||||
case <-time.After(100 * time.Millisecond):
|
||||
}
|
||||
|
||||
openGate()
|
||||
|
||||
for range calls {
|
||||
err := <-results
|
||||
if err != nil {
|
||||
t.Errorf("Process() error = %v, want nil once a slot is free", err)
|
||||
}
|
||||
}
|
||||
|
||||
if most := counter.mostReading(); most > limit {
|
||||
t.Errorf("%d Process calls worked at once, want at most %d", most, limit)
|
||||
}
|
||||
}
|
||||
|
||||
// TestProcessWaitsThenFailsWhenNoSlotFrees holds the only slot and checks
|
||||
// that another call waits the whole wait timeout, then fails with
|
||||
// ErrTooManyImages instead of processing anyway.
|
||||
func TestProcessWaitsThenFailsWhenNoSlotFrees(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
proc := New(Params{MaxConcurrentProcessing: 1})
|
||||
proc.processingWaitTimeout = 100 * time.Millisecond
|
||||
|
||||
input := createTestJPEG(t, 10, 10)
|
||||
|
||||
gate := make(chan struct{})
|
||||
entered := make(chan struct{}, 1)
|
||||
held := make(chan error, 1)
|
||||
|
||||
openGate := sync.OnceFunc(func() { close(gate) })
|
||||
t.Cleanup(openGate)
|
||||
|
||||
processInBackground(proc, &gatedReader{
|
||||
data: bytes.NewReader(input), gate: gate, entered: entered,
|
||||
counter: &readingCounter{},
|
||||
}, held)
|
||||
waitForEntries(t, entered, 1)
|
||||
|
||||
start := time.Now()
|
||||
|
||||
_, err := proc.Process(context.Background(), bytes.NewReader(input),
|
||||
smallJPEGRequest())
|
||||
if !errors.Is(err, ErrTooManyImages) {
|
||||
t.Fatalf("Process() error = %v, want ErrTooManyImages", err)
|
||||
}
|
||||
|
||||
if waited := time.Since(start); waited < proc.processingWaitTimeout {
|
||||
t.Errorf("Process() failed after %v, before waiting %v",
|
||||
waited, proc.processingWaitTimeout)
|
||||
}
|
||||
|
||||
openGate()
|
||||
|
||||
err = <-held
|
||||
if err != nil {
|
||||
t.Errorf("Process() holding the slot: error = %v, want nil", err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestProcessReleasesSlotOnError checks that Process gives its slot back
|
||||
// when it fails, whether it fails early or late: with one slot, the slot
|
||||
// must be free after the failure and the next call must succeed.
|
||||
func TestProcessReleasesSlotOnError(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
valid := createTestJPEG(t, 10, 10)
|
||||
|
||||
unsupported := smallJPEGRequest()
|
||||
unsupported.Format = "bmp"
|
||||
|
||||
cases := []struct {
|
||||
name string
|
||||
input io.Reader
|
||||
req *Request
|
||||
// want is the error Process must return; nil means any error.
|
||||
want error
|
||||
}{
|
||||
{
|
||||
name: "input cannot be read",
|
||||
input: iotest.ErrReader(errTestReadFailed),
|
||||
req: smallJPEGRequest(),
|
||||
want: errTestReadFailed,
|
||||
},
|
||||
{
|
||||
name: "input over the byte limit",
|
||||
input: bytes.NewReader(createTestJPEG(t, 800, 600)),
|
||||
req: smallJPEGRequest(),
|
||||
want: ErrInputDataTooLarge,
|
||||
},
|
||||
{
|
||||
name: "input not an image",
|
||||
input: strings.NewReader("not an image"),
|
||||
req: smallJPEGRequest(),
|
||||
},
|
||||
{
|
||||
name: "output format not supported",
|
||||
input: bytes.NewReader(valid),
|
||||
req: unsupported,
|
||||
want: ErrUnsupportedOutputFormat,
|
||||
},
|
||||
}
|
||||
|
||||
for _, tc := range cases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
proc := New(Params{MaxInputBytes: 4096, MaxConcurrentProcessing: 1})
|
||||
proc.processingWaitTimeout = 100 * time.Millisecond
|
||||
|
||||
_, err := proc.Process(context.Background(), tc.input, tc.req)
|
||||
if err == nil || (tc.want != nil && !errors.Is(err, tc.want)) {
|
||||
t.Fatalf("Process() error = %v, want %v", err, tc.want)
|
||||
}
|
||||
|
||||
if held := len(proc.processingSemaphore); held != 0 {
|
||||
t.Fatalf("slot still held after the error: %d held", held)
|
||||
}
|
||||
|
||||
result, err := proc.Process(context.Background(), bytes.NewReader(valid),
|
||||
smallJPEGRequest())
|
||||
if err != nil {
|
||||
t.Fatalf("Process() after the error = %v, want nil", err)
|
||||
}
|
||||
|
||||
_ = result.Content.Close()
|
||||
})
|
||||
}
|
||||
}
|
||||
+22
-43
@@ -159,13 +159,15 @@ type LookupResult struct {
|
||||
CacheStatus CacheStatus
|
||||
}
|
||||
|
||||
// Lookup checks if a processed variant exists on disk. Hits touch the
|
||||
// variant's LRU timestamp; a disabled cache always misses.
|
||||
// Lookup checks if a processed variant exists on disk: a variant held
|
||||
// in metaCache counts as present without a check of the disk. Hits
|
||||
// touch the variant's LRU timestamp; a disabled cache always misses.
|
||||
func (c *Cache) Lookup(ctx context.Context, req *ImageRequest) (*LookupResult, error) {
|
||||
cacheKey := CacheKey(req)
|
||||
|
||||
// Check variant storage directly - no DB needed for cache hits
|
||||
if !c.disabled && c.variants.Exists(cacheKey) {
|
||||
// Check memory, then variant storage - no DB needed for cache hits
|
||||
if !c.disabled &&
|
||||
(c.metaCache.Contains(cacheKey) || c.variants.Exists(cacheKey)) {
|
||||
c.touchVariant(ctx, cacheKey)
|
||||
|
||||
return &LookupResult{
|
||||
@@ -184,19 +186,14 @@ func (c *Cache) Lookup(ctx context.Context, req *ImageRequest) (*LookupResult, e
|
||||
|
||||
// GetVariant returns a reader, size, and content type for a cached
|
||||
// variant. The content type comes from metaCache, or else from the
|
||||
// variant's .meta file and is then kept in metaCache. A variant with
|
||||
// no .meta file is served as application/octet-stream, which is not
|
||||
// kept.
|
||||
// variant's .meta file and is then kept in metaCache.
|
||||
func (c *Cache) GetVariant(cacheKey VariantKey) (io.ReadCloser, int64, string, error) {
|
||||
if c.disabled {
|
||||
return nil, 0, "", ErrNotFound
|
||||
}
|
||||
|
||||
contentType, known := c.metaCache.Get(cacheKey)
|
||||
if !known {
|
||||
return c.loadVariantWithMeta(cacheKey)
|
||||
}
|
||||
|
||||
if known {
|
||||
reader, size, err := c.variants.LoadWithSize(cacheKey)
|
||||
if err != nil {
|
||||
// The file is gone, e.g. deleted outside pixa
|
||||
@@ -205,6 +202,16 @@ func (c *Cache) GetVariant(cacheKey VariantKey) (io.ReadCloser, int64, string, e
|
||||
return nil, 0, "", err
|
||||
}
|
||||
|
||||
return reader, size, contentType, nil
|
||||
}
|
||||
|
||||
reader, size, contentType, err := c.variants.LoadWithMeta(cacheKey)
|
||||
if err != nil {
|
||||
return nil, 0, "", err
|
||||
}
|
||||
|
||||
c.metaCache.Add(cacheKey, contentType)
|
||||
|
||||
return reader, size, contentType, nil
|
||||
}
|
||||
|
||||
@@ -407,16 +414,13 @@ func (c *Cache) GetSourceMetadataID(
|
||||
return id, nil
|
||||
}
|
||||
|
||||
// GetSourceContent returns a reader for cached source content by its hash,
|
||||
// and the content's size in bytes.
|
||||
func (c *Cache) GetSourceContent(
|
||||
contentHash ContentHash,
|
||||
) (io.ReadCloser, int64, error) {
|
||||
// GetSourceContent returns a reader for cached source content by its hash.
|
||||
func (c *Cache) GetSourceContent(contentHash ContentHash) (io.ReadCloser, error) {
|
||||
if c.disabled {
|
||||
return nil, 0, ErrNotFound
|
||||
return nil, ErrNotFound
|
||||
}
|
||||
|
||||
return c.srcContent.LoadWithSize(contentHash)
|
||||
return c.srcContent.Load(contentHash)
|
||||
}
|
||||
|
||||
// CleanExpired removes expired entries from the cache.
|
||||
@@ -523,31 +527,6 @@ func (c *Cache) IncrementTransformCount(ctx context.Context) {
|
||||
}
|
||||
}
|
||||
|
||||
// loadVariantWithMeta is GetVariant for a variant metaCache does not
|
||||
// hold: it reads the content type from the variant's .meta file and
|
||||
// keeps it in metaCache, unless a StoreVariant has put one there
|
||||
// meanwhile, as the store's is newer. A read that finds no .meta file,
|
||||
// as one can between a store's writing of the variant file and of its
|
||||
// .meta file, serves application/octet-stream and keeps nothing, so
|
||||
// metaCache only ever holds a type read from a .meta file or passed to
|
||||
// StoreVariant.
|
||||
func (c *Cache) loadVariantWithMeta(
|
||||
cacheKey VariantKey,
|
||||
) (io.ReadCloser, int64, string, error) {
|
||||
reader, size, contentType, err := c.variants.LoadWithMeta(cacheKey)
|
||||
if err != nil {
|
||||
return nil, 0, "", err
|
||||
}
|
||||
|
||||
if contentType == "" {
|
||||
return reader, size, fallbackContentType, nil
|
||||
}
|
||||
|
||||
c.metaCache.ContainsOrAdd(cacheKey, contentType)
|
||||
|
||||
return reader, size, contentType, nil
|
||||
}
|
||||
|
||||
// writeMetadataSidecar writes the JSON metadata sidecar of a stored source.
|
||||
// A failure is logged and is otherwise non-fatal; the metadata is in the
|
||||
// database.
|
||||
|
||||
@@ -39,8 +39,8 @@ const tempFilePrefix = ".tmp-"
|
||||
// to each variant file.
|
||||
const variantMetaSuffix = ".meta"
|
||||
|
||||
// fallbackContentType is the content type given to a variant file that
|
||||
// has no readable .meta sidecar, when it is served or reconciled.
|
||||
// fallbackContentType is recorded when a reconciled variant file has
|
||||
// no readable .meta sidecar.
|
||||
const fallbackContentType = "application/octet-stream"
|
||||
|
||||
// UsageBytes returns the total number of bytes of cache content
|
||||
|
||||
@@ -1,125 +0,0 @@
|
||||
package imgcache
|
||||
|
||||
import (
|
||||
"image/color"
|
||||
"image/jpeg"
|
||||
"io"
|
||||
"os"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"sneak.berlin/go/pixa/internal/imageprocessor"
|
||||
)
|
||||
|
||||
// widthOnlyRequest asks for the test photo at width, its height scaled to
|
||||
// keep the photo's aspect ratio.
|
||||
func widthOnlyRequest(fixtures *TestFixtures, width int) *ImageRequest {
|
||||
return &ImageRequest{
|
||||
SourceHost: fixtures.GoodHost,
|
||||
SourcePath: testPathPhoto,
|
||||
Size: Size{Width: width},
|
||||
Format: FormatJPEG,
|
||||
Quality: 85,
|
||||
FitMode: FitCover,
|
||||
}
|
||||
}
|
||||
|
||||
// holdProcessingSlot takes one of proc's processing slots and returns the
|
||||
// func that gives it back. Process takes its slot before it reads its input,
|
||||
// so once it has read a byte from the pipe it holds the slot, until the pipe
|
||||
// is closed.
|
||||
func holdProcessingSlot(
|
||||
t *testing.T, proc *imageprocessor.ImageProcessor,
|
||||
) func() {
|
||||
t.Helper()
|
||||
|
||||
input, feed := io.Pipe()
|
||||
|
||||
go func() {
|
||||
_, _ = proc.Process(t.Context(), input, &imageprocessor.Request{})
|
||||
}()
|
||||
|
||||
_, err := feed.Write([]byte{0})
|
||||
if err != nil {
|
||||
t.Fatalf("Process call to hold the slot did not start: %v", err)
|
||||
}
|
||||
|
||||
release := func() { _ = feed.Close() }
|
||||
t.Cleanup(release)
|
||||
|
||||
return release
|
||||
}
|
||||
|
||||
// TestService_Get_WaitsForSlotBeforeReadingCachedSource checks that a
|
||||
// request whose source is cached holds none of it while it waits for a
|
||||
// processing slot: it reads the cached file only once it has a slot. With
|
||||
// the only slot held, a request for a new width of the cached 100x100 photo
|
||||
// waits; the cached file is then rewritten as a 100x50 image before the slot
|
||||
// is freed, so the request must answer with that image scaled to 40x20.
|
||||
func TestService_Get_WaitsForSlotBeforeReadingCachedSource(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
svc, fixtures := SetupTestService(t)
|
||||
svc.processor = imageprocessor.New(
|
||||
imageprocessor.Params{MaxConcurrentProcessing: 1},
|
||||
)
|
||||
|
||||
// A first request caches the photo as a source.
|
||||
resp, err := svc.Get(t.Context(), widthOnlyRequest(fixtures, 50))
|
||||
if err != nil {
|
||||
t.Fatalf("first Get() error = %v", err)
|
||||
}
|
||||
|
||||
_ = resp.Content.Close()
|
||||
|
||||
contentHash, _, err := svc.cache.LookupSource(t.Context(),
|
||||
widthOnlyRequest(fixtures, 50))
|
||||
if err != nil || contentHash == "" {
|
||||
t.Fatalf("LookupSource() = %q, %v; want the cached source",
|
||||
contentHash, err)
|
||||
}
|
||||
|
||||
release := holdProcessingSlot(t, svc.processor)
|
||||
|
||||
var (
|
||||
waited *ImageResponse
|
||||
waitedErr error
|
||||
)
|
||||
|
||||
done := make(chan struct{})
|
||||
|
||||
go func() {
|
||||
defer close(done)
|
||||
|
||||
waited, waitedErr = svc.Get(t.Context(), widthOnlyRequest(fixtures, 40))
|
||||
}()
|
||||
|
||||
// Give the request time to reach the slot: had it read the cached source
|
||||
// before waiting, it would have read it by now.
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
|
||||
err = os.WriteFile(svc.cache.srcContent.hashToPath(contentHash),
|
||||
generateTestJPEG(t, 100, 50, color.RGBA{0, 0, 255, 255}), 0o600)
|
||||
if err != nil {
|
||||
t.Fatalf("failed to rewrite the cached source: %v", err)
|
||||
}
|
||||
|
||||
release()
|
||||
<-done
|
||||
|
||||
if waitedErr != nil {
|
||||
t.Fatalf("Get() error = %v", waitedErr)
|
||||
}
|
||||
|
||||
defer func() { _ = waited.Content.Close() }()
|
||||
|
||||
output, err := jpeg.DecodeConfig(waited.Content)
|
||||
if err != nil {
|
||||
t.Fatalf("failed to decode the response: %v", err)
|
||||
}
|
||||
|
||||
if output.Width != 40 || output.Height != 20 {
|
||||
t.Errorf("response is %dx%d, want 40x20: the request read the cached "+
|
||||
"source before it had a processing slot", output.Width, output.Height)
|
||||
}
|
||||
}
|
||||
@@ -2,7 +2,6 @@ package imgcache
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
@@ -84,16 +83,6 @@ func assertLookupMisses(t *testing.T, cache *Cache, request *ImageRequest) {
|
||||
}
|
||||
}
|
||||
|
||||
// renameFile renames the file at from to to.
|
||||
func renameFile(t *testing.T, from, to string) {
|
||||
t.Helper()
|
||||
|
||||
err := os.Rename(from, to)
|
||||
if err != nil {
|
||||
t.Fatalf("renaming %s: %v", from, err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestSecondHitDoesNotReadMetaFile checks that once a variant has been
|
||||
// stored or read, a hit takes its content type from memory: with the
|
||||
// .meta file deleted, GetVariant must still return the stored content
|
||||
@@ -125,122 +114,6 @@ func TestSecondHitDoesNotReadMetaFile(t *testing.T) {
|
||||
assertVariantServed(t, restarted, testVariantKeyOne, content)
|
||||
}
|
||||
|
||||
// TestReadDuringStoreKeepsStoredContentType checks that a GetVariant
|
||||
// which began before StoreVariant finished cannot replace the content
|
||||
// type the store kept in memory. Such a read can find the variant file
|
||||
// but not yet its .meta file, and so gets application/octet-stream. The
|
||||
// test deletes the .meta file after the store, then runs the part of
|
||||
// GetVariant that comes after its check of memory.
|
||||
func TestReadDuringStoreKeepsStoredContentType(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
cache, _ := newEvictionTestCache(t, 1<<20)
|
||||
content := []byte("webp variant bytes")
|
||||
|
||||
storeEvictionTestVariant(t, cache, testVariantKeyOne, content)
|
||||
|
||||
err := os.Remove(cache.variants.keyToPath(testVariantKeyOne) + ".meta")
|
||||
if err != nil {
|
||||
t.Fatalf("removing .meta file: %v", err)
|
||||
}
|
||||
|
||||
reader, _, contentType, err := cache.loadVariantWithMeta(testVariantKeyOne)
|
||||
if err != nil {
|
||||
t.Fatalf("loadVariantWithMeta(%s) error = %v", testVariantKeyOne, err)
|
||||
}
|
||||
|
||||
_ = reader.Close()
|
||||
|
||||
t.Logf("the read without a .meta file got content type %q", contentType)
|
||||
|
||||
assertVariantServed(t, cache, testVariantKeyOne, content)
|
||||
}
|
||||
|
||||
// TestReadOfOlderMetaFileKeepsStoredContentType checks that a read
|
||||
// which got its content type from a .meta file that StoreVariant had
|
||||
// not yet rewritten cannot replace the type the store kept in memory.
|
||||
// The test writes such a .meta file, with a different content type,
|
||||
// after the store, then runs the part of GetVariant that comes after
|
||||
// its check of memory.
|
||||
func TestReadOfOlderMetaFileKeepsStoredContentType(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
cache, _ := newEvictionTestCache(t, 1<<20)
|
||||
content := []byte("webp variant bytes")
|
||||
|
||||
storeEvictionTestVariant(t, cache, testVariantKeyOne, content)
|
||||
|
||||
olderMeta, err := json.Marshal(VariantMeta{
|
||||
ContentType: testContentTypeJPEG,
|
||||
Size: int64(len(content)),
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("encoding .meta file: %v", err)
|
||||
}
|
||||
|
||||
metaPath := cache.variants.keyToPath(testVariantKeyOne) + ".meta"
|
||||
|
||||
err = os.WriteFile(metaPath, olderMeta, StorageFilePerm)
|
||||
if err != nil {
|
||||
t.Fatalf("writing .meta file: %v", err)
|
||||
}
|
||||
|
||||
reader, _, contentType, err := cache.loadVariantWithMeta(testVariantKeyOne)
|
||||
if err != nil {
|
||||
t.Fatalf("loadVariantWithMeta(%s) error = %v", testVariantKeyOne, err)
|
||||
}
|
||||
|
||||
_ = reader.Close()
|
||||
|
||||
if contentType != testContentTypeJPEG {
|
||||
t.Fatalf("loadVariantWithMeta(%s) content type = %q, want %q from the .meta file",
|
||||
testVariantKeyOne, contentType, testContentTypeJPEG)
|
||||
}
|
||||
|
||||
kept, _ := cache.metaCache.Get(testVariantKeyOne)
|
||||
if kept != testContentTypeWebP {
|
||||
t.Errorf("content type in memory = %q, want the stored %q",
|
||||
kept, testContentTypeWebP)
|
||||
}
|
||||
}
|
||||
|
||||
// TestFailedReadDuringStoreKeepsStoredContentType checks that a read
|
||||
// which found no .meta file cannot leave application/octet-stream in
|
||||
// memory, even when another read has removed the content type
|
||||
// StoreVariant kept there. In this order: the store; a read that finds
|
||||
// the variant in memory but cannot open its file, and so removes it
|
||||
// from memory; a read that opened the variant file before the store
|
||||
// wrote its .meta file. Later hits must get the stored content type.
|
||||
func TestFailedReadDuringStoreKeepsStoredContentType(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
cache, _ := newEvictionTestCache(t, 1<<20)
|
||||
content := []byte("webp variant bytes")
|
||||
variantPath := cache.variants.keyToPath(testVariantKeyOne)
|
||||
metaPath := variantPath + ".meta"
|
||||
|
||||
storeEvictionTestVariant(t, cache, testVariantKeyOne, content)
|
||||
|
||||
renameFile(t, variantPath, variantPath+".hidden")
|
||||
assertVariantNotFound(t, cache, testVariantKeyOne)
|
||||
renameFile(t, variantPath+".hidden", variantPath)
|
||||
|
||||
renameFile(t, metaPath, metaPath+".hidden")
|
||||
|
||||
reader, _, contentType, err := cache.GetVariant(testVariantKeyOne)
|
||||
if err != nil {
|
||||
t.Fatalf("GetVariant(%s) error = %v", testVariantKeyOne, err)
|
||||
}
|
||||
|
||||
_ = reader.Close()
|
||||
|
||||
t.Logf("the read without a .meta file got content type %q", contentType)
|
||||
|
||||
renameFile(t, metaPath+".hidden", metaPath)
|
||||
|
||||
assertVariantServed(t, cache, testVariantKeyOne, content)
|
||||
}
|
||||
|
||||
// TestEvictedVariantIsNotServed checks that a variant the evictor
|
||||
// removed is a miss and cannot be read, although it had been stored
|
||||
// and served before.
|
||||
|
||||
@@ -43,9 +43,6 @@ type ServiceConfig struct {
|
||||
SigningKey string
|
||||
// Allowlist is the list of hosts that don't require signatures
|
||||
Allowlist []string
|
||||
// MaxConcurrentProcessing is the most images processed at once; zero
|
||||
// uses the image processor's default, one per CPU
|
||||
MaxConcurrentProcessing int
|
||||
// Logger for logging
|
||||
Logger *slog.Logger
|
||||
}
|
||||
@@ -94,10 +91,9 @@ func NewService(cfg *ServiceConfig) (*Service, error) {
|
||||
}
|
||||
|
||||
maxResponseSize := fetcherCfg.MaxResponseSize
|
||||
processor := imageprocessor.New(imageprocessor.Params{
|
||||
MaxInputBytes: maxResponseSize,
|
||||
MaxConcurrentProcessing: cfg.MaxConcurrentProcessing,
|
||||
})
|
||||
processor := imageprocessor.New(
|
||||
imageprocessor.Params{MaxInputBytes: maxResponseSize},
|
||||
)
|
||||
|
||||
return &Service{
|
||||
cache: cfg.Cache,
|
||||
@@ -241,37 +237,38 @@ func (s *Service) GenerateSignedURL(
|
||||
baseURL, path, sig, exp, req.Quality, req.FitMode), nil
|
||||
}
|
||||
|
||||
// loadCachedSource opens source content from cache, without reading it, and
|
||||
// returns it with its size; nil if the cached data is unavailable, empty or
|
||||
// exceeds maxResponseSize.
|
||||
func (s *Service) loadCachedSource(
|
||||
contentHash ContentHash,
|
||||
) (io.ReadCloser, int64) {
|
||||
reader, size, err := s.cache.GetSourceContent(contentHash)
|
||||
// loadCachedSource attempts to load source content from cache, returning nil
|
||||
// if the cached data is unavailable or exceeds maxResponseSize.
|
||||
func (s *Service) loadCachedSource(contentHash ContentHash) []byte {
|
||||
reader, err := s.cache.GetSourceContent(contentHash)
|
||||
if err != nil {
|
||||
s.log.Warn("failed to load cached source, fetching", "error", err)
|
||||
|
||||
return nil, 0
|
||||
return nil
|
||||
}
|
||||
|
||||
if size > s.maxResponseSize {
|
||||
// Bound the read to maxResponseSize to prevent unbounded memory use
|
||||
// from unexpectedly large cached files.
|
||||
limited := io.LimitReader(reader, s.maxResponseSize+1)
|
||||
data, err := io.ReadAll(limited)
|
||||
_ = reader.Close()
|
||||
|
||||
if err != nil {
|
||||
s.log.Warn("failed to read cached source, fetching", "error", err)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
if int64(len(data)) > s.maxResponseSize {
|
||||
s.log.Warn("cached source exceeds max response size, discarding",
|
||||
"hash", contentHash,
|
||||
"max_bytes", s.maxResponseSize,
|
||||
)
|
||||
|
||||
return nil, 0
|
||||
return nil
|
||||
}
|
||||
|
||||
if size == 0 {
|
||||
_ = reader.Close()
|
||||
|
||||
return nil, 0
|
||||
}
|
||||
|
||||
return reader, size
|
||||
return data
|
||||
}
|
||||
|
||||
// processFromSourceOrFetch processes an image, using cached source content
|
||||
@@ -288,27 +285,22 @@ func (s *Service) processFromSourceOrFetch(
|
||||
s.log.Warn("source lookup failed", "error", err)
|
||||
}
|
||||
|
||||
var (
|
||||
source io.ReadCloser
|
||||
sourceSize int64
|
||||
)
|
||||
var sourceData []byte
|
||||
|
||||
if contentHash != "" {
|
||||
s.log.Debug("using cached source", "hash", contentHash)
|
||||
source, sourceSize = s.loadCachedSource(contentHash)
|
||||
sourceData = s.loadCachedSource(contentHash)
|
||||
}
|
||||
|
||||
// Fetch from upstream if we don't have source data or it's empty
|
||||
if source == nil {
|
||||
if len(sourceData) == 0 {
|
||||
return s.fetchAndProcess(ctx, req, cacheKey)
|
||||
}
|
||||
|
||||
defer func() { _ = source.Close() }()
|
||||
|
||||
// Process using cached source; nothing was fetched from upstream. The
|
||||
// image processor reads the source only once it has a processing slot,
|
||||
// so a request waiting for one holds none of it in memory.
|
||||
resp, err := s.processAndStore(ctx, req, cacheKey, source, sourceSize)
|
||||
// Process using cached source; nothing was fetched from upstream
|
||||
resp, err := s.processAndStore(
|
||||
ctx, req, cacheKey, sourceData, int64(len(sourceData)),
|
||||
)
|
||||
|
||||
return resp, 0, err
|
||||
}
|
||||
@@ -342,10 +334,6 @@ func (s *Service) fetchAndProcess(
|
||||
return nil, 0, fmt.Errorf("upstream fetch failed: %w", err)
|
||||
}
|
||||
|
||||
// Closing the body frees the upstream connection. It is closed only
|
||||
// after processing, so the fetcher's connection limit also bounds the
|
||||
// fetched sources held in memory while their requests wait for a
|
||||
// processing slot.
|
||||
defer func() { _ = fetchResult.Content.Close() }()
|
||||
|
||||
// Read and validate the source content
|
||||
@@ -391,20 +379,17 @@ func (s *Service) fetchAndProcess(
|
||||
// Continue even if caching fails
|
||||
}
|
||||
|
||||
resp, err := s.processAndStore(
|
||||
ctx, req, cacheKey, bytes.NewReader(sourceData), fetchBytes,
|
||||
)
|
||||
resp, err := s.processAndStore(ctx, req, cacheKey, sourceData, fetchBytes)
|
||||
|
||||
return resp, fetchBytes, err
|
||||
}
|
||||
|
||||
// processAndStore processes the image read from source and stores the
|
||||
// result.
|
||||
// processAndStore processes an image and stores the result.
|
||||
func (s *Service) processAndStore(
|
||||
ctx context.Context,
|
||||
req *ImageRequest,
|
||||
cacheKey VariantKey,
|
||||
source io.Reader,
|
||||
sourceData []byte,
|
||||
fetchBytes int64,
|
||||
) (*ImageResponse, error) {
|
||||
// Process the image
|
||||
@@ -417,7 +402,7 @@ func (s *Service) processAndStore(
|
||||
FitMode: imageprocessor.FitMode(req.FitMode),
|
||||
}
|
||||
|
||||
processResult, err := s.processor.Process(ctx, source, processReq)
|
||||
processResult, err := s.processor.Process(ctx, bytes.NewReader(sourceData), processReq)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("image processing failed: %w", err)
|
||||
}
|
||||
|
||||
@@ -531,8 +531,7 @@ func (s *VariantStorage) LoadWithSize(key VariantKey) (io.ReadCloser, int64, err
|
||||
}
|
||||
|
||||
// LoadWithMeta returns a reader, size, and content type for the content at
|
||||
// the given key. The content type is read from the .meta file, and is
|
||||
// empty when that file is missing or unreadable.
|
||||
// the given key.
|
||||
func (s *VariantStorage) LoadWithMeta(
|
||||
key VariantKey,
|
||||
) (io.ReadCloser, int64, string, error) {
|
||||
@@ -541,8 +540,8 @@ func (s *VariantStorage) LoadWithMeta(
|
||||
return nil, 0, "", err
|
||||
}
|
||||
|
||||
var contentType string
|
||||
|
||||
// Load metadata for content type
|
||||
contentType := "application/octet-stream" // fallback
|
||||
metaPath := s.keyToPath(key) + ".meta"
|
||||
|
||||
metaData, err := os.ReadFile(metaPath) //nolint:gosec // path derived from cache key
|
||||
|
||||
@@ -172,7 +172,7 @@ func (s *Middleware) Logging() func(http.Handler) http.Handler {
|
||||
// CORS returns a CORS middleware.
|
||||
func (s *Middleware) CORS() func(http.Handler) http.Handler {
|
||||
return cors.Handler(cors.Options{
|
||||
AllowedOrigins: []string{s.config.AccessControlAllowOrigin},
|
||||
AllowedOrigins: []string{"*"},
|
||||
AllowedMethods: []string{"GET", "HEAD", "OPTIONS"},
|
||||
AllowedHeaders: []string{"Accept", "Authorization", "Content-Type"},
|
||||
ExposedHeaders: []string{"Link"},
|
||||
|
||||
@@ -9,53 +9,6 @@ import (
|
||||
"sneak.berlin/go/pixa/internal/config"
|
||||
)
|
||||
|
||||
// TestCORSAnswersWithConfiguredOrigin checks that the CORS middleware
|
||||
// answers with access_control_allow_origin, where "*" lets any origin read
|
||||
// responses and a single origin lets only that origin read them.
|
||||
func TestCORSAnswersWithConfiguredOrigin(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
const appOrigin = "https://app.example.com"
|
||||
|
||||
cases := []struct {
|
||||
configured string
|
||||
requestOrigin string
|
||||
want string
|
||||
}{
|
||||
{"*", "https://any.example.com", "*"},
|
||||
{appOrigin, appOrigin, appOrigin},
|
||||
{appOrigin, "https://other.example.com", ""},
|
||||
}
|
||||
|
||||
testHandler := http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
|
||||
w.WriteHeader(http.StatusOK)
|
||||
})
|
||||
|
||||
for _, tc := range cases {
|
||||
mw := &Middleware{
|
||||
log: slog.Default(),
|
||||
config: &config.Config{AccessControlAllowOrigin: tc.configured},
|
||||
}
|
||||
|
||||
handler := mw.CORS()(testHandler)
|
||||
|
||||
req := httptest.NewRequestWithContext(
|
||||
t.Context(), http.MethodGet, "/v1/image/example.com/a.jpg/1x1.png", nil)
|
||||
req.Header.Set("Origin", tc.requestOrigin)
|
||||
|
||||
rec := httptest.NewRecorder()
|
||||
|
||||
handler.ServeHTTP(rec, req)
|
||||
|
||||
got := rec.Header().Get("Access-Control-Allow-Origin")
|
||||
if got != tc.want {
|
||||
t.Errorf("configured %q, request from %q: "+
|
||||
"Access-Control-Allow-Origin = %q, want %q",
|
||||
tc.configured, tc.requestOrigin, got, tc.want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestSecurityHeaders(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
|
||||
@@ -14,6 +14,7 @@ const (
|
||||
// short, so a slowloris client dribbling headers is dropped well
|
||||
// before it ties up a connection for the whole ReadTimeout window.
|
||||
HTTPReadHeaderTimeout = 10 * time.Second
|
||||
HTTPWriteTimeout = 60 * time.Second
|
||||
// HTTPIdleTimeout bounds how long an idle keep-alive connection is
|
||||
// held open, so idle connections cannot accumulate without limit on a
|
||||
// service targeting high concurrency.
|
||||
@@ -29,7 +30,7 @@ func (s *Server) newHTTPServer() *http.Server {
|
||||
Addr: fmt.Sprintf(":%d", s.config.Port),
|
||||
ReadTimeout: HTTPReadTimeout,
|
||||
ReadHeaderTimeout: HTTPReadHeaderTimeout,
|
||||
WriteTimeout: s.config.DownstreamTimeout,
|
||||
WriteTimeout: HTTPWriteTimeout,
|
||||
IdleTimeout: HTTPIdleTimeout,
|
||||
MaxHeaderBytes: HTTPMaxHeaderBytes,
|
||||
Handler: s,
|
||||
|
||||
@@ -11,15 +11,11 @@ import (
|
||||
// carries every hardening timeout wired onto it, including the slowloris
|
||||
// defense (ReadHeaderTimeout) and the keep-alive bound (IdleTimeout). This
|
||||
// guards against a field being defined but never set on the server, so
|
||||
// each assertion compares the server field to its constant, or, for
|
||||
// WriteTimeout, to downstream_timeout from the config.
|
||||
// each assertion compares the server field to its constant.
|
||||
func TestNewHTTPServerTimeouts(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
s := &Server{config: &config.Config{
|
||||
Port: 8080,
|
||||
DownstreamTimeout: 45 * time.Second,
|
||||
}}
|
||||
s := &Server{config: &config.Config{Port: 8080}}
|
||||
|
||||
srv := s.newHTTPServer()
|
||||
|
||||
@@ -30,7 +26,7 @@ func TestNewHTTPServerTimeouts(t *testing.T) {
|
||||
}{
|
||||
{"ReadTimeout", srv.ReadTimeout, HTTPReadTimeout},
|
||||
{"ReadHeaderTimeout", srv.ReadHeaderTimeout, HTTPReadHeaderTimeout},
|
||||
{"WriteTimeout", srv.WriteTimeout, 45 * time.Second},
|
||||
{"WriteTimeout", srv.WriteTimeout, HTTPWriteTimeout},
|
||||
{"IdleTimeout", srv.IdleTimeout, HTTPIdleTimeout},
|
||||
}
|
||||
|
||||
|
||||
@@ -18,7 +18,6 @@ import (
|
||||
"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/logger"
|
||||
"sneak.berlin/go/pixa/internal/middleware"
|
||||
)
|
||||
@@ -57,7 +56,6 @@ func newTestServer(t *testing.T) *Server {
|
||||
StateDir: stateDir,
|
||||
DBURL: "file:" + filepath.Join(stateDir, "state.sqlite3"),
|
||||
TrustedProxies: []netip.Prefix{netip.MustParsePrefix("10.0.0.0/8")},
|
||||
DownstreamTimeout: config.DefaultDownstreamTimeout,
|
||||
}
|
||||
|
||||
lc := fxtest.NewLifecycle(t)
|
||||
@@ -72,15 +70,8 @@ func newTestServer(t *testing.T) *Server {
|
||||
t.Fatalf("database.New() error = %v", err)
|
||||
}
|
||||
|
||||
hc, err := healthcheck.New(lc, healthcheck.Params{
|
||||
Globals: &globals.Globals{}, Config: cfg, Logger: log, Database: db,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("healthcheck.New() error = %v", err)
|
||||
}
|
||||
|
||||
h, err := handlers.New(lc, handlers.Params{
|
||||
Logger: log, Healthcheck: hc, Database: db, Config: cfg,
|
||||
Logger: log, Database: db, Config: cfg,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("handlers.New() error = %v", err)
|
||||
|
||||
@@ -1,169 +0,0 @@
|
||||
package server
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strconv"
|
||||
"testing"
|
||||
|
||||
"sneak.berlin/go/pixa/internal/healthcheck"
|
||||
)
|
||||
|
||||
// unsignedImagePath is an image URL that carries no signature.
|
||||
const unsignedImagePath = "/v1/image/cdn.example.com/cat.jpg/100x100.jpeg"
|
||||
|
||||
// TestMaintenanceModeRefusesImageRequests verifies that while maintenance
|
||||
// mode is on, both image routes answer 503 Service Unavailable with a
|
||||
// Retry-After header and the JSON error body the image handlers send.
|
||||
func TestMaintenanceModeRefusesImageRequests(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
s := newTestServer(t)
|
||||
s.config.MaintenanceMode = true
|
||||
|
||||
requests := []struct {
|
||||
method string
|
||||
path string
|
||||
}{
|
||||
{http.MethodGet, unsignedImagePath},
|
||||
{http.MethodHead, unsignedImagePath},
|
||||
{http.MethodGet, "/v1/e/token/cat.jpg"},
|
||||
}
|
||||
|
||||
for _, tc := range requests {
|
||||
t.Run(tc.method+" "+tc.path, func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
rec := httptest.NewRecorder()
|
||||
s.ServeHTTP(rec, httptest.NewRequestWithContext(
|
||||
t.Context(), tc.method, tc.path, nil))
|
||||
t.Logf("status %d, body %s", rec.Code, rec.Body.String())
|
||||
|
||||
if rec.Code != http.StatusServiceUnavailable {
|
||||
t.Fatalf("status = %d, want %d",
|
||||
rec.Code, http.StatusServiceUnavailable)
|
||||
}
|
||||
|
||||
retryAfter := rec.Header().Get("Retry-After")
|
||||
|
||||
seconds, err := strconv.Atoi(retryAfter)
|
||||
if err != nil || seconds <= 0 {
|
||||
t.Errorf("Retry-After = %q, want a positive number of seconds",
|
||||
retryAfter)
|
||||
}
|
||||
|
||||
// A HEAD response carries no body.
|
||||
if tc.method == http.MethodHead {
|
||||
return
|
||||
}
|
||||
|
||||
var body struct {
|
||||
Error string `json:"error"`
|
||||
Status int `json:"status"`
|
||||
Timestamp string `json:"timestamp"`
|
||||
}
|
||||
|
||||
err = json.NewDecoder(rec.Body).Decode(&body)
|
||||
if err != nil {
|
||||
t.Fatalf("body is not JSON: %v", err)
|
||||
}
|
||||
|
||||
if body.Error == "" || body.Status != http.StatusServiceUnavailable ||
|
||||
body.Timestamp == "" {
|
||||
t.Errorf("body = %+v, want an error, status %d and a timestamp",
|
||||
body, http.StatusServiceUnavailable)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// TestImageRequestsServedWithoutMaintenanceMode verifies that while
|
||||
// maintenance mode is off, image requests reach the image handlers instead
|
||||
// of the 503. The handlers refuse an unsigned image URL with 401 and a token
|
||||
// they cannot decrypt with 400, so either status shows a request got through.
|
||||
func TestImageRequestsServedWithoutMaintenanceMode(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
s := newTestServer(t)
|
||||
s.config.MaintenanceMode = false
|
||||
|
||||
requests := []struct {
|
||||
method string
|
||||
path string
|
||||
want int
|
||||
}{
|
||||
{http.MethodGet, unsignedImagePath, http.StatusUnauthorized},
|
||||
{http.MethodHead, unsignedImagePath, http.StatusUnauthorized},
|
||||
{http.MethodGet, "/v1/e/token/cat.jpg", http.StatusBadRequest},
|
||||
}
|
||||
|
||||
for _, tc := range requests {
|
||||
t.Run(tc.method+" "+tc.path, func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
rec := httptest.NewRecorder()
|
||||
s.ServeHTTP(rec, httptest.NewRequestWithContext(
|
||||
t.Context(), tc.method, tc.path, nil))
|
||||
t.Logf("status %d, body %s", rec.Code, rec.Body.String())
|
||||
|
||||
if rec.Code != tc.want {
|
||||
t.Errorf("status = %d, want %d from the image handler",
|
||||
rec.Code, tc.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// TestMaintenanceModeKeepsOtherRoutes verifies that while maintenance mode
|
||||
// is on, the health check still answers 200 and reports it, and the login
|
||||
// page and /metrics still answer 200. The image's Docker HEALTHCHECK
|
||||
// requests the health check: a 503 there would make the container
|
||||
// unhealthy, and upaas marks a deploy failed when its container is
|
||||
// unhealthy.
|
||||
func TestMaintenanceModeKeepsOtherRoutes(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
s := newTestServer(t)
|
||||
s.config.MaintenanceMode = true
|
||||
|
||||
// /metrics is routed only when its username is set.
|
||||
s.config.MetricsUsername = "metrics"
|
||||
s.config.MetricsPassword = "metrics-password"
|
||||
s.SetupRoutes()
|
||||
|
||||
rec := httptest.NewRecorder()
|
||||
s.ServeHTTP(rec, httptest.NewRequestWithContext(t.Context(),
|
||||
http.MethodGet, "/.well-known/healthcheck.json", nil))
|
||||
t.Logf("health check status %d, body %s", rec.Code, rec.Body.String())
|
||||
|
||||
if rec.Code != http.StatusOK {
|
||||
t.Fatalf("health check status = %d, want %d", rec.Code, http.StatusOK)
|
||||
}
|
||||
|
||||
var health healthcheck.Response
|
||||
|
||||
err := json.NewDecoder(rec.Body).Decode(&health)
|
||||
if err != nil || !health.Maintenance {
|
||||
t.Errorf("health check maintenance_mode = %v (error %v), want true",
|
||||
health.Maintenance, err)
|
||||
}
|
||||
|
||||
rec = httptest.NewRecorder()
|
||||
s.ServeHTTP(rec, clientRequest(t, http.MethodGet, nil, firstClient, ""))
|
||||
|
||||
if rec.Code != http.StatusOK {
|
||||
t.Errorf("login page status = %d, want %d", rec.Code, http.StatusOK)
|
||||
}
|
||||
|
||||
req := httptest.NewRequestWithContext(t.Context(),
|
||||
http.MethodGet, "/metrics", nil)
|
||||
req.SetBasicAuth(s.config.MetricsUsername, s.config.MetricsPassword)
|
||||
|
||||
rec = httptest.NewRecorder()
|
||||
s.ServeHTTP(rec, req)
|
||||
|
||||
if rec.Code != http.StatusOK {
|
||||
t.Errorf("/metrics status = %d, want %d", rec.Code, http.StatusOK)
|
||||
}
|
||||
}
|
||||
@@ -1,9 +1,7 @@
|
||||
package server
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"strconv"
|
||||
"time"
|
||||
|
||||
sentryhttp "github.com/getsentry/sentry-go/http"
|
||||
@@ -19,10 +17,6 @@ import (
|
||||
// make per minute; the next is refused with 429 Too Many Requests.
|
||||
const LoginAttemptsPerMinute = 5
|
||||
|
||||
// MaintenanceRetryAfterSeconds is the Retry-After, in seconds, sent with
|
||||
// the 503 that the image routes answer while maintenance mode is on.
|
||||
const MaintenanceRetryAfterSeconds = 300
|
||||
|
||||
// SetupRoutes configures all HTTP routes.
|
||||
func (s *Server) SetupRoutes() {
|
||||
s.router = chi.NewRouter()
|
||||
@@ -39,7 +33,7 @@ func (s *Server) SetupRoutes() {
|
||||
}
|
||||
|
||||
s.router.Use(s.mw.CORS())
|
||||
s.router.Use(middleware.Timeout(s.config.DownstreamTimeout))
|
||||
s.router.Use(middleware.Timeout(HTTPWriteTimeout))
|
||||
|
||||
if s.sentryEnabled {
|
||||
sentryHandler := sentryhttp.New(sentryhttp.Options{
|
||||
@@ -74,23 +68,15 @@ func (s *Server) SetupRoutes() {
|
||||
|
||||
s.router.Get("/logout", s.h.HandleLogout())
|
||||
|
||||
// Image routes, refused while maintenance mode is on. Only these: the
|
||||
// image's Docker HEALTHCHECK requests the health check, a 503 there
|
||||
// would make the container unhealthy, and upaas marks a deploy failed
|
||||
// when its container is unhealthy.
|
||||
s.router.Group(func(r chi.Router) {
|
||||
r.Use(s.refuseDuringMaintenance)
|
||||
|
||||
// Main image proxy route
|
||||
// /v1/image/<host>/<path>/<width>x<height>.<format>
|
||||
r.Get("/v1/image/*", s.h.HandleImage())
|
||||
r.Head("/v1/image/*", s.h.HandleImage())
|
||||
s.router.Get("/v1/image/*", s.h.HandleImage())
|
||||
s.router.Head("/v1/image/*", s.h.HandleImage())
|
||||
|
||||
// Encrypted image URL route
|
||||
// The trailing filename (e.g., /img.jpg) is ignored but helps
|
||||
// browsers with content type
|
||||
r.Get("/v1/e/{token}/*", s.h.HandleImageEnc())
|
||||
})
|
||||
s.router.Get("/v1/e/{token}/*", s.h.HandleImageEnc())
|
||||
|
||||
// Metrics endpoint with auth
|
||||
if s.config.MetricsUsername != "" {
|
||||
@@ -100,30 +86,3 @@ func (s *Server) SetupRoutes() {
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// refuseDuringMaintenance answers a request with 503 Service Unavailable,
|
||||
// a Retry-After header and a JSON error body while maintenance mode is on,
|
||||
// and passes it on otherwise. The body has the fields of the JSON errors
|
||||
// the image handlers send.
|
||||
func (s *Server) refuseDuringMaintenance(next http.Handler) http.Handler {
|
||||
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
if !s.MaintenanceMode() {
|
||||
next.ServeHTTP(w, r)
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
w.Header().Set("Retry-After", strconv.Itoa(MaintenanceRetryAfterSeconds))
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
w.WriteHeader(http.StatusServiceUnavailable)
|
||||
|
||||
err := json.NewEncoder(w).Encode(map[string]any{
|
||||
"error": "down for maintenance, try again later",
|
||||
"status": http.StatusServiceUnavailable,
|
||||
"timestamp": time.Now().UTC().Format(time.RFC3339),
|
||||
})
|
||||
if err != nil {
|
||||
s.log.Error("json encode error", "error", err)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
+2
-9
@@ -5,10 +5,8 @@
|
||||
# or apk (detected in that order); assumes NOTHING is present (not git,
|
||||
# make, or go). The linter is never installed on the host: golangci-lint
|
||||
# runs only inside a container, Dockerfile.lint or the Dockerfile lint
|
||||
# stage (see script/lint). A C compiler and the CGO image libraries
|
||||
# (pkg-config, vips, libheif) are installed for the govips bindings.
|
||||
# Both Dockerfiles run this script too, so their build dependencies are
|
||||
# the ones listed here.
|
||||
# stage (see script/lint). CGO image libraries (pkg-config, vips,
|
||||
# libheif) are installed for the govips bindings.
|
||||
set -eu
|
||||
|
||||
ROOT="$(cd "$(dirname "$0")/.." && pwd -P)"
|
||||
@@ -55,11 +53,6 @@ missing() {
|
||||
|
||||
# CGO dependencies for govips (image processing)
|
||||
ensure_cgo_deps() {
|
||||
# cgo compiles with gcc on Linux; build-base and build-essential
|
||||
# also bring the C library headers.
|
||||
if missing gcc; then
|
||||
pkg_install gcc build-essential gcc build-base
|
||||
fi
|
||||
if missing pkg-config; then
|
||||
pkg_install pkg-config pkg-config pkg-config pkgconfig
|
||||
fi
|
||||
|
||||
Reference in New Issue
Block a user