diff --git a/.mise.toml b/.mise.toml index 7309b4c..e3c2286 100644 --- a/.mise.toml +++ b/.mise.toml @@ -3,7 +3,7 @@ go = "1.27.2" [tasks.fmt] description = "Check Go formatting" -run = "test -z \"$(gofmt -l cmd)\"" +run = "test -z \"$(gofmt -l cmd internal)\"" [tasks.vet] description = "Run static analysis" diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index cc8fd84..80f7ee2 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -26,15 +26,25 @@ Run the unit tests with: make test ``` -They need neither ExifTool nor network access. To also run the tests against -the pinned ExifTool release: +They need neither ExifTool nor network access: a fake stripper covers raw and +multipart uploads, request limits, error responses, deadlines and cancellation, +bounded uploads and concurrency, temporary file cleanup, and shutdown. To also +run the tests against the pinned ExifTool release: ```sh make test-integration ``` This downloads and verifies ExifTool if it is missing. The integration suite -fails rather than silently skipping when ExifTool is unavailable. +tags JPEG, PNG, TIFF, GIF, WebP, AVIF, and HEIC files with a location, camera, +and capture time, strips them, and checks that every private tag is gone while +the orientation is kept. A JPEG also checks that its ICC profile is kept byte +for byte. The non-JPEG and non-PNG fixtures in `internal/exiftool/testdata` +are 32x24 gradients made with ImageMagick 7.1 +(`magick -size 32x24 gradient:red-blue photo.`). It also covers files ExifTool cannot rewrite, recovery after a timeout +or an exited process, concurrent requests, and that a failed startup leaves no +processes behind. It fails rather than silently skipping when ExifTool is +unavailable. ## Upgrading ExifTool @@ -48,4 +58,5 @@ checksums in the author's ```text cmd/exiftool-server/ HTTP server, lifecycle, and telemetry +internal/exiftool/ ExifTool process pool and metadata removal ``` diff --git a/README.md b/README.md index 726926e..4efac78 100644 --- a/README.md +++ b/README.md @@ -5,6 +5,15 @@ writing image, video, and document metadata. The service checks for ExifTool at startup and refuses to start without it. +It removes metadata such as location, camera, and capture time from images +while keeping the ICC color profile and orientation, so a stripped photo looks +the same as the original. Pixel data is never re-encoded. JPEG, PNG, TIFF, GIF, +WebP, AVIF, and HEIC are tested. A fixed pool of ExifTool processes runs in +`-stay_open` mode, so each request skips Perl's startup and memory does not +grow with the number of callers. Uploads are written straight to a temporary +file and the response is sent from it, so memory does not grow with file size +either. + ## Requirements - Go 1.27 or newer @@ -32,8 +41,17 @@ The server listens on `:8080`. These environment variables are available: | --- | --- | --- | | `ADDR` | `:8080` | HTTP listen address | | `EXIFTOOL_PATH` | `exiftool` | ExifTool executable, either a path or a name looked up on `PATH` | +| `MAX_CONCURRENT_STRIPS` | available CPUs | Number of ExifTool processes, and so of files stripped concurrently | +| `STRIP_TIMEOUT` | `30s` | Maximum lifetime of a strip request, including upload and queue waits | +| `MAX_REQUEST_SIZE` | `25MiB` | Maximum `/strip` request body, including multipart overhead. Uploads are kept on disk, so this caps temporary files, not memory | | `SHUTDOWN_TIMEOUT` | `30s` | Grace period for active requests during shutdown | +Each ExifTool process uses about 50 MiB, so memory depends on +`MAX_CONCURRENT_STRIPS`, not on how many clients call the service or how +large their files are. With eight processes and 16 concurrent 18 MB uploads, +the HTTP server itself peaked at 14 MiB, and 340 requests for 320 KB JPEGs, 34 +at a time, completed at about 385 per second with a p95 of about 110 ms. + ## Docker The image downloads ExifTool from CPAN, verifies its checksum during the build, @@ -76,7 +94,40 @@ curl -sS http://localhost:8080/readyz `/livez` reports that the process is running. `/readyz` reports whether the service accepts requests and returns 503 as soon as graceful shutdown starts. -`/healthz` is an alias for `/readyz`. +`/healthz` is an alias for `/readyz`. The service only starts listening after +every ExifTool process has started. + +Send a file to strip as the raw request body: + +```sh +curl -sS -X POST http://localhost:8080/strip \ + -H 'Content-Type: application/octet-stream' \ + --data-binary '@photo.jpg' -o stripped.jpg +``` + +Or as a multipart `file` field: + +```sh +curl -sS -X POST http://localhost:8080/strip \ + -F 'file=@photo.jpg' -o stripped.jpg +``` + +A successful response is `200` with the stripped file as +`application/octet-stream`. The format is detected from the content, so a +renamed file is still stripped. Errors are JSON with an `error` message: + +| Status | Meaning | +| --- | --- | +| `400` | Missing or invalid body | +| `413` | Body is larger than `MAX_REQUEST_SIZE` | +| `422` | ExifTool cannot rewrite the file: the format is not writable or the file is corrupt | +| `499` | The client cancelled the request | +| `500` | ExifTool failed; its process is replaced before the next request uses it | +| `503` | The service is shutting down; retry after `Retry-After` | +| `504` | The request took longer than `STRIP_TIMEOUT` | + +Callers that must not lose an upload should keep the original file on any +response other than `200`. ## Telemetry and shutdown @@ -86,13 +137,18 @@ letters, digits, `_`, or `-` and is at most 64 characters. Logs never include request bodies, query strings, or request headers. Prometheus metrics are exposed at `/metrics`. They cover HTTP request rates and -latency, Go runtime and process statistics, and build information including the -ExifTool version. Keep this endpoint private at the ingress layer. +latency, strip stage latency and outcomes, active ExifTool commands, +cancellations, Go runtime and process statistics, and build information +including the ExifTool version. Keep this endpoint private at the ingress layer. + +Every strip request has a deadline. A request that times out or is cancelled +stops its ExifTool process, which is started again the next time it is needed, +so one stuck file never blocks other requests. On SIGINT or SIGTERM, readiness becomes false immediately, and active requests have `SHUTDOWN_TIMEOUT` to finish. After that, request contexts are cancelled -and connections are closed. Set the orchestrator's termination grace period -longer than `SHUTDOWN_TIMEOUT`. +and connections are closed, and the ExifTool processes are stopped. Set the +orchestrator's termination grace period longer than `SHUTDOWN_TIMEOUT`. ## Contributing diff --git a/cmd/exiftool-server/integration_test.go b/cmd/exiftool-server/integration_test.go index f0860da..e735aa8 100644 --- a/cmd/exiftool-server/integration_test.go +++ b/cmd/exiftool-server/integration_test.go @@ -3,9 +3,20 @@ package main import ( + "bytes" "context" + "encoding/json" + "image" + "image/color" + "image/jpeg" + "net/http" "os" + "os/exec" + "path/filepath" "testing" + "time" + + "exiftool/internal/exiftool" ) func TestDetectExiftoolVersionWithRealExiftool(t *testing.T) { @@ -21,3 +32,98 @@ func TestDetectExiftoolVersionWithRealExiftool(t *testing.T) { t.Fatalf("version = %q, want %q", got, want) } } + +func TestStripWithRealExiftool(t *testing.T) { + path := os.Getenv("EXIFTOOL_PATH") + if path == "" { + t.Fatal("EXIFTOOL_PATH must point to an ExifTool executable") + } + pool, err := exiftool.New(context.Background(), exiftool.Options{Binary: path, Size: 2, StartupTimeout: time.Minute}) + if err != nil { + t.Fatal(err) + } + t.Cleanup(pool.Close) + app := newStripApplication(t, pool, 2) + original := taggedJPEG(t, path) + + for _, tc := range []struct { + name string + req *http.Request + }{ + {"raw body", stripRequest(bytes.NewReader(original))}, + {"multipart file", multipartRequest(t, "file", original)}, + } { + t.Run(tc.name, func(t *testing.T) { + rec := serveRequest(app, tc.req) + if rec.Code != http.StatusOK { + t.Fatalf("status = %d, want %d; body = %s", rec.Code, http.StatusOK, rec.Body.String()) + } + stripped := rec.Body.Bytes() + if len(stripped) >= len(original) { + t.Fatalf("stripped size = %d, want less than %d", len(stripped), len(original)) + } + tags := readTags(t, path, stripped) + for _, tag := range []string{"GPSLatitude", "Make"} { + if value, ok := tags[tag]; ok { + t.Errorf("%s = %v was not removed", tag, value) + } + } + if got := tags["Orientation"]; got != float64(6) { + t.Errorf("Orientation = %v, want 6", got) + } + }) + } + + t.Run("unsupported file", func(t *testing.T) { + rec := serveRequest(app, stripRequest(bytes.NewReader([]byte("plain text, not an image")))) + if rec.Code != http.StatusUnprocessableEntity { + t.Fatalf("status = %d, want %d; body = %s", rec.Code, http.StatusUnprocessableEntity, rec.Body.String()) + } + }) +} + +// taggedJPEG returns a JPEG carrying a location, camera, and orientation. +func taggedJPEG(t *testing.T, exiftoolPath string) []byte { + t.Helper() + img := image.NewRGBA(image.Rect(0, 0, 16, 12)) + for x := range 16 { + for y := range 12 { + img.Set(x, y, color.RGBA{R: uint8(x * 15), G: uint8(y * 20), B: 120, A: 255}) + } + } + var buffer bytes.Buffer + if err := jpeg.Encode(&buffer, img, nil); err != nil { + t.Fatal(err) + } + path := filepath.Join(t.TempDir(), "photo.jpg") + if err := os.WriteFile(path, buffer.Bytes(), 0o600); err != nil { + t.Fatal(err) + } + output, err := exec.Command(exiftoolPath, "-GPSLatitude=37.7749", "-GPSLatitudeRef=N", + "-Make=Canon", "-Orientation#=6", "-overwrite_original", path).CombinedOutput() + if err != nil { + t.Fatalf("tag fixture: %v: %s", err, output) + } + data, err := os.ReadFile(path) + if err != nil { + t.Fatal(err) + } + return data +} + +func readTags(t *testing.T, exiftoolPath string, data []byte) map[string]any { + t.Helper() + path := filepath.Join(t.TempDir(), "result") + if err := os.WriteFile(path, data, 0o600); err != nil { + t.Fatal(err) + } + output, err := exec.Command(exiftoolPath, "-j", "-n", path).Output() + if err != nil { + t.Fatalf("read tags: %v", err) + } + var tags []map[string]any + if err := json.Unmarshal(output, &tags); err != nil || len(tags) != 1 { + t.Fatalf("decode tags: %v", err) + } + return tags[0] +} diff --git a/cmd/exiftool-server/main.go b/cmd/exiftool-server/main.go index dc51949..a3fa993 100644 --- a/cmd/exiftool-server/main.go +++ b/cmd/exiftool-server/main.go @@ -5,20 +5,30 @@ import ( "encoding/json" "errors" "fmt" + "io" "log/slog" + "math" + "mime" "net" "net/http" "os" "os/exec" "os/signal" + "runtime" + "strconv" "strings" "sync/atomic" "syscall" "time" + + "exiftool/internal/exiftool" ) const ( defaultExiftoolPath = "exiftool" + defaultMaxRequestBytes = 25 << 20 // 25 MiB: Appwrite strips files up to 20 MB, plus multipart overhead. + maxConcurrentUploads = 4 // Bounds uploads in progress without reserving a process. + defaultStripTimeout = 30 * time.Second defaultShutdownTimeout = 30 * time.Second exiftoolStartupTimeout = 10 * time.Second ) @@ -28,9 +38,20 @@ var ( commit = "unknown" ) +type stripper interface { + // Strip rewrites the file at path in place without its metadata. + Strip(context.Context, string) error +} + type application struct { exiftoolVersion string + stripper stripper + uploadSlots chan struct{} + stripSlots chan struct{} telemetry *telemetry + stripTimeout time.Duration + maxRequestBytes int64 + tempDir string draining atomic.Bool } @@ -53,10 +74,22 @@ func main() { func run() error { addr := envOrDefault("ADDR", ":8080") exiftoolPath := envOrDefault("EXIFTOOL_PATH", defaultExiftoolPath) + maxConcurrentStrips, err := positiveEnvInt("MAX_CONCURRENT_STRIPS", runtime.GOMAXPROCS(0)) + if err != nil { + return fmt.Errorf("startup: %w", err) + } + stripTimeout, err := positiveEnvDuration("STRIP_TIMEOUT", defaultStripTimeout) + if err != nil { + return fmt.Errorf("startup: %w", err) + } shutdownTimeout, err := positiveEnvDuration("SHUTDOWN_TIMEOUT", defaultShutdownTimeout) if err != nil { return fmt.Errorf("startup: %w", err) } + maxRequestBytes, err := positiveEnvBytes("MAX_REQUEST_SIZE", defaultMaxRequestBytes) + if err != nil { + return fmt.Errorf("startup: %w", err) + } versionCtx, cancelVersion := context.WithTimeout(context.Background(), exiftoolStartupTimeout) exiftoolVersion, err := detectExiftoolVersion(versionCtx, exiftoolPath) @@ -66,13 +99,28 @@ func run() error { } slog.Info("exiftool found", "path", exiftoolPath, "version", exiftoolVersion) - app := newApplication(exiftoolVersion) + started := time.Now() + pool, err := exiftool.New(context.Background(), exiftool.Options{ + Binary: exiftoolPath, + Size: maxConcurrentStrips, + StartupTimeout: exiftoolStartupTimeout, + }) + if err != nil { + return fmt.Errorf("startup: %w", err) + } + defer pool.Close() + slog.Info("exiftool processes started", "processes", maxConcurrentStrips, + "duration_ms", time.Since(started).Milliseconds()) + + app := newApplication(exiftoolVersion, pool, maxConcurrentStrips) + app.stripTimeout = stripTimeout + app.maxRequestBytes = maxRequestBytes server := &http.Server{ Addr: addr, Handler: app.telemetry.instrument(app.routes()), ReadHeaderTimeout: 5 * time.Second, - ReadTimeout: 30 * time.Second, - WriteTimeout: 35 * time.Second, + ReadTimeout: stripTimeout, + WriteTimeout: stripTimeout + 5*time.Second, IdleTimeout: 60 * time.Second, } serviceCtx, cancelService := context.WithCancel(context.Background()) @@ -84,7 +132,7 @@ func run() error { defer signal.Stop(shutdownSignals) serverErrors := make(chan error, 1) go func() { - slog.Info("listening", "address", addr) + slog.Info("listening", "address", addr, "max_request_bytes", maxRequestBytes) if err := server.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) { serverErrors <- err } @@ -110,15 +158,21 @@ func run() error { return nil } -func newApplication(exiftoolVersion string) *application { +func newApplication(exiftoolVersion string, stripper stripper, maxConcurrentStrips int) *application { return &application{ exiftoolVersion: exiftoolVersion, - telemetry: newTelemetry(exiftoolVersion), + stripper: stripper, + uploadSlots: make(chan struct{}, maxConcurrentUploads), + stripSlots: make(chan struct{}, maxConcurrentStrips), + telemetry: newTelemetry(exiftoolVersion, maxConcurrentStrips), + stripTimeout: defaultStripTimeout, + maxRequestBytes: defaultMaxRequestBytes, } } func (app *application) routes() http.Handler { mux := http.NewServeMux() + mux.HandleFunc("/strip", app.handleStrip) mux.HandleFunc("/livez", app.handleLivez) mux.HandleFunc("/readyz", app.handleReadyz) mux.HandleFunc("/healthz", app.handleReadyz) @@ -153,13 +207,235 @@ func (app *application) handleReadyz(w http.ResponseWriter, r *http.Request) { writeError(w, http.StatusMethodNotAllowed, "method not allowed") return } - if app.exiftoolVersion == "" || app.draining.Load() { + if app.exiftoolVersion == "" || app.stripper == nil || app.draining.Load() { writeError(w, http.StatusServiceUnavailable, "service not ready") return } writeJSON(w, http.StatusOK, healthResponse{Status: "ok"}) } +func (app *application) handleStrip(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + w.Header().Set("Allow", http.MethodPost) + writeError(w, http.StatusMethodNotAllowed, "method not allowed") + return + } + if app.draining.Load() { + w.Header().Set("Retry-After", "1") + writeError(w, http.StatusServiceUnavailable, "service is shutting down") + return + } + ctx, cancel := context.WithTimeout(r.Context(), app.stripTimeout) + defer cancel() + r = r.WithContext(ctx) + outcome := "success" + defer func() { app.telemetry.strips.WithLabelValues(outcome).Inc() }() + observeCancellation := func() { + if errors.Is(ctx.Err(), context.DeadlineExceeded) { + outcome = "timeout" + app.telemetry.cancellations.WithLabelValues("deadline").Inc() + } else { + outcome = "cancelled" + app.telemetry.cancellations.WithLabelValues("client").Inc() + } + } + uploadWaitStarted := time.Now() + select { + case app.uploadSlots <- struct{}{}: + app.telemetry.stageDuration.WithLabelValues("upload_queue").Observe(time.Since(uploadWaitStarted).Seconds()) + case <-ctx.Done(): + observeCancellation() + writeContextError(w, ctx) + return + } + uploadSlotHeld := true + defer func() { + if uploadSlotHeld { + <-app.uploadSlots + } + }() + + // No extension, so ExifTool detects the format from the content and a + // renamed file is still stripped. + file, err := os.CreateTemp(app.tempDir, "strip-*") + if err != nil { + outcome = "internal_error" + slog.Error("create temporary file failed", "error", err) + writeError(w, http.StatusInternalServerError, "strip failed") + return + } + path := file.Name() + defer os.Remove(path) + + // The upload goes straight to the file ExifTool rewrites, so memory does not + // grow with file size. + r.Body = http.MaxBytesReader(w, r.Body, app.maxRequestBytes) + defer r.Body.Close() + upload := &fileWriter{file: file} + receiveErr := receiveFile(r, upload) + if err := errors.Join(upload.err, file.Close()); err != nil { + outcome = "internal_error" + slog.Error("write temporary file failed", "error", err) + writeError(w, http.StatusInternalServerError, "strip failed") + return + } + if receiveErr != nil { + // The server's read deadline starts before this handler, so a slow upload + // can hit it first. The server then also cancels the request context, + // which would otherwise look like the client leaving. + var netErr net.Error + if errors.As(receiveErr, &netErr) && netErr.Timeout() { + outcome = "timeout" + app.telemetry.cancellations.WithLabelValues("deadline").Inc() + writeError(w, http.StatusGatewayTimeout, "strip timed out") + return + } + if ctx.Err() != nil { + observeCancellation() + writeContextError(w, ctx) + return + } + outcome = "invalid_request" + var tooLarge *http.MaxBytesError + if errors.As(receiveErr, &tooLarge) { + writeError(w, http.StatusRequestEntityTooLarge, + fmt.Sprintf("file exceeds the %s request limit", formatByteSize(app.maxRequestBytes))) + return + } + writeError(w, http.StatusBadRequest, receiveErr.Error()) + return + } + stripWaitStarted := time.Now() + select { + case app.stripSlots <- struct{}{}: + app.telemetry.stageDuration.WithLabelValues("strip_queue").Observe(time.Since(stripWaitStarted).Seconds()) + defer func() { <-app.stripSlots }() + case <-ctx.Done(): + observeCancellation() + writeContextError(w, ctx) + return + } + // Keep the upload slot while waiting for a process so the uploads waiting on + // disk remain bounded. Release it once the strip slot is secured. + <-app.uploadSlots + uploadSlotHeld = false + + stripStarted := time.Now() + app.telemetry.stripActive.Inc() + err = app.stripper.Strip(ctx, path) + app.telemetry.stripActive.Dec() + app.telemetry.stageDuration.WithLabelValues("strip").Observe(time.Since(stripStarted).Seconds()) + if err != nil { + if ctx.Err() != nil { + observeCancellation() + writeContextError(w, ctx) + return + } + if errors.Is(err, exiftool.ErrNotStripped) { + outcome = "not_stripped" + writeError(w, http.StatusUnprocessableEntity, "file format is not supported or the file is corrupt") + return + } + outcome = "strip_error" + slog.Error("strip failed", "error", err) + writeError(w, http.StatusInternalServerError, "strip failed") + return + } + + stripped, err := os.Open(path) + if err != nil { + outcome = "internal_error" + slog.Error("open stripped file failed", "error", err) + writeError(w, http.StatusInternalServerError, "strip failed") + return + } + defer stripped.Close() + info, err := stripped.Stat() + if err != nil || info.Size() == 0 { + outcome = "internal_error" + slog.Error("stripped file is unreadable or empty", "error", err) + writeError(w, http.StatusInternalServerError, "strip failed") + return + } + w.Header().Set("Content-Type", "application/octet-stream") + w.Header().Set("Content-Length", strconv.FormatInt(info.Size(), 10)) + w.WriteHeader(http.StatusOK) + if _, err := io.Copy(w, stripped); err != nil { + slog.Error("write response failed", "error", err) + } +} + +// fileWriter keeps the temporary file's write error apart from upload errors, +// so a full disk is not reported as a bad request. +type fileWriter struct { + file *os.File + err error +} + +func (w *fileWriter) Write(data []byte) (int, error) { + n, err := w.file.Write(data) + if err != nil { + w.err = err + } + return n, err +} + +func writeContextError(w http.ResponseWriter, ctx context.Context) { + if errors.Is(ctx.Err(), context.DeadlineExceeded) { + writeError(w, http.StatusGatewayTimeout, "strip timed out") + return + } + writeError(w, 499, "request cancelled") +} + +func receiveFile(r *http.Request, dst io.Writer) error { + contentType, params, err := mime.ParseMediaType(r.Header.Get("Content-Type")) + if err != nil { + return errors.New("invalid Content-Type header") + } + if contentType != "multipart/form-data" { + written, err := io.Copy(dst, r.Body) + if err != nil { + return fmt.Errorf("read file: %w", err) + } + if written == 0 { + return errors.New("empty file body") + } + return nil + } + + if params["boundary"] == "" { + return errors.New("multipart boundary is missing") + } + reader, err := r.MultipartReader() + if err != nil { + return fmt.Errorf("read multipart body: %w", err) + } + for { + part, err := reader.NextPart() + if errors.Is(err, io.EOF) { + break + } + if err != nil { + return fmt.Errorf("read multipart body: %w", err) + } + if part.FormName() != "file" { + part.Close() + continue + } + written, copyErr := io.Copy(dst, part) + part.Close() + if copyErr != nil { + return fmt.Errorf("read file field: %w", copyErr) + } + if written == 0 { + return errors.New("file field is empty") + } + return nil + } + return errors.New("multipart field \"file\" is required") +} + func writeError(w http.ResponseWriter, status int, message string) { writeJSON(w, status, errorResponse{Error: message}) } @@ -179,6 +455,18 @@ func envOrDefault(key, fallback string) string { return fallback } +func positiveEnvInt(key string, fallback int) (int, error) { + value := strings.TrimSpace(os.Getenv(key)) + if value == "" { + return fallback, nil + } + parsed, err := strconv.Atoi(value) + if err != nil || parsed < 1 { + return 0, fmt.Errorf("%s must be a positive integer, got %q", key, value) + } + return parsed, nil +} + func positiveEnvDuration(key string, fallback time.Duration) (time.Duration, error) { value := strings.TrimSpace(os.Getenv(key)) if value == "" { @@ -190,3 +478,68 @@ func positiveEnvDuration(key string, fallback time.Duration) (time.Duration, err } return parsed, nil } + +func positiveEnvBytes(key string, fallback int64) (int64, error) { + value := strings.TrimSpace(os.Getenv(key)) + if value == "" { + return fallback, nil + } + parsed, err := parseByteSize(value) + if err != nil { + return 0, fmt.Errorf("%s must be a positive byte size such as 26214400 or 25MiB, got %q", key, value) + } + return parsed, nil +} + +func parseByteSize(value string) (int64, error) { + value = strings.TrimSpace(value) + if value == "" { + return 0, errors.New("empty byte size") + } + upper := strings.ToUpper(value) + for _, unit := range []struct { + suffix string + bytes int64 + }{ + {"GIB", 1 << 30}, + {"MIB", 1 << 20}, + {"KIB", 1 << 10}, + {"GB", 1_000_000_000}, + {"MB", 1_000_000}, + {"KB", 1_000}, + {"B", 1}, + } { + if !strings.HasSuffix(upper, unit.suffix) { + continue + } + number := strings.TrimSpace(value[:len(value)-len(unit.suffix)]) + parsed, err := strconv.ParseInt(number, 10, 64) + if err != nil || parsed < 1 { + return 0, fmt.Errorf("invalid byte size %q", value) + } + if parsed > math.MaxInt64/unit.bytes { + return 0, fmt.Errorf("byte size %q overflows", value) + } + return parsed * unit.bytes, nil + } + parsed, err := strconv.ParseInt(value, 10, 64) + if err != nil || parsed < 1 { + return 0, fmt.Errorf("invalid byte size %q", value) + } + return parsed, nil +} + +func formatByteSize(n int64) string { + switch { + case n <= 0: + return fmt.Sprintf("%d bytes", n) + case n%(1<<30) == 0: + return fmt.Sprintf("%d GiB", n>>30) + case n%(1<<20) == 0: + return fmt.Sprintf("%d MiB", n>>20) + case n%(1<<10) == 0: + return fmt.Sprintf("%d KiB", n>>10) + default: + return fmt.Sprintf("%d bytes", n) + } +} diff --git a/cmd/exiftool-server/main_test.go b/cmd/exiftool-server/main_test.go index 3a4490d..be4de1e 100644 --- a/cmd/exiftool-server/main_test.go +++ b/cmd/exiftool-server/main_test.go @@ -1,16 +1,23 @@ package main import ( + "bytes" "context" "encoding/json" + "errors" "io" + "mime/multipart" "net/http" "net/http/httptest" "os" "path/filepath" "strings" + "sync" + "sync/atomic" "testing" "time" + + "exiftool/internal/exiftool" ) func serve(app *application, method, path string, header http.Header) *httptest.ResponseRecorder { @@ -24,7 +31,7 @@ func serve(app *application, method, path string, header http.Header) *httptest. } func TestHealthEndpoints(t *testing.T) { - app := newApplication("13.55") + app := newApplication("13.55", fakeStripper(), 1) for _, path := range []string{"/livez", "/readyz", "/healthz"} { t.Run(path, func(t *testing.T) { rec := serve(app, http.MethodGet, path, nil) @@ -46,7 +53,7 @@ func TestHealthEndpoints(t *testing.T) { } func TestHealthEndpointsRejectOtherMethods(t *testing.T) { - app := newApplication("13.55") + app := newApplication("13.55", fakeStripper(), 1) for _, path := range []string{"/livez", "/readyz", "/healthz"} { t.Run(path, func(t *testing.T) { rec := serve(app, http.MethodPost, path, nil) @@ -61,7 +68,7 @@ func TestHealthEndpointsRejectOtherMethods(t *testing.T) { } func TestReadinessDuringShutdown(t *testing.T) { - app := newApplication("13.55") + app := newApplication("13.55", fakeStripper(), 1) app.draining.Store(true) if rec := serve(app, http.MethodGet, "/readyz", nil); rec.Code != http.StatusServiceUnavailable { t.Fatalf("readyz status = %d, want %d", rec.Code, http.StatusServiceUnavailable) @@ -72,14 +79,14 @@ func TestReadinessDuringShutdown(t *testing.T) { } func TestReadinessWithoutExiftool(t *testing.T) { - app := newApplication("") + app := newApplication("", fakeStripper(), 1) if rec := serve(app, http.MethodGet, "/readyz", nil); rec.Code != http.StatusServiceUnavailable { t.Fatalf("status = %d, want %d", rec.Code, http.StatusServiceUnavailable) } } func TestRequestID(t *testing.T) { - app := newApplication("13.55") + app := newApplication("13.55", fakeStripper(), 1) rec := serve(app, http.MethodGet, "/livez", http.Header{"X-Request-Id": {"caller-id_1"}}) if got := rec.Header().Get("X-Request-ID"); got != "caller-id_1" { t.Fatalf("X-Request-ID = %q, want caller-id_1", got) @@ -91,7 +98,7 @@ func TestRequestID(t *testing.T) { } func TestMetrics(t *testing.T) { - app := newApplication("13.55") + app := newApplication("13.55", fakeStripper(), 1) serve(app, http.MethodGet, "/unknown", nil) rec := serve(app, http.MethodGet, "/metrics", nil) if rec.Code != http.StatusOK { @@ -102,7 +109,7 @@ func TestMetrics(t *testing.T) { t.Fatal(err) } for _, want := range []string{ - `exiftool_build_info{commit="unknown",exiftool_version="13.55",version="dev"} 1`, + `exiftool_build_info{commit="unknown",exiftool_version="13.55",max_concurrent_strips="1",version="dev"} 1`, `exiftool_http_requests_total{method="GET",route="unmatched",status="404"} 1`, } { if !strings.Contains(string(body), want) { @@ -165,3 +172,453 @@ func TestPositiveEnvDuration(t *testing.T) { }) } } + +const ( + testFile = "IMAGE-DATA;GPS=37.7749" + testStripped = "IMAGE-DATA" +) + +type stripperFunc func(context.Context, string) error + +func (f stripperFunc) Strip(ctx context.Context, path string) error { + return f(ctx, path) +} + +// fakeStripper keeps everything before the first ';', standing in for ExifTool. +func fakeStripper() stripperFunc { + return func(_ context.Context, path string) error { + data, err := os.ReadFile(path) + if err != nil { + return err + } + if index := bytes.IndexByte(data, ';'); index >= 0 { + data = data[:index] + } + return os.WriteFile(path, data, 0o600) + } +} + +type gatedStripper struct { + started chan struct{} + release chan struct{} + once sync.Once +} + +func (s *gatedStripper) Strip(ctx context.Context, path string) error { + s.once.Do(func() { close(s.started) }) + <-s.release + return fakeStripper()(ctx, path) +} + +type countingReader struct { + reader io.Reader + started *atomic.Int32 + once sync.Once +} + +func (r *countingReader) Read(data []byte) (int, error) { + r.once.Do(func() { r.started.Add(1) }) + return r.reader.Read(data) +} + +func newStripApplication(t *testing.T, stripper stripper, maxConcurrentStrips int) *application { + t.Helper() + app := newApplication("13.55", stripper, maxConcurrentStrips) + app.tempDir = t.TempDir() + return app +} + +func stripRequest(body io.Reader) *http.Request { + req := httptest.NewRequest(http.MethodPost, "/strip", body) + req.Header.Set("Content-Type", "application/octet-stream") + return req +} + +func multipartRequest(t *testing.T, field string, data []byte) *http.Request { + t.Helper() + var body bytes.Buffer + writer := multipart.NewWriter(&body) + part, err := writer.CreateFormFile(field, "photo.jpg") + if err != nil { + t.Fatal(err) + } + if _, err := part.Write(data); err != nil { + t.Fatal(err) + } + if err := writer.Close(); err != nil { + t.Fatal(err) + } + req := httptest.NewRequest(http.MethodPost, "/strip", &body) + req.Header.Set("Content-Type", writer.FormDataContentType()) + return req +} + +func serveRequest(app *application, req *http.Request) *httptest.ResponseRecorder { + rec := httptest.NewRecorder() + app.telemetry.instrument(app.routes()).ServeHTTP(rec, req) + return rec +} + +func TestStrip(t *testing.T) { + for _, tc := range []struct { + name string + req func(t *testing.T) *http.Request + }{ + {"raw body", func(*testing.T) *http.Request { return stripRequest(strings.NewReader(testFile)) }}, + {"multipart file", func(t *testing.T) *http.Request { return multipartRequest(t, "file", []byte(testFile)) }}, + } { + t.Run(tc.name, func(t *testing.T) { + rec := serveRequest(newStripApplication(t, fakeStripper(), 1), tc.req(t)) + if rec.Code != http.StatusOK { + t.Fatalf("status = %d, want %d; body = %s", rec.Code, http.StatusOK, rec.Body.String()) + } + if got := rec.Body.String(); got != testStripped { + t.Fatalf("body = %q, want %q", got, testStripped) + } + if got := rec.Header().Get("Content-Type"); got != "application/octet-stream" { + t.Fatalf("Content-Type = %q, want application/octet-stream", got) + } + if got := rec.Header().Get("Content-Length"); got != "10" { + t.Fatalf("Content-Length = %q, want 10", got) + } + }) + } +} + +func TestStripRemovesTemporaryFile(t *testing.T) { + for _, tc := range []struct { + name string + stripper stripper + }{ + {"success", fakeStripper()}, + {"failure", stripperFunc(func(context.Context, string) error { return exiftool.ErrNotStripped })}, + } { + t.Run(tc.name, func(t *testing.T) { + app := newStripApplication(t, tc.stripper, 1) + serveRequest(app, stripRequest(strings.NewReader(testFile))) + + entries, err := os.ReadDir(app.tempDir) + if err != nil { + t.Fatal(err) + } + if len(entries) != 0 { + t.Fatalf("temporary directory has %d entries, want 0", len(entries)) + } + }) + } +} + +func TestStripErrors(t *testing.T) { + for _, tc := range []struct { + name string + method string + contentType string + body []byte + stripper stripper + want int + }{ + {name: "method", method: http.MethodGet, want: http.StatusMethodNotAllowed}, + {name: "missing content type", method: http.MethodPost, body: []byte(testFile), want: http.StatusBadRequest}, + {name: "empty body", method: http.MethodPost, contentType: "application/octet-stream", want: http.StatusBadRequest}, + {name: "multipart without boundary", method: http.MethodPost, contentType: "multipart/form-data", body: []byte(testFile), want: http.StatusBadRequest}, + {name: "request too large", method: http.MethodPost, contentType: "application/octet-stream", body: make([]byte, defaultMaxRequestBytes+1), want: http.StatusRequestEntityTooLarge}, + { + name: "not stripped", method: http.MethodPost, contentType: "application/octet-stream", body: []byte(testFile), + stripper: stripperFunc(func(context.Context, string) error { return exiftool.ErrNotStripped }), + want: http.StatusUnprocessableEntity, + }, + { + name: "process failure", method: http.MethodPost, contentType: "application/octet-stream", body: []byte(testFile), + stripper: stripperFunc(func(context.Context, string) error { return errors.New("exiftool exited") }), + want: http.StatusInternalServerError, + }, + { + name: "empty result", method: http.MethodPost, contentType: "application/octet-stream", body: []byte(testFile), + stripper: stripperFunc(func(_ context.Context, path string) error { return os.WriteFile(path, nil, 0o600) }), + want: http.StatusInternalServerError, + }, + } { + t.Run(tc.name, func(t *testing.T) { + stripper := tc.stripper + if stripper == nil { + stripper = fakeStripper() + } + req := httptest.NewRequest(tc.method, "/strip", bytes.NewReader(tc.body)) + if tc.contentType != "" { + req.Header.Set("Content-Type", tc.contentType) + } + rec := serveRequest(newStripApplication(t, stripper, 1), req) + if rec.Code != tc.want { + t.Fatalf("status = %d, want %d; body = %s", rec.Code, tc.want, rec.Body.String()) + } + var body errorResponse + if err := json.NewDecoder(rec.Body).Decode(&body); err != nil || body.Error == "" { + t.Fatalf("error response = %+v, %v; want a message", body, err) + } + }) + } +} + +func TestStripMultipartErrors(t *testing.T) { + for _, tc := range []struct { + name, field string + data []byte + }{ + {"missing file field", "image", []byte(testFile)}, + {"empty file field", "file", nil}, + } { + t.Run(tc.name, func(t *testing.T) { + rec := serveRequest(newStripApplication(t, fakeStripper(), 1), multipartRequest(t, tc.field, tc.data)) + if rec.Code != http.StatusBadRequest { + t.Fatalf("status = %d, want %d; body = %s", rec.Code, http.StatusBadRequest, rec.Body.String()) + } + }) + } +} + +func TestStripConfiguredRequestLimit(t *testing.T) { + app := newStripApplication(t, fakeStripper(), 1) + app.maxRequestBytes = int64(len(testFile)) + + if rec := serveRequest(app, stripRequest(strings.NewReader(testFile))); rec.Code != http.StatusOK { + t.Fatalf("status at the limit = %d, want %d", rec.Code, http.StatusOK) + } + if rec := serveRequest(app, stripRequest(strings.NewReader(testFile+"!"))); rec.Code != http.StatusRequestEntityTooLarge { + t.Fatalf("status above the limit = %d, want %d", rec.Code, http.StatusRequestEntityTooLarge) + } +} + +func TestStripDeadlineCancelsStrip(t *testing.T) { + cancelled := make(chan struct{}) + app := newStripApplication(t, stripperFunc(func(ctx context.Context, _ string) error { + <-ctx.Done() + close(cancelled) + return ctx.Err() + }), 1) + app.stripTimeout = 20 * time.Millisecond + + rec := serveRequest(app, stripRequest(strings.NewReader(testFile))) + + if rec.Code != http.StatusGatewayTimeout { + t.Fatalf("status = %d, want %d; body = %s", rec.Code, http.StatusGatewayTimeout, rec.Body.String()) + } + select { + case <-cancelled: + default: + t.Fatal("strip context was not cancelled") + } +} + +func TestStripClientCancellation(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + app := newStripApplication(t, stripperFunc(func(ctx context.Context, _ string) error { + cancel() + <-ctx.Done() + return ctx.Err() + }), 1) + + rec := serveRequest(app, stripRequest(strings.NewReader(testFile)).WithContext(ctx)) + + if rec.Code != 499 { + t.Fatalf("status = %d, want 499; body = %s", rec.Code, rec.Body.String()) + } +} + +func TestStripBoundsConcurrentWork(t *testing.T) { + var inFlight, maxInFlight atomic.Int32 + unblock := make(chan struct{}) + sawTwo := make(chan struct{}) + var sawTwoOnce, releaseOnce sync.Once + release := func() { releaseOnce.Do(func() { close(unblock) }) } + stripper := stripperFunc(func(ctx context.Context, path string) error { + n := inFlight.Add(1) + defer inFlight.Add(-1) + for { + maximum := maxInFlight.Load() + if n <= maximum || maxInFlight.CompareAndSwap(maximum, n) { + break + } + } + if n >= 2 { + sawTwoOnce.Do(func() { close(sawTwo) }) + } + select { + case <-unblock: + case <-time.After(3 * time.Second): + } + return fakeStripper()(ctx, path) + }) + const concurrency = 2 + app := newStripApplication(t, stripper, concurrency) + + var wait sync.WaitGroup + for range 8 { + wait.Go(func() { + if rec := serveRequest(app, stripRequest(strings.NewReader(testFile))); rec.Code != http.StatusOK { + t.Errorf("status = %d, want %d", rec.Code, http.StatusOK) + } + }) + } + select { + case <-sawTwo: + case <-time.After(3 * time.Second): + release() + wait.Wait() + t.Fatalf("maximum concurrent strips = %d, want %d", maxInFlight.Load(), concurrency) + } + release() + wait.Wait() + if got := maxInFlight.Load(); got != concurrency { + t.Fatalf("maximum concurrent strips = %d, want %d", got, concurrency) + } +} + +func TestStripBoundsBufferedUploads(t *testing.T) { + stripper := &gatedStripper{started: make(chan struct{}), release: make(chan struct{})} + // The only strip slot is occupied, so this isolates upload admission. + app := newStripApplication(t, stripper, 1) + + var wait sync.WaitGroup + startRequest := func(body io.Reader) { + wait.Go(func() { + if rec := serveRequest(app, stripRequest(body)); rec.Code != http.StatusOK { + t.Errorf("status = %d, want %d", rec.Code, http.StatusOK) + } + }) + } + + startRequest(strings.NewReader(testFile)) + <-stripper.started // The strip slot is occupied, but its upload slot is free. + + var bodiesStarted atomic.Int32 + for range maxConcurrentUploads + 1 { + startRequest(&countingReader{reader: strings.NewReader(testFile), started: &bodiesStarted}) + } + deadline := time.After(time.Second) + for bodiesStarted.Load() < maxConcurrentUploads { + select { + case <-deadline: + t.Fatalf("only %d request bodies started", bodiesStarted.Load()) + default: + time.Sleep(time.Millisecond) + } + } + time.Sleep(20 * time.Millisecond) + if got := bodiesStarted.Load(); got != maxConcurrentUploads { + t.Fatalf("buffered request bodies = %d, want %d", got, maxConcurrentUploads) + } + + close(stripper.release) + wait.Wait() +} + +func TestStripDuringShutdown(t *testing.T) { + app := newStripApplication(t, fakeStripper(), 1) + app.draining.Store(true) + + rec := serveRequest(app, stripRequest(strings.NewReader(testFile))) + + if rec.Code != http.StatusServiceUnavailable || rec.Header().Get("Retry-After") != "1" { + t.Fatalf("status = %d, Retry-After = %q; want %d, 1", rec.Code, rec.Header().Get("Retry-After"), http.StatusServiceUnavailable) + } +} + +func TestReadinessWithoutStripper(t *testing.T) { + app := newApplication("13.55", nil, 1) + if rec := serve(app, http.MethodGet, "/readyz", nil); rec.Code != http.StatusServiceUnavailable { + t.Fatalf("status = %d, want %d", rec.Code, http.StatusServiceUnavailable) + } +} + +func TestStripMetrics(t *testing.T) { + app := newStripApplication(t, fakeStripper(), 1) + serveRequest(app, stripRequest(strings.NewReader(testFile))) + rec := serve(app, http.MethodGet, "/metrics", nil) + body, err := io.ReadAll(rec.Body) + if err != nil { + t.Fatal(err) + } + for _, want := range []string{ + `exiftool_strips_total{outcome="success"} 1`, + `exiftool_http_requests_total{method="POST",route="/strip",status="200"} 1`, + } { + if !strings.Contains(string(body), want) { + t.Errorf("metrics missing %s", want) + } + } +} + +func TestPositiveEnvInt(t *testing.T) { + for _, tc := range []struct { + value string + want int + invalid bool + }{ + {"", 3, false}, + {"8", 8, false}, + {"0", 0, true}, + {"-1", 0, true}, + {"many", 0, true}, + } { + t.Run(tc.value, func(t *testing.T) { + t.Setenv("MAX_CONCURRENT_STRIPS", tc.value) + got, err := positiveEnvInt("MAX_CONCURRENT_STRIPS", 3) + if (err != nil) != tc.invalid || got != tc.want { + t.Fatalf("positiveEnvInt() = %v, %v; want %v, invalid=%v", got, err, tc.want, tc.invalid) + } + }) + } +} + +func TestParseByteSize(t *testing.T) { + for _, tc := range []struct { + value string + want int64 + invalid bool + }{ + {"1024", 1024, false}, + {"25MiB", 25 << 20, false}, + {"20MB", 20_000_000, false}, + {"1 KiB", 1 << 10, false}, + {"0", 0, true}, + {"-5MB", 0, true}, + {"lots", 0, true}, + {"9999999999GiB", 0, true}, + } { + t.Run(tc.value, func(t *testing.T) { + got, err := parseByteSize(tc.value) + if (err != nil) != tc.invalid || got != tc.want { + t.Fatalf("parseByteSize() = %v, %v; want %v, invalid=%v", got, err, tc.want, tc.invalid) + } + }) + } +} + +func TestStripSlowUploadTimesOut(t *testing.T) { + app := newStripApplication(t, fakeStripper(), 1) + server := httptest.NewUnstartedServer(app.telemetry.instrument(app.routes())) + // Shorter than the strip timeout, so the server's read deadline fires first. + server.Config.ReadTimeout = 100 * time.Millisecond + server.Start() + t.Cleanup(server.Close) + + body, writer := io.Pipe() + t.Cleanup(func() { writer.Close() }) + go func() { + _, _ = writer.Write([]byte("IMAGE")) + }() + req, err := http.NewRequest(http.MethodPost, server.URL+"/strip", body) + if err != nil { + t.Fatal(err) + } + req.Header.Set("Content-Type", "application/octet-stream") + + res, err := server.Client().Do(req) + if err != nil { + t.Fatal(err) + } + defer res.Body.Close() + if res.StatusCode != http.StatusGatewayTimeout { + t.Fatalf("status = %d, want %d", res.StatusCode, http.StatusGatewayTimeout) + } +} diff --git a/cmd/exiftool-server/telemetry.go b/cmd/exiftool-server/telemetry.go index d6ac0f8..1fd91bc 100644 --- a/cmd/exiftool-server/telemetry.go +++ b/cmd/exiftool-server/telemetry.go @@ -14,13 +14,17 @@ import ( ) type telemetry struct { - registry *prometheus.Registry - httpRequests *prometheus.CounterVec - httpDuration *prometheus.HistogramVec - httpActive prometheus.Gauge + registry *prometheus.Registry + httpRequests *prometheus.CounterVec + httpDuration *prometheus.HistogramVec + httpActive prometheus.Gauge + stageDuration *prometheus.HistogramVec + strips *prometheus.CounterVec + stripActive prometheus.Gauge + cancellations *prometheus.CounterVec } -func newTelemetry(exiftoolVersion string) *telemetry { +func newTelemetry(exiftoolVersion string, maxConcurrentStrips int) *telemetry { registry := prometheus.NewRegistry() t := &telemetry{ registry: registry, @@ -36,14 +40,32 @@ func newTelemetry(exiftoolVersion string) *telemetry { Namespace: "exiftool", Subsystem: "http", Name: "requests_active", Help: "HTTP requests currently being served.", }), + stageDuration: prometheus.NewHistogramVec(prometheus.HistogramOpts{ + Namespace: "exiftool", Name: "stage_duration_seconds", + Help: "Strip pipeline stage duration.", + }, []string{"stage"}), + strips: prometheus.NewCounterVec(prometheus.CounterOpts{ + Namespace: "exiftool", Name: "strips_total", + Help: "Strip requests by outcome.", + }, []string{"outcome"}), + stripActive: prometheus.NewGauge(prometheus.GaugeOpts{ + Namespace: "exiftool", Name: "strip_active", + Help: "ExifTool commands currently running.", + }), + cancellations: prometheus.NewCounterVec(prometheus.CounterOpts{ + Namespace: "exiftool", Name: "cancellations_total", + Help: "Cancelled strips by reason.", + }, []string{"reason"}), } registry.MustRegister(t.httpRequests, t.httpDuration, t.httpActive, + t.stageDuration, t.strips, t.stripActive, t.cancellations, collectors.NewGoCollector(), collectors.NewProcessCollector(collectors.ProcessCollectorOpts{})) registry.MustRegister(prometheus.NewGaugeFunc(prometheus.GaugeOpts{ - Namespace: "exiftool", Name: "build_info", Help: "Build and ExifTool version.", + Namespace: "exiftool", Name: "build_info", Help: "Build, ExifTool version, and pool size.", ConstLabels: prometheus.Labels{ "version": version, "commit": commit, "exiftool_version": exiftoolVersion, + "max_concurrent_strips": strconv.Itoa(maxConcurrentStrips), }, }, func() float64 { return 1 })) return t @@ -106,7 +128,7 @@ func (t *telemetry) instrument(next http.Handler) http.Handler { func telemetryRoute(path string) string { switch path { - case "/livez", "/readyz", "/healthz", "/metrics": + case "/strip", "/livez", "/readyz", "/healthz", "/metrics": return path default: return "unmatched" diff --git a/internal/exiftool/exiftool.go b/internal/exiftool/exiftool.go new file mode 100644 index 0000000..025815c --- /dev/null +++ b/internal/exiftool/exiftool.go @@ -0,0 +1,334 @@ +// Package exiftool removes metadata from files with ExifTool processes kept +// running in -stay_open mode, so each request skips Perl's startup. +package exiftool + +import ( + "bufio" + "context" + _ "embed" + "errors" + "fmt" + "io" + "os" + "os/exec" + "path/filepath" + "strconv" + "strings" + "sync" + "time" +) + +// ErrNotStripped means ExifTool ran but could not rewrite the file, for +// example because the format is not writable or the file is corrupt. +var ErrNotStripped = errors.New("exiftool could not rewrite the file") + +var errClosed = errors.New("exiftool pool is closed") + +// A 1x1 JPEG stripped once per process at startup, so ExifTool loads its +// writer modules before the first real request. +// +//go:embed warmup.jpg +var warmUpJPEG []byte + +const closeTimeout = 5 * time.Second + +// Options controls the processes started by the pool. +type Options struct { + // Binary is the ExifTool executable. Defaults to "exiftool" on PATH. + Binary string + // Size is the number of ExifTool processes, and so the number of files + // stripped concurrently. + Size int + // StartupTimeout bounds starting a process and stripping the warm-up file. + StartupTimeout time.Duration +} + +// Pool owns a fixed number of ExifTool processes. A process that fails or +// times out is killed and replaced on its next use. +type Pool struct { + binary string + size int + startupTimeout time.Duration + // Holds exactly size entries while idle. A nil entry is a slot whose + // process must be started before use. + processes chan *process + // Closed by Close, so callers waiting for a process stop waiting. + done chan struct{} + closeOnce sync.Once +} + +// New starts and warms up every process. It fails if any process fails, so a +// misconfigured service never reports ready. +func New(ctx context.Context, options Options) (*Pool, error) { + if options.Binary == "" { + options.Binary = "exiftool" + } + if options.Size < 1 { + return nil, errors.New("pool size must be at least one") + } + if options.StartupTimeout <= 0 { + return nil, errors.New("startup timeout must be positive") + } + + pool := &Pool{ + binary: options.Binary, + size: options.Size, + startupTimeout: options.StartupTimeout, + processes: make(chan *process, options.Size), + done: make(chan struct{}), + } + + results := make(chan error, options.Size) + for range options.Size { + go func() { + proc, err := pool.start(ctx) + pool.processes <- proc + results <- err + }() + } + // Processes usually fail for the same reason, so one error is enough. + var startErr error + for range options.Size { + if err := <-results; err != nil && startErr == nil { + startErr = err + } + } + if startErr != nil { + pool.Close() + return nil, startErr + } + return pool, nil +} + +// Strip removes all metadata from the file at path, keeping the ICC color +// profile and orientation. It returns ErrNotStripped when ExifTool cannot +// rewrite the file; the file is left unchanged in that case. +func (p *Pool) Strip(ctx context.Context, path string) error { + if ctx == nil { + return errors.New("strip context is required") + } + // Each argument is one line of ExifTool's argument file. + if !filepath.IsAbs(path) || strings.ContainsAny(path, "\r\n") { + return errors.New("path must be absolute and must not contain line breaks") + } + if p.isClosed() { + return errClosed + } + if err := ctx.Err(); err != nil { + return err + } + + var proc *process + select { + case proc = <-p.processes: + case <-ctx.Done(): + return ctx.Err() + case <-p.done: + return errClosed + } + if p.isClosed() { + p.processes <- proc + return errClosed + } + + if proc == nil { + var err error + if proc, err = p.start(ctx); err != nil { + p.processes <- nil + return err + } + } + + output, err := proc.execute(ctx, stripArguments(path)) + if err != nil { + // The process may still be writing this file or holding its output, so it + // can't serve another request. + proc.kill() + p.processes <- nil + return err + } + p.processes <- proc + + return parseStatus(output) +} + +// Close stops every process, waiting for strips in progress to finish. +func (p *Pool) Close() { + p.closeOnce.Do(func() { + close(p.done) + for range p.size { + if proc := <-p.processes; proc != nil { + proc.close() + } + } + }) +} + +func (p *Pool) isClosed() bool { + select { + case <-p.done: + return true + default: + return false + } +} + +func (p *Pool) start(ctx context.Context) (*process, error) { + ctx, cancel := context.WithTimeout(ctx, p.startupTimeout) + defer cancel() + + proc, err := startProcess(p.binary) + if err != nil { + return nil, err + } + if err := proc.warmUp(ctx); err != nil { + proc.kill() + return nil, fmt.Errorf("warm up exiftool: %w", err) + } + return proc, nil +} + +func stripArguments(path string) []string { + return []string{ + // Print only the exit status: 0 when the file was written or had + // nothing to remove, 1 when it could not be rewritten. + "-q", "-q", + "-echo3", "${status}", + "-all=", + "-tagsfromfile", "@", + "-icc_profile", + "-orientation", + // -all= can't delete a TIFF's first IFD, which holds the image data, so + // this clears the camera and other common tags left in it. + "-CommonIFD0=", + "-overwrite_original", + path, + } +} + +func parseStatus(output string) error { + switch strings.TrimSpace(output) { + case "0": + return nil + case "1": + return ErrNotStripped + default: + return fmt.Errorf("unexpected exiftool output %q", output) + } +} + +type process struct { + cmd *exec.Cmd + stdin io.WriteCloser + stdout *bufio.Reader + sequence int + waitOnce sync.Once +} + +func startProcess(binary string) (*process, error) { + cmd := exec.Command(binary, "-stay_open", "True", "-@", "-") + stdin, err := cmd.StdinPipe() + if err != nil { + return nil, fmt.Errorf("start exiftool: %w", err) + } + stdout, err := cmd.StdoutPipe() + if err != nil { + return nil, fmt.Errorf("start exiftool: %w", err) + } + // Stderr is discarded: failures are detected from the summary on stdout. + if err := cmd.Start(); err != nil { + return nil, fmt.Errorf("start exiftool: %w", err) + } + return &process{cmd: cmd, stdin: stdin, stdout: bufio.NewReader(stdout)}, nil +} + +func (p *process) warmUp(ctx context.Context) error { + file, err := os.CreateTemp("", "exiftool-warmup-*") + if err != nil { + return err + } + defer os.Remove(file.Name()) + _, writeErr := file.Write(warmUpJPEG) + if err := errors.Join(writeErr, file.Close()); err != nil { + return err + } + + output, err := p.execute(ctx, stripArguments(file.Name())) + if err != nil { + return err + } + if err := parseStatus(output); err != nil { + return fmt.Errorf("strip the warm-up image: %w", err) + } + return nil +} + +// execute runs one command and returns its output. A numbered -execute makes +// ExifTool end the output with a matching {readyN} marker. +func (p *process) execute(ctx context.Context, args []string) (string, error) { + if err := ctx.Err(); err != nil { + return "", err + } + p.sequence++ + marker := "{ready" + strconv.Itoa(p.sequence) + "}" + command := strings.Join(args, "\n") + "\n-execute" + strconv.Itoa(p.sequence) + "\n" + + type result struct { + output string + err error + } + done := make(chan result, 1) + go func() { + if _, err := io.WriteString(p.stdin, command); err != nil { + done <- result{err: fmt.Errorf("send command to exiftool: %w", err)} + return + } + var output strings.Builder + for { + line, err := p.stdout.ReadString('\n') + if strings.TrimRight(line, "\r\n") == marker { + done <- result{output: output.String()} + return + } + output.WriteString(line) + if err != nil { + done <- result{err: fmt.Errorf("read exiftool output: %w", err)} + return + } + } + }() + + select { + case r := <-done: + return r.output, r.err + case <-ctx.Done(): + // Killing the process closes its pipes, which unblocks the goroutine. + p.kill() + <-done + return "", ctx.Err() + } +} + +func (p *process) kill() { + _ = p.cmd.Process.Kill() + p.wait() +} + +func (p *process) close() { + _, _ = io.WriteString(p.stdin, "-stay_open\nFalse\n") + _ = p.stdin.Close() + exited := make(chan struct{}) + go func() { + p.wait() + close(exited) + }() + select { + case <-exited: + case <-time.After(closeTimeout): + p.kill() + } +} + +func (p *process) wait() { + p.waitOnce.Do(func() { _ = p.cmd.Wait() }) +} diff --git a/internal/exiftool/exiftool_test.go b/internal/exiftool/exiftool_test.go new file mode 100644 index 0000000..39bf816 --- /dev/null +++ b/internal/exiftool/exiftool_test.go @@ -0,0 +1,370 @@ +//go:build integration + +package exiftool + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "image" + "image/color" + "image/png" + "os" + "os/exec" + "path/filepath" + "strings" + "sync" + "testing" + "time" +) + +// Tags written to every fixture. ExifTool must remove all of them. +var privateTags = []string{ + "-GPSLatitude=37.7749", "-GPSLatitudeRef=N", + "-GPSLongitude=122.4194", "-GPSLongitudeRef=W", + "-Make=Canon", "-Model=EOS 5D", + "-DateTimeOriginal=2026:01:01 10:00:00", + "-XMP-dc:Creator=Jane Doe", +} + +func exiftoolPath(t *testing.T) string { + t.Helper() + path := os.Getenv("EXIFTOOL_PATH") + if path == "" { + t.Fatal("EXIFTOOL_PATH must point to an ExifTool executable") + } + return path +} + +func newPool(t *testing.T, size int) *Pool { + t.Helper() + pool, err := New(context.Background(), Options{Binary: exiftoolPath(t), Size: size, StartupTimeout: time.Minute}) + if err != nil { + t.Fatal(err) + } + t.Cleanup(pool.Close) + return pool +} + +func writeFixture(t *testing.T, name string, data []byte, tags ...string) string { + t.Helper() + path := filepath.Join(t.TempDir(), name) + if err := os.WriteFile(path, data, 0o600); err != nil { + t.Fatal(err) + } + if len(tags) > 0 { + args := append(append([]string{}, tags...), "-overwrite_original", path) + if output, err := exec.Command(exiftoolPath(t), args...).CombinedOutput(); err != nil { + t.Fatalf("tag fixture: %v: %s", err, output) + } + } + return path +} + +func testPNG(t *testing.T) []byte { + t.Helper() + img := image.NewRGBA(image.Rect(0, 0, 8, 6)) + for x := range 8 { + for y := range 6 { + img.Set(x, y, color.RGBA{R: uint8(x * 30), G: uint8(y * 40), B: 200, A: 255}) + } + } + var buffer bytes.Buffer + if err := png.Encode(&buffer, img); err != nil { + t.Fatal(err) + } + return buffer.Bytes() +} + +func readTags(t *testing.T, path string) map[string]any { + t.Helper() + output, err := exec.Command(exiftoolPath(t), "-j", "-G0", "-n", path).Output() + if err != nil { + t.Fatalf("read tags: %v", err) + } + var tags []map[string]any + if err := json.Unmarshal(output, &tags); err != nil || len(tags) != 1 { + t.Fatalf("decode tags: %v", err) + } + return tags[0] +} + +// assertNoPrivateTags checks every group, since a format without EXIF, like +// GIF, stores the same tags in XMP. +func assertNoPrivateTags(t *testing.T, tags map[string]any) { + t.Helper() + for key, value := range tags { + for _, name := range []string{"GPSLatitude", "GPSLongitude", "Make", "Model", "DateTimeOriginal", "Creator"} { + if strings.HasSuffix(key, ":"+name) { + t.Errorf("%s = %v was not removed", key, value) + } + } + } +} + +func TestStripRemovesMetadata(t *testing.T) { + pool := newPool(t, 1) + // The formats Appwrite strips; HEIC covers HEIF. + for _, tc := range []struct { + name string + data []byte + }{ + {"jpeg", warmUpJPEG}, + {"png", testPNG(t)}, + {"tiff", readFixture(t, "photo.tiff")}, + {"gif", readFixture(t, "photo.gif")}, + {"webp", readFixture(t, "photo.webp")}, + {"avif", readFixture(t, "photo.avif")}, + {"heic", readFixture(t, "photo.heic")}, + } { + t.Run(tc.name, func(t *testing.T) { + // No extension: the format must come from the content. + path := writeFixture(t, "upload", tc.data, append([]string{"-Orientation#=6"}, privateTags...)...) + if !hasTag(readTags(t, path), "Make", "Canon") { + t.Fatal("fixture was not tagged") + } + + if err := pool.Strip(context.Background(), path); err != nil { + t.Fatal(err) + } + + tags := readTags(t, path) + assertNoPrivateTags(t, tags) + if !hasTag(tags, "Orientation", float64(6)) { + t.Errorf("orientation was removed: %v", tags) + } + }) + } +} + +// readFixture reads an untagged image generated with ImageMagick. +func readFixture(t *testing.T, name string) []byte { + t.Helper() + data, err := os.ReadFile(filepath.Join("testdata", name)) + if err != nil { + t.Fatal(err) + } + return data +} + +// hasTag reports whether any group, such as EXIF or XMP, holds the tag with this value. +func hasTag(tags map[string]any, name string, want any) bool { + for key, value := range tags { + if strings.HasSuffix(key, ":"+name) && value == want { + return true + } + } + return false +} + +func TestStripKeepsOrientationAndColorProfile(t *testing.T) { + pool := newPool(t, 1) + path := writeFixture(t, "photo.jpg", warmUpJPEG, append([]string{"-Orientation#=6"}, privateTags...)...) + profile := writeICCProfile(t) + if output, err := exec.Command(exiftoolPath(t), "-icc_profile<="+profile, "-overwrite_original", path).CombinedOutput(); err != nil { + t.Fatalf("add profile: %v: %s", err, output) + } + want := readProfile(t, path) + if len(want) == 0 { + t.Fatal("fixture has no ICC profile") + } + + if err := pool.Strip(context.Background(), path); err != nil { + t.Fatal(err) + } + + tags := readTags(t, path) + assertNoPrivateTags(t, tags) + if got := tags["EXIF:Orientation"]; got != float64(6) { + t.Errorf("EXIF:Orientation = %v, want 6", got) + } + if got := readProfile(t, path); !bytes.Equal(got, want) { + t.Errorf("ICC profile = %d bytes after strip, want the original %d bytes", len(got), len(want)) + } +} + +func readProfile(t *testing.T, path string) []byte { + t.Helper() + output, err := exec.Command(exiftoolPath(t), "-b", "-icc_profile", path).Output() + if err != nil { + t.Fatalf("read profile: %v", err) + } + return output +} + +// writeICCProfile writes the smallest ICC header ExifTool accepts as a profile. +func writeICCProfile(t *testing.T) string { + t.Helper() + header := make([]byte, 132) + size := len(header) + header[0], header[1], header[2], header[3] = byte(size>>24), byte(size>>16), byte(size>>8), byte(size) + copy(header[4:], "none") + header[8] = 4 // Version 4.0. + copy(header[12:], "mntr") + copy(header[16:], "RGB ") + copy(header[20:], "XYZ ") + copy(header[36:], "acsp") + path := filepath.Join(t.TempDir(), "profile.icc") + if err := os.WriteFile(path, header, 0o600); err != nil { + t.Fatal(err) + } + return path +} + +func TestStripLeavesUnsupportedFileUnchanged(t *testing.T) { + pool := newPool(t, 1) + data := []byte("GPSLatitude 37.7749 is just text here") + path := writeFixture(t, "notes", data) + + err := pool.Strip(context.Background(), path) + + if !errors.Is(err, ErrNotStripped) { + t.Fatalf("Strip() error = %v, want ErrNotStripped", err) + } + got, readErr := os.ReadFile(path) + if readErr != nil { + t.Fatal(readErr) + } + if !bytes.Equal(got, data) { + t.Fatal("unsupported file was modified") + } +} + +func TestStripRejectsUnsafePaths(t *testing.T) { + pool := newPool(t, 1) + for _, path := range []string{"relative.jpg", " /tmp/photo.jpg", "/tmp/line\nbreak.jpg", "/tmp/line\rbreak.jpg"} { + if err := pool.Strip(context.Background(), path); err == nil { + t.Errorf("Strip(%q) succeeded, want an error", path) + } + } +} + +func TestStripRecoversAfterTimeout(t *testing.T) { + pool := newPool(t, 1) + path := writeFixture(t, "photo.jpg", warmUpJPEG, privateTags...) + expired, cancel := context.WithTimeout(context.Background(), time.Nanosecond) + defer cancel() + <-expired.Done() + + // An expired context fails before a process is used. + if err := pool.Strip(expired, path); !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("Strip() error = %v, want deadline exceeded", err) + } + + // Cancelling mid-command kills the process; the slot must be replaced on next use. + slow, cancelSlow := context.WithCancel(context.Background()) + go func() { + time.Sleep(time.Millisecond) + cancelSlow() + }() + _ = pool.Strip(slow, path) + + if err := pool.Strip(context.Background(), path); err != nil { + t.Fatalf("Strip() after cancellation: %v", err) + } + assertNoPrivateTags(t, readTags(t, path)) +} + +func TestStripReplacesExitedProcess(t *testing.T) { + pool := newPool(t, 1) + proc := <-pool.processes + proc.kill() + pool.processes <- proc + path := writeFixture(t, "photo.jpg", warmUpJPEG, privateTags...) + + // The dead process fails this request and is discarded. + if err := pool.Strip(context.Background(), path); err == nil { + t.Fatal("Strip() with an exited process succeeded") + } + + if err := pool.Strip(context.Background(), path); err != nil { + t.Fatalf("Strip() after replacement: %v", err) + } + assertNoPrivateTags(t, readTags(t, path)) +} + +func TestStripConcurrentRequests(t *testing.T) { + pool := newPool(t, 2) + var wait sync.WaitGroup + for i := range 12 { + path := writeFixture(t, "photo", warmUpJPEG, privateTags...) + wait.Add(1) + go func() { + defer wait.Done() + if err := pool.Strip(context.Background(), path); err != nil { + t.Errorf("request %d: %v", i, err) + return + } + assertNoPrivateTags(t, readTags(t, path)) + }() + } + wait.Wait() +} + +func TestStripAfterClose(t *testing.T) { + pool := newPool(t, 1) + pool.Close() + path := writeFixture(t, "photo.jpg", warmUpJPEG) + + if err := pool.Strip(context.Background(), path); !errors.Is(err, errClosed) { + t.Fatalf("Strip() error = %v, want errClosed", err) + } +} + +func TestNewFailsWithoutExifTool(t *testing.T) { + _, err := New(context.Background(), Options{Binary: "/nonexistent/exiftool", Size: 2, StartupTimeout: time.Second}) + if err == nil { + t.Fatal("New() succeeded with a missing binary") + } +} + +func TestNewStopsProcessesWhenWarmUpTimesOut(t *testing.T) { + before := countProcesses(t) + + _, err := New(context.Background(), Options{Binary: exiftoolPath(t), Size: 2, StartupTimeout: time.Microsecond}) + + if err == nil { + t.Fatal("New() succeeded with an impossible startup timeout") + } + if after := countProcesses(t); after != before { + t.Fatalf("exiftool processes = %d after failed startup, want %d", after, before) + } +} + +func countProcesses(t *testing.T) int { + t.Helper() + entries, err := os.ReadDir("/proc") + if err != nil { + t.Fatalf("read /proc: %v", err) + } + count := 0 + for _, entry := range entries { + cmdline, err := os.ReadFile(filepath.Join("/proc", entry.Name(), "cmdline")) + if err == nil && bytes.Contains(cmdline, []byte("exiftool")) && bytes.Contains(cmdline, []byte("-stay_open")) { + count++ + } + } + return count +} + +func TestCloseWakesWaitingStrip(t *testing.T) { + pool := newPool(t, 1) + path := writeFixture(t, "photo.jpg", warmUpJPEG) + proc := <-pool.processes // Hold the only process so the next Strip waits. + + result := make(chan error, 1) + go func() { result <- pool.Strip(context.Background(), path) }() + time.Sleep(50 * time.Millisecond) + go pool.Close() + + select { + case err := <-result: + if !errors.Is(err, errClosed) { + t.Fatalf("Strip() error = %v, want errClosed", err) + } + case <-time.After(5 * time.Second): + t.Fatal("Strip() kept waiting after Close") + } + pool.processes <- proc +} diff --git a/internal/exiftool/testdata/photo.avif b/internal/exiftool/testdata/photo.avif new file mode 100644 index 0000000..879c4b2 Binary files /dev/null and b/internal/exiftool/testdata/photo.avif differ diff --git a/internal/exiftool/testdata/photo.gif b/internal/exiftool/testdata/photo.gif new file mode 100644 index 0000000..7db72b4 Binary files /dev/null and b/internal/exiftool/testdata/photo.gif differ diff --git a/internal/exiftool/testdata/photo.heic b/internal/exiftool/testdata/photo.heic new file mode 100644 index 0000000..6ae7401 Binary files /dev/null and b/internal/exiftool/testdata/photo.heic differ diff --git a/internal/exiftool/testdata/photo.tiff b/internal/exiftool/testdata/photo.tiff new file mode 100644 index 0000000..016ae16 Binary files /dev/null and b/internal/exiftool/testdata/photo.tiff differ diff --git a/internal/exiftool/testdata/photo.webp b/internal/exiftool/testdata/photo.webp new file mode 100644 index 0000000..4e56583 Binary files /dev/null and b/internal/exiftool/testdata/photo.webp differ diff --git a/internal/exiftool/warmup.jpg b/internal/exiftool/warmup.jpg new file mode 100644 index 0000000..cd0c205 Binary files /dev/null and b/internal/exiftool/warmup.jpg differ