From 14fa9c8000be74d0f30b4d5c9e744ab3bd5754ce Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Sun, 31 May 2026 17:11:53 +0000 Subject: [PATCH] Add S3-compatible storage backend --- README.md | 2 +- docs/cli.md | 2 +- docs/config.md | 35 +- docs/internal/app.md | 2 +- docs/internal/config.md | 4 +- docs/internal/publish.md | 6 +- docs/internal/storage.md | 7 +- docs/operations.md | 18 +- docs/policy/architecture.md | 3 +- docs/policy/development.md | 10 +- docs/troubleshooting.md | 49 ++- examples/s3-destination.yml | 22 ++ go.mod | 18 + go.sum | 36 ++ internal/adapters/s3/backend.go | 430 +++++++++++++++++++++++ internal/adapters/s3/backend_test.go | 357 +++++++++++++++++++ internal/adapters/s3/integration_test.go | 39 ++ internal/adapters/s3/options.go | 42 +++ internal/app/backends.go | 76 +++- internal/app/backends_test.go | 84 ++++- internal/config/config.go | 4 +- internal/config/defaults.go | 19 + internal/config/load_test.go | 69 ++++ internal/config/s3.go | 19 + internal/config/secrets.go | 3 + internal/config/secrets_test.go | 11 + internal/config/validate.go | 12 +- 27 files changed, 1334 insertions(+), 45 deletions(-) create mode 100644 examples/s3-destination.yml create mode 100644 internal/adapters/s3/backend.go create mode 100644 internal/adapters/s3/backend_test.go create mode 100644 internal/adapters/s3/integration_test.go create mode 100644 internal/adapters/s3/options.go create mode 100644 internal/config/s3.go diff --git a/README.md b/README.md index 060d954..292ce40 100644 --- a/README.md +++ b/README.md @@ -2,7 +2,7 @@ `distributor` validates manifested report bundles and publishes selected source or generated artifacts to configured destinations. -It is a local-first CLI with SSH/SFTP support: source bundles can be read from local or SSH storage, destinations can be local directories or SSH paths, and Markdown files can be rendered to HTML sidecars. +It is a local-first CLI with SSH/SFTP and S3-compatible storage support: source bundles can be read from local or remote storage, destinations can be local directories or remote paths, and Markdown files can be rendered to HTML sidecars. Run the local example pipeline: diff --git a/docs/cli.md b/docs/cli.md index 03be4c8..89dbf96 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -23,7 +23,7 @@ distributor inspect - `validate`: validates a local source bundle directory or a local tree containing source bundles. - `inspect`: validates local source bundles and prints normalized bundle metadata. -`validate` and `inspect` accept local paths only. `run` executes `local` and `ssh` backends. S3 config can be parsed and validated, but configured S3 execution fails with a clear unsupported-execution error. +`validate` and `inspect` accept local paths only. `run` executes `local`, `ssh`, and `s3` backends. ## Flag reference diff --git a/docs/config.md b/docs/config.md index 8f8a576..1d3ee80 100644 --- a/docs/config.md +++ b/docs/config.md @@ -10,7 +10,7 @@ If `--config` is omitted, `run` uses: /usr/local/etc/distributor/config.yml ``` -Config parsing rejects unknown YAML fields. The executable backends are `local` and `ssh`. S3 config fields are accepted by config validation, but runtime execution for S3 is unavailable. +Config parsing rejects unknown YAML fields. The executable backends are `local`, `ssh`, and `s3`. ## Minimal Local Config @@ -94,9 +94,9 @@ Source backend: - `host_key_policy`: optional for `ssh`; defaults to `accept-new`. - `endpoint`: required for `s3`. - `bucket`: required for `s3`. -- `prefix`: optional for `s3`. -- `region`: optional for `s3`. -- `force_path_style`: optional for `s3`. +- `prefix`: optional for `s3`; leading and trailing slashes are trimmed. +- `region`: optional for `s3`; defaults to `us-east-1`. +- `force_path_style`: optional for `s3`; defaults to `true`. Set `false` only for services that require virtual-host addressing. - `credentials.access_key_id_env`: optional S3 credential environment variable name. - `credentials.secret_access_key_env`: optional S3 credential environment variable name. @@ -112,7 +112,7 @@ Accepted backend names: - `local`: executable; requires `path`. - `ssh`: executable; requires `host` and `path`. -- `s3`: config validation only; execution is unavailable. +- `s3`: executable; requires `endpoint` and `bucket`. ## SSH Backend @@ -139,6 +139,26 @@ Host key policies: `accept-new` and `strict` use `known_hosts` when configured. If omitted, distributor uses the current service user's default OpenSSH `known_hosts` path where practical. `accept-new` fails when it needs to persist a new host key and no writable `known_hosts` path is available. It does not create a missing parent `.ssh` directory. +## S3 Backend + +S3 uses the AWS SDK for Go v2 and supports S3-compatible endpoints: + +```yaml +backend: s3 +endpoint: https://s3.example.com +bucket: reports +prefix: archive +region: us-east-1 +force_path_style: true +credentials: + access_key_id_env: DISTRIBUTOR_S3_ACCESS_KEY_ID + secret_access_key_env: DISTRIBUTOR_S3_SECRET_ACCESS_KEY +``` + +`endpoint` and `bucket` are required. `prefix` is an optional backend root; it is treated as an object-key prefix, not a real directory. Prefixes must be clean slash-separated paths after trimming leading and trailing slashes. `http://` endpoints are allowed for explicitly configured local development or local S3-compatible test services. + +If either credential environment variable name is configured, both must be configured and both referenced variables must resolve to non-empty values through the real process environment or `secrets.directory`. Explicit credentials take precedence over the AWS SDK default credential chain. If credential environment variable names are omitted, the SDK default credential chain is used and `secrets.directory` values are not injected into the process environment. + Publish policy: - `publish.source`: publish source artifacts. @@ -160,6 +180,8 @@ Defaults are applied after YAML decoding and before validation: - `validation.on_digest_mismatch: fail` - SSH `port: 22` - SSH `host_key_policy: accept-new` +- S3 `region: us-east-1` +- S3 `force_path_style: true` - `publish.source: true` - `publish.html: false` - `transfer.on_destination_same: skip` @@ -185,8 +207,6 @@ S3 credentials may name environment variables: - `credentials.access_key_id_env` - `credentials.secret_access_key_env` -S3 execution is unavailable; these fields are accepted so config shape can be validated. - ## Examples Maintained examples live under [examples](../examples/): @@ -196,3 +216,4 @@ Maintained examples live under [examples](../examples/): - `local-html.yml`: runnable local HTML publication. - `fan-out.yml`: runnable local fan-out publication to source and HTML destinations. - `ssh-destination.yml`: environment-gated local-to-SSH publication example. +- `s3-destination.yml`: environment-gated local-to-S3 publication example. diff --git a/docs/internal/app.md b/docs/internal/app.md index aeb4ffa..38a5e3f 100644 --- a/docs/internal/app.md +++ b/docs/internal/app.md @@ -27,7 +27,7 @@ Destination failures are collected while later destinations continue to run. Sou ## Backend and transform wiring -The app-level backend factory registers local and SSH backends for execution. Config validation accepts S3 shape, but `Run` cannot execute S3 sources or destinations. +The app-level backend factory registers local, SSH, and S3 backends for execution. S3 explicit credential references are resolved through the config environment resolver before adapter construction. The app-level transform registry registers Markdown-to-HTML using `internal/transform/markdown`. Lower-level publish code receives a resolver and does not import concrete transform implementations. diff --git a/docs/internal/config.md b/docs/internal/config.md index 1169e6d..526385a 100644 --- a/docs/internal/config.md +++ b/docs/internal/config.md @@ -37,10 +37,12 @@ Validation requires at least one pipeline, slug-like unique pipeline ids, one so ## Executable support boundary -Config validation accepts `local`, `ssh`, and `s3` backend shapes so config files can be validated as schemas. Runtime execution opens local and SSH backends through `internal/app`; S3 remains accepted by validation but unavailable at execution. +Config validation accepts `local`, `ssh`, and `s3` backend shapes. Runtime execution opens all three through `internal/app`. SSH config uses structured fields: `host`, optional `user`, optional `port`, `path`, optional `ssh_key_file`, optional `known_hosts`, and optional `host_key_policy`. `host_key_policy` accepts YAML booleans and strings and normalizes `true`/`strict`, `accept-new`, and `false`/`off`. +S3 config requires `endpoint` and `bucket`, normalizes optional `prefix`, defaults `region` to `us-east-1`, and defaults omitted `force_path_style` to `true` while preserving explicit `false`. + ## Secrets and credential resolution `secrets.directory` points to a directory of credential files. `LoadSecretEnvironment` reads regular files and symlinks to regular files, rejects invalid filenames, trims exactly one trailing LF or CRLF, and returns an `Environment` resolver plus conflict metadata. diff --git a/docs/internal/publish.md b/docs/internal/publish.md index eceebc1..4e2c608 100644 --- a/docs/internal/publish.md +++ b/docs/internal/publish.md @@ -22,17 +22,17 @@ Execution fails if a write, delete, state serialization, or context check fails. ## Boundaries -The current implementation publishes source files and Markdown-to-HTML sidecar outputs. Backend behavior is supplied through `internal/storage`; app runtime currently supplies local backends. +The current implementation publishes source files and Markdown-to-HTML sidecar outputs. Backend behavior is supplied through `internal/storage`; app runtime currently supplies local, SSH, and S3 backends. The package uses `internal/state` for destination comparison, `internal/storage` for IO, and the shared `internal/config` publish/transform policy helper for request validation. It resolves transforms through a narrow resolver supplied by the caller; concrete transform registration is owned by the app layer. It does not parse CLI flags or load config files. ## Safety -Replacement deletes only outputs recorded in existing destination state plus `.distributor.json`. Failed local writes trigger cleanup of outputs written during the failed attempt. +Replacement deletes only outputs recorded in existing destination state plus `.distributor.json`. Failed writes trigger cleanup of outputs written during the failed attempt where practical. ## Tests -Before changing publish behavior, inspect tests under `internal/publish` and local run tests under `internal/app`. +Before changing publish behavior, inspect tests under `internal/publish` and run tests under `internal/app`. ## Invariants diff --git a/docs/internal/storage.md b/docs/internal/storage.md index afb71a1..6ac8544 100644 --- a/docs/internal/storage.md +++ b/docs/internal/storage.md @@ -14,7 +14,7 @@ Entries report a logical path, type, and size when available. Entry types are `f Core packages should depend on `internal/storage`, not adapter packages. Adapter-specific path handling stays behind backend implementations. -The local adapter lives in `internal/adapters/local`. The SSH/SFTP adapter lives in `internal/adapters/ssh`. Runtime backend construction is wired through the app-level backend factory and storage registry. The fake backend lives in `internal/storage/fake` for tests and is not registered for runtime use. +The local adapter lives in `internal/adapters/local`. The SSH/SFTP adapter lives in `internal/adapters/ssh`. The S3-compatible adapter lives in `internal/adapters/s3`. Runtime backend construction is wired through the app-level backend factory and storage registry. The fake backend lives in `internal/storage/fake` for tests and is not registered for runtime use. ## Paths @@ -30,12 +30,14 @@ Backends may wrap implementation-specific errors, but callers should receive sto Backends expose guarded managed deletion only. `DeleteManagedBundle` may delete listed managed outputs plus `.distributor.json`; it does not provide broad recursive deletion. -## Local, SSH, and fake backends +## Local, SSH, S3, and fake backends The local adapter maps logical paths to a configured filesystem root and keeps adapter-specific path handling behind the storage interface. The SSH adapter maps logical paths to a configured remote SFTP root. It uses native SSH and SFTP libraries, supports SSH agent and key-file authentication, applies host-key policies, rejects unsafe logical paths, reports symlink entries from `Lstat`, and limits deletion to managed targets. +The S3 adapter maps logical paths to object keys below a configured bucket and optional prefix. It uses the AWS SDK for Go v2, treats prefixes as object trees, requires exact objects for `Stat`, paginates traversal, applies conservative overwrite checks with `HeadObject`, infers basic content types, and limits deletion to managed target objects. + The fake backend is an in-memory implementation for package tests. It is not registered for runtime use. ## Tests @@ -46,6 +48,7 @@ Before changing storage behavior, inspect tests under: - `internal/storage/fake` - `internal/adapters/local` - `internal/adapters/ssh` +- `internal/adapters/s3` ## Invariants diff --git a/docs/operations.md b/docs/operations.md index 4f48adb..5d7c7ab 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -38,6 +38,12 @@ Preview an environment-gated SSH destination config after editing it for an SSH/ go run ./cmd/distributor run --config examples/ssh-destination.yml --dry-run ``` +Preview an environment-gated S3 destination config after editing it for an S3-compatible endpoint and bucket you control: + +```sh +go run ./cmd/distributor run --config examples/s3-destination.yml --dry-run +``` + ## Filesystem Layout Source bundles are discovered beneath the configured source root. Each bundle is a directory containing `manifest.json`. @@ -48,6 +54,8 @@ The maintained local examples write under `workspace/`, which is ignored by Git. SSH backends use the configured remote `path` as the backend root. Source bundle discovery and destination bundle paths are relative to that root, using the same logical path rules as local storage. +S3 backends use the configured bucket plus optional `prefix` as the backend root. Source bundle discovery and destination bundle paths are relative to that object-key prefix. Prefixes are object-key prefixes, not real directories. + ## Destination State Each published destination bundle contains `.distributor.json`. This file is the managed sentinel and destination state record. It stores: @@ -96,6 +104,14 @@ The default host key policy is `accept-new`. New host keys are written to `known Recovery boundaries are the same as local storage: replacement deletes only managed output paths recorded in `.distributor.json` plus the state file, and failed writes are cleaned up where practical. Distributor never performs broad recursive remote deletion. +## S3 Operation Notes + +S3 execution uses the AWS SDK for Go v2. Configure `endpoint`, `bucket`, optional `prefix`, optional `region`, and optional explicit credential environment variable names. + +When explicit credential env names are configured, both variables must resolve to non-empty values through the real process environment or `secrets.directory`. When they are omitted, the AWS SDK default credential chain is used as-is. + +Replacement and failed-write cleanup delete only managed output objects recorded in `.distributor.json` plus the state object. Distributor does not perform recursive prefix deletion and does not manage bucket versioning or delete markers. + ## Secrets Directory Configure `secrets.directory` when credential values should come from mounted files, such as deployment secrets: @@ -111,6 +127,6 @@ Real process environment values take precedence over files with the same name. I ## Caveats -S3 execution, external notification adapters, and force overwrite behavior are unavailable. +External notification adapters and force overwrite behavior are unavailable. For symptom-oriented fixes, see [troubleshooting](troubleshooting.md). For config details, see [configuration](config.md). For command syntax, see [CLI](cli.md). diff --git a/docs/policy/architecture.md b/docs/policy/architecture.md index aa224d3..3b5583b 100644 --- a/docs/policy/architecture.md +++ b/docs/policy/architecture.md @@ -166,7 +166,7 @@ For example, one destination may publish source files only as a long-term archiv ## Backend Abstraction -Sources and destinations use the same storage abstraction. Current runtime execution uses the local filesystem and SSH/SFTP backends. Additional storage backends should be peer implementations behind the same interface, and any backend-specific execution limitation must be documented. +Sources and destinations use the same storage abstraction. Current runtime execution uses the local filesystem, SSH/SFTP, and S3-compatible backends. Additional storage backends should be peer implementations behind the same interface, and any backend-specific execution limitation must be documented. Application logic must interact with storage through internal backend interfaces. Backend-specific behavior belongs in adapter packages. Pipeline, bundle, state, publish, and transform packages must not import service-specific or filesystem adapter implementation details. @@ -195,6 +195,7 @@ Use this current layout unless the project has a documented reason to differ: - `internal/storage`: backend interfaces, shared path/resource types, backend registry, and storage errors. - `internal/adapters/local`: local filesystem backend. - `internal/adapters/ssh`: SSH/SFTP backend. +- `internal/adapters/s3`: S3-compatible object storage backend. - `internal/transform`: transform interfaces, registry, planning, and shared transform models. - `internal/transform/markdown`: Markdown-to-HTML implementation. - `internal/publish`: destination planning, reconciliation, safety checks, and publish execution. diff --git a/docs/policy/development.md b/docs/policy/development.md index b2397e3..aae8e25 100644 --- a/docs/policy/development.md +++ b/docs/policy/development.md @@ -14,6 +14,7 @@ Use it with `docs/policy/architecture.md` and `docs/policy/documentation.md`. - `internal/storage`: backend interface, registry, logical path rules, typed errors, and shared storage helpers. - `internal/adapters/local`: local filesystem backend. - `internal/adapters/ssh`: SSH/SFTP backend. +- `internal/adapters/s3`: S3-compatible object storage backend. - `internal/storage/fake`: in-memory backend for tests. - `internal/publish`: destination inspection, output planning, reconciliation, execution, and managed cleanup. - `internal/transform`: transform interface and registry. @@ -85,6 +86,7 @@ The project currently depends on: - `github.com/yuin/goldmark` for Markdown rendering. - `golang.org/x/crypto/ssh`, `golang.org/x/crypto/ssh/agent`, and `golang.org/x/crypto/ssh/knownhosts` for native SSH support. - `github.com/pkg/sftp` for native SFTP support. +- `github.com/aws/aws-sdk-go-v2/...` packages for S3-compatible storage support. Add external dependencies only when they materially improve correctness, security, interoperability, or implementation complexity. Avoid dependencies @@ -105,7 +107,7 @@ When adding or changing configuration: Config validation may accept fields for backends that are not executable yet, but user-facing docs and examples must clearly state execution support. At the -time of this policy, local and SSH backends are executable. +time of this policy, local, SSH, and S3 backends are executable. Credential-consuming code must use the config-owned environment resolver for explicit credential environment variable references. Do not call `os.Getenv` @@ -126,7 +128,7 @@ When adding or changing commands or flags: 4. Update `docs/cli.md` if syntax, flags, output expectations, or workflows change. `validate` and `inspect` are local path commands. `run` loads configured -pipelines and currently executes local and SSH backends. +pipelines and currently executes local, SSH, and S3 backends. ## Storage Backends @@ -142,8 +144,8 @@ When adding a backend: 5. Add focused adapter tests and app-level wiring tests. 6. Update user docs, operations docs, examples, and internal docs only for behavior that is actually implemented. -Do not document S3 execution as available until the corresponding adapter -package and app wiring exist. +Do not document future backend execution as available until the corresponding +adapter package and app wiring exist. ## Transforms diff --git a/docs/troubleshooting.md b/docs/troubleshooting.md index 9c819be..8e18aaf 100644 --- a/docs/troubleshooting.md +++ b/docs/troubleshooting.md @@ -34,11 +34,11 @@ Diagnostic: rg -n "backend:" ``` -Safe fix: use `backend: local` or `backend: ssh` for executable workflows. S3 config shape is accepted only for validation; runtime execution is unavailable. +Safe fix: use `backend: local`, `backend: ssh`, or `backend: s3` for executable workflows. -## `backend s3 is not implemented for execution` +## `prefix must be a clean relative slash-separated path` -Likely cause: the config validates but `run` tried to execute S3, which is not implemented. +Likely cause: S3 `prefix` contains traversal, dot segments, empty segments, or backslashes after leading and trailing slashes are trimmed. Diagnostic: @@ -46,7 +46,44 @@ Diagnostic: go run ./cmd/distributor run --config --dry-run ``` -Safe fix: use `local` or `ssh` for executable workflows. See [configuration](config.md). +Safe fix: use a clean relative prefix such as `reports/archive`, or omit `prefix`. + +## `NoSuchBucket`, `InvalidBucketName`, or `not_found` + +Likely cause: the S3 bucket, endpoint, or prefix is wrong, or the configured credentials cannot see the requested object. + +Diagnostic: + +```sh +go run ./cmd/distributor run --config --dry-run +``` + +Safe fix: verify `endpoint`, `bucket`, `region`, `force_path_style`, and `prefix`. For S3-compatible services, keep `force_path_style: true` unless the service requires virtual-host addressing. + +## `AccessDenied`, `InvalidAccessKeyId`, or `SignatureDoesNotMatch` + +Likely cause: S3 credentials are missing, wrong, empty, or lack permission for the bucket or prefix. + +Diagnostic: + +```sh +env | cut -d= -f1 | rg '^(|)$' +ls -l +``` + +Safe fix: provide both configured credential environment variables through the real environment or `secrets.directory`, or omit explicit credential fields to use the AWS SDK default credential chain. + +## S3 endpoint connection failures + +Likely cause: the endpoint URL is unreachable, uses the wrong scheme, or does not match the configured path-style mode. + +Diagnostic: + +```sh +curl -I +``` + +Safe fix: correct `endpoint`, network routing, TLS settings outside distributor, or `force_path_style`. Distributor does not provide insecure TLS bypass configuration. ## `load secrets directory ... no such file or directory` @@ -92,7 +129,7 @@ Likely cause: a backend credential field references an environment variable that Diagnostic: ```sh -printenv +env | cut -d= -f1 | rg '^$' ls -l / ``` @@ -105,7 +142,7 @@ Likely cause: the real process environment and secrets directory both define the Diagnostic: ```sh -printenv +env | cut -d= -f1 | rg '^$' ls -l / ``` diff --git a/examples/s3-destination.yml b/examples/s3-destination.yml new file mode 100644 index 0000000..91ba01c --- /dev/null +++ b/examples/s3-destination.yml @@ -0,0 +1,22 @@ +# Environment-gated example. +# Replace endpoint, bucket, prefix, and credential environment variable names +# with values for an S3-compatible service you control before running this config. +pipelines: + - id: example-s3-destination + source: + backend: local + path: examples/source-bundle + destinations: + - id: s3-archive + backend: s3 + endpoint: https://s3.example.com + bucket: reports + prefix: distributor/archive + region: us-east-1 + force_path_style: true + credentials: + access_key_id_env: DISTRIBUTOR_S3_ACCESS_KEY_ID + secret_access_key_env: DISTRIBUTOR_S3_SECRET_ACCESS_KEY + publish: + source: true + html: false diff --git a/go.mod b/go.mod index b44c1e5..0278d29 100644 --- a/go.mod +++ b/go.mod @@ -3,6 +3,11 @@ module gitea.maximumdirect.net/eric/distributor go 1.26 require ( + github.com/aws/aws-sdk-go-v2 v1.41.9 + github.com/aws/aws-sdk-go-v2/config v1.32.20 + github.com/aws/aws-sdk-go-v2/credentials v1.19.19 + github.com/aws/aws-sdk-go-v2/service/s3 v1.102.2 + github.com/aws/smithy-go v1.26.0 github.com/pkg/sftp v1.13.10 github.com/yuin/goldmark v1.8.2 golang.org/x/crypto v0.52.0 @@ -10,6 +15,19 @@ require ( ) require ( + github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.11 // indirect + github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.25 // indirect + github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.25 // indirect + github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.25 // indirect + github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.26 // indirect + github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.10 // indirect + github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.9.18 // indirect + github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.25 // indirect + github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.25 // indirect + github.com/aws/aws-sdk-go-v2/service/signin v1.1.1 // indirect + github.com/aws/aws-sdk-go-v2/service/sso v1.30.19 // indirect + github.com/aws/aws-sdk-go-v2/service/ssooidc v1.36.2 // indirect + github.com/aws/aws-sdk-go-v2/service/sts v1.42.3 // indirect github.com/kr/fs v0.1.0 // indirect golang.org/x/sys v0.45.0 // indirect ) diff --git a/go.sum b/go.sum index a513948..17c298e 100644 --- a/go.sum +++ b/go.sum @@ -1,3 +1,39 @@ +github.com/aws/aws-sdk-go-v2 v1.41.9 h1:/rYeyO2+HrMztAmxAq9++XJtFMqSIpSsNA0yDGALYq4= +github.com/aws/aws-sdk-go-v2 v1.41.9/go.mod h1:+HsoOEX80qAVUitj1A2DhCNTjmb3edVyuDypb6LNEeo= +github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.11 h1:h5+3VT69KUBK24grGuuA5saDJTj2IIjLb9au668Fo5I= +github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.11/go.mod h1:dnakxebH6UwFvcvujL0LVggYQ8nEvBGjU4G/V79Nv94= +github.com/aws/aws-sdk-go-v2/config v1.32.20 h1:8VMDnWc/kEzxsI/1ngGM9mG81a8IGmIHD8KLcYGwagc= +github.com/aws/aws-sdk-go-v2/config v1.32.20/go.mod h1:PuwEpciweIXGULWeOeSTXtSbH4CW9mWdWrhdCKQI1sM= +github.com/aws/aws-sdk-go-v2/credentials v1.19.19 h1:yuFzSV1U0aRNYCQGVaTY2zW2M/L93pYHnXnrJUphYhU= +github.com/aws/aws-sdk-go-v2/credentials v1.19.19/go.mod h1:7y63L1kGzeoDlJaQ3Z578KrnmfBut96JjvJUzGwR+YE= +github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.25 h1:0w6dCiO8iez+YKwRhRBlL1CH/E3GTfdkuzrwj1by8vo= +github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.25/go.mod h1:9FDWUothyr5RCRAHc45XOiVCzUR8n/IhCYX+uVqw6vk= +github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.25 h1:Uii3frf9ztec/ABM2/FSH9/z7PLzxfpG8h4RpkUFflQ= +github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.25/go.mod h1:G6kntsA2GorAxDPbap6xgB2F+amSLUF8GJTi7PUoX44= +github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.25 h1:r1+/l6m+WaUJF9HISEsNOLHSNj5EXYQxK8VX6Cz9NlA= +github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.25/go.mod h1:cKf+D+NMDK1LndD7BowHbBZPgR9V0/5HubH0PFWvA+c= +github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.26 h1:A1PmWU2zfkIm9EyFlJncFXL4W4phML+h8KjltUsCvNQ= +github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.26/go.mod h1:dY4MRzXEizrD4hqtpKvWVGPX7QleSGGVY+EBolo1RmM= +github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.10 h1:d5/908OJ4bXg8lyjeMPvXetEKqoDoLi5Owy1zNue3yg= +github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.10/go.mod h1:a57l7Hwh+FWI+we50g5NPJHYUKeJKfXbc4w8SyXu8Ig= +github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.9.18 h1:W/EyPFl9A5rXrtoilfwHYEvzHER+K4SpBPtMXi24Mos= +github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.9.18/go.mod h1:UG50K+pvd/uy6xExbobg0rjqFBFZe6I3l75EPDZw4tg= +github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.25 h1:dD3dhHNglpd98gs72my22Ndqi1hqQGllFFg1F+twfxg= +github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.25/go.mod h1:0yAbjPfd64gG7mj85RW+fMEYdfBgCRZw8g/oWcL1pjc= +github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.25 h1:2pQEbwf+/6EDbiit/GcBE2K4IUpMZymaA0kOz3xK978= +github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.25/go.mod h1:KvT6NCcQ0EZ+ZkVRrlBMt04Po3ok23YELEp7WimhLhM= +github.com/aws/aws-sdk-go-v2/service/s3 v1.102.2 h1:ie4ElCmUKS26pzrZcIk/lmt4yWjAqLLcawstyQCh298= +github.com/aws/aws-sdk-go-v2/service/s3 v1.102.2/go.mod h1:zjsomFeX5duj+4PlMB+o4JoWTIx+G0XMyzjYrUbQkN0= +github.com/aws/aws-sdk-go-v2/service/signin v1.1.1 h1:1VwbP3qMNfxUDEXWki4rCE5iA+44VA1lokTz9HasGzw= +github.com/aws/aws-sdk-go-v2/service/signin v1.1.1/go.mod h1:vUtyoSj0OPji3kjIVSc/GlKuWEiL33f/WFxl6dmpy/A= +github.com/aws/aws-sdk-go-v2/service/sso v1.30.19 h1:N6pIsdFOW1Kd9S4KyFKXdGRBojPPxkP32+uHFWLv4Hc= +github.com/aws/aws-sdk-go-v2/service/sso v1.30.19/go.mod h1:3gt5WJArFooNmyLONS+h/R4J+o86II8du38IgCwj9dE= +github.com/aws/aws-sdk-go-v2/service/ssooidc v1.36.2 h1:hc+lBYiiTr8Zk4MTzIsQ92MeDWCIDvWGmzKUWOaBcOg= +github.com/aws/aws-sdk-go-v2/service/ssooidc v1.36.2/go.mod h1:hU6fqB3OJA6/ePheD47LQnxvjYk6br6PtQxs+Q9ojvk= +github.com/aws/aws-sdk-go-v2/service/sts v1.42.3 h1:ErklX/7uhSbkAAeyQD/Y1OoQ9hO3SJXQNEgksORW3Js= +github.com/aws/aws-sdk-go-v2/service/sts v1.42.3/go.mod h1:ULe4HCzfKPiR6R3HEurE3b1upEkuk8AkMrOKtaOxKO8= +github.com/aws/smithy-go v1.26.0 h1:9ouqbi+NyKP7fV3Te7UElCwdAb6Y8uk7LGwPE5tVe/s= +github.com/aws/smithy-go v1.26.0/go.mod h1:YE2RhdIuDbA5E5bTdciG9KrW3+TiEONeUWCqxX9i1Fc= github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/kr/fs v0.1.0 h1:Jskdu9ieNAYnjxsi0LbQp1ulIKZV1LAFgK1tWhpZgl8= diff --git a/internal/adapters/s3/backend.go b/internal/adapters/s3/backend.go new file mode 100644 index 0000000..746519e --- /dev/null +++ b/internal/adapters/s3/backend.go @@ -0,0 +1,430 @@ +package s3 + +import ( + "bytes" + "context" + "errors" + "fmt" + "io" + "path" + "sort" + "strings" + + "gitea.maximumdirect.net/eric/distributor/internal/storage" + "github.com/aws/aws-sdk-go-v2/aws" + awscfg "github.com/aws/aws-sdk-go-v2/config" + "github.com/aws/aws-sdk-go-v2/credentials" + awss3 "github.com/aws/aws-sdk-go-v2/service/s3" + "github.com/aws/aws-sdk-go-v2/service/s3/types" + "github.com/aws/smithy-go" +) + +type Client interface { + HeadObject(ctx context.Context, input *awss3.HeadObjectInput, optFns ...func(*awss3.Options)) (*awss3.HeadObjectOutput, error) + GetObject(ctx context.Context, input *awss3.GetObjectInput, optFns ...func(*awss3.Options)) (*awss3.GetObjectOutput, error) + PutObject(ctx context.Context, input *awss3.PutObjectInput, optFns ...func(*awss3.Options)) (*awss3.PutObjectOutput, error) + ListObjectsV2(ctx context.Context, input *awss3.ListObjectsV2Input, optFns ...func(*awss3.Options)) (*awss3.ListObjectsV2Output, error) + DeleteObject(ctx context.Context, input *awss3.DeleteObjectInput, optFns ...func(*awss3.Options)) (*awss3.DeleteObjectOutput, error) +} + +type Backend struct { + client Client + bucket string + prefix string +} + +func New(ctx context.Context, options Options) (*Backend, error) { + if err := ctx.Err(); err != nil { + return nil, err + } + options, err := options.normalized() + if err != nil { + return nil, storage.NewError(storage.OpOpenBackend, BackendName, options.Prefix, storage.ErrInvalidPath, err) + } + loadOptions := []func(*awscfg.LoadOptions) error{ + awscfg.WithRegion(options.Region), + } + if options.AccessKeyID != "" { + loadOptions = append(loadOptions, awscfg.WithCredentialsProvider(credentials.NewStaticCredentialsProvider(options.AccessKeyID, options.SecretAccessKey, ""))) + } + cfg, err := awscfg.LoadDefaultConfig(ctx, loadOptions...) + if err != nil { + return nil, storage.NewError(storage.OpOpenBackend, BackendName, options.Endpoint, storage.ErrUnknown, err) + } + client := awss3.NewFromConfig(cfg, func(o *awss3.Options) { + o.BaseEndpoint = aws.String(options.Endpoint) + o.UsePathStyle = options.ForcePathStyle + }) + return NewWithClient(client, options) +} + +func NewWithClient(client Client, options Options) (*Backend, error) { + options, err := options.normalized() + if err != nil { + return nil, storage.NewError(storage.OpOpenBackend, BackendName, options.Prefix, storage.ErrInvalidPath, err) + } + if client == nil { + return nil, storage.NewError(storage.OpOpenBackend, BackendName, options.Bucket, storage.ErrInvalidPath, fmt.Errorf("client is required")) + } + return &Backend{ + client: client, + bucket: options.Bucket, + prefix: options.Prefix, + }, nil +} + +func (b *Backend) ReadFile(ctx context.Context, logicalPath string) ([]byte, error) { + reader, err := b.OpenReader(ctx, logicalPath) + if err != nil { + return nil, err + } + defer reader.Close() + data, err := io.ReadAll(reader) + if err != nil { + return nil, storage.NewError(storage.OpReadFile, BackendName, logicalPath, storage.ErrUnknown, err) + } + return data, nil +} + +func (b *Backend) OpenReader(ctx context.Context, logicalPath string) (io.ReadCloser, error) { + if err := ctx.Err(); err != nil { + return nil, err + } + key, err := b.objectKey(logicalPath, false) + if err != nil { + return nil, err + } + output, err := b.client.GetObject(ctx, &awss3.GetObjectInput{ + Bucket: aws.String(b.bucket), + Key: aws.String(key), + }) + if err != nil { + return nil, b.translateError(storage.OpOpenReader, logicalPath, err) + } + return output.Body, nil +} + +func (b *Backend) WriteFile(ctx context.Context, logicalPath string, data []byte, opts storage.WriteOptions) (storage.Entry, error) { + opts.Size = int64(len(data)) + opts.SizeKnown = true + return b.WriteFrom(ctx, logicalPath, bytes.NewReader(data), opts) +} + +func (b *Backend) WriteFrom(ctx context.Context, logicalPath string, r io.Reader, opts storage.WriteOptions) (storage.Entry, error) { + if err := ctx.Err(); err != nil { + return storage.Entry{}, err + } + key, err := b.objectKey(logicalPath, false) + if err != nil { + return storage.Entry{}, err + } + if !opts.Overwrite { + _, err := b.client.HeadObject(ctx, &awss3.HeadObjectInput{ + Bucket: aws.String(b.bucket), + Key: aws.String(key), + }) + if err == nil { + return storage.Entry{}, storage.NewError(storage.OpWriteFrom, BackendName, logicalPath, storage.ErrAlreadyExist, nil) + } + if !isNotFound(err) { + return storage.Entry{}, b.translateError(storage.OpWriteFrom, logicalPath, err) + } + } + data, err := io.ReadAll(r) + if err != nil { + return storage.Entry{}, storage.NewError(storage.OpWriteFrom, BackendName, logicalPath, storage.ErrUnknown, err) + } + if opts.SizeKnown && int64(len(data)) != opts.Size { + return storage.Entry{}, storage.NewError(storage.OpWriteFrom, BackendName, logicalPath, storage.ErrConflict, fmt.Errorf("stream size %d does not match expected size %d", len(data), opts.Size)) + } + contentType := opts.ContentType + if contentType == "" { + contentType = ContentType(logicalPath) + } + _, err = b.client.PutObject(ctx, &awss3.PutObjectInput{ + Bucket: aws.String(b.bucket), + Key: aws.String(key), + Body: bytes.NewReader(data), + ContentLength: aws.Int64(int64(len(data))), + ContentType: aws.String(contentType), + }) + if err != nil { + return storage.Entry{}, b.translateError(storage.OpWriteFrom, logicalPath, err) + } + return storage.Entry{Path: logicalPath, Type: storage.EntryTypeFile, Size: int64(len(data))}, nil +} + +func (b *Backend) Stat(ctx context.Context, logicalPath string) (storage.Entry, error) { + if err := ctx.Err(); err != nil { + return storage.Entry{}, err + } + key, err := b.objectKey(logicalPath, false) + if err != nil { + return storage.Entry{}, err + } + output, err := b.client.HeadObject(ctx, &awss3.HeadObjectInput{ + Bucket: aws.String(b.bucket), + Key: aws.String(key), + }) + if err != nil { + return storage.Entry{}, b.translateError(storage.OpStat, logicalPath, err) + } + size := int64(0) + if output.ContentLength != nil { + size = *output.ContentLength + } + return storage.Entry{Path: logicalPath, Type: storage.EntryTypeFile, Size: size}, nil +} + +func (b *Backend) Walk(ctx context.Context, logicalPrefix string, opts storage.WalkOptions, fn storage.WalkFunc) error { + if err := ctx.Err(); err != nil { + return err + } + if err := storage.ValidatePrefix(logicalPrefix); err != nil { + return err + } + visited := 0 + emit := func(entry storage.Entry) error { + if err := ctx.Err(); err != nil { + return err + } + if opts.Limit > 0 && visited >= opts.Limit { + return storage.ErrStopWalk + } + visited++ + if err := fn(entry); err != nil { + if errors.Is(err, storage.ErrStopWalk) { + return storage.ErrStopWalk + } + return storage.NewError(storage.OpWalk, BackendName, entry.Path, storage.ErrUnknown, err) + } + return nil + } + if logicalPrefix != "" { + entry, err := b.Stat(ctx, logicalPrefix) + if err == nil { + if err := emit(entry); errors.Is(err, storage.ErrStopWalk) { + return nil + } else if err != nil { + return err + } + return nil + } + if !storage.IsNotFound(err) { + return err + } + } + err := b.walkObjects(ctx, logicalPrefix, opts, emit) + if errors.Is(err, storage.ErrStopWalk) { + return nil + } + return err +} + +func (b *Backend) HasAny(ctx context.Context, logicalPrefix string) (bool, error) { + found := false + err := b.Walk(ctx, logicalPrefix, storage.WalkOptions{Recursive: false, Limit: 1}, func(storage.Entry) error { + found = true + return storage.ErrStopWalk + }) + if err != nil { + return false, err + } + return found, nil +} + +func (b *Backend) DeleteManagedBundle(ctx context.Context, bundlePath string, managedOutputPaths []string, opts storage.DeleteOptions) error { + if err := ctx.Err(); err != nil { + return err + } + targets, err := storage.ManagedBundleTargets(bundlePath, managedOutputPaths) + if err != nil { + return err + } + for _, target := range targets { + key, err := b.objectKey(target, false) + if err != nil { + return err + } + if key == b.prefix { + return storage.NewError(storage.OpDeleteManagedBundle, BackendName, target, storage.ErrInvalidPath, nil) + } + if !opts.IgnoreMissing { + _, err := b.client.HeadObject(ctx, &awss3.HeadObjectInput{ + Bucket: aws.String(b.bucket), + Key: aws.String(key), + }) + if err != nil { + return b.translateError(storage.OpDeleteManagedBundle, target, err) + } + } + _, err = b.client.DeleteObject(ctx, &awss3.DeleteObjectInput{ + Bucket: aws.String(b.bucket), + Key: aws.String(key), + }) + if err != nil { + if opts.IgnoreMissing && isNotFound(err) { + continue + } + return b.translateError(storage.OpDeleteManagedBundle, target, err) + } + } + return nil +} + +func (b *Backend) walkObjects(ctx context.Context, logicalPrefix string, opts storage.WalkOptions, emit func(storage.Entry) error) error { + listPrefix, err := b.listPrefix(logicalPrefix) + if err != nil { + return err + } + delimiter := "" + if !opts.Recursive { + delimiter = "/" + } + var token *string + for { + output, err := b.client.ListObjectsV2(ctx, &awss3.ListObjectsV2Input{ + Bucket: aws.String(b.bucket), + Prefix: aws.String(listPrefix), + Delimiter: aws.String(delimiter), + ContinuationToken: token, + }) + if err != nil { + return b.translateError(storage.OpWalk, logicalPrefix, err) + } + entries := entriesFromList(logicalPrefix, b.prefix, output) + sort.Slice(entries, func(i, j int) bool { return entries[i].Path < entries[j].Path }) + for _, entry := range entries { + if entry.Path == "" { + continue + } + if err := emit(entry); err != nil { + return err + } + } + if output.IsTruncated == nil || !*output.IsTruncated { + return nil + } + token = output.NextContinuationToken + } +} + +func entriesFromList(logicalPrefix, rootPrefix string, output *awss3.ListObjectsV2Output) []storage.Entry { + seen := make(map[string]storage.Entry) + for _, object := range output.Contents { + if object.Key == nil { + continue + } + logicalPath := logicalPathFromKey(rootPrefix, *object.Key) + if logicalPath == "" || logicalPath == logicalPrefix { + continue + } + size := int64(0) + if object.Size != nil { + size = *object.Size + } + seen[logicalPath] = storage.Entry{Path: logicalPath, Type: storage.EntryTypeFile, Size: size} + } + for _, commonPrefix := range output.CommonPrefixes { + if commonPrefix.Prefix == nil { + continue + } + logicalPath := strings.TrimSuffix(logicalPathFromKey(rootPrefix, *commonPrefix.Prefix), "/") + if logicalPath == "" || logicalPath == logicalPrefix { + continue + } + seen[logicalPath] = storage.Entry{Path: logicalPath, Type: storage.EntryTypeDirectory} + } + entries := make([]storage.Entry, 0, len(seen)) + for _, entry := range seen { + entries = append(entries, entry) + } + return entries +} + +func (b *Backend) objectKey(logicalPath string, allowEmpty bool) (string, error) { + if logicalPath == "" { + if !allowEmpty { + return "", storage.NewError(storage.OpValidatePath, BackendName, logicalPath, storage.ErrInvalidPath, nil) + } + return b.prefix, nil + } + if err := storage.ValidatePath(logicalPath); err != nil { + return "", err + } + if b.prefix == "" { + return logicalPath, nil + } + return b.prefix + "/" + logicalPath, nil +} + +func (b *Backend) listPrefix(logicalPrefix string) (string, error) { + key, err := b.objectKey(logicalPrefix, true) + if err != nil { + return "", err + } + if key != "" { + key = strings.TrimSuffix(key, "/") + "/" + } + return key, nil +} + +func logicalPathFromKey(rootPrefix, key string) string { + if rootPrefix == "" { + return key + } + if key == rootPrefix { + return "" + } + return strings.TrimPrefix(key, rootPrefix+"/") +} + +func ContentType(logicalPath string) string { + switch strings.ToLower(path.Ext(logicalPath)) { + case ".md": + return "text/markdown; charset=utf-8" + case ".html": + return "text/html; charset=utf-8" + case ".json": + return "application/json" + case ".txt": + return "text/plain; charset=utf-8" + default: + return "application/octet-stream" + } +} + +func isNotFound(err error) bool { + var notFound *types.NotFound + if errors.As(err, ¬Found) { + return true + } + var apiErr smithy.APIError + if errors.As(err, &apiErr) { + switch apiErr.ErrorCode() { + case "NotFound", "NoSuchKey", "404": + return true + } + } + return false +} + +func (b *Backend) translateError(op, logicalPath string, err error) error { + kind := storage.ErrUnknown + if isNotFound(err) { + kind = storage.ErrNotFound + } else { + var apiErr smithy.APIError + if errors.As(err, &apiErr) { + switch apiErr.ErrorCode() { + case "AccessDenied", "InvalidAccessKeyId", "SignatureDoesNotMatch": + kind = storage.ErrPermission + case "SlowDown", "RequestTimeout", "ServiceUnavailable", "InternalError": + kind = storage.ErrTemporary + case "InvalidBucketName", "NoSuchBucket": + kind = storage.ErrInvalidPath + } + } + } + return storage.NewError(op, BackendName, logicalPath, kind, err) +} diff --git a/internal/adapters/s3/backend_test.go b/internal/adapters/s3/backend_test.go new file mode 100644 index 0000000..11a69ff --- /dev/null +++ b/internal/adapters/s3/backend_test.go @@ -0,0 +1,357 @@ +package s3 + +import ( + "context" + "io" + "sort" + "strings" + "testing" + + "gitea.maximumdirect.net/eric/distributor/internal/storage" + "github.com/aws/aws-sdk-go-v2/aws" + awss3 "github.com/aws/aws-sdk-go-v2/service/s3" + "github.com/aws/aws-sdk-go-v2/service/s3/types" +) + +func TestKeyAndPrefixNormalization(t *testing.T) { + backend := newTestBackend(t, "root/prefix", nil) + key, err := backend.objectKey("bundle/report.md", false) + if err != nil { + t.Fatalf("objectKey() error = %v", err) + } + if got, want := key, "root/prefix/bundle/report.md"; got != want { + t.Fatalf("objectKey() = %q, want %q", got, want) + } + listPrefix, err := backend.listPrefix("bundle") + if err != nil { + t.Fatalf("listPrefix() error = %v", err) + } + if got, want := listPrefix, "root/prefix/bundle/"; got != want { + t.Fatalf("listPrefix() = %q, want %q", got, want) + } +} + +func TestPathTraversalRejected(t *testing.T) { + backend := newTestBackend(t, "", nil) + for _, logicalPath := range []string{"/absolute", "../escape", "a/../b", `a\b`} { + t.Run(logicalPath, func(t *testing.T) { + if _, err := backend.objectKey(logicalPath, false); err == nil || !storage.IsInvalidPath(err) { + t.Fatalf("objectKey() error = %v, want invalid path", err) + } + }) + } +} + +func TestContentType(t *testing.T) { + tests := map[string]string{ + "report.md": "text/markdown; charset=utf-8", + "report.html": "text/html; charset=utf-8", + "state.json": "application/json", + "summary.txt": "text/plain; charset=utf-8", + "data.bin": "application/octet-stream", + } + for path, want := range tests { + if got := ContentType(path); got != want { + t.Fatalf("ContentType(%q) = %q, want %q", path, got, want) + } + } +} + +func TestStatRequiresExactObject(t *testing.T) { + client := newFakeClient(map[string]string{"root/dir/file.txt": "data"}) + backend := newTestBackend(t, "root", client) + _, err := backend.Stat(context.Background(), "dir") + if err == nil || !storage.IsNotFound(err) { + t.Fatalf("Stat() error = %v, want not found", err) + } +} + +func TestWriteFromChecksOverwriteBeforePut(t *testing.T) { + client := newFakeClient(map[string]string{"root/report.md": "old"}) + backend := newTestBackend(t, "root", client) + _, err := backend.WriteFile(context.Background(), "report.md", []byte("new"), storage.WriteOptions{}) + if err == nil || !storage.IsAlreadyExists(err) { + t.Fatalf("WriteFile() error = %v, want already exists", err) + } + if len(client.putKeys) != 0 { + t.Fatalf("put keys = %v, want none", client.putKeys) + } +} + +func TestWriteFromPutsNewObjectWithContentType(t *testing.T) { + client := newFakeClient(nil) + backend := newTestBackend(t, "root", client) + entry, err := backend.WriteFile(context.Background(), "report.html", []byte("

ok

"), storage.WriteOptions{}) + if err != nil { + t.Fatalf("WriteFile() error = %v", err) + } + if entry.Path != "report.html" || entry.Size != 9 || entry.Type != storage.EntryTypeFile { + t.Fatalf("entry = %#v", entry) + } + if got, want := client.objects["root/report.html"], "

ok

"; got != want { + t.Fatalf("object = %q, want %q", got, want) + } + if got, want := client.contentTypes["root/report.html"], "text/html; charset=utf-8"; got != want { + t.Fatalf("content type = %q, want %q", got, want) + } +} + +func TestWalkUsesPagination(t *testing.T) { + client := newFakeClient(nil) + client.listPages = []awss3.ListObjectsV2Output{ + { + Contents: []types.Object{{Key: aws.String("root/a.txt"), Size: aws.Int64(1)}}, + IsTruncated: aws.Bool(true), + NextContinuationToken: aws.String("next"), + }, + { + Contents: []types.Object{{Key: aws.String("root/b.txt"), Size: aws.Int64(2)}}, + IsTruncated: aws.Bool(false), + }, + } + backend := newTestBackend(t, "root", client) + entries, err := storage.List(context.Background(), backend, "", storage.WalkOptions{Recursive: true}) + if err != nil { + t.Fatalf("List() error = %v", err) + } + paths := entryPaths(entries) + if got, want := paths, []string{"a.txt", "b.txt"}; !equalStrings(got, want) { + t.Fatalf("paths = %v, want %v", got, want) + } + if got, want := client.tokens, []string{"", "next"}; !equalStrings(got, want) { + t.Fatalf("tokens = %v, want %v", got, want) + } +} + +func TestHasAnyStopsAfterFirstPage(t *testing.T) { + client := newFakeClient(nil) + client.listPages = []awss3.ListObjectsV2Output{ + { + Contents: []types.Object{{Key: aws.String("root/a.txt"), Size: aws.Int64(1)}}, + IsTruncated: aws.Bool(true), + NextContinuationToken: aws.String("next"), + }, + { + Contents: []types.Object{{Key: aws.String("root/b.txt"), Size: aws.Int64(2)}}, + IsTruncated: aws.Bool(false), + }, + } + backend := newTestBackend(t, "root", client) + found, err := backend.HasAny(context.Background(), "") + if err != nil { + t.Fatalf("HasAny() error = %v", err) + } + if !found { + t.Fatal("HasAny() = false, want true") + } + if got, want := len(client.tokens), 1; got != want { + t.Fatalf("list calls = %d, want %d", got, want) + } +} + +func TestWalkNonRecursiveUsesPrefixBoundary(t *testing.T) { + client := newFakeClient(map[string]string{ + "base/dir/file.txt": "nested", + "base/file.txt": "file", + "baseball/file.txt": "wrong", + }) + backend := newTestBackend(t, "base", client) + entries, err := storage.List(context.Background(), backend, "", storage.WalkOptions{}) + if err != nil { + t.Fatalf("List() error = %v", err) + } + paths := entryPaths(entries) + if got, want := paths, []string{"dir", "file.txt"}; !equalStrings(got, want) { + t.Fatalf("paths = %v, want %v", got, want) + } +} + +func TestDeleteManagedBundleDeletesOnlyManagedTargets(t *testing.T) { + client := newFakeClient(map[string]string{ + "root/report.md": "report", + "root/.distributor.json": "state", + "root/keep.txt": "keep", + }) + backend := newTestBackend(t, "root", client) + err := backend.DeleteManagedBundle(context.Background(), "", []string{"report.md"}, storage.DeleteOptions{IgnoreMissing: true}) + if err != nil { + t.Fatalf("DeleteManagedBundle() error = %v", err) + } + if _, ok := client.objects["root/report.md"]; ok { + t.Fatal("managed output still exists") + } + if _, ok := client.objects["root/.distributor.json"]; ok { + t.Fatal("state file still exists") + } + if _, ok := client.objects["root/keep.txt"]; !ok { + t.Fatal("unmanaged object was deleted") + } + if got, want := sortedStrings(client.deleteKeys), []string{"root/.distributor.json", "root/report.md"}; !equalStrings(got, want) { + t.Fatalf("deleted keys = %v, want %v", got, want) + } +} + +func newTestBackend(t *testing.T, prefix string, client *fakeClient) *Backend { + t.Helper() + if client == nil { + client = newFakeClient(nil) + } + backend, err := NewWithClient(client, Options{ + Endpoint: "https://s3.example.com", + Bucket: "reports", + Prefix: prefix, + Region: DefaultRegion, + ForcePathStyle: true, + }) + if err != nil { + t.Fatalf("NewWithClient() error = %v", err) + } + return backend +} + +type fakeClient struct { + objects map[string]string + contentTypes map[string]string + listPages []awss3.ListObjectsV2Output + tokens []string + putKeys []string + deleteKeys []string +} + +func newFakeClient(objects map[string]string) *fakeClient { + copied := make(map[string]string) + for key, value := range objects { + copied[key] = value + } + return &fakeClient{ + objects: copied, + contentTypes: make(map[string]string), + } +} + +func (c *fakeClient) HeadObject(ctx context.Context, input *awss3.HeadObjectInput, optFns ...func(*awss3.Options)) (*awss3.HeadObjectOutput, error) { + if err := ctx.Err(); err != nil { + return nil, err + } + value, ok := c.objects[aws.ToString(input.Key)] + if !ok { + return nil, &types.NotFound{} + } + return &awss3.HeadObjectOutput{ContentLength: aws.Int64(int64(len(value)))}, nil +} + +func (c *fakeClient) GetObject(ctx context.Context, input *awss3.GetObjectInput, optFns ...func(*awss3.Options)) (*awss3.GetObjectOutput, error) { + if err := ctx.Err(); err != nil { + return nil, err + } + value, ok := c.objects[aws.ToString(input.Key)] + if !ok { + return nil, &types.NotFound{} + } + return &awss3.GetObjectOutput{ + Body: io.NopCloser(stringsReader(value)), + ContentLength: aws.Int64(int64(len(value))), + }, nil +} + +func (c *fakeClient) PutObject(ctx context.Context, input *awss3.PutObjectInput, optFns ...func(*awss3.Options)) (*awss3.PutObjectOutput, error) { + if err := ctx.Err(); err != nil { + return nil, err + } + data, err := io.ReadAll(input.Body) + if err != nil { + return nil, err + } + key := aws.ToString(input.Key) + c.objects[key] = string(data) + c.contentTypes[key] = aws.ToString(input.ContentType) + c.putKeys = append(c.putKeys, key) + return &awss3.PutObjectOutput{}, nil +} + +func (c *fakeClient) ListObjectsV2(ctx context.Context, input *awss3.ListObjectsV2Input, optFns ...func(*awss3.Options)) (*awss3.ListObjectsV2Output, error) { + if err := ctx.Err(); err != nil { + return nil, err + } + c.tokens = append(c.tokens, aws.ToString(input.ContinuationToken)) + if len(c.listPages) > 0 { + index := len(c.tokens) - 1 + if index >= len(c.listPages) { + return &awss3.ListObjectsV2Output{IsTruncated: aws.Bool(false)}, nil + } + page := c.listPages[index] + return &page, nil + } + return c.dynamicList(input), nil +} + +func (c *fakeClient) DeleteObject(ctx context.Context, input *awss3.DeleteObjectInput, optFns ...func(*awss3.Options)) (*awss3.DeleteObjectOutput, error) { + if err := ctx.Err(); err != nil { + return nil, err + } + key := aws.ToString(input.Key) + delete(c.objects, key) + c.deleteKeys = append(c.deleteKeys, key) + return &awss3.DeleteObjectOutput{}, nil +} + +func (c *fakeClient) dynamicList(input *awss3.ListObjectsV2Input) *awss3.ListObjectsV2Output { + prefix := aws.ToString(input.Prefix) + delimiter := aws.ToString(input.Delimiter) + var contents []types.Object + commonPrefixes := make(map[string]struct{}) + for key, value := range c.objects { + if !strings.HasPrefix(key, prefix) { + continue + } + remainder := strings.TrimPrefix(key, prefix) + if delimiter != "" { + if index := strings.Index(remainder, delimiter); index >= 0 { + commonPrefixes[prefix+remainder[:index+1]] = struct{}{} + continue + } + } + contents = append(contents, types.Object{Key: aws.String(key), Size: aws.Int64(int64(len(value)))}) + } + sort.Slice(contents, func(i, j int) bool { return aws.ToString(contents[i].Key) < aws.ToString(contents[j].Key) }) + prefixes := make([]types.CommonPrefix, 0, len(commonPrefixes)) + for prefix := range commonPrefixes { + prefixes = append(prefixes, types.CommonPrefix{Prefix: aws.String(prefix)}) + } + sort.Slice(prefixes, func(i, j int) bool { return aws.ToString(prefixes[i].Prefix) < aws.ToString(prefixes[j].Prefix) }) + return &awss3.ListObjectsV2Output{ + Contents: contents, + CommonPrefixes: prefixes, + IsTruncated: aws.Bool(false), + } +} + +func stringsReader(value string) io.Reader { + return strings.NewReader(value) +} + +func entryPaths(entries []storage.Entry) []string { + paths := make([]string, 0, len(entries)) + for _, entry := range entries { + paths = append(paths, entry.Path) + } + return paths +} + +func sortedStrings(values []string) []string { + copied := append([]string(nil), values...) + sort.Strings(copied) + return copied +} + +func equalStrings(a, b []string) bool { + if len(a) != len(b) { + return false + } + for index := range a { + if a[index] != b[index] { + return false + } + } + return true +} diff --git a/internal/adapters/s3/integration_test.go b/internal/adapters/s3/integration_test.go new file mode 100644 index 0000000..a1972b8 --- /dev/null +++ b/internal/adapters/s3/integration_test.go @@ -0,0 +1,39 @@ +package s3 + +import ( + "context" + "os" + "strconv" + "testing" +) + +func TestIntegrationS3BackendHasAny(t *testing.T) { + endpoint := os.Getenv("DISTRIBUTOR_TEST_S3_ENDPOINT") + bucket := os.Getenv("DISTRIBUTOR_TEST_S3_BUCKET") + if endpoint == "" || bucket == "" { + t.Skip("DISTRIBUTOR_TEST_S3_ENDPOINT and DISTRIBUTOR_TEST_S3_BUCKET are not set") + } + forcePathStyle := true + if raw := os.Getenv("DISTRIBUTOR_TEST_S3_FORCE_PATH_STYLE"); raw != "" { + parsed, err := strconv.ParseBool(raw) + if err != nil { + t.Fatalf("parse DISTRIBUTOR_TEST_S3_FORCE_PATH_STYLE: %v", err) + } + forcePathStyle = parsed + } + backend, err := New(context.Background(), Options{ + Endpoint: endpoint, + Bucket: bucket, + Prefix: os.Getenv("DISTRIBUTOR_TEST_S3_PREFIX"), + Region: os.Getenv("DISTRIBUTOR_TEST_S3_REGION"), + ForcePathStyle: forcePathStyle, + AccessKeyID: os.Getenv("DISTRIBUTOR_TEST_S3_ACCESS_KEY_ID"), + SecretAccessKey: os.Getenv("DISTRIBUTOR_TEST_S3_SECRET_ACCESS_KEY"), + }) + if err != nil { + t.Fatalf("New() error = %v", err) + } + if _, err := backend.HasAny(context.Background(), ""); err != nil { + t.Fatalf("HasAny(root) error = %v", err) + } +} diff --git a/internal/adapters/s3/options.go b/internal/adapters/s3/options.go new file mode 100644 index 0000000..b8fe80d --- /dev/null +++ b/internal/adapters/s3/options.go @@ -0,0 +1,42 @@ +package s3 + +import ( + "fmt" + "strings" + + "gitea.maximumdirect.net/eric/distributor/internal/storage" +) + +const BackendName = "s3" + +const DefaultRegion = "us-east-1" + +type Options struct { + Endpoint string + Bucket string + Prefix string + Region string + ForcePathStyle bool + AccessKeyID string + SecretAccessKey string +} + +func (o Options) normalized() (Options, error) { + if o.Endpoint == "" { + return Options{}, fmt.Errorf("endpoint is required") + } + if o.Bucket == "" { + return Options{}, fmt.Errorf("bucket is required") + } + if o.Region == "" { + o.Region = DefaultRegion + } + o.Prefix = strings.Trim(o.Prefix, "/") + if err := storage.ValidatePrefix(o.Prefix); err != nil { + return Options{}, fmt.Errorf("prefix: %w", err) + } + if (o.AccessKeyID == "") != (o.SecretAccessKey == "") { + return Options{}, fmt.Errorf("access key id and secret access key must be configured together") + } + return o, nil +} diff --git a/internal/app/backends.go b/internal/app/backends.go index 186ce99..5ceb2ba 100644 --- a/internal/app/backends.go +++ b/internal/app/backends.go @@ -6,6 +6,7 @@ import ( "strconv" "gitea.maximumdirect.net/eric/distributor/internal/adapters/local" + s3adapter "gitea.maximumdirect.net/eric/distributor/internal/adapters/s3" sshadapter "gitea.maximumdirect.net/eric/distributor/internal/adapters/ssh" "gitea.maximumdirect.net/eric/distributor/internal/config" "gitea.maximumdirect.net/eric/distributor/internal/storage" @@ -20,6 +21,13 @@ const ( sshKeyFileKey = "ssh_key_file" sshKnownHostsKey = "known_hosts" sshHostKeyPolicyKey = "host_key_policy" + s3EndpointKey = "endpoint" + s3BucketKey = "bucket" + s3PrefixKey = "prefix" + s3RegionKey = "region" + s3ForcePathStyleKey = "force_path_style" + s3AccessKeyIDKey = "access_key_id" + s3SecretAccessKey = "secret_access_key" ) type backendFactory struct { @@ -54,21 +62,44 @@ func newBackendFactoryWithEnvironment(environment config.Environment) *backendFa HostKeyPolicy: sshadapter.HostKeyPolicy(cfg[sshHostKeyPolicyKey]), }) }) + _ = registry.Register(config.BackendS3, func(ctx context.Context, cfg storage.OpenConfig) (storage.Backend, error) { + forcePathStyle, err := strconv.ParseBool(cfg[s3ForcePathStyleKey]) + if err != nil { + return nil, fmt.Errorf("s3 force_path_style: %w", err) + } + return s3adapter.New(ctx, s3adapter.Options{ + Endpoint: cfg[s3EndpointKey], + Bucket: cfg[s3BucketKey], + Prefix: cfg[s3PrefixKey], + Region: cfg[s3RegionKey], + ForcePathStyle: forcePathStyle, + AccessKeyID: cfg[s3AccessKeyIDKey], + SecretAccessKey: cfg[s3SecretAccessKey], + }) + }) return &backendFactory{registry: registry, environment: environment} } func (f *backendFactory) openSource(ctx context.Context, source config.Backend) (storage.Backend, error) { - if source.Backend != config.BackendLocal && source.Backend != config.BackendSSH { + if source.Backend != config.BackendLocal && source.Backend != config.BackendSSH && source.Backend != config.BackendS3 { return nil, fmt.Errorf("source backend %s is not implemented for execution", source.Backend) } - return f.registry.Open(ctx, source.Backend, sourceOpenConfig(source)) + openConfig, err := f.sourceOpenConfig(source) + if err != nil { + return nil, err + } + return f.registry.Open(ctx, source.Backend, openConfig) } func (f *backendFactory) openDestination(ctx context.Context, destination config.Destination) (storage.Backend, error) { - if destination.Backend != config.BackendLocal && destination.Backend != config.BackendSSH { + if destination.Backend != config.BackendLocal && destination.Backend != config.BackendSSH && destination.Backend != config.BackendS3 { return nil, fmt.Errorf("backend %s is not implemented for execution", destination.Backend) } - return f.registry.Open(ctx, destination.Backend, destinationOpenConfig(destination)) + openConfig, err := f.destinationOpenConfig(destination) + if err != nil { + return nil, err + } + return f.registry.Open(ctx, destination.Backend, openConfig) } func (f *backendFactory) openLocalPath(ctx context.Context, path string) (storage.Backend, error) { @@ -79,6 +110,43 @@ func (f *backendFactory) resolveCredentials(creds config.Credentials) (config.Re return f.environment.ResolveCredentials(creds) } +func (f *backendFactory) sourceOpenConfig(source config.Backend) (storage.OpenConfig, error) { + cfg := sourceOpenConfig(source) + if source.Backend == config.BackendS3 { + if err := f.addS3Config(cfg, source.Endpoint, source.Bucket, source.Prefix, source.Region, source.ForcePath, source.Creds); err != nil { + return nil, err + } + } + return cfg, nil +} + +func (f *backendFactory) destinationOpenConfig(destination config.Destination) (storage.OpenConfig, error) { + cfg := destinationOpenConfig(destination) + if destination.Backend == config.BackendS3 { + if err := f.addS3Config(cfg, destination.Endpoint, destination.Bucket, destination.Prefix, destination.Region, destination.ForcePath, destination.Creds); err != nil { + return nil, err + } + } + return cfg, nil +} + +func (f *backendFactory) addS3Config(cfg storage.OpenConfig, endpoint, bucket, prefix, region string, forcePath *bool, creds config.Credentials) error { + cfg[s3EndpointKey] = endpoint + cfg[s3BucketKey] = bucket + cfg[s3PrefixKey] = prefix + cfg[s3RegionKey] = region + cfg[s3ForcePathStyleKey] = strconv.FormatBool(config.ForcePathStyle(forcePath)) + if creds.AccessKeyIDEnv != "" || creds.SecretAccessKeyEnv != "" { + resolved, err := f.resolveCredentials(creds) + if err != nil { + return err + } + cfg[s3AccessKeyIDKey] = resolved.AccessKeyID + cfg[s3SecretAccessKey] = resolved.SecretAccessKey + } + return nil +} + func sourceOpenConfig(source config.Backend) storage.OpenConfig { cfg := storage.OpenConfig{storagePathKey: source.Path} if source.Backend == config.BackendSSH { diff --git a/internal/app/backends_test.go b/internal/app/backends_test.go index 09ed03c..026e041 100644 --- a/internal/app/backends_test.go +++ b/internal/app/backends_test.go @@ -110,11 +110,9 @@ func TestBackendFactoryOpensSSHDestinationWithRegisteredOpener(t *testing.T) { func TestBackendFactoryRejectsUnsupportedSource(t *testing.T) { factory := newBackendFactory() _, err := factory.openSource(context.Background(), config.Backend{ - Backend: config.BackendS3, - Endpoint: "https://s3.example.com", - Bucket: "reports", + Backend: "ftp", }) - if err == nil || !strings.Contains(err.Error(), "source backend s3 is not implemented for execution") { + if err == nil || !strings.Contains(err.Error(), "source backend ftp is not implemented for execution") { t.Fatalf("openSource() error = %v, want not implemented", err) } } @@ -122,15 +120,85 @@ func TestBackendFactoryRejectsUnsupportedSource(t *testing.T) { func TestBackendFactoryRejectsUnsupportedDestination(t *testing.T) { factory := newBackendFactory() _, err := factory.openDestination(context.Background(), config.Destination{ - Backend: config.BackendS3, - Endpoint: "https://s3.example.com", - Bucket: "reports", + Backend: "ftp", }) - if err == nil || !strings.Contains(err.Error(), "backend s3 is not implemented for execution") { + if err == nil || !strings.Contains(err.Error(), "backend ftp is not implemented for execution") { t.Fatalf("openDestination() error = %v, want not implemented", err) } } +func TestBackendFactoryOpensS3DestinationWithRegisteredOpener(t *testing.T) { + factory := &backendFactory{ + registry: storage.NewRegistry(), + environment: config.NewEnvironment(nil, func(string) (string, bool) { return "", false }), + } + var got storage.OpenConfig + if err := factory.registry.Register(config.BackendS3, func(ctx context.Context, cfg storage.OpenConfig) (storage.Backend, error) { + got = cfg + return fake.New(), nil + }); err != nil { + t.Fatalf("Register() error = %v", err) + } + forcePathStyle := false + backend, err := factory.openDestination(context.Background(), config.Destination{ + Backend: config.BackendS3, + Endpoint: "https://s3.example.com", + Bucket: "reports", + Prefix: "archive", + Region: config.DefaultS3Region, + ForcePath: &forcePathStyle, + }) + if err != nil { + t.Fatalf("openDestination() error = %v", err) + } + if backend == nil { + t.Fatal("openDestination() backend = nil") + } + assertOpenConfig(t, got, map[string]string{ + s3EndpointKey: "https://s3.example.com", + s3BucketKey: "reports", + s3PrefixKey: "archive", + s3RegionKey: config.DefaultS3Region, + s3ForcePathStyleKey: "false", + }) +} + +func TestBackendFactoryResolvesS3CredentialsThroughSecretsAwareEnvironment(t *testing.T) { + factory := &backendFactory{ + registry: storage.NewRegistry(), + environment: config.NewEnvironment(map[string]string{ + "ACCESS_KEY_ID": "secret-access", + "SECRET_ACCESS_KEY": "secret-secret", + }, func(string) (string, bool) { return "", false }), + } + var got storage.OpenConfig + if err := factory.registry.Register(config.BackendS3, func(ctx context.Context, cfg storage.OpenConfig) (storage.Backend, error) { + got = cfg + return fake.New(), nil + }); err != nil { + t.Fatalf("Register() error = %v", err) + } + forcePathStyle := true + _, err := factory.openSource(context.Background(), config.Backend{ + Backend: config.BackendS3, + Endpoint: "https://s3.example.com", + Bucket: "reports", + Region: config.DefaultS3Region, + ForcePath: &forcePathStyle, + Creds: config.Credentials{ + AccessKeyIDEnv: "ACCESS_KEY_ID", + SecretAccessKeyEnv: "SECRET_ACCESS_KEY", + }, + }) + if err != nil { + t.Fatalf("openSource() error = %v", err) + } + assertOpenConfig(t, got, map[string]string{ + s3AccessKeyIDKey: "secret-access", + s3SecretAccessKey: "secret-secret", + }) +} + func TestBackendFactoryResolvesCredentialsThroughEnvironment(t *testing.T) { factory := newBackendFactoryWithEnvironment(config.NewEnvironment(map[string]string{ "ACCESS_KEY_ID": "secret-access", diff --git a/internal/config/config.go b/internal/config/config.go index 8689ce2..f1439a3 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -28,7 +28,7 @@ type Destination struct { Bucket string `yaml:"bucket"` Prefix string `yaml:"prefix"` Region string `yaml:"region"` - ForcePath bool `yaml:"force_path_style"` + ForcePath *bool `yaml:"force_path_style"` Creds Credentials `yaml:"credentials"` SSH SSH `yaml:",inline"` Publish *PublishPolicy `yaml:"publish"` @@ -47,7 +47,7 @@ type Backend struct { Bucket string `yaml:"bucket"` Prefix string `yaml:"prefix"` Region string `yaml:"region"` - ForcePath bool `yaml:"force_path_style"` + ForcePath *bool `yaml:"force_path_style"` Creds Credentials `yaml:"credentials"` SSH SSH `yaml:",inline"` } diff --git a/internal/config/defaults.go b/internal/config/defaults.go index 287022d..39f9c52 100644 --- a/internal/config/defaults.go +++ b/internal/config/defaults.go @@ -22,6 +22,8 @@ const ( TransformModeSidecar = "sidecar" ) +const DefaultS3Region = "us-east-1" + func ApplyDefaults(cfg *Config) { for pipelineIndex := range cfg.Pipelines { pipeline := &cfg.Pipelines[pipelineIndex] @@ -60,6 +62,9 @@ func applyBackendDefaults(backend *Backend) { backend.SSH.HostKeyPolicy = HostKeyPolicyAcceptNew } } + if backend.Backend == BackendS3 { + applyS3Defaults(&backend.Region, &backend.Prefix, &backend.ForcePath) + } } func applyDestinationDefaults(destination *Destination) { @@ -71,4 +76,18 @@ func applyDestinationDefaults(destination *Destination) { destination.SSH.HostKeyPolicy = HostKeyPolicyAcceptNew } } + if destination.Backend == BackendS3 { + applyS3Defaults(&destination.Region, &destination.Prefix, &destination.ForcePath) + } +} + +func applyS3Defaults(region, prefix *string, forcePath **bool) { + if *region == "" { + *region = DefaultS3Region + } + *prefix = NormalizeS3Prefix(*prefix) + if *forcePath == nil { + defaultForcePath := true + *forcePath = &defaultForcePath + } } diff --git a/internal/config/load_test.go b/internal/config/load_test.go index cdda4aa..7f5111e 100644 --- a/internal/config/load_test.go +++ b/internal/config/load_test.go @@ -155,6 +155,61 @@ pipelines: } } +func TestLoadFileDefaultsS3Config(t *testing.T) { + cfg := loadConfig(t, ` +pipelines: + - id: s3-defaults + source: + backend: s3 + endpoint: http://127.0.0.1:9000 + bucket: source + prefix: /incoming/reports/ + destinations: + - id: archive + backend: s3 + endpoint: http://127.0.0.1:9000 + bucket: destination +`) + + source := cfg.Pipelines[0].Source + if got, want := source.Region, DefaultS3Region; got != want { + t.Fatalf("source region = %q, want %q", got, want) + } + if got, want := source.Prefix, "incoming/reports"; got != want { + t.Fatalf("source prefix = %q, want %q", got, want) + } + if !ForcePathStyle(source.ForcePath) { + t.Fatal("source force_path_style = false, want true") + } + destination := cfg.Pipelines[0].Destinations[0] + if got, want := destination.Region, DefaultS3Region; got != want { + t.Fatalf("destination region = %q, want %q", got, want) + } + if !ForcePathStyle(destination.ForcePath) { + t.Fatal("destination force_path_style = false, want true") + } +} + +func TestLoadFilePreservesExplicitS3ForcePathStyleFalse(t *testing.T) { + cfg := loadConfig(t, ` +pipelines: + - id: s3-force-path + source: + backend: s3 + endpoint: https://s3.example.com + bucket: source + force_path_style: false + destinations: + - id: archive + backend: local + path: /archive +`) + + if ForcePathStyle(cfg.Pipelines[0].Source.ForcePath) { + t.Fatal("force_path_style = true, want explicit false") + } +} + func TestLoadFileRejectsDuplicatePipelineIDs(t *testing.T) { assertLoadError(t, ` pipelines: @@ -214,6 +269,19 @@ func TestLoadFileRejectsMissingRequiredFields(t *testing.T) { } } +func TestLoadFileRejectsInvalidS3Config(t *testing.T) { + tests := map[string]string{ + "prefix traversal": `pipelines: [{id: reports, source: {backend: s3, endpoint: "https://s3.example.com", bucket: source, prefix: "../reports"}, destinations: [{id: archive, backend: local, path: /archive}]}]`, + "prefix backslash": `pipelines: [{id: reports, source: {backend: s3, endpoint: "https://s3.example.com", bucket: source, prefix: 'a\b'}, destinations: [{id: archive, backend: local, path: /archive}]}]`, + "partial creds": `pipelines: [{id: reports, source: {backend: s3, endpoint: "https://s3.example.com", bucket: source, credentials: {access_key_id_env: ACCESS_KEY_ID}}, destinations: [{id: archive, backend: local, path: /archive}]}]`, + } + for name, body := range tests { + t.Run(name, func(t *testing.T) { + assertLoadError(t, body, "") + }) + } +} + func TestLoadFileDefaultsSSHConfig(t *testing.T) { cfg := loadConfig(t, ` pipelines: @@ -398,6 +466,7 @@ func TestExampleConfigsLoad(t *testing.T) { "../../examples/local-html.yml", "../../examples/fan-out.yml", "../../examples/ssh-destination.yml", + "../../examples/s3-destination.yml", } { t.Run(path, func(t *testing.T) { if _, err := LoadFile(path); err != nil { diff --git a/internal/config/s3.go b/internal/config/s3.go new file mode 100644 index 0000000..c0bb03f --- /dev/null +++ b/internal/config/s3.go @@ -0,0 +1,19 @@ +package config + +import ( + "strings" + + "gitea.maximumdirect.net/eric/distributor/internal/storage" +) + +func NormalizeS3Prefix(prefix string) string { + return strings.Trim(prefix, "/") +} + +func ValidateS3Prefix(prefix string) error { + return storage.ValidatePrefix(prefix) +} + +func ForcePathStyle(value *bool) bool { + return value == nil || *value +} diff --git a/internal/config/secrets.go b/internal/config/secrets.go index 68c1d46..fe10624 100644 --- a/internal/config/secrets.go +++ b/internal/config/secrets.go @@ -80,6 +80,9 @@ func (e Environment) required(name string) (string, error) { if !ok { return "", fmt.Errorf("credential environment variable %s is not set", name) } + if value == "" { + return "", fmt.Errorf("credential environment variable %s is empty", name) + } return value, nil } diff --git a/internal/config/secrets_test.go b/internal/config/secrets_test.go index cb6c29a..0fdbb44 100644 --- a/internal/config/secrets_test.go +++ b/internal/config/secrets_test.go @@ -207,6 +207,17 @@ func TestResolveCredentialsMissingReferenceFailsWithoutSecretValue(t *testing.T) } } +func TestResolveCredentialsRejectsEmptyReferencedValue(t *testing.T) { + env := NewEnvironment(map[string]string{"EMPTY": ""}, emptyLookup) + _, err := env.ResolveCredentials(Credentials{AccessKeyIDEnv: "EMPTY"}) + if err == nil { + t.Fatal("ResolveCredentials() error = nil, want error") + } + if !strings.Contains(err.Error(), "EMPTY") || !strings.Contains(err.Error(), "empty") { + t.Fatalf("ResolveCredentials() error = %q, want empty variable context", err) + } +} + func writeSecret(t *testing.T, directory, name, value string) { t.Helper() if err := os.WriteFile(filepath.Join(directory, name), []byte(value), 0o600); err != nil { diff --git a/internal/config/validate.go b/internal/config/validate.go index bf35cd2..a1f3313 100644 --- a/internal/config/validate.go +++ b/internal/config/validate.go @@ -69,14 +69,14 @@ func Validate(cfg Config) error { } func validateSourceBackend(errs ValidationErrors, context string, backend Backend) ValidationErrors { - return validateBackend(errs, context, backend.Backend, backend.Host, backend.Port, backend.Path, backend.URI, backend.Endpoint, backend.Bucket, backend.SSH.HostKeyPolicy) + return validateBackend(errs, context, backend.Backend, backend.Host, backend.Port, backend.Path, backend.URI, backend.Endpoint, backend.Bucket, backend.Prefix, backend.SSH.HostKeyPolicy, backend.Creds) } func validateDestinationBackend(errs ValidationErrors, context string, destination Destination) ValidationErrors { - return validateBackend(errs, context, destination.Backend, destination.Host, destination.Port, destination.Path, destination.URI, destination.Endpoint, destination.Bucket, destination.SSH.HostKeyPolicy) + return validateBackend(errs, context, destination.Backend, destination.Host, destination.Port, destination.Path, destination.URI, destination.Endpoint, destination.Bucket, destination.Prefix, destination.SSH.HostKeyPolicy, destination.Creds) } -func validateBackend(errs ValidationErrors, context, backend, host string, port int, path, uri, endpoint, bucket string, hostKeyPolicy HostKeyPolicy) ValidationErrors { +func validateBackend(errs ValidationErrors, context, backend, host string, port int, path, uri, endpoint, bucket, prefix string, hostKeyPolicy HostKeyPolicy, creds Credentials) ValidationErrors { switch backend { case "": errs = append(errs, context+".backend is required") @@ -112,6 +112,12 @@ func validateBackend(errs ValidationErrors, context, backend, host string, port if bucket == "" { errs = append(errs, context+".bucket is required for s3 backend") } + if err := ValidateS3Prefix(prefix); err != nil { + errs = append(errs, context+".prefix must be a clean relative slash-separated path") + } + if (creds.AccessKeyIDEnv == "") != (creds.SecretAccessKeyEnv == "") { + errs = append(errs, context+".credentials.access_key_id_env and credentials.secret_access_key_env must be configured together") + } default: errs = append(errs, context+".backend "+backend+" is unsupported") }