Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
6b7d939234 |
@@ -26,11 +26,6 @@ make docker
|
||||
docker run -p 8080:8080 -e PIXA_SIGNING_KEY="$(openssl rand -base64 32)" pixa:latest
|
||||
```
|
||||
|
||||
The example config allowlists no host (`allowlist_hosts`), so pixa serves an
|
||||
image only for a signed or encrypted URL. Every image on an allowlisted host is
|
||||
served to anyone without a signature, at any size and in any format, so
|
||||
allowlist only hosts you control, where nobody else can put images.
|
||||
|
||||
A container takes its settings from environment variables (see Configuration
|
||||
below for the list). Only `PIXA_SIGNING_KEY` is required; if it is unset the
|
||||
container exits at startup naming the variable. Everything else has a built-in
|
||||
@@ -120,9 +115,7 @@ What the [upaas](https://git.eeqj.de/sneak/upaas) app for pixa needs:
|
||||
- `PIXA_SIGNING_KEY` (required): secret for signed and encrypted URLs and
|
||||
login, 32+ characters, for example from `openssl rand -base64 32`
|
||||
- `PIXA_ALLOWLIST_HOSTS`: upstream hosts served without a signature,
|
||||
comma-separated; default none. Every image on a listed host is served to
|
||||
anyone without a signature, so list only hosts you control, where nobody
|
||||
else can put images.
|
||||
comma-separated
|
||||
- `PIXA_CACHE_MAX_BYTES`: disk cache limit in bytes; `0` disables it;
|
||||
default 75% of (free space + what the cache holds)
|
||||
- the rest are in the table under Configuration below
|
||||
@@ -419,11 +412,8 @@ HMAC-SHA256(secret, "host:path:query:width:height:format:expiration:quality:fit"
|
||||
Where:
|
||||
|
||||
- `host` — source origin hostname (e.g. `cdn.example.com`)
|
||||
- `path` — source path (e.g. `/photos/cat.jpg`), with each `%` written as `%25`
|
||||
and each `:` as `%3A`, so `/photos/a:b.jpg` is signed as `/photos/a%3Ab.jpg`
|
||||
- `query` — source query string, empty string if none, with `%` and `:` written
|
||||
as in `path`; neither then holds a `:`, so path `/a:b` with no query and path
|
||||
`/a` with query `b:` are signed differently
|
||||
- `path` — source path (e.g. `/photos/cat.jpg`)
|
||||
- `query` — source query string, empty string if none
|
||||
- `width` — requested width in pixels, `0` for original
|
||||
- `height` — requested height in pixels, `0` for original
|
||||
- `format` — output format, one of those listed under Routes, with `original`
|
||||
|
||||
@@ -30,42 +30,6 @@ P2: security: per-origin rate limiting
|
||||
|
||||
# Completed Steps
|
||||
|
||||
- 2026-10-15 a `:` in the source path or query no longer moves where the path
|
||||
ends and the query begins (closes #247): the signed data and the cache key
|
||||
joined host, path and query with `:` and no escaping, so a URL signed for
|
||||
`https://h/a:b` verified for `https://h/a?b:`; the two also shared a cache
|
||||
key, and whichever was fetched first was served for both. In both, the path
|
||||
and the query now have each `%` written as `%25` and each `:` as `%3A`; a path
|
||||
and query holding neither are signed and keyed as before. Signature
|
||||
Specification in `README.md` says so.
|
||||
- 2026-10-15 expired negative cache rows are removed on a schedule (closes
|
||||
#244): the row of a failed fetch, which makes pixa answer 502 for that source
|
||||
URL for 5 minutes, was deleted only when the same URL was requested again
|
||||
after that, so a client asking for a new missing path each time added rows
|
||||
that stayed for good. The eviction goroutine now calls `CleanExpired` at
|
||||
startup and on each periodic tick, every 5 minutes in pixad, so a row stays at
|
||||
most about 10 minutes. `CleanExpired` deletes at most 1000 rows per statement,
|
||||
so a request waits behind 1000 deletions at most, however many rows have
|
||||
expired. The goroutine runs with the disk cache disabled
|
||||
(`cache_max_bytes: 0`) too, where it does nothing else, as the negative cache
|
||||
is used either way.
|
||||
- 2026-10-15 the example config allowlists no host (closes #249): it allowlisted
|
||||
`github.com` and `user-images.githubusercontent.com`, where anyone can put an
|
||||
image, so a deployment set up as Getting Started says served those images to
|
||||
anyone without a signature, at any size and in any format.
|
||||
`configs/config.example.yml` now leaves `allowlist_hosts` out, with one
|
||||
commented entry under `example.com`. Getting Started and "Running under upaas"
|
||||
in `README.md` say that every image on an allowlisted host is served without a
|
||||
signature, so an allowlisted host must be one the operator controls.
|
||||
- 2026-10-15 a chunked upstream body of exactly `upstream_max_response_size`
|
||||
bytes is accepted (closes #248): once the limit was used up, the fetcher's
|
||||
next read refused the body without looking for its end. Go's HTTP client
|
||||
returns the end of a body sent with a `Content-Length` together with its last
|
||||
bytes, but the end of a chunked body can come in a later read, so such a body
|
||||
was refused after it had been downloaded whole, and the image routes
|
||||
answered 500. Once the limit is used up, the fetcher now reads one more byte:
|
||||
if the body has ended it is accepted, and if there is more it is refused with
|
||||
`ErrResponseTooLarge`, as before.
|
||||
- 2026-10-13 `fit=outside` no longer makes an output above 8192 on a side
|
||||
(closes #257): it scales the image to cover the requested size, and the side
|
||||
scaled past that size was held to no maximum, so `1x8192?fit=outside` of a
|
||||
|
||||
@@ -45,14 +45,15 @@ state_dir: ./data
|
||||
signing_key: "CHANGE_ME_generate_with_openssl_rand_base64_32"
|
||||
|
||||
# Hosts that don't require signatures (default: none)
|
||||
# Every image on an allowlisted host is served to anyone without a
|
||||
# signature, at any size and in any format, so list only hosts you
|
||||
# control, where nobody else can put images.
|
||||
# Use "." prefix for wildcard subdomain matching (e.g., ".example.com" matches "cdn.example.com")
|
||||
# An entry that is neither a host name nor an IP address (IPv6 without
|
||||
# brackets), such as one with a port or a "*." wildcard, aborts startup.
|
||||
# allowlist_hosts:
|
||||
# - images.example.com
|
||||
allowlist_hosts:
|
||||
- s3.sneak.cloud
|
||||
- static.sneak.cloud
|
||||
- sneak.berlin
|
||||
- github.com
|
||||
- user-images.githubusercontent.com
|
||||
|
||||
# Hosts whose pages may not show pixa's images, written as for
|
||||
# allowlist_hosts. A request to /v1/image/ or /v1/e/ whose Referer header
|
||||
|
||||
@@ -32,6 +32,28 @@ func TestOmittedConcurrencyLimitsUseDefaults(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// TestOmittedMaxConcurrentProcessingIsAtMostFour checks that an omitted
|
||||
// max_concurrent_processing is the number of CPUs Go uses, but never more
|
||||
// than 4, so a container with no CPU limit on a host with many CPUs does not
|
||||
// process one image per CPU at once.
|
||||
func TestOmittedMaxConcurrentProcessingIsAtMostFour(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
c, err := configFromYAML(t, signingKeyLine)
|
||||
if err != nil {
|
||||
t.Fatalf("minimal config should be valid, got error: %v", err)
|
||||
}
|
||||
|
||||
want := min(runtime.GOMAXPROCS(0), 4)
|
||||
t.Logf("GOMAXPROCS %d, MaxConcurrentProcessing %d",
|
||||
runtime.GOMAXPROCS(0), c.MaxConcurrentProcessing)
|
||||
|
||||
if c.MaxConcurrentProcessing != want {
|
||||
t.Errorf("MaxConcurrentProcessing = %d, want %d",
|
||||
c.MaxConcurrentProcessing, want)
|
||||
}
|
||||
}
|
||||
|
||||
// TestExplicitConcurrencyLimitsAreUsed checks that valid values for the
|
||||
// two limits are used as given.
|
||||
func TestExplicitConcurrencyLimitsAreUsed(t *testing.T) {
|
||||
|
||||
@@ -92,32 +92,6 @@ func TestGetStringSlice_Empty(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// TestExampleConfigAllowlistsNoHost loads configs/config.example.yml
|
||||
// with a real signing key, as Getting Started in README.md has it copied
|
||||
// and edited, and checks that it allowlists no host: every image on an
|
||||
// allowlisted host is served to anyone without a signature.
|
||||
func TestExampleConfigAllowlistsNoHost(t *testing.T) {
|
||||
t.Setenv("PIXA_SIGNING_KEY", validTestSigningKey)
|
||||
|
||||
examplePath := filepath.Join("..", "..", "configs", "config.example.yml")
|
||||
|
||||
sc, err := loadTestConfig(examplePath)
|
||||
if err != nil {
|
||||
t.Fatalf("failed to load %s: %v", examplePath, err)
|
||||
}
|
||||
|
||||
c, err := newFromSmartConfig(sc)
|
||||
if err != nil {
|
||||
t.Fatalf("%s is not a valid config: %v", examplePath, err)
|
||||
}
|
||||
|
||||
t.Logf("allowlist_hosts in %s: %v", examplePath, c.AllowlistHosts)
|
||||
|
||||
if len(c.AllowlistHosts) != 0 {
|
||||
t.Errorf("%s allowlists %v, want no host", examplePath, c.AllowlistHosts)
|
||||
}
|
||||
}
|
||||
|
||||
// loadTestConfig is a helper to load a config file for testing.
|
||||
func loadTestConfig(path string) (*smartconfig.Config, error) {
|
||||
return smartconfig.NewFromConfigPath(path)
|
||||
|
||||
@@ -124,8 +124,7 @@ func (s *Handlers) initImageService() error {
|
||||
s.imgCache = cache
|
||||
|
||||
// Background eviction: startup reconciliation, then periodic and
|
||||
// write-pressure passes, and the removal of expired negative cache
|
||||
// rows, which is all it does when the disk cache is disabled.
|
||||
// write-pressure passes. No-op when the disk cache is disabled.
|
||||
cache.StartEviction(imgcache.DefaultEvictionInterval)
|
||||
|
||||
// Writes the counts requests could not write by their deadline
|
||||
|
||||
@@ -9,7 +9,6 @@ import (
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"slices"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
@@ -298,139 +297,6 @@ func TestFetchMaxResponseSizeEnforced(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// fetchWithMaxResponseSize serves handler as the fake upstream and fetches
|
||||
// its /image route with MaxResponseSize set to maxResponseSize.
|
||||
func fetchWithMaxResponseSize(
|
||||
t *testing.T,
|
||||
maxResponseSize int64,
|
||||
handler http.HandlerFunc,
|
||||
) *FetchResult {
|
||||
t.Helper()
|
||||
|
||||
srv := httptest.NewServer(handler)
|
||||
t.Cleanup(srv.Close)
|
||||
|
||||
cfg := DefaultConfig()
|
||||
cfg.MaxResponseSize = maxResponseSize
|
||||
|
||||
f, _ := newServerFetcher(t, srv, cfg)
|
||||
|
||||
res := fetchImage(t, f, "/image")
|
||||
t.Cleanup(func() { _ = res.Content.Close() })
|
||||
|
||||
return res
|
||||
}
|
||||
|
||||
// TestFetchContentLengthBodyAtMaxResponseSize checks that a body sent with a
|
||||
// Content-Length is read whole at exactly MaxResponseSize bytes and refused
|
||||
// at one byte more.
|
||||
func TestFetchContentLengthBodyAtMaxResponseSize(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
const maxResponseSize = 8
|
||||
|
||||
for _, tc := range []struct {
|
||||
size int
|
||||
wantErr error
|
||||
}{
|
||||
{maxResponseSize, nil},
|
||||
{maxResponseSize + 1, ErrResponseTooLarge},
|
||||
} {
|
||||
t.Run(strconv.Itoa(tc.size)+" bytes", func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
body := strings.Repeat("x", tc.size)
|
||||
|
||||
res := fetchWithMaxResponseSize(t, maxResponseSize,
|
||||
func(w http.ResponseWriter, _ *http.Request) {
|
||||
w.Header().Set("Content-Type", contentTypeJPEG)
|
||||
w.Header().Set("Content-Length", strconv.Itoa(tc.size))
|
||||
_, _ = io.WriteString(w, body)
|
||||
})
|
||||
|
||||
if res.ContentLength != int64(tc.size) {
|
||||
t.Fatalf("ContentLength = %d, want %d", res.ContentLength, tc.size)
|
||||
}
|
||||
|
||||
data, err := io.ReadAll(res.Content)
|
||||
t.Logf("body of %d bytes: read %d, error %v", tc.size, len(data), err)
|
||||
|
||||
if !errors.Is(err, tc.wantErr) {
|
||||
t.Fatalf("read error = %v, want %v", err, tc.wantErr)
|
||||
}
|
||||
|
||||
if err == nil && string(data) != body {
|
||||
t.Errorf("read %q, want %q", data, body)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// TestFetchChunkedBodyAtMaxResponseSize checks that a chunked body is read
|
||||
// whole at exactly MaxResponseSize bytes and refused at one byte more. The
|
||||
// upstream sends the end of the body only once the test has read
|
||||
// MaxResponseSize bytes, so the read after them comes before the end of the
|
||||
// body has arrived, as it can from a real upstream.
|
||||
func TestFetchChunkedBodyAtMaxResponseSize(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
const maxResponseSize = 8
|
||||
|
||||
for _, tc := range []struct {
|
||||
size int
|
||||
wantErr error
|
||||
}{
|
||||
{maxResponseSize, nil},
|
||||
{maxResponseSize + 1, ErrResponseTooLarge},
|
||||
} {
|
||||
t.Run(strconv.Itoa(tc.size)+" bytes", func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
body := strings.Repeat("x", tc.size)
|
||||
endBody := make(chan struct{})
|
||||
|
||||
res := fetchWithMaxResponseSize(t, maxResponseSize,
|
||||
func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Content-Type", contentTypeJPEG)
|
||||
_, _ = io.WriteString(w, body)
|
||||
// With no Content-Length, flushing sends the body
|
||||
// chunked; its end is sent when the handler returns.
|
||||
_ = http.NewResponseController(w).Flush()
|
||||
|
||||
select {
|
||||
case <-endBody:
|
||||
case <-r.Context().Done():
|
||||
}
|
||||
})
|
||||
|
||||
if res.ContentLength != -1 {
|
||||
t.Fatalf("ContentLength = %d, want -1 (chunked)", res.ContentLength)
|
||||
}
|
||||
|
||||
head := make([]byte, maxResponseSize)
|
||||
|
||||
_, err := io.ReadFull(res.Content, head)
|
||||
if err != nil {
|
||||
t.Fatalf("read the first %d bytes: %v", maxResponseSize, err)
|
||||
}
|
||||
|
||||
close(endBody)
|
||||
|
||||
rest, err := io.ReadAll(res.Content)
|
||||
data := string(head) + string(rest)
|
||||
t.Logf("body of %d bytes: read %d, error %v", tc.size, len(data), err)
|
||||
|
||||
if !errors.Is(err, tc.wantErr) {
|
||||
t.Fatalf("read error = %v, want %v", err, tc.wantErr)
|
||||
}
|
||||
|
||||
if err == nil && data != body {
|
||||
t.Errorf("read %q, want %q", data, body)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestFetchSemaphoreReleasedOnError(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
|
||||
@@ -660,33 +660,14 @@ func dialSSRFSafe(
|
||||
return nil, fmt.Errorf("%w to %s", errConnectFailed, host)
|
||||
}
|
||||
|
||||
// limitedReader wraps a reader and refuses a body longer than remaining
|
||||
// bytes with ErrResponseTooLarge. remaining goes below 0 once the body is
|
||||
// found to be longer.
|
||||
// limitedReader wraps a reader and limits the number of bytes read.
|
||||
type limitedReader struct {
|
||||
reader io.Reader
|
||||
remaining int64
|
||||
}
|
||||
|
||||
// Read reads at most the remaining bytes. Once they are read, it reads one
|
||||
// more byte to tell the end of the body from more data: at the end it
|
||||
// returns the wrapped reader's error, io.EOF; with more data it returns
|
||||
// ErrResponseTooLarge, as does every later Read.
|
||||
func (r *limitedReader) Read(p []byte) (int, error) {
|
||||
if r.remaining < 0 {
|
||||
return 0, ErrResponseTooLarge
|
||||
}
|
||||
|
||||
if r.remaining == 0 {
|
||||
var extra [1]byte
|
||||
|
||||
n, err := r.reader.Read(extra[:])
|
||||
r.remaining -= int64(n)
|
||||
|
||||
if n == 0 {
|
||||
return 0, err
|
||||
}
|
||||
|
||||
if r.remaining <= 0 {
|
||||
return 0, ErrResponseTooLarge
|
||||
}
|
||||
|
||||
|
||||
@@ -132,6 +132,27 @@ func TestNewDefaultsMaxConcurrentProcessingToCPUs(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// TestNewDefaultMaxConcurrentProcessingIsAtMostFour checks that New, given
|
||||
// no MaxConcurrentProcessing, makes one slot per CPU Go uses, but never more
|
||||
// than 4.
|
||||
func TestNewDefaultMaxConcurrentProcessingIsAtMostFour(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
want := min(runtime.GOMAXPROCS(0), 4)
|
||||
|
||||
for _, limit := range []int{0, -1} {
|
||||
proc := New(Params{MaxConcurrentProcessing: limit})
|
||||
got := cap(proc.processingSemaphore)
|
||||
t.Logf("MaxConcurrentProcessing %d, GOMAXPROCS %d: %d slots",
|
||||
limit, runtime.GOMAXPROCS(0), got)
|
||||
|
||||
if got != want {
|
||||
t.Errorf("MaxConcurrentProcessing %d: %d slots, want %d",
|
||||
limit, got, want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 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
|
||||
|
||||
+11
-52
@@ -32,12 +32,6 @@ const httpStatusOK = 200
|
||||
// variant not among them is served as before, reading its .meta file.
|
||||
const metaCacheSize = 10000
|
||||
|
||||
// cleanExpiredBatchSize is the most expired negative cache rows one
|
||||
// statement of CleanExpired deletes. Each batch is a statement of its
|
||||
// own, so a request waits behind one batch at most, however many rows
|
||||
// have expired.
|
||||
const cleanExpiredBatchSize = 1000
|
||||
|
||||
// CacheConfig holds cache configuration.
|
||||
type CacheConfig struct {
|
||||
StateDir string
|
||||
@@ -56,9 +50,8 @@ type CacheConfig struct {
|
||||
|
||||
// DisableDiskCache turns the disk cache off entirely: no cache
|
||||
// directories are created, lookups always miss, stores are
|
||||
// no-ops, and the eviction goroutine only removes the expired
|
||||
// negative cache rows. The config layer sets this when the
|
||||
// operator configures cache_max_bytes: 0.
|
||||
// no-ops, and no eviction machinery runs. The config layer sets
|
||||
// this when the operator configures cache_max_bytes: 0.
|
||||
DisableDiskCache bool
|
||||
|
||||
// Logger receives accounting and eviction log output. A nil
|
||||
@@ -489,27 +482,17 @@ func (c *Cache) GetSourceContent(
|
||||
return c.srcContent.LoadWithSize(contentHash)
|
||||
}
|
||||
|
||||
// CleanExpired removes the negative cache rows past their expiry, one
|
||||
// batch per statement, until none is left. Once ctx ends it starts no
|
||||
// further batch and returns ctx's error. The eviction goroutine calls it
|
||||
// at startup and on every periodic tick, so a row stays about
|
||||
// NegativeTTL plus one eviction interval, whether or not its URL is
|
||||
// requested again.
|
||||
// CleanExpired removes expired entries from the cache.
|
||||
func (c *Cache) CleanExpired(ctx context.Context) error {
|
||||
for {
|
||||
if ctx.Err() != nil {
|
||||
return ctx.Err()
|
||||
}
|
||||
|
||||
deleted, err := c.cleanExpiredBatch(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if deleted < cleanExpiredBatchSize {
|
||||
return nil
|
||||
}
|
||||
// Clean expired negative cache entries
|
||||
_, err := c.db.ExecContext(ctx, `
|
||||
DELETE FROM negative_cache WHERE expires_at < CURRENT_TIMESTAMP
|
||||
`)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to clean negative cache: %w", err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// Stats returns cache statistics.
|
||||
@@ -773,27 +756,3 @@ func (c *Cache) checkNegativeCache(
|
||||
|
||||
return true, nil
|
||||
}
|
||||
|
||||
// cleanExpiredBatch deletes up to cleanExpiredBatchSize negative cache
|
||||
// rows past their expiry in one statement, and returns how many it
|
||||
// deleted. DELETE ... LIMIT needs a SQLite build option, so the subquery
|
||||
// picks the rows.
|
||||
func (c *Cache) cleanExpiredBatch(ctx context.Context) (int64, error) {
|
||||
result, err := c.db.ExecContext(ctx, `
|
||||
DELETE FROM negative_cache WHERE id IN (
|
||||
SELECT id FROM negative_cache
|
||||
WHERE expires_at < CURRENT_TIMESTAMP
|
||||
LIMIT ?
|
||||
)
|
||||
`, cleanExpiredBatchSize)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("failed to clean negative cache: %w", err)
|
||||
}
|
||||
|
||||
deleted, err := result.RowsAffected()
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("failed to count cleaned negative cache rows: %w", err)
|
||||
}
|
||||
|
||||
return deleted, nil
|
||||
}
|
||||
|
||||
@@ -437,16 +437,14 @@ func (c *Cache) notifyWritePressure() {
|
||||
}
|
||||
|
||||
// StartEviction launches the background eviction goroutine, which
|
||||
// reconciles the database accounting with the cache directories and
|
||||
// removes the expired negative cache rows at startup and again on every
|
||||
// periodic tick thereafter, and evicts to the configured limit on the
|
||||
// given periodic interval and on write-pressure notifications. On a
|
||||
// disabled cache it only removes the expired negative cache rows, as
|
||||
// the negative cache is used either way. It is a no-op when already
|
||||
// started. The goroutine outlives the caller, so it runs with its own
|
||||
// context, which StopEviction cancels.
|
||||
// reconciles the database accounting with the cache directories at
|
||||
// startup and again on every periodic tick thereafter, and evicts to
|
||||
// the configured limit on the given periodic interval and on
|
||||
// write-pressure notifications. It is a no-op on a disabled cache or
|
||||
// when already started. The goroutine outlives the caller, so it runs
|
||||
// with its own context, which StopEviction cancels.
|
||||
func (c *Cache) StartEviction(interval time.Duration) {
|
||||
if c.evictionCancel != nil {
|
||||
if c.disabled || c.evictionCancel != nil {
|
||||
return
|
||||
}
|
||||
|
||||
@@ -482,7 +480,6 @@ func (c *Cache) evictionLoop(ctx context.Context, interval time.Duration) {
|
||||
defer close(c.evictionDone)
|
||||
|
||||
c.runReconciliationPass(ctx)
|
||||
c.runCleanExpired(ctx)
|
||||
c.runEvictionPass(ctx)
|
||||
|
||||
ticker := time.NewTicker(interval)
|
||||
@@ -503,7 +500,6 @@ func (c *Cache) evictionLoop(ctx context.Context, interval time.Duration) {
|
||||
// unaccounted for to one interval, on a process that is
|
||||
// already running this loop regardless.
|
||||
c.runReconciliationPass(ctx)
|
||||
c.runCleanExpired(ctx)
|
||||
case <-c.evictionPressure:
|
||||
}
|
||||
|
||||
@@ -539,20 +535,6 @@ func (c *Cache) runReconciliationPass(ctx context.Context) {
|
||||
}
|
||||
}
|
||||
|
||||
// runCleanExpired runs CleanExpired, logging a failure instead of
|
||||
// propagating it (the loop must keep running). It does nothing once ctx
|
||||
// is cancelled.
|
||||
func (c *Cache) runCleanExpired(ctx context.Context) {
|
||||
if ctx.Err() != nil {
|
||||
return
|
||||
}
|
||||
|
||||
err := c.CleanExpired(ctx)
|
||||
if err != nil {
|
||||
c.log.Warn("negative cache cleanup failed", "error", err)
|
||||
}
|
||||
}
|
||||
|
||||
// reconcileAccounting synchronizes the database size accounting with
|
||||
// the actual contents of the cache directories. It runs at startup and
|
||||
// again on every periodic eviction tick thereafter, off the request
|
||||
|
||||
@@ -3,7 +3,6 @@ package imgcache
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"strconv"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
@@ -95,242 +94,6 @@ func TestNegativeCache_Expired(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// TestEvictionLoopRemovesExpiredNegativeCacheRows stores a failed fetch
|
||||
// that expires a second later and starts the eviction loop with a short
|
||||
// interval: a later tick must remove its row although its URL is never
|
||||
// requested again, and must keep a row that has not expired. It runs
|
||||
// with the disk cache enabled and disabled, as the negative cache is
|
||||
// used either way.
|
||||
func TestEvictionLoopRemovesExpiredNegativeCacheRows(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
tests := []struct {
|
||||
name string
|
||||
disableDiskCache bool
|
||||
}{
|
||||
{name: "disk cache enabled", disableDiskCache: false},
|
||||
{name: "disk cache disabled", disableDiskCache: true},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
cache, err := NewCache(evictionTestDB(t), CacheConfig{
|
||||
StateDir: t.TempDir(),
|
||||
NegativeTTL: time.Second,
|
||||
DisableDiskCache: tt.disableDiskCache,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("failed to create cache: %v", err)
|
||||
}
|
||||
|
||||
expired := &ImageRequest{
|
||||
SourceHost: testHostExample, SourcePath: "/gone.jpg",
|
||||
}
|
||||
|
||||
err = cache.StoreNegative(t.Context(), expired, 404, "not found")
|
||||
if err != nil {
|
||||
t.Fatalf("StoreNegative failed: %v", err)
|
||||
}
|
||||
|
||||
const notExpiredPath = "/later.jpg"
|
||||
|
||||
_, err = cache.db.ExecContext(t.Context(), `
|
||||
INSERT INTO negative_cache
|
||||
(source_host, source_path, status_code, expires_at)
|
||||
VALUES (?, ?, 404, datetime('now', '+1 hour'))
|
||||
`, testHostExample, notExpiredPath)
|
||||
if err != nil {
|
||||
t.Fatalf("failed to insert a row that has not expired: %v", err)
|
||||
}
|
||||
|
||||
started := time.Now()
|
||||
|
||||
cache.StartEviction(100 * time.Millisecond)
|
||||
defer func() { _ = cache.StopEviction(t.Context()) }()
|
||||
|
||||
for negativeCacheRowCount(t, cache, expired.SourcePath) > 0 {
|
||||
if time.Since(started) > 10*time.Second {
|
||||
t.Fatal("the expired row is still in negative_cache")
|
||||
}
|
||||
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
}
|
||||
|
||||
t.Logf("the expired row was removed %s after the loop started",
|
||||
time.Since(started))
|
||||
|
||||
if negativeCacheRowCount(t, cache, notExpiredPath) != 1 {
|
||||
t.Error("the row that has not expired was removed")
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// negativeCacheRowCount returns how many negative_cache rows hold
|
||||
// sourcePath.
|
||||
func negativeCacheRowCount(tb testing.TB, cache *Cache, sourcePath string) int {
|
||||
tb.Helper()
|
||||
|
||||
var count int
|
||||
|
||||
err := cache.db.QueryRowContext(tb.Context(),
|
||||
`SELECT COUNT(*) FROM negative_cache WHERE source_path = ?`,
|
||||
sourcePath,
|
||||
).Scan(&count)
|
||||
if err != nil {
|
||||
tb.Fatalf("failed to count negative_cache rows: %v", err)
|
||||
}
|
||||
|
||||
return count
|
||||
}
|
||||
|
||||
// TestCleanExpiredDeletesInBatches stores one expired negative cache row
|
||||
// more than one batch holds, and one row that has not expired. One
|
||||
// statement must delete one batch and no more, so a request waits behind
|
||||
// one batch at most; CleanExpired must then remove the expired row left
|
||||
// over and keep the row that has not expired.
|
||||
func TestCleanExpiredDeletesInBatches(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
cache, err := NewCache(evictionTestDB(t), CacheConfig{StateDir: t.TempDir()})
|
||||
if err != nil {
|
||||
t.Fatalf("failed to create cache: %v", err)
|
||||
}
|
||||
|
||||
ctx := t.Context()
|
||||
|
||||
insertExpiredNegativeCacheRows(t, cache, cleanExpiredBatchSize+1)
|
||||
|
||||
const notExpiredPath = "/later.jpg"
|
||||
|
||||
_, err = cache.db.ExecContext(ctx, `
|
||||
INSERT INTO negative_cache
|
||||
(source_host, source_path, status_code, expires_at)
|
||||
VALUES (?, ?, 404, datetime('now', '+1 hour'))
|
||||
`, testHostExample, notExpiredPath)
|
||||
if err != nil {
|
||||
t.Fatalf("failed to insert a row that has not expired: %v", err)
|
||||
}
|
||||
|
||||
deleted, err := cache.cleanExpiredBatch(ctx)
|
||||
if err != nil {
|
||||
t.Fatalf("cleanExpiredBatch failed: %v", err)
|
||||
}
|
||||
|
||||
t.Logf("one statement deleted %d of %d expired rows",
|
||||
deleted, cleanExpiredBatchSize+1)
|
||||
|
||||
if deleted != cleanExpiredBatchSize {
|
||||
t.Errorf("one statement deleted %d rows, want %d",
|
||||
deleted, cleanExpiredBatchSize)
|
||||
}
|
||||
|
||||
err = cache.CleanExpired(ctx)
|
||||
if err != nil {
|
||||
t.Fatalf("CleanExpired failed: %v", err)
|
||||
}
|
||||
|
||||
if count := negativeCacheTableRowCount(t, cache); count != 1 {
|
||||
t.Errorf("negative_cache holds %d rows after CleanExpired, want 1", count)
|
||||
}
|
||||
|
||||
if negativeCacheRowCount(t, cache, notExpiredPath) != 1 {
|
||||
t.Error("the row that has not expired was removed")
|
||||
}
|
||||
}
|
||||
|
||||
// TestCleanExpiredRemovesMoreThanOneBatch stores one expired negative
|
||||
// cache row more than one batch holds and calls CleanExpired once: it
|
||||
// must remove them all, so it fails if CleanExpired stops after its
|
||||
// first batch.
|
||||
func TestCleanExpiredRemovesMoreThanOneBatch(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
cache, err := NewCache(evictionTestDB(t), CacheConfig{StateDir: t.TempDir()})
|
||||
if err != nil {
|
||||
t.Fatalf("failed to create cache: %v", err)
|
||||
}
|
||||
|
||||
insertExpiredNegativeCacheRows(t, cache, cleanExpiredBatchSize+1)
|
||||
|
||||
before := negativeCacheTableRowCount(t, cache)
|
||||
|
||||
err = cache.CleanExpired(t.Context())
|
||||
if err != nil {
|
||||
t.Fatalf("CleanExpired failed: %v", err)
|
||||
}
|
||||
|
||||
after := negativeCacheTableRowCount(t, cache)
|
||||
|
||||
t.Logf("CleanExpired removed %d of %d expired rows, %d per batch",
|
||||
before-after, before, cleanExpiredBatchSize)
|
||||
|
||||
if after != 0 {
|
||||
t.Errorf("negative_cache holds %d rows after CleanExpired, want 0", after)
|
||||
}
|
||||
}
|
||||
|
||||
// TestCleanExpiredStopsOnceContextEnds runs CleanExpired with a context
|
||||
// that has already ended: it must return the context's error and delete
|
||||
// no row.
|
||||
func TestCleanExpiredStopsOnceContextEnds(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
cache, err := NewCache(evictionTestDB(t), CacheConfig{StateDir: t.TempDir()})
|
||||
if err != nil {
|
||||
t.Fatalf("failed to create cache: %v", err)
|
||||
}
|
||||
|
||||
insertExpiredNegativeCacheRows(t, cache, 1)
|
||||
|
||||
ctx, cancel := context.WithCancel(t.Context())
|
||||
cancel()
|
||||
|
||||
err = cache.CleanExpired(ctx)
|
||||
if !errors.Is(err, context.Canceled) {
|
||||
t.Errorf("CleanExpired() error = %v, want context.Canceled", err)
|
||||
}
|
||||
|
||||
if count := negativeCacheTableRowCount(t, cache); count != 1 {
|
||||
t.Errorf("negative_cache holds %d rows, want the expired row kept", count)
|
||||
}
|
||||
}
|
||||
|
||||
// insertExpiredNegativeCacheRows inserts count negative_cache rows that
|
||||
// expired an hour ago, each for a different source path.
|
||||
func insertExpiredNegativeCacheRows(t *testing.T, cache *Cache, count int) {
|
||||
t.Helper()
|
||||
|
||||
for i := range count {
|
||||
_, err := cache.db.ExecContext(t.Context(), `
|
||||
INSERT INTO negative_cache
|
||||
(source_host, source_path, status_code, expires_at)
|
||||
VALUES (?, ?, 404, datetime('now', '-1 hour'))
|
||||
`, testHostExample, "/expired/"+strconv.Itoa(i)+".jpg")
|
||||
if err != nil {
|
||||
t.Fatalf("failed to insert an expired row: %v", err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// negativeCacheTableRowCount returns how many rows negative_cache holds.
|
||||
func negativeCacheTableRowCount(tb testing.TB, cache *Cache) int {
|
||||
tb.Helper()
|
||||
|
||||
var count int
|
||||
|
||||
err := cache.db.QueryRowContext(tb.Context(),
|
||||
`SELECT COUNT(*) FROM negative_cache`,
|
||||
).Scan(&count)
|
||||
if err != nil {
|
||||
tb.Fatalf("failed to count negative_cache rows: %v", err)
|
||||
}
|
||||
|
||||
return count
|
||||
}
|
||||
|
||||
func TestService_Get_ReturnsErrorForNegativeCachedURL(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
|
||||
@@ -10,7 +10,6 @@ import (
|
||||
"log/slog"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
@@ -372,13 +371,11 @@ func HashPath(path string) PathHash {
|
||||
|
||||
// CacheKey generates a unique key for a request variant.
|
||||
// Format: sha256(host:path:query:width:height:format:quality:fit_mode)
|
||||
// The path and query go through escapePercentAndColon, so that a ":" in
|
||||
// either cannot be taken for the one between them.
|
||||
func CacheKey(req *ImageRequest) VariantKey {
|
||||
data := fmt.Sprintf("%s:%s:%s:%d:%d:%s:%d:%s",
|
||||
req.SourceHost,
|
||||
escapePercentAndColon(req.SourcePath),
|
||||
escapePercentAndColon(req.SourceQuery),
|
||||
req.SourcePath,
|
||||
req.SourceQuery,
|
||||
req.Size.Width,
|
||||
req.Size.Height,
|
||||
req.Format,
|
||||
@@ -390,14 +387,6 @@ func CacheKey(req *ImageRequest) VariantKey {
|
||||
return VariantKey(hex.EncodeToString(h[:]))
|
||||
}
|
||||
|
||||
// escapePercentAndColon returns s with each "%" written as "%25" and each
|
||||
// ":" as "%3A", so that the result holds no ":".
|
||||
func escapePercentAndColon(s string) string {
|
||||
s = strings.ReplaceAll(s, "%", "%25")
|
||||
|
||||
return strings.ReplaceAll(s, ":", "%3A")
|
||||
}
|
||||
|
||||
// VariantStorage handles key-based file storage for processed image variants.
|
||||
// Files are stored at: <basedir>/<ab>/<cd>/<cache_key>
|
||||
// Metadata is stored at: <basedir>/<ab>/<cd>/<cache_key>.meta
|
||||
|
||||
@@ -407,62 +407,6 @@ func TestCacheKey(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// TestCacheKey_ColonOrPercentInPathOrQuery checks that two different source
|
||||
// URLs whose path or query holds ":" or "%" get different keys.
|
||||
func TestCacheKey_ColonOrPercentInPathOrQuery(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
cases := []struct {
|
||||
name string
|
||||
path1, query1 string
|
||||
path2, query2 string
|
||||
}{
|
||||
{
|
||||
name: "colon moved from path to query",
|
||||
path1: "/a:b", query1: "",
|
||||
path2: "/a", query2: "b:",
|
||||
},
|
||||
{
|
||||
name: "colon in path and its percent-encoding",
|
||||
path1: "/a:b", query1: "",
|
||||
path2: "/a%3Ab", query2: "",
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range cases {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
key1 := CacheKey(&ImageRequest{
|
||||
SourceHost: testHostCDN,
|
||||
SourcePath: tt.path1,
|
||||
SourceQuery: tt.query1,
|
||||
Size: Size{Width: 800, Height: 600},
|
||||
Format: FormatWebP,
|
||||
Quality: 85,
|
||||
FitMode: FitCover,
|
||||
})
|
||||
key2 := CacheKey(&ImageRequest{
|
||||
SourceHost: testHostCDN,
|
||||
SourcePath: tt.path2,
|
||||
SourceQuery: tt.query2,
|
||||
Size: Size{Width: 800, Height: 600},
|
||||
Format: FormatWebP,
|
||||
Quality: 85,
|
||||
FitMode: FitCover,
|
||||
})
|
||||
|
||||
t.Logf("CacheKey() of path %q query %q = %s", tt.path1, tt.query1, key1)
|
||||
t.Logf("CacheKey() of path %q query %q = %s", tt.path2, tt.query2, key2)
|
||||
|
||||
if key1 == key2 {
|
||||
t.Errorf("CacheKey() gave path %q query %q and path %q query %q "+
|
||||
"the same key", tt.path1, tt.query1, tt.path2, tt.query2)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// TestVariantStorage_StoreLogsFailedMetaWrite verifies that a .meta write
|
||||
// that fails is logged at warn and the store still succeeds.
|
||||
func TestVariantStorage_StoreLogsFailedMetaWrite(t *testing.T) {
|
||||
|
||||
@@ -1,113 +0,0 @@
|
||||
package signature_test
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"sneak.berlin/go/pixa/internal/signature"
|
||||
)
|
||||
|
||||
// TestSigner_Verify_ColonCannotMovePathQueryBoundary checks that a URL
|
||||
// signed for path "/a:b" with no query does not verify for path "/a" with
|
||||
// query "b:", a different resource on the same host.
|
||||
func TestSigner_Verify_ColonCannotMovePathQueryBoundary(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
signer := signature.New(goldenSigningKey)
|
||||
|
||||
signed := &signature.Request{
|
||||
SourceHost: testHost,
|
||||
SourcePath: "/a:b",
|
||||
Width: 800,
|
||||
Height: 600,
|
||||
Format: testFormatWebP,
|
||||
Quality: 85,
|
||||
FitMode: testFitCover,
|
||||
Expires: time.Now().Add(1 * time.Hour),
|
||||
}
|
||||
signed.Signature = signer.Sign(signed)
|
||||
|
||||
replayed := *signed
|
||||
replayed.SourcePath = "/a"
|
||||
replayed.SourceQuery = "b:"
|
||||
|
||||
t.Logf("signature for path %q query %q: %s",
|
||||
signed.SourcePath, signed.SourceQuery, signed.Signature)
|
||||
t.Logf("signature for path %q query %q: %s",
|
||||
replayed.SourcePath, replayed.SourceQuery, signer.Sign(&replayed))
|
||||
|
||||
err := signer.Verify(&replayed)
|
||||
if !errors.Is(err, signature.ErrInvalid) {
|
||||
t.Errorf("Verify() of path %q query %q with the signature for "+
|
||||
"path %q = %v, want %v",
|
||||
replayed.SourcePath, replayed.SourceQuery, signed.SourcePath,
|
||||
err, signature.ErrInvalid)
|
||||
}
|
||||
}
|
||||
|
||||
// TestSigner_EscapedGoldenVectors pins the signature of requests whose path
|
||||
// or query holds ":" or "%", which are signed with "%" written as "%25" and
|
||||
// ":" as "%3A". The expected values were computed once, outside pixa, from
|
||||
// the signed data in each comment.
|
||||
func TestSigner_EscapedGoldenVectors(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
signer := signature.New(goldenSigningKey)
|
||||
|
||||
vectors := []struct {
|
||||
name string
|
||||
path string
|
||||
query string
|
||||
wantSignature string
|
||||
}{
|
||||
{
|
||||
name: "colon in path",
|
||||
path: "/photos/a:b.jpg",
|
||||
// Signed data:
|
||||
// "cdn.example.com:/photos/a%3Ab.jpg::800:600:webp:1704067200:85:cover"
|
||||
wantSignature: "WnEpZ7bQp1Zvw8WDatlj2vu-4wYKWE3BFBizV8m__6k=",
|
||||
},
|
||||
{
|
||||
name: "colon in query",
|
||||
path: testPath,
|
||||
query: "t=1:2",
|
||||
// Signed data:
|
||||
// "cdn.example.com:/photos/cat.jpg:t=1%3A2:800:600:webp:1704067200:85:cover"
|
||||
wantSignature: "85kRoN8hhr4cCUkEUjNvUsvVYKjYYfWdeffx5TjJjqA=",
|
||||
},
|
||||
{
|
||||
name: "percent in path and query",
|
||||
path: "/photos/a%20b.jpg",
|
||||
query: "q=50%",
|
||||
// Signed data:
|
||||
// "cdn.example.com:/photos/a%2520b.jpg:q=50%25:800:600:webp:1704067200:85:cover"
|
||||
wantSignature: "MjDiiRdgXMrsjqPMsENtD4aG1hp1e-a6zy67vKTX4CQ=",
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range vectors {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
req := &signature.Request{
|
||||
SourceHost: testHost,
|
||||
SourcePath: tt.path,
|
||||
SourceQuery: tt.query,
|
||||
Width: 800,
|
||||
Height: 600,
|
||||
Format: testFormatWebP,
|
||||
Quality: 85,
|
||||
FitMode: testFitCover,
|
||||
Expires: time.Unix(goldenExpiresUnix, 0),
|
||||
}
|
||||
|
||||
gotSignature := signer.Sign(req)
|
||||
t.Logf("Sign() of path %q query %q = %s", tt.path, tt.query, gotSignature)
|
||||
|
||||
if gotSignature != tt.wantSignature {
|
||||
t.Errorf("Sign() = %q, want %q", gotSignature, tt.wantSignature)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -10,7 +10,6 @@ import (
|
||||
"fmt"
|
||||
"net/url"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
@@ -155,17 +154,15 @@ func (s *Signer) GenerateSignedURL(
|
||||
|
||||
// buildSignatureData creates the string to be signed.
|
||||
// Format: "host:path:query:width:height:format:expiration:quality:fit"
|
||||
// The path and query go through escapePercentAndColon, so that a ":" in
|
||||
// either cannot be taken for the one between them; every other component
|
||||
// is used verbatim. No normalization, suffix matching, or wildcard
|
||||
// expansion is performed. Quality and fit are the effective transform
|
||||
// values, so replaying a signed URL with a different quality or fit mode
|
||||
// fails verification.
|
||||
// All components are used verbatim (exact match). No normalization,
|
||||
// suffix matching, or wildcard expansion is performed. Quality and fit
|
||||
// are the effective transform values, so replaying a signed URL with a
|
||||
// different quality or fit mode fails verification.
|
||||
func (s *Signer) buildSignatureData(req *Request) string {
|
||||
return fmt.Sprintf("%s:%s:%s:%d:%d:%s:%d:%d:%s",
|
||||
req.SourceHost,
|
||||
escapePercentAndColon(req.SourcePath),
|
||||
escapePercentAndColon(req.SourceQuery),
|
||||
req.SourcePath,
|
||||
req.SourceQuery,
|
||||
req.Width,
|
||||
req.Height,
|
||||
req.Format,
|
||||
@@ -175,14 +172,6 @@ func (s *Signer) buildSignatureData(req *Request) string {
|
||||
)
|
||||
}
|
||||
|
||||
// escapePercentAndColon returns s with each "%" written as "%25" and each
|
||||
// ":" as "%3A", so that the result holds no ":".
|
||||
func escapePercentAndColon(s string) string {
|
||||
s = strings.ReplaceAll(s, "%", "%25")
|
||||
|
||||
return strings.ReplaceAll(s, ":", "%3A")
|
||||
}
|
||||
|
||||
// ParseParams extracts signature and expiration from query parameters.
|
||||
func ParseParams(sig, expStr string) (string, time.Time, error) {
|
||||
if expStr == "" {
|
||||
|
||||
Reference in New Issue
Block a user