diff --git a/CHANGELOG.md b/CHANGELOG.md index c7be659b..08788e98 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,45 @@ All notable changes to virtkit will be documented in this file. ## [Unreleased] +### Added + +- **The registry's plain-file area is also served as S3.** Under `/s3/` on `vk-registry`, + each top-level directory of the WebDAV `files/` area is a bucket, so `sccache`'s S3 backend + and other S3 clients can share a compiler cache or files with it: point them at + `https:///s3` with any region (a bucket is made by the first write into it, so + `sccache` needs none made beforehand), and use the registry credential (an API key, a hub + token or the shared bearer) as both the access key and the secret access key, or the shared + Basic user and its password. Objects can be written, read with conditional headers and + checksums (in every algorithm S3 has, checked on upload and kept with the object, and a + CRC64NVME computed for an upload that sends none, as S3 does), given + metadata and deleted, and listed page by page (as object versions too, for + tools that empty a bucket that way), and are the same files WebDAV and `vk registry get` + see; a key limited to a directory of a bucket can list that directory. + The chunked uploads AWS SDKs and the CLI send are taken, their signatures and checksums + checked, and objects can be read in byte ranges, as the AWS CLI downloads large files. + Buckets can be created and removed, objects copied without their bytes being stored twice + (or with the checksum a copy asks for computed from them) and deleted in batches — only while still the ETag, modification time or size a delete + names, when it names one, as S3's conditional deletes — and the configuration probes tools send before using a bucket + (versioning, ACLs, tagging and the like) are answered as for a bucket with none set, and a + write asking for one of them — encryption under the client's own key, tags, an ACL, a + storage class, a redirect, a retention lock — is refused rather than stored without it, and + a refused upload sees its refusal rather than a dropped connection; every refusal carries + the error code, status and message S3 answers with. Large + files can be uploaded in parts, as the AWS CLI uploads them, every part but the last at + least 5 MiB as S3 requires, each part and the completion + held to the checksum the upload was created with, and the object given S3's composite or + full-object checksum of them, completed once however many + completions of the upload are sent, at once or as retries (each answered as the first while + the object is still the one it made), read back a part at a time, + parts copied from other objects, + and unfinished uploads listed, aborted, or cleaned up by `gc` once idle; `status` counts + them. Set `s3 = false` in the server config to turn it off. A newly created API key is also + shown as the S3 key pair, ready to paste. With wildcard DNS, setting `s3_host` also serves + each bucket at its own host name, as clients using virtual-hosted style address it. Jobs a + fleet hub places with their own registry token reach it too through the guest registry + proxy, holding no credential: `sccache` there takes bucket `job` with its area's alias as the + key prefix. + ## [0.106.0] - 2026-11-05 ### Added diff --git a/Cargo.lock b/Cargo.lock index bcc53cc9..0ebcdcf5 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4304,13 +4304,17 @@ dependencies = [ "anyhow", "bytes", "clap", + "crc", + "crc32fast", "futures", + "hmac", "http-body-util", "httpdate", "hyper", "hyper-util", "jemallocator", "libc", + "md5", "oci-client", "rand", "redb", @@ -4318,6 +4322,7 @@ dependencies = [ "rustls", "serde", "serde_json", + "sha1", "sha2", "tokio", "tokio-rustls", @@ -4328,6 +4333,8 @@ dependencies = [ "vk-oidc", "vk-peer", "vk-selfupdate", + "xxhash-rust", + "zeroize", "zstd", ] @@ -4821,6 +4828,12 @@ dependencies = [ "rustix", ] +[[package]] +name = "xxhash-rust" +version = "0.8.19" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "550a2b930b62486a393c52d5c3b84bff264b28aa437ed64694d31e93b1757af7" + [[package]] name = "yoke" version = "0.8.3" diff --git a/Cargo.toml b/Cargo.toml index 8124b74f..b77de3d7 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -128,9 +128,20 @@ oci-client = { version = "0.18", default-features = false } flate2 = "1" # docker-hash: content hashing of a (Dockerfile, stage) for image-tag resolution sha2 = "0.11" -# vk-tpm's SHA-1 PCR bank (TPM 2.0 keeps one for compatibility). RustCrypto, like sha2; already -# in the tree via russh. +# vk-tpm's SHA-1 PCR bank (TPM 2.0 keeps one for compatibility), and vk-registry's check of an +# S3 upload's x-amz-checksum-sha1. RustCrypto, like sha2; already in the tree via russh. sha1 = "0.11" +# vk-registry: the MD5 an S3 object's ETag is, and the check of a Content-MD5. Already in the +# tree via russh. +md5 = "0.8" +# vk-registry: the check of an S3 upload's x-amz-checksum-crc32. Already in the tree via flate2. +crc32fast = "1.5" +# vk-registry: the check of an S3 upload's x-amz-checksum-crc32c and -crc64nvme, from its +# catalog. Already in the tree via lzma-rs. +crc = "3" +# vk-registry: the check of an S3 upload's x-amz-checksum-xxhash64, -xxhash3 and -xxhash128. +# Pure Rust, no dependencies of its own; XXH3 is too intricate to rewrite safely. +xxhash-rust = { version = "0.8", features = ["xxh64", "xxh3"] } # vk-tpm: the TPM's entropy (TPM2_GetRandom, its seeds) from the OS. Already in the tree via rand. getrandom = "0.4" # vk-tpm: constant-time comparison of authorization values. Already in the tree via the diff --git a/NOTICE b/NOTICE index cf283b38..dacd4dbb 100644 --- a/NOTICE +++ b/NOTICE @@ -74,6 +74,12 @@ LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. +vk-registry's SigV4 tests use AWS's signing test suite, as vendored in awslabs/aws-c-auth +(tests/aws-signing-test-suite/v4), copied under vk-registry/src/s3/testdata/sigv4/: +https://github.com/awslabs/aws-c-auth/tree/3281f8692e6fd10562c4585a4dded5c16b322698/tests/aws-signing-test-suite/v4 +Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +Apache License 2.0; see vk-registry/src/s3/testdata/sigv4/LICENSE. + The variable expansion in vk-gitlab/src/spec.rs (`expand`) and vk-driver/src/node/jobs/vars.rs is adapted from Go's os.Expand (src/os/env.go) and its tests, as gitlab-runner uses it: https://github.com/golang/go/blob/go1.24.0/src/os/env.go diff --git a/README.md b/README.md index fe7872ab..d803d259 100644 --- a/README.md +++ b/README.md @@ -855,6 +855,22 @@ with `read:files//*` only reads it. Cached files expire under `vk-registry retention like any other tag, unless their manifest keeps or expires them. Set `webdav = false` in the server config to disable WebDAV. +The same plain-file area is served as S3 under `/s3/`, a bucket being a top-level directory +of `/dav/files/`, so `sccache`'s S3 backend, the AWS CLI and SDKs and other S3 clients can use +it. The access key is the registry credential — an API key, a hub token, or the shared bearer — +and the secret access key is the same string; with a shared Basic user, the user name and its +password. For `sccache`: + +```bash +SCCACHE_BUCKET=sccache SCCACHE_ENDPOINT=https:///s3 SCCACHE_REGION=auto \ +SCCACHE_S3_KEY_PREFIX= AWS_ACCESS_KEY_ID=$KEY AWS_SECRET_ACCESS_KEY=$KEY sccache … +``` + +Object keys follow the store's names (`A-Z a-z 0-9 . _ -` between slashes, at most 1024 +bytes); objects are at most 4 GiB; presigned URLs are refused. Set `s3 = false` to disable +it. With wildcard DNS, `s3_host = "s3.example.com"` also serves each bucket virtual-hosted +style, as `https://.s3.example.com/`. + `vk registry put [--dir ] [--ttl 7d | --keep] ...` stores files there, each once by digest (`/dav/files/sha256/`) and, with `--dir`, by name (`/dav/files//`), and `vk registry get ` fetches one back, checked against its digest; @@ -883,7 +899,7 @@ See [`docs/registry-hub.md`](docs/registry-hub.md). | --- | --- | | `vk` | Host CLI, VMM, image builder, userspace network, compose runner, and GitLab executor. It embeds the default guest kernel and `vk-agent`. | | `vk-agent` | Guest PID 1 and command server. It configures mounts, networking, hostname, shared directories, optional SSH, and host-driven execution over vsock. | -| `vk-registry` | Optional OCI-distribution server with a pull-through cache, a WebDAV view of the store with a plain-file area for compiler caches, and build-once locking. | +| `vk-registry` | Optional OCI-distribution server with a pull-through cache, a WebDAV view of the store with a plain-file area for compiler caches, also served as S3, and build-once locking. | | `vk-hub` | Experimental hub for a fleet of `vk node` hosts, and a web UI for the VMs on this machine: `vk-hub local` serves it on loopback. See the [fleet design](docs/fleet-design.md) and [prototype reference](docs/fleet-prototype.md). | | `vk-gitlab` | Experimental GitLab runner for a `vk-hub` fleet: takes jobs from GitLab and has the hub run each on a `vk node`. See [GitLab dispatch](docs/gitlab-dispatch.md). | | `vk-runnerctl` | Optional root-side helper that adjusts GitLab runner concurrency within an administrator-configured range. | diff --git a/docs/gitlab-dispatch.md b/docs/gitlab-dispatch.md index fb4cea7b..75d349eb 100644 --- a/docs/gitlab-dispatch.md +++ b/docs/gitlab-dispatch.md @@ -664,10 +664,21 @@ job and injects the job's token, never the node's credential. A job with no toke proxy that refuses everything; so does every job a local gitlab-runner gives the node, which under `"job"` reaches the registry only by its own means. -With a token, the proxy forwards the lock API (`/lock/`) and plain files (`/dav/files/`) as -well as `/v2/`; the registry enforces the token's grants. Under `/dav/files/` only GET, -HEAD, PROPFIND, PUT, MKCOL, DELETE and OPTIONS pass: MOVE and COPY would name the proxy in -their `Destination`. The job's steps find the proxy in `VK_REGISTRY` (`http://registry.vk`) +With a token, the proxy forwards the lock API (`/lock/`), plain files (`/dav/files/`) and their +S3 view (`/s3`) as well as `/v2/`; the registry enforces the token's grants. Under +`/dav/files/` only GET, HEAD, PROPFIND, PUT, MKCOL, DELETE and OPTIONS pass: MOVE and COPY +would name the proxy in their `Destination`. On `/s3`, the guest's own signature is replaced by +the token, so a job points `sccache` at its area with no key at all: + +```sh +export SCCACHE_BUCKET=job SCCACHE_S3_KEY_PREFIX=/ SCCACHE_ENDPOINT=$VK_REGISTRY/s3 +export SCCACHE_REGION=auto SCCACHE_S3_NO_CREDENTIALS=true +``` + +`job/` there stands for the area's prefix, as on `/dav/files/job/` +([4b](registry-hub.md#4b-registry-access-from-inside-a-job)); `SCCACHE_S3_NO_CREDENTIALS` +sends the requests unsigned, and dummy `AWS_ACCESS_KEY_ID`/`AWS_SECRET_ACCESS_KEY` work as +well. The job's steps find the proxy in `VK_REGISTRY` (`http://registry.vk`) and their grants in `VK_REGISTRY_NAMESPACES`, space-separated `=` with `r`, `w` and `l` for read, write and lock. The proxy needs `net.mode = "switch"`, the only network that serves `registry.vk`; under any other, a job gets neither the proxy nor these variables. diff --git a/docs/registry-hub.md b/docs/registry-hub.md index 0191231a..522369ef 100644 --- a/docs/registry-hub.md +++ b/docs/registry-hub.md @@ -632,8 +632,8 @@ though within them it can claim protection. ### 4b: registry access from inside a job -- **The proxy.** When it injects a job token, the credential proxy forwards `/lock/` and - `/dav/files/` as well as `/v2/`. With the node's credential it keeps refusing anything +- **The proxy.** When it injects a job token, the credential proxy forwards `/lock/`, + `/dav/files/` and the S3 view (`/s3`) as well as `/v2/`. With the node's credential it keeps refusing anything outside `/v2/`, since that credential is not scoped. The proxy filters no namespace itself: the registry enforces the token. - It sets no response timeout of its own: `/lock/acquire` long-polls for up to an hour by @@ -674,12 +674,21 @@ though within them it can claim protection. (`name=job//…:`, decoded as the registry decodes it). The mapped URL goes through the proxy's checks again. Any other path is forwarded as it is; `job/…` naming no alias of the job is refused with 403 and not forwarded. + - On the S3 view, `/s3/job//` is likewise the key below the area's prefix, and + `x-amz-copy-source: /job//` the same. A prefix `a/b/c` is bucket `a` and key + prefix `b/c/`. `/s3/job/` alone is the area as a bucket: its listings go to bucket + `a` with `b/c/` put before their `prefix`, `start-after`, `marker` and `key-marker` and + into their `continuation-token` (base64url of a key), so they stay in the area; when the + prefix has more than one component, only reads pass there. The guest's SigV4 + `Authorization`, with dummy keys or none, is replaced by the token, as any other. - No other query parameter is mapped: an OCI cross-repository mount's `from=job//…` reaches the registry as it is. - Only requests are mapped. The registry answers with real names, an upload's `Location` among them, which the proxy forwards unchanged, so a client continues on the real path. A `PROPFIND`'s hrefs are mapped back to the alias, since a WebDAV client resolves a - listing against the root it asked for. + listing against the root it asked for, and so are the names in an S3 document answering + a request through an area — the bucket, keys, prefixes, markers, continuation tokens, a + `Location` and an error's `Resource` — and a created bucket's `Location`. - No namespace's path may start with `job`. The hub refuses an alias while one stored before that rule has such a path, and gives no job any alias until it moves. - The hub sends a job's areas (`JobRegistry.areas`) only to a node announcing @@ -694,7 +703,17 @@ though within them it can claim protection. - **In-job tools.** task-rs takes `vk: http://registry.vk/job/task-cache` as its cache: entries under the alias and lock names `job/task-cache/:`. sccache uses `$VK_REGISTRY/dav/files/job/` as its WebDAV endpoint (the alias in the endpoint, not - in `key_prefix`, so it never creates `/dav/files/job/` itself). Neither holds a credential. + in `key_prefix`, so it never creates `/dav/files/job/` itself), or the S3 view: + + ```sh + SCCACHE_BUCKET=job SCCACHE_S3_KEY_PREFIX=/ SCCACHE_ENDPOINT=$VK_REGISTRY/s3 \ + SCCACHE_REGION=auto SCCACHE_S3_NO_CREDENTIALS=true + ``` + + `SCCACHE_S3_NO_CREDENTIALS` sends unsigned requests; dummy `AWS_ACCESS_KEY_ID` and + `AWS_SECRET_ACCESS_KEY` do as well, since the proxy drops the signature. Other S3 clients + list an area as the bucket `` at the endpoint `$VK_REGISTRY/s3/job`, path-style + (`aws --endpoint-url $VK_REGISTRY/s3/job s3 ls s3:///`). None holds a credential. ### 4c: pipeline caches and captures diff --git a/tests/gitlab-e2e.sh b/tests/gitlab-e2e.sh index ac699432..7c0ffac8 100755 --- a/tests/gitlab-e2e.sh +++ b/tests/gitlab-e2e.sh @@ -33,7 +33,7 @@ # Environment: VK_GITLAB_E2E_CACHE (default ~/.cache/vk-gitlab-e2e) holds the GitLab # instance; GL_MEM (6G) and GL_CPUS (4) size its VM; GL_BOOT_TIMEOUT (1500 s). E2E_TOKENS=0 # runs the fleet without hub registry tokens: the registry not enrolled, the node's own -# credential for caches, and the `registry` and `pipeline_build` scenarios skipped. +# credential for caches, and the `registry`, `pipeline_build` and `registry_s3` scenarios skipped. # # Needs: KVM, curl, jq, git, openssl, ss, ip, shuf, network to Docker Hub; about 9 GiB of free RAM # (GitLab 6 GiB, job VMs 1 GiB each) and 15 GiB of disk for the GitLab image and its state. @@ -52,7 +52,7 @@ REPO=$(cd "$E2E/../.." && pwd) # shellcheck source=gitlab/scenarios.sh . "$E2E/scenarios.sh" -ALL_SCENARIOS=(basic scripts variables exit_codes artifacts cache registry pipeline_build services +ALL_SCENARIOS=(basic scripts variables exit_codes artifacts cache registry pipeline_build registry_s3 services timeout cancel restart) usage() { @@ -210,7 +210,7 @@ pids=() for s in "${SCENARIOS[@]}"; do case $s in restart) ;; - registry | pipeline_build) + registry | pipeline_build | registry_s3) if [ -z "$VK_REGISTRY" ] || [ "$E2E_TOKENS" = 0 ]; then echo "SKIP|needs a vk-registry taking hub tokens" >"$RUN/results/$s" else diff --git a/tests/gitlab/scenarios.sh b/tests/gitlab/scenarios.sh index 09bb10e1..f218dedc 100644 --- a/tests/gitlab/scenarios.sh +++ b/tests/gitlab/scenarios.sh @@ -391,6 +391,59 @@ EOF SC_DETAIL="built in $p1 ($ref), refused in $p2" } +# A job reaches the S3 view through the guest proxy by an area's alias, the area three +# components deep (bucket e2e-s3, keys below e2e//): it writes, reads, lists and +# probes an object there unsigned, the proxy lending the job's token, and is refused another +# alias and a bucket its token does not cover. busybox wget sends neither PUT nor HEAD, so the +# requests go over nc. The token stays out of the trace. +sc_registry_s3() { + local ci p j + hub namespaces add "$REG_ID" e2e-s3 --path 'e2e-s3/{gitlab}/{project_id}' \ + --alias s3area --grant 'rw:own project' >>"$LOGS/hub-cli.log" 2>&1 || { + bad "vk-hub namespaces add e2e-s3 failed" + return 1 + } + ci=$({ + ci_header + cat <<'EOF' +s3: + script: + - | + H=${VK_REGISTRY#http://} + HOST=${H%%:*} PORT=80 + case $H in *:*) PORT=${H##*:} ;; esac + req() { + printf '%s %s HTTP/1.0\r\nHost: %s\r\nContent-Length: %s\r\nConnection: close\r\n\r\n%s' \ + "$1" "$2" "$H" "${#3}" "${3:-}" | nc "$HOST" "$PORT" | tr -d '\r' + } + code() { req "$@" | awk 'NR == 1 { print $2 }'; } + BODY=s3-probe-$CI_PIPELINE_ID + echo "put=$(code PUT /s3/job/s3area/probe "$BODY")" + echo "get=$(code GET /s3/job/s3area/probe)" + echo "get-body=$(req GET /s3/job/s3area/probe | sed '1,/^$/d')" + LIST=$(req GET '/s3/job/s3area?list-type=2') + echo "list=$(printf '%s\n' "$LIST" | awk 'NR == 1 { print $2 }')" + echo "list-key=$(printf '%s' "$LIST" | grep -c 'probe')" + echo "list-name=$(printf '%s' "$LIST" | sed -n 's:.*\([^<]*\).*:\1:p')" + echo "head=$(code HEAD /s3/job/s3area/probe)" + echo "head-length=$(req HEAD /s3/job/s3area/probe | sed -n 's/^[Cc]ontent-[Ll]ength: *//p')" + echo "put-nope=$(code PUT /s3/job/nope/x x)" + echo "put-images=$(code PUT /s3/images/x x)" +EOF + }) + p=$(push_scenario registry_s3 <<<"$ci") || return 1 + wait_pipeline "$p" success || return 1 + j=$(job "$p" s3) + keep_job "$j" + local body=s3-probe-$p want + for want in put=200 get=200 "get-body=$body" list=200 list-key=1 list-name=s3area head=200 \ + "head-length=${#body}" put-nope=403 put-images=403; do + grep -qE "${LINE}${want}\$" "$RUN/jobs/$j.log" || bad "job $j's trace lacks a line '$want'" + done + trace_lacks "$j" vkt_ + SC_DETAIL="pipeline $p: an area's S3 alias written, read, listed and probed; others refused" +} + sc_services() { local p j p=$({ diff --git a/tests/s3-e2e.sh b/tests/s3-e2e.sh new file mode 100755 index 00000000..da76ae6d --- /dev/null +++ b/tests/s3-e2e.sh @@ -0,0 +1,166 @@ +#!/usr/bin/env bash +# vk-registry's S3 view (/s3/) driven by real clients in a microVM: sccache, the AWS CLI and +# rclone (tests/s3/Dockerfile), against a `vk-registry serve` in accounts mode on this host. +# +# 1. The registry listens with TLS on the host's LAN address (a guest cannot reach the host's +# loopback, and credentials off loopback need TLS), with a certificate from a CA made for +# the run; a `write:files/*` and a `read:files/*` key are minted over the admin socket. +# 2. The image boots detached with --net; the guest trusts the run's CA. +# 3. sccache: one miss then one hit; a read-only key hits and writes nothing; a wrong secret +# fails its startup check. +# 4. AWS CLI: mb, cp of 1 KiB, 20 MiB and 100 MiB each way (multipart, aws-chunked bodies with +# CRC32 trailers, ranged GETs), a paged listing, sync twice, head, ranged get, copy, batch +# delete, a multipart upload by hand, a presigned URL refused, rb --force. +# 5. rclone: copy, ls, check, sync twice (the second copies nothing), delete. +# 6. On the host, with the registry stopped: `vk registry get` reads an object the CLI wrote, +# and `vk-registry gc` keeps it. +# +# Run: VK=vk VK_REGISTRY=./dist/vk-registry tests/s3-e2e.sh +# VK=./dist/vk tests/release-e2e.sh tests/s3-e2e.sh +# VK is the vk that runs the guest (default: vk on PATH); VK_REGISTRY the vk-registry under +# test (default: the one beside $VK, else ./dist/vk-registry; build it with +# `./build.sh --fast --only=vk-registry`). WORK_DIR is where the store and logs go (default +# $HOME: the store holds a few hundred MiB, too much for a tmpfs /tmp); KEEP_WORK=1 keeps it. +# Needs: KVM, network (the image's packages), a host LAN IPv4 address, openssl, curl, python3. +set -euo pipefail + +VK=$(command -v "${VK:-vk}" || true) +[ -n "$VK" ] && [ -x "$VK" ] || { echo "no usable vk (build one: ./build.sh --fast)"; exit 2; } +VK=$(cd "$(dirname "$VK")" && pwd)/$(basename "$VK") +if [ -z "${VK_REGISTRY:-}" ]; then + VK_REGISTRY=./dist/vk-registry + [ -x "$(dirname "$VK")/vk-registry" ] && VK_REGISTRY=$(dirname "$VK")/vk-registry +fi +[ -x "$VK_REGISTRY" ] || { echo "no $VK_REGISTRY (build it: ./build.sh --fast --only=vk-registry)"; exit 2; } +VK_REGISTRY=$(cd "$(dirname "$VK_REGISTRY")" && pwd)/$(basename "$VK_REGISTRY") +cd "$(dirname "$0")/.." +for t in openssl curl python3; do command -v "$t" >/dev/null || { echo "need $t"; exit 2; }; done + +# The guest reaches the host's private LAN address through ordinary egress. +IP=$(ip -4 route get 1.1.1.1 2>/dev/null | awk '{ for (i = 1; i < NF; i++) if ($i == "src") { print $(i + 1); exit } }') +[ -n "$IP" ] || IP=$(hostname -I 2>/dev/null | awk '{ print $1 }') +case $IP in + '' | 127.* | 169.254.*) echo "no LAN IPv4 address on this host for the guest to reach the registry"; exit 2 ;; +esac + +WORK=$(mktemp -d "${WORK_DIR:-$HOME}/vk-s3-e2e.XXXXXX") +# Unix sockets (the admin socket, the VM's vsock) need a short path. +SOCKS=$(mktemp -d /tmp/vk-s3-e2e.XXXXXX) +VM=$SOCKS/vm +SHARE=$WORK/share +LOGS=$SHARE/logs +mkdir -p "$WORK/pki" "$WORK/store" "$LOGS" +reg_pid= +cleanup() { + "$VK" stop "$VM" >/dev/null 2>&1 || true + [ -z "$reg_pid" ] || kill "$reg_pid" 2>/dev/null || true + wait 2>/dev/null || true + rm -rf "$SOCKS" + if [ -n "${KEEP_WORK:-}" ]; then echo "kept $WORK"; else rm -rf "$WORK"; fi +} +trap cleanup EXIT +fail() { + echo "FAIL: $*" + echo "--- registry log"; tail -n 40 "$WORK/registry.log" || true + for f in "$LOGS"/*; do + [ -f "$f" ] && [ -s "$f" ] || continue + echo "--- $(basename "$f")"; tail -n 40 "$f" + done + exit 1 +} +wait_for() { + local secs=$1; shift + for _ in $(seq "$((secs * 2))"); do + if "$@" >/dev/null 2>&1; then return 0; fi + sleep 0.5 + done + return 1 +} +port() { python3 -c 'import socket; s=socket.socket(); s.bind(("0.0.0.0",0)); print(s.getsockname()[1])'; } +# One client's scenario in the guest (tests/s3/clients.sh), its output kept in logs/. +client() { + echo "== $1 ==" + if ! "$VK" exec "$VM" -- bash /e2e/clients.sh "$1" 2>&1 | tee "$LOGS/$1.out"; then fail "$1"; fi +} + +echo "== 1. the registry, with TLS on $IP, and two keys ==" +d=$WORK/pki +openssl req -x509 -newkey ec -pkeyopt ec_paramgen_curve:P-256 -nodes -days 2 \ + -subj /CN=vk-s3-e2e-ca -keyout "$d/ca.key" -out "$d/ca.pem" \ + -addext basicConstraints=critical,CA:TRUE -addext keyUsage=critical,keyCertSign 2>/dev/null +openssl req -newkey ec -pkeyopt ec_paramgen_curve:P-256 -nodes -subj /CN=vk-registry \ + -keyout "$d/server.key" -out "$d/server.csr" 2>/dev/null +printf '%s\n' "subjectAltName=IP:$IP" 'basicConstraints=critical,CA:FALSE' \ + 'keyUsage=critical,digitalSignature' 'extendedKeyUsage=serverAuth' >"$d/server.ext" +openssl x509 -req -in "$d/server.csr" -CA "$d/ca.pem" -CAkey "$d/ca.key" \ + -CAcreateserial -days 2 -extfile "$d/server.ext" -out "$d/server.pem" 2>/dev/null +cp "$d/ca.pem" "$SHARE/ca.pem" + +PORT=$(port) +# Accounts mode needs an [oidc] table; no one signs in, so its provider is never asked. +printf 'unused\n' >"$WORK/oidc-secret" +cat >"$WORK/registry.toml" <"$WORK/registry.log" 2>&1 & +reg_pid=$! +URL=https://$IP:$PORT +wait_for 30 curl -sS --cacert "$d/ca.pem" -o /dev/null "$URL/v2/" || fail "the registry did not come up" +key() { + "$VK_REGISTRY" accounts create-key --config "$WORK/registry.toml" --name "$1" --scope "$2" \ + 2>>"$WORK/keys.log" || fail "create-key $1" +} +WKEY=$(key s3-e2e-write 'write:files/*') +RKEY=$(key s3-e2e-read 'read:files/*') +code=$(curl -sS --cacert "$d/ca.pem" -o /dev/null -w '%{http_code}' "$URL/s3/") +[ "$code" = 403 ] || fail "an unsigned ListBuckets got $code, not 403" +cat >"$SHARE/env" <"$WORK/vm.log" 2>&1 || { cat "$WORK/vm.log"; fail "the guest did not boot"; } +client trust + +echo "== 3. sccache ==" +client sccache +echo "== 4. AWS CLI ==" +client awscli +echo "== 5. rclone ==" +client rclone + +echo "== 6. the host side: vk registry get, and gc keeps it ==" +"$VK" stop "$VM" >/dev/null +kill "$reg_pid" +wait "$reg_pid" 2>/dev/null || true +reg_pid= +got() { + rm -f "$WORK/got" + "$VK" registry get --cache-registry "$WORK/store" -o "$WORK/got" keep/from-cli.bin \ + >>"$LOGS/vk-registry-get.log" 2>&1 && + [ "$(sha256sum <"$WORK/got" | cut -d' ' -f1)" = "$(cat "$SHARE/keep.sha256")" ] +} +got || fail "vk registry get does not fetch what the AWS CLI wrote" +"$VK_REGISTRY" gc --root "$WORK/store" >"$LOGS/gc.log" 2>&1 || fail "vk-registry gc" +got || fail "the gc did not keep what the AWS CLI wrote" + +echo "PASS: sccache, the AWS CLI and rclone over /s3/, and the store keeps what they wrote, in $(($(date +%s) - start))s" diff --git a/tests/s3/Dockerfile b/tests/s3/Dockerfile new file mode 100644 index 00000000..b5f5e447 --- /dev/null +++ b/tests/s3/Dockerfile @@ -0,0 +1,24 @@ +# The S3 clients tests/s3-e2e.sh drives against vk-registry's /s3/: sccache, the AWS CLI and +# rclone, plus gcc for sccache to wrap. No credential is baked in: the script passes keys and +# the run's CA at run time. +# +# The digest is trixie-slim's multi-platform index on Docker Hub, read with +# docker manifest inspect -v debian:trixie-slim (or, without docker, the +# Docker-Content-Digest of GET registry-1.docker.io/v2/library/debian/manifests/trixie-slim) +FROM debian:trixie-slim@sha256:a72b4433de1cf7d6bae0c4cee74d0547f58601a643b3375f070c51600d63b6b2 +# awscli 2.23 and rclone 1.60 from trixie. +RUN apt-get update \ + && apt-get install -y --no-install-recommends \ + awscli rclone gcc libc6-dev ca-certificates coreutils curl \ + && rm -rf /var/lib/apt/lists/* +# trixie's sccache (0.10) is built without its S3 backend, so the upstream static release, +# checked against the sha256 published beside it. +ARG SCCACHE_VERSION=0.18.0 +ARG SCCACHE_SHA256=45f1447fbe231e3037bde351ef70677dd212216c8d62ae7ca409fecc4d6acc89 +RUN cd /tmp \ + && t=sccache-v$SCCACHE_VERSION-x86_64-unknown-linux-musl \ + && curl -fsSLO https://github.com/mozilla/sccache/releases/download/v$SCCACHE_VERSION/$t.tar.gz \ + && echo "$SCCACHE_SHA256 $t.tar.gz" | sha256sum -c - \ + && tar xzf $t.tar.gz $t/sccache \ + && install -m 0755 $t/sccache /usr/local/bin/sccache \ + && rm -rf $t $t.tar.gz diff --git a/tests/s3/clients.sh b/tests/s3/clients.sh new file mode 100644 index 00000000..00800cbc --- /dev/null +++ b/tests/s3/clients.sh @@ -0,0 +1,244 @@ +#!/usr/bin/env bash +# The guest side of tests/s3-e2e.sh: one client's scenario against the registry's /s3/, run as +# bash /e2e/clients.sh +# in the tests/s3/Dockerfile image. /e2e is the host's share: `env` (S3_ENDPOINT, WKEY, RKEY, +# written by the host script), `ca.pem` (the run's CA) and `logs/`, where each +# client's own output goes. A failed check prints FAIL and exits non-zero. +set -euo pipefail +. /e2e/env +LOGS=/e2e/logs +mkdir -p "$LOGS" +T=$(mktemp -d) +trap 'rm -rf "$T"' EXIT + +fail() { echo "FAIL: $*"; exit 1; } +secs() { awk -v a="$1" -v b="$(date +%s.%N)" 'BEGIN { printf "%.2fs", b - a }'; } +same() { [ "$(sha256sum <"$1" | cut -d' ' -f1)" = "$(sha256sum <"$2" | cut -d' ' -f1)" ]; } + +# The run's CA in the system store, which curl, rclone and sccache read; the AWS CLI does not +# (it takes AWS_CA_BUNDLE), and each client is also given the file explicitly below +# (sccache SSL_CERT_FILE, rclone --ca-cert). +trust() { + cp /e2e/ca.pem /usr/local/share/ca-certificates/vk-s3-e2e.crt + update-ca-certificates >/dev/null 2>&1 + curl -sS -o /dev/null "$S3_ENDPOINT/" || fail "curl does not reach $S3_ENDPOINT over TLS" +} + +# ---------------------------------------------------------------------------- sccache +sccache_env() { + export SCCACHE_BUCKET=sccache SCCACHE_ENDPOINT=$S3_ENDPOINT SCCACHE_S3_USE_SSL=true + export SCCACHE_REGION=auto SCCACHE_S3_KEY_PREFIX=ci/ SCCACHE_IDLE_TIMEOUT=0 + export SSL_CERT_FILE=/e2e/ca.pem SCCACHE_LOG=debug + export AWS_ACCESS_KEY_ID=$1 AWS_SECRET_ACCESS_KEY=$2 + export SCCACHE_ERROR_LOG=$LOGS/sccache-$3.server.log +} +# The value of one `--show-stats` counter, e.g. `stat "Cache hits"`. +stat() { sccache --show-stats | awk -v k="$1" 'index($0, k) == 1 { print $NF; exit }'; } +compile() { + sccache gcc -c "$T/hello.c" -o "$T/hello.o" + sccache --show-stats >"$LOGS/sccache-$1.stats" +} + +# The bucket's entries, listed with the write key. +keys() { + AWS_ACCESS_KEY_ID=$WKEY AWS_SECRET_ACCESS_KEY=$WKEY AWS_DEFAULT_REGION=us-east-1 \ + AWS_ENDPOINT_URL=$S3_ENDPOINT AWS_CA_BUNDLE=/e2e/ca.pem aws s3 ls --recursive s3://sccache/ | + awk '{ print $NF }' +} + +run_sccache() { + # SCCACHE_LOG makes the client talk too: its stderr goes to a log of its own. + exec 2>>"$LOGS/sccache-client.log" + cat >"$T/hello.c" <<'EOF' +int hello(int x) { return x * 42; } +EOF + # sccache never creates its bucket. + AWS_ACCESS_KEY_ID=$WKEY AWS_SECRET_ACCESS_KEY=$WKEY AWS_DEFAULT_REGION=us-east-1 \ + AWS_ENDPOINT_URL=$S3_ENDPOINT AWS_CA_BUNDLE=/e2e/ca.pem aws s3 mb s3://sccache + + sccache_env "$WKEY" "$WKEY" write + sccache --start-server + compile miss + [ "$(stat "Cache misses")" = 1 ] || fail "the first compile was not one miss" + [ "$(stat "Cache hits")" = 0 ] || fail "the first compile hit" + [ "$(stat "Cache write errors")" = 0 ] || fail "sccache could not write its entry" + sccache --stop-server >/dev/null + sccache --start-server + compile hit + [ "$(stat "Cache hits")" = 1 ] || fail "the second compile was not one hit" + [ "$(stat "Cache misses")" = 0 ] || fail "the second compile missed" + sccache --stop-server >/dev/null + + # A read-only key: the startup check finds the bucket read-only (its write of + # .sccache_check is refused), the entry is still a hit, and a new one is not written. + entries=$(keys) + sccache_env "$RKEY" "$RKEY" read + sccache --start-server + grep -q 'server has setup with ReadOnly' "$SCCACHE_ERROR_LOG" || + fail "sccache did not find the bucket read-only with a read key" + compile read-hit + [ "$(stat "Cache hits")" = 1 ] || fail "the read-only key did not hit" + echo 'int other(int x) { return x - 1; }' >"$T/hello.c" + compile read-miss + [ "$(stat "Cache misses")" = 1 ] || fail "the read-only key's new compile was not a miss" + sccache --stop-server >/dev/null + [ "$(keys)" = "$entries" ] || fail "an entry was written with the read-only key" + + # A wrong secret: the storage check at startup fails, and so does the server. + sccache_env "$WKEY" "${WKEY}x" wrong + if sccache --start-server >"$LOGS/sccache-wrong.out" 2>&1; then + sccache --stop-server >/dev/null 2>&1 || true + fail "sccache started with a wrong secret" + fi + grep -q SignatureDoesNotMatch "$LOGS/sccache-wrong.out" || + fail "sccache's startup failure does not name the signature" + echo "sccache: a miss then a hit, read-only with a read key, refused with a wrong secret" +} + +# ---------------------------------------------------------------------------- AWS CLI +run_awscli() { + export AWS_ACCESS_KEY_ID=$WKEY AWS_SECRET_ACCESS_KEY=$WKEY AWS_DEFAULT_REGION=us-east-1 + export AWS_ENDPOINT_URL=$S3_ENDPOINT AWS_CA_BUNDLE=/e2e/ca.pem + aws s3 mb s3://cli + for size in 1K 20M 100M; do + head -c "$size" /dev/urandom >"$T/$size.bin" + s=$(date +%s.%N) + aws s3 cp --no-progress "$T/$size.bin" "s3://cli/obj/$size.bin" + put=$(secs "$s") + s=$(date +%s.%N) + aws s3 cp --no-progress "s3://cli/obj/$size.bin" "$T/$size.back" + echo "timing: $size put $put, get $(secs "$s")" + same "$T/$size.bin" "$T/$size.back" || fail "$size came back different" + done + # The CLI's default checksum, CRC64NVME, is taken unchecked: these the server checks, in a + # PutObject's trailer and in each part's of a multipart upload. + for algo in CRC32 SHA1 SHA256; do + for size in 1K 20M; do + aws s3 cp --no-progress --checksum-algorithm "$algo" "$T/$size.bin" "s3://cli/sum/$algo-$size" + aws s3 cp --no-progress "s3://cli/sum/$algo-$size" "$T/sum.back" + same "$T/$size.bin" "$T/sum.back" || fail "$size with $algo came back different" + done + done + aws s3 ls s3://cli/obj/ | tee "$T/ls" + [ "$(wc -l <"$T/ls")" = 3 ] || fail "s3 ls does not list 3 objects" + + # Seven keys, two a page. + for i in 1 2 3 4 5 6 7; do echo "$i" | aws s3 cp - "s3://cli/page/k$i"; done + token= pages=0 + : >"$T/keys" + while :; do + aws s3api list-objects-v2 --bucket cli --prefix page/ --max-keys 2 --no-paginate \ + ${token:+--continuation-token "$token"} --output json >"$T/page.json" + python3 -I -c ' +import json, sys +p = json.load(open(sys.argv[1])) +for o in p.get("Contents", []): print(o["Key"], file=open(sys.argv[2], "a")) +print(p.get("NextContinuationToken", "") if p["IsTruncated"] else "")' \ + "$T/page.json" "$T/keys" >"$T/next" + pages=$((pages + 1)) + token=$(cat "$T/next") + [ -n "$token" ] || break + [ "$pages" -lt 10 ] || fail "the listing does not end" + done + [ "$pages" = 4 ] || fail "7 keys came in $pages pages of 2" + [ "$(printf 'page/k%s\n' 1 2 3 4 5 6 7)" = "$(cat "$T/keys")" ] || fail "the pages: $(cat "$T/keys")" + + mkdir -p "$T/tree/sub" + for f in a b sub/c; do head -c 3000 /dev/urandom >"$T/tree/$f"; done + # sync uploads a file newer than its object, whose LastModified is in whole seconds, as + # S3's is: a file written in the second its upload lands would go again. + touch -d '-1 minute' "$T/tree/a" "$T/tree/b" "$T/tree/sub/c" + aws s3 sync --no-progress "$T/tree" s3://cli/tree | tee "$T/sync1" + [ "$(grep -c '^upload:' "$T/sync1")" = 3 ] || fail "the first sync did not upload 3 files" + aws s3 sync --no-progress "$T/tree" s3://cli/tree >"$T/sync2" + [ ! -s "$T/sync2" ] || fail "the second sync did something: $(cat "$T/sync2")" + + aws s3api head-object --bucket cli --key obj/1K.bin --output json >"$T/head.json" + grep -q '"ContentLength": 1024' "$T/head.json" || fail "head-object: $(cat "$T/head.json")" + aws s3api get-object --bucket cli --key obj/100M.bin --range bytes=0-9 "$T/range" >/dev/null + [ "$(head -c 10 "$T/100M.bin" | od -An -tx1)" = "$(od -An -tx1 <"$T/range")" ] || + fail "get-object --range bytes=0-9 is not the first 10 bytes" + s=$(date +%s.%N) + aws s3api get-object --bucket cli --key obj/100M.bin --range bytes=52428800- "$T/tail" >/dev/null + echo "timing: 100M ranged get of the second half $(secs "$s")" + tail -c 52428800 "$T/100M.bin" | same /dev/stdin "$T/tail" || fail "the ranged second half differs" + aws s3api copy-object --bucket cli --key obj/copy.bin --copy-source cli/obj/20M.bin >/dev/null + aws s3 cp --no-progress s3://cli/obj/copy.bin "$T/copy.back" + same "$T/20M.bin" "$T/copy.back" || fail "copy-object made other bytes" + aws s3api delete-objects --bucket cli --output json \ + --delete '{"Objects":[{"Key":"page/k1"},{"Key":"page/k2"},{"Key":"page/none"}]}' >"$T/del.json" + [ "$(grep -c '"Key"' "$T/del.json")" = 3 ] || fail "delete-objects: $(cat "$T/del.json")" + [ "$(aws s3 ls s3://cli/page/ | wc -l)" = 5 ] || fail "delete-objects left the wrong keys" + + # A multipart upload by hand: three parts of 5 MiB, 5 MiB and 1 MiB. + head -c 11534336 /dev/urandom >"$T/mp.bin" + split -b 5242880 -d -a 1 "$T/mp.bin" "$T/mp.part" + id=$(aws s3api create-multipart-upload --bucket cli --key obj/manual.bin --query UploadId --output text) + parts= + for n in 1 2 3; do + etag=$(aws s3api upload-part --bucket cli --key obj/manual.bin --upload-id "$id" \ + --part-number "$n" --body "$T/mp.part$((n - 1))" --query ETag --output text) + parts="$parts${parts:+,}{\"PartNumber\":$n,\"ETag\":$etag}" + done + aws s3api complete-multipart-upload --bucket cli --key obj/manual.bin --upload-id "$id" \ + --multipart-upload "{\"Parts\":[$parts]}" --output json >"$T/complete.json" + grep -q -- '-3\\"' "$T/complete.json" || fail "complete-multipart-upload: $(cat "$T/complete.json")" + aws s3 cp --no-progress s3://cli/obj/manual.bin "$T/manual.back" + same "$T/mp.bin" "$T/manual.back" || fail "the manual multipart object differs" + + # Presigned URLs put a credential in a URL: refused. + url=$(aws s3 presign s3://cli/obj/1K.bin) + code=$(curl -sS --cacert /e2e/ca.pem -o "$T/presigned" -w '%{http_code}' "$url") + [ "$code" = 400 ] && grep -q AuthorizationQueryParametersError "$T/presigned" || + fail "a presigned URL got $code: $(cat "$T/presigned")" + + # Kept for the host side: vk registry get, and the gc keeping it. + aws s3 mb s3://keep + aws s3 cp --no-progress "$T/1K.bin" s3://keep/from-cli.bin + sha256sum <"$T/1K.bin" | cut -d' ' -f1 >/e2e/keep.sha256 + + aws s3 rb --force s3://cli + if aws s3api head-bucket --bucket cli 2>/dev/null; then fail "s3 rb --force left the bucket"; fi + echo "awscli: cp up and down to 100 MiB, CRC32/SHA-1/SHA-256 checksums, paged listing, sync, ranges, copy, batch delete, multipart, presign refused, rb" +} + +# ---------------------------------------------------------------------------- rclone +run_rclone() { + export RCLONE_CONFIG_VKR_TYPE=s3 RCLONE_CONFIG_VKR_PROVIDER=Other + export RCLONE_CONFIG_VKR_ENDPOINT=$S3_ENDPOINT RCLONE_CONFIG_VKR_FORCE_PATH_STYLE=true + export RCLONE_CONFIG_VKR_ACCESS_KEY_ID=$WKEY RCLONE_CONFIG_VKR_SECRET_ACCESS_KEY=$WKEY + export RCLONE_CONFIG_VKR_REGION=us-east-1 + r() { rclone --ca-cert /e2e/ca.pem --log-file "$LOGS/rclone.log" -v "$@"; } + mkdir -p "$T/src/d" + for f in one two d/three; do head -c 70000 /dev/urandom >"$T/src/$f"; done + # A multipart upload of rclone's own (5 MiB parts, its MD5 kept in metadata). + head -c 12M /dev/urandom >"$T/src/d/big" + touch -d '2024-01-02 03:04:05' "$T/src/one" + r mkdir vkr:rcl + r copy --s3-upload-cutoff 5M --s3-chunk-size 5M "$T/src" vkr:rcl/data + r ls vkr:rcl/data | tee "$T/ls" + [ "$(wc -l <"$T/ls")" = 4 ] || fail "rclone ls does not list 4 files" + r check "$T/src" vkr:rcl/data || fail "rclone check" + r lsl vkr:rcl/data/one | grep -q '2024-01-02 03:04:05' || fail "rclone lost the modification time" + head -c 500 /dev/urandom >"$T/src/four" + : >"$LOGS/rclone.log" + r sync "$T/src" vkr:rcl/data + [ "$(grep -c ': Copied' "$LOGS/rclone.log")" = 1 ] || fail "the first sync did not copy one file" + : >"$LOGS/rclone.log" + r sync "$T/src" vkr:rcl/data + if grep -q ': Copied\|: Updated' "$LOGS/rclone.log"; then fail "the second sync copied again"; fi + r delete vkr:rcl/data + [ -z "$(r ls vkr:rcl)" ] || fail "rclone delete left files" + # Lists the bucket's multipart uploads to abort the stale ones. + r cleanup vkr:rcl + r rmdirs vkr:rcl + echo "rclone: copy (multipart too), ls, check, sync twice, delete, cleanup" +} + +case ${1:-} in + trust) trust ;; + sccache) run_sccache ;; + awscli) run_awscli ;; + rclone) run_rclone ;; + *) echo "usage: $0 " >&2; exit 2 ;; +esac diff --git a/vk-driver/src/registry.rs b/vk-driver/src/registry.rs index f3999db4..bc22884b 100644 --- a/vk-driver/src/registry.rs +++ b/vk-driver/src/registry.rs @@ -5660,6 +5660,8 @@ pub(crate) mod tests { auth: vk_registry::Authenticator::Shared(auth), tls: None, webdav: true, + s3: true, + s3_host: None, tokens: Default::default(), handover: Default::default(), })) @@ -5707,6 +5709,8 @@ pub(crate) mod tests { auth: vk_registry::Authenticator::Shared(vk_registry::auth::Auth::None), tls: None, webdav: true, + s3: true, + s3_host: None, tokens: Default::default(), handover: Default::default(), })); @@ -5742,6 +5746,8 @@ pub(crate) mod tests { auth: vk_registry::Authenticator::Shared(vk_registry::auth::Auth::None), tls: None, webdav: true, + s3: true, + s3_host: None, tokens: Default::default(), handover: Default::default(), })); @@ -5855,6 +5861,8 @@ pub(crate) mod tests { auth: vk_registry::Authenticator::Shared(vk_registry::auth::Auth::None), tls: None, webdav: true, + s3: true, + s3_host: None, tokens: Default::default(), handover: Default::default(), })); @@ -6104,6 +6112,8 @@ pub(crate) mod tests { auth: vk_registry::Authenticator::Shared(vk_registry::auth::Auth::None), tls: None, webdav: true, + s3: true, + s3_host: None, tokens: Default::default(), handover: Default::default(), })); @@ -6162,6 +6172,8 @@ pub(crate) mod tests { auth: vk_registry::Authenticator::Shared(vk_registry::auth::Auth::None), tls: None, webdav: true, + s3: true, + s3_host: None, tokens: Default::default(), handover: Default::default(), })); @@ -6203,6 +6215,8 @@ pub(crate) mod tests { auth: vk_registry::Authenticator::Shared(vk_registry::auth::Auth::None), tls: None, webdav: true, + s3: true, + s3_host: None, tokens: Default::default(), handover: Default::default(), })); @@ -6239,6 +6253,8 @@ pub(crate) mod tests { auth: vk_registry::Authenticator::Shared(vk_registry::auth::Auth::None), tls: None, webdav: true, + s3: true, + s3_host: None, tokens: Default::default(), handover: Default::default(), })); diff --git a/vk-driver/src/regproxy.rs b/vk-driver/src/regproxy.rs index edeebce9..575c8818 100644 --- a/vk-driver/src/regproxy.rs +++ b/vk-driver/src/regproxy.rs @@ -25,9 +25,9 @@ //! What the proxy lends decides what it forwards ([`Scope`]). The node's own credential is //! not scoped, so it reaches the OCI API only (`/v2/`). A job's registry token, which the hub //! scoped to the job's namespaces (`[registry] guest_credential = "job"`), also reaches the -//! lock API (`/lock/`) and plain files (`/dav/files/`): the registry enforces the token's -//! grants on each. A job meant to have a token and given none gets a proxy that forwards -//! nothing. +//! lock API (`/lock/`), plain files (`/dav/files/`) and their S3 view (`/s3`): the registry +//! enforces the token's grants on each. A job meant to have a token and given none gets a proxy +//! that forwards nothing. //! //! A job's token may come with areas ([`Area`]): `job/` in an OCI or plain-file path, or //! before the `:` of a lock name, stands for the area's prefix, so a guest names a fixed path @@ -36,6 +36,13 @@ //! (an upload's `Location`, a lock's holders), which the proxy forwards too. A `PROPFIND`'s //! hrefs are the exception, mapped back to the alias, since a WebDAV client resolves a listing //! against the root it asked for. +//! +//! Under `/s3/`, `job//` names a key below the area's prefix, and `job/` alone +//! names the area as a bucket of its own ([`map_s3`]): the prefix's first component is the real +//! bucket, the rest a key prefix the proxy adds to a listing's query and takes off the names in +//! the documents it answers ([`S3Names`]). The guest's own `Authorization`, SigV4 with whatever +//! keys or none, is replaced by the token, so the registry decodes an `aws-chunked` body without +//! checking chunk signatures made with keys it does not know. use std::convert::Infallible; use std::net::SocketAddr; @@ -71,12 +78,17 @@ const DAV_FILES_METHODS: &[&str] = &[ "GET", "HEAD", "PROPFIND", "PUT", "MKCOL", "DELETE", "OPTIONS", ]; +/// The S3 view's path, and the methods forwarded under it: vk-registry's for `/s3`. +const S3: &str = "/s3"; +const S3_METHODS: &[&str] = &["GET", "HEAD", "PUT", "POST", "DELETE", "OPTIONS"]; + /// Which of the registry's paths the proxy forwards. #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub enum Scope { /// The OCI API only, `/v2/`: what an unscoped credential may be lent for. Oci, - /// `/v2/`, `/lock/` and `/dav/files/`: a job token's, whose grants the registry enforces. + /// `/v2/`, `/lock/`, `/dav/files/` and `/s3`: a job token's, whose grants the registry + /// enforces. Job, /// Nothing: a job that should have had a token and has none. Deny, @@ -87,7 +99,7 @@ impl Scope { fn paths(self) -> &'static [&'static str] { match self { Scope::Oci => &["/v2/"], - Scope::Job => &["/v2/", "/lock/", DAV_FILES], + Scope::Job => &["/v2/", "/lock/", DAV_FILES, S3], Scope::Deny => &[], } } @@ -304,19 +316,23 @@ async fn handle( })) } +/// Whether `path` lies under the forwarded path `scoped`: below it when it ends in `/`, else +/// also the path itself, as [`S3`] alone names the service, with or without a query. +fn within(path: &str, scoped: &str) -> bool { + match path.strip_prefix(scoped) { + Some(_) if scoped.ends_with('/') => true, + Some(rest) => rest.is_empty() || rest.starts_with(['/', '?']), + None => false, + } +} + /// The upstream URL a guest's request target names, or `None` when, once resolved, it is not /// under one of the scope's paths below the upstream's own. Checked after resolving, not on /// the text: the URL parser resolves `..` and its percent-encoded spellings (`%2e%2e`), so /// `/v2/../lock/acquire` starts with `/v2/` and still reaches `/lock/acquire`, carrying the /// credential. fn upstream_url(cfg: &ProxyCfg, path_and_query: &str) -> Option { - let scoped = |p: &str| { - cfg.scope - .paths() - .iter() - .find(|s| p.starts_with(*s)) - .copied() - }; + let scoped = |p: &str| cfg.scope.paths().iter().find(|s| within(p, s)).copied(); let prefix = scoped(path_and_query)?; // An encoded separator stays encoded through the parser, so `/v2/..%2Flock` passes the // check below — and reaches `/lock` through a front proxy that decodes `%2F`/`%5C` once @@ -332,7 +348,7 @@ fn upstream_url(cfg: &ProxyCfg, path_and_query: &str) -> Option { } let url = reqwest::Url::parse(&format!("{}{path_and_query}", cfg.upstream)).ok()?; let resolved = url.path().strip_prefix(cfg.root.as_str())?; - (url.origin() == cfg.base.origin() && resolved.starts_with(prefix)).then_some(url) + (url.origin() == cfg.base.origin() && within(resolved, prefix)).then_some(url) } /// What a request becomes under the job's areas. @@ -344,11 +360,13 @@ enum Mapped<'a> { To(String, &'a Area), /// `job/…` naming no area of this job. Unknown, + /// An area named in a way the request cannot take, refused with this message. + Refused(&'static [u8]), } /// `url` (an accepted upstream URL) under `cfg`'s areas: `job/` at the start of an -/// OCI or plain-file path, or of a lock name's path part, becomes the area's prefix. -fn map_aliases<'a>(cfg: &'a ProxyCfg, url: &reqwest::Url) -> Mapped<'a> { +/// OCI, plain-file or S3 path, or of a lock name's path part, becomes the area's prefix. +fn map_aliases<'a>(cfg: &'a ProxyCfg, method: &Method, url: &reqwest::Url) -> Mapped<'a> { if cfg.scope != Scope::Job { return Mapped::Unchanged; } @@ -363,6 +381,9 @@ fn map_aliases<'a>(cfg: &'a ProxyCfg, url: &reqwest::Url) -> Mapped<'a> { Err(()) => Mapped::Unknown, }; } + if let Some(rest) = path.strip_prefix(S3).and_then(|p| p.strip_prefix('/')) { + return map_s3(&cfg.areas, method, rest, url.query()); + } for base in ["/v2/", DAV_FILES] { let Some(rest) = path.strip_prefix(base) else { continue; @@ -398,6 +419,202 @@ fn map_name<'a>( Ok(Some((format!("{}{tail}", area.prefix), area))) } +/// `/s3/` under the job's areas. `job//` is the key below the area's +/// prefix, as on `/dav/files/`. `job/` alone, with or without a `/`, is the area as a +/// bucket: the real bucket, with a listing's query given the key prefix ([`listing_query`]). +/// When the area lies below its bucket, only reads pass there: creating or deleting the bucket +/// would act on the whole real bucket, and a batch delete's keys sit in a body the proxy does +/// not rewrite, under checksums it could not keep. +fn map_s3<'a>(areas: &'a [Area], method: &Method, rest: &str, query: Option<&str>) -> Mapped<'a> { + let (name, area) = match map_name(areas, rest, '/') { + Ok(None) => return Mapped::Unchanged, + Ok(Some(mapped)) => mapped, + Err(()) => return Mapped::Unknown, + }; + let tail = name.strip_prefix(area.prefix.as_str()).unwrap_or_default(); + if !matches!(tail, "" | "/") { + let query = query.map(|q| format!("?{q}")).unwrap_or_default(); + return Mapped::To(format!("{S3}/{name}{query}"), area); + } + let names = S3Names::new(area); + let query = match query { + _ if names.keys.is_empty() => query.map(str::to_string), + _ if !matches!(method.as_str(), "GET" | "HEAD" | "OPTIONS") => { + return Mapped::Refused( + b"registry proxy: an area below its bucket takes reads only, not bucket writes", + ); + } + _ if *method == Method::GET => match listing_query(&names, query.unwrap_or_default()) { + Some(query) => Some(query), + None => { + return Mapped::Refused( + b"registry proxy: a listing through an area takes only a listing's parameters", + ); + } + }, + _ => query.map(str::to_string), + }; + let query = query + .filter(|q| !q.is_empty()) + .map(|q| format!("?{q}")) + .unwrap_or_default(); + Mapped::To(format!("{S3}/{}{query}", names.bucket), area) +} + +/// The parameters a bucket listing takes, of either version, and a listing of its multipart +/// uploads, without a `delimiter`, which the registry does not serve there; an SDK's `x-id` +/// hint too. +const LISTING_PARAMS: &[&str] = &[ + "list-type", + "prefix", + "delimiter", + "max-keys", + "continuation-token", + "start-after", + "marker", + "encoding-type", + "fetch-owner", + "x-id", +]; +const UPLOADS_PARAMS: &[&str] = &[ + "uploads", + "prefix", + "key-marker", + "upload-id-marker", + "max-uploads", + "encoding-type", + "x-id", +]; + +/// A bucket `GET`'s query through an area below its bucket: a listing's keys are the area's, +/// so the key prefix goes before each `prefix`, `start-after`, `marker` and `key-marker`, and +/// into the key a `continuation-token` names; a `prefix` is added when none is given, so the +/// listing stays in the area. A query naming no listing parameter, a configuration's, is +/// kept as sent. `None` for one naming a listing's parameters with others, which the prefix +/// could not be trusted to confine. Names are compared decoded, as the registry takes them. +fn listing_query(names: &S3Names, query: &str) -> Option { + let name = |p: &str| vk_registry::percent_decode(p.split_once('=').map_or(p, |(n, _)| n)); + let pairs: Vec<&str> = query.split('&').filter(|p| !p.is_empty()).collect(); + let known = match pairs.iter().any(|p| name(p) == "uploads") { + true => UPLOADS_PARAMS, + false => LISTING_PARAMS, + }; + let listing = |n: &str| n != "x-id" && (LISTING_PARAMS.contains(&n) || n == "uploads"); + if !pairs.iter().all(|p| known.contains(&name(p).as_str())) { + return match pairs.iter().any(|p| listing(&name(p))) { + true => None, + false => Some(query.to_string()), + }; + } + // A prefix is lowercase alphanumerics and `._-/`, so encoding escapes the `/` alone. + let keys = vk_registry::percent_encode(&names.keys); + let value = |p: &str| p.split_once('=').map_or("", |(_, v)| v).to_string(); + let mut out: Vec = pairs + .iter() + .map(|p| match name(p).as_str() { + n @ ("prefix" | "start-after" | "marker" | "key-marker") => { + format!("{n}={keys}{}", value(p)) + } + "continuation-token" => format!("continuation-token={}", names.token_in(&value(p))), + _ => p.to_string(), + }) + .collect(); + if !pairs.iter().any(|p| name(p) == "prefix") { + out.push(format!("prefix={keys}")); + } + Some(out.join("&")) +} + +/// An `x-amz-copy-source` under the job's areas: `[/]job//`, escaped as a whole +/// or not, becomes `//`, unescaped as the registry also takes it. `Ok(None)` for +/// a source not under [`ALIAS_ROOT`], forwarded as sent; `Err` for no such alias. +fn map_copy_source(areas: &[Area], source: &str) -> Result, ()> { + let decoded = vk_registry::percent_decode(source); + let name = decoded.strip_prefix('/').unwrap_or(&decoded); + Ok(map_name(areas, name, '/')?.map(|(name, _)| format!("/{name}"))) +} + +/// An area as the S3 view sees it, and its names mapped back from the registry's. +#[derive(Clone)] +struct S3Names { + /// The prefix's first component, the real bucket + bucket: String, + /// The rest of the prefix with a `/`, or nothing when the prefix is the bucket alone + keys: String, + /// `/s3/` and `/s3/`, as the registry names a request's path + real: String, + real_bucket: String, + alias: String, + /// `/s3/job/`, which a guest names the area by + aliased: String, +} + +impl S3Names { + fn new(area: &Area) -> Self { + let (bucket, keys) = match area.prefix.split_once('/') { + Some((bucket, rest)) => (bucket, format!("{rest}/")), + None => (area.prefix.as_str(), String::new()), + }; + S3Names { + bucket: bucket.to_string(), + keys, + real: format!("{S3}/{}", area.prefix), + real_bucket: format!("{S3}/{bucket}"), + alias: area.alias.clone(), + aliased: format!("{S3}/{ALIAS_ROOT}/{}", area.alias), + } + } + + /// A guest's continuation token, the key it names put below the key prefix; one that does + /// not decode is kept for the registry to refuse. + fn token_in(&self, token: &str) -> String { + use vk_hub_proto::token::{from_base64url, to_base64url}; + match from_base64url(&vk_registry::percent_decode(token)) { + Some(key) if !token.is_empty() => to_base64url(&[self.keys.as_bytes(), &key].concat()), + _ => token.to_string(), + } + } + + /// The registry's continuation token, the key it names taken from below the key prefix. + fn token_out(&self, token: &[u8]) -> Option> { + use vk_hub_proto::token::{from_base64url, to_base64url}; + let key = from_base64url(std::str::from_utf8(token).ok()?)?; + let key = key.strip_prefix(self.keys.as_bytes())?; + Some(to_base64url(key).into_bytes()) + } + + /// A path the registry names, `/s3/[/…]` or the bucket `/s3/[/]`, as the + /// guest names it, `/s3/job/[/…]`. + fn path_out(&self, path: &[u8]) -> Option> { + let tail = path + .strip_prefix(self.real.as_bytes()) + .filter(|r| r.is_empty() || r.starts_with(b"/")) + .or_else(|| { + path.strip_prefix(self.real_bucket.as_bytes()) + .filter(|r| matches!(r, [] | [b'/'])) + })?; + Some([self.aliased.as_bytes(), tail].concat()) + } + + /// The text of the element `name` as the guest names it, or `None` to keep it: a bucket + /// becomes the alias, a key or key prefix loses the area's key prefix, a continuation + /// token the same in the key it names, and a path the alias form. A name outside the area + /// is kept. Every one of these is the registry's escaped XML text, which the prefix's + /// characters pass through unchanged, under `encoding-type=url` too. + fn text(&self, name: &[u8], text: &[u8]) -> Option> { + match name { + b"Name" | b"Bucket" => { + (text == self.bucket.as_bytes()).then(|| self.alias.clone().into_bytes()) + } + b"Prefix" | b"StartAfter" | b"Marker" | b"NextMarker" | b"Key" | b"KeyMarker" + | b"NextKeyMarker" => text.strip_prefix(self.keys.as_bytes()).map(<[u8]>::to_vec), + b"ContinuationToken" | b"NextContinuationToken" => self.token_out(text), + b"Location" | b"Resource" => self.path_out(text), + _ => None, + } + } +} + /// A lock request's query with each `name=` under [`ALIAS_ROOT`] mapped, decoded as the /// registry decodes it and encoded again; every other pair is kept as it came, in order. /// `Ok(None)` when no name was mapped. The area returned is the first mapped name's. One @@ -421,10 +638,18 @@ fn map_lock_names<'a>(areas: &'a [Area], query: &str) -> Result Response { + refusal(StatusCode::FORBIDDEN, msg) +} + +/// A `status` answer with `msg`, never forwarded. +fn refusal(status: StatusCode, msg: &'static [u8]) -> Response { Response::builder() - .status(StatusCode::FORBIDDEN) + .status(status) .body( Full::new(Bytes::from_static(msg)) .map_err(|never| match never {}) @@ -445,6 +670,14 @@ struct HrefMap { carry: Vec, } +/// A rewrite of a response body as it streams, holding back at most a bounded tail. +trait BodyMap { + /// The bytes that can be sent now, mapped, once `chunk` has arrived. + fn push(&mut self, chunk: &[u8]) -> Bytes; + /// The rest, mapped, once the body has ended. + fn finish(&mut self) -> Bytes; +} + impl HrefMap { fn new(area: &Area) -> Self { let href = |p: &str| format!("{DAV_FILES}{p}").into_bytes(); @@ -455,18 +688,6 @@ impl HrefMap { } } - /// The bytes of `chunk` that can be sent now, mapped; a tail as long as the pattern waits - /// for the next chunk, which holds the byte after a match. - fn push(&mut self, chunk: &[u8]) -> Bytes { - self.carry.extend_from_slice(chunk); - self.emit(self.carry.len().saturating_sub(self.from.len())) - } - - /// The rest, mapped, once the body has ended. - fn finish(&mut self) -> Bytes { - self.emit(self.carry.len()) - } - /// The carry up to `upto`, mapped; a match starting before `upto` is taken whole. fn emit(&mut self, upto: usize) -> Bytes { let mut out = Vec::with_capacity(upto); @@ -486,36 +707,171 @@ impl HrefMap { self.carry.drain(..i.min(self.carry.len())); Bytes::from(out) } +} - /// `body` with its hrefs mapped; a chunk that leaves nothing to send yet yields no frame. - fn over(self, body: S) -> impl futures::Stream> - where - S: futures::Stream> + Send + 'static, - { - use futures::StreamExt; - futures::stream::unfold((Box::pin(body), Some(self)), |(mut body, map)| async move { - let mut map = map?; - loop { - match body.next().await { - Some(Ok(chunk)) => { - let out = map.push(&chunk); - if !out.is_empty() { - return Some((Ok(out), (body, Some(map)))); - } - } - Some(Err(e)) => return Some((Err(e), (body, None))), - None => { - let out = map.finish(); - return (!out.is_empty()).then_some((Ok(out), (body, None))); - } +impl BodyMap for HrefMap { + /// A tail as long as the pattern waits for the next chunk, which holds the byte after a + /// match. + fn push(&mut self, chunk: &[u8]) -> Bytes { + self.carry.extend_from_slice(chunk); + self.emit(self.carry.len().saturating_sub(self.from.len())) + } + + fn finish(&mut self) -> Bytes { + self.emit(self.carry.len()) + } +} + +/// The longest element [`S3Map`] waits for whole: past it, the element streams on unmapped. +/// A key is at most 1024 bytes, 3 KiB once URL-encoded, and a token or path not much more. +const MAX_ELEMENT: usize = 8 << 10; + +/// The elements of an S3 document [`S3Names::text`] maps. +const S3_ELEMENTS: &[&[u8]] = &[ + b"Name", + b"Bucket", + b"Prefix", + b"StartAfter", + b"Marker", + b"NextMarker", + b"Key", + b"KeyMarker", + b"NextKeyMarker", + b"ContinuationToken", + b"NextContinuationToken", + b"Location", + b"Resource", +]; + +/// Maps an S3 document answering a request through an area back to the area's names, as the +/// body streams: each `text` of [`S3_ELEMENTS`] has its text mapped by [`S3Names::text`], +/// whatever element holds it (a common prefix's `Prefix` as the listing's). vk-registry writes +/// these elements with no attribute and their text escaped, so no `<` is in it. At most +/// [`MAX_ELEMENT`] bytes wait for the next chunk. +struct S3Map { + names: S3Names, + /// Bytes not sent yet: an element, or the start of a tag, still arriving. + carry: Vec, +} + +/// What the carry holds from a `<` on. +enum Scan { + /// One of [`S3_ELEMENTS`], whole: its bytes mapped, and how many it took. + Mapped(Vec, usize), + /// Something else, sent as it is. + Other, + /// Too little to tell yet. + Short, +} + +impl S3Map { + fn new(names: S3Names) -> Self { + S3Map { + names, + carry: Vec::new(), + } + } + + /// `at`, which starts with `<`, read as one of [`S3_ELEMENTS`]. + fn scan(&self, at: &[u8]) -> Scan { + let Some(gt) = at.iter().position(|&b| b == b'>') else { + return Scan::Short; + }; + let name = at.get(1..gt).unwrap_or_default(); + if !S3_ELEMENTS.contains(&name) { + return Scan::Other; + } + let rest = at.get(gt + 1..).unwrap_or_default(); + let Some(lt) = rest.iter().position(|&b| b == b'<') else { + return Scan::Short; + }; + let (text, after) = rest.split_at(lt); + let close = [b""].concat(); + if !after.starts_with(&close) { + return match close.starts_with(after) { + true => Scan::Short, + false => Scan::Other, + }; + } + let text = self.names.text(name, text).unwrap_or_else(|| text.to_vec()); + let open = at.get(..=gt).unwrap_or_default(); + let used = gt + 1 + lt + close.len(); + Scan::Mapped([open, &text, &close].concat(), used) + } + + /// The carry, mapped, up to an element still arriving unless the body has `ended`. + fn emit(&mut self, ended: bool) -> Bytes { + let mut out = Vec::with_capacity(self.carry.len()); + let mut i = 0; + while let Some(rest) = self.carry.get(i..).filter(|r| !r.is_empty()) { + let Some(lt) = rest.iter().position(|&b| b == b'<') else { + out.extend_from_slice(rest); + i = self.carry.len(); + break; + }; + out.extend_from_slice(rest.get(..lt).unwrap_or_default()); + i += lt; + let at = self.carry.get(i..).unwrap_or_default(); + match self.scan(at) { + Scan::Mapped(bytes, used) => { + out.extend_from_slice(&bytes); + i += used; + } + Scan::Short if !ended && at.len() <= MAX_ELEMENT => break, + Scan::Short | Scan::Other => { + out.push(b'<'); + i += 1; } } - }) + } + self.carry.drain(..i.min(self.carry.len())); + Bytes::from(out) } } +impl BodyMap for S3Map { + fn push(&mut self, chunk: &[u8]) -> Bytes { + self.carry.extend_from_slice(chunk); + self.emit(false) + } + + fn finish(&mut self) -> Bytes { + self.emit(true) + } +} + +/// `body` with `map` applied; a chunk that leaves nothing to send yet yields no frame. +fn mapped( + map: Box, + body: S, +) -> impl futures::Stream> +where + S: futures::Stream> + Send + 'static, +{ + use futures::StreamExt; + futures::stream::unfold((Box::pin(body), Some(map)), |(mut body, map)| async move { + let mut map = map?; + loop { + match body.next().await { + Some(Ok(chunk)) => { + let out = map.push(&chunk); + if !out.is_empty() { + return Some((Ok(out), (body, Some(map)))); + } + } + Some(Err(e)) => return Some((Err(e), (body, None))), + None => { + let out = map.finish(); + return (!out.is_empty()).then_some((Ok(out), (body, None))); + } + } + } + }) +} + /// Headers that must not be copied across the proxy (connection-scoped, credential-bearing, -/// or ones we set ourselves). +/// or ones we set ourselves). An answer's `content-length` is kept when its body is not +/// rewritten ([`forward`]). fn is_skipped(name: &str) -> bool { matches!( name, @@ -526,6 +882,8 @@ fn is_skipped(name: &str) -> bool { | "content-length" | "connection" | "transfer-encoding" + // a SigV4 session token, meaningless once the guest's signature is replaced + | "x-amz-security-token" ) } @@ -538,13 +896,13 @@ async fn forward(req: Request, cfg: &ProxyCfg) -> Result b"registry proxy: only /v2/ registry paths are proxied", - Scope::Job => b"registry proxy: only /v2/, /lock/ and /dav/files/ are proxied", + Scope::Job => b"registry proxy: only /v2/, /lock/, /dav/files/ and /s3/ are proxied", Scope::Deny => b"registry proxy: this job has no registry credential", })); }; @@ -552,33 +910,35 @@ async fn forward(req: Request, cfg: &ProxyCfg) -> Result (url, None), Mapped::To(target, area) => match upstream_url(cfg, &target) { Some(url) => (url, Some(area)), None => return Ok(forbidden(b"registry proxy: an area's path is refused")), }, - Mapped::Unknown => { - return Ok(forbidden( - b"registry proxy: this job has no area of that alias", - )); - } + Mapped::Unknown => return Ok(forbidden(NO_SUCH_AREA)), + Mapped::Refused(msg) => return Ok(forbidden(msg)), }; let path = url.path().strip_prefix(cfg.root.as_str()).unwrap_or(""); // A listing's hrefs go back to the alias the guest asked for (see the module's doc). let listing = method.as_str() == "PROPFIND" && path.starts_with(DAV_FILES); let hrefs = area.filter(|_| listing).map(HrefMap::new); + let s3 = cfg.scope == Scope::Job && within(path, S3); + // An S3 answer through an area names the area back (see [`S3Map`]). + let s3_names = area.filter(|_| s3).map(S3Names::new); // OCI writes and WebDAV PROPFIND always stream a body upstream, even when empty. // Forward other bodies for the registry to validate: dropping a MKCOL body // would create the collection instead of rejecting the request (RFC 4918 §9.3). @@ -586,13 +946,29 @@ async fn forward(req: Request, cfg: &ProxyCfg) -> Result = parts + let mut headers: Vec<_> = parts .headers .iter() .filter(|(k, _)| !is_skipped(&k.as_str().to_ascii_lowercase())) .map(|(k, v)| (k.clone(), v.clone())) .collect(); - let sent = match (repeatable(&method, path), bodyful) { + // A copy's source names its area as a path does. One that is not text is forwarded for + // the registry to refuse. + for (_, v) in headers + .iter_mut() + .filter(|(k, _)| s3 && k.as_str() == "x-amz-copy-source") + { + let Ok(source) = v.to_str() else { continue }; + match map_copy_source(&cfg.areas, source) { + Ok(None) => {} + Ok(Some(mapped)) => match hyper::header::HeaderValue::from_str(&mapped) { + Ok(mapped) => *v = mapped, + Err(_) => return Ok(forbidden(b"registry proxy: an area's path is refused")), + }, + Err(()) => return Ok(forbidden(NO_SUCH_AREA)), + } + } + let sent = match (repeatable(&method, path, &parts.headers), bodyful) { (true, false) => Outgoing::Whole(Bytes::new()), (true, true) => held(incoming, MAX_HELD).await?, (false, false) => Outgoing::Once(None), @@ -632,11 +1008,24 @@ async fn forward(req: Request, cfg: &ProxyCfg) -> Result> = match (hrefs, &s3_names) { + _ if method == Method::HEAD => None, + (Some(map), _) => Some(Box::new(map)), + (None, Some(names)) if xml => Some(Box::new(S3Map::new(names.clone()))), + _ => None, + }; + // A rewritten body's length is not the upstream's; any other is the upstream's bytes, + // whole or resumed, so its length holds. + let rewritten = body_map.is_some(); // stream the upstream response back without buffering (blob pulls). let stream = resp.bytes_stream(); - let body = match (hrefs, resume) { + let body = match (body_map, resume) { (Some(map), _) => StreamBody::new( - map.over(stream) + mapped(map, stream) .map_ok(Frame::data) .map_err(|e| std::io::Error::other(e.to_string())), ) @@ -683,9 +1072,17 @@ async fn forward(req: Request, cfg: &ProxyCfg) -> Result bool { +/// reads, manifest PUTs by tag or digest, whole-file PUTs, S3 PUTs of an object, a part or a +/// copy, and upload starts (which leave the first upload unused). A retried PUT lands last, +/// like a later PUT from another writer. Excludes upload chunks (`PATCH`), upload completion +/// (`PUT` under `blobs/uploads/`, whose upload is then gone), deletions, collection +/// creation, a PUT under a precondition (`If-Match`, `If-None-Match`, `If-Unmodified-Since` +/// or a copy source's `x-amz-copy-source-if-*`), which its own first write would fail +/// again, and all `/lock/` requests regardless of method. +fn repeatable(method: &Method, path: &str, headers: &hyper::HeaderMap) -> bool { if path.starts_with("/lock/") { return false; } + let conditional = headers.keys().any(|k| { + matches!( + k.as_str(), + "if-match" | "if-none-match" | "if-unmodified-since" + ) || k.as_str().starts_with("x-amz-copy-source-if-") + }); + if *method == Method::PUT && conditional { + return false; + } let oci = path.strip_prefix("/v2/"); match method.as_str() { "GET" | "HEAD" | "OPTIONS" | "PROPFIND" => true, "PUT" => { oci.is_some_and(|p| p.rsplit('/').nth(1) == Some("manifests")) || path.starts_with(DAV_FILES) + || path + .strip_prefix("/s3/") + .is_some_and(|p| p.trim_end_matches('/').contains('/')) } "POST" => oci.is_some_and(|p| p.ends_with("/blobs/uploads/")), _ => false, @@ -934,20 +1345,30 @@ fn resumed_at( mod tests { use super::*; - // A fake upstream that records the Authorization header + path and query of the last - // request and replies 200 with a marker body. It runs on its own dedicated thread + runtime + /// What [`fake_upstream`] saw of the last request: its `Authorization`, its path and + /// query, and every header forwarded. + type Seen = (String, String, hyper::HeaderMap); + + // A fake upstream that records the Authorization header, path and query, and headers of + // the last request and replies 200 with a marker body. It runs on its own dedicated thread + runtime // (like the proxy's `spawn_blocking` and the vk-registry e2e servers) so its accept // loop never competes with the test's runtime threads — which caused acceptance stalls // and flakes when many of these servers ran concurrently under `cargo test`. - fn fake_upstream() -> (SocketAddr, Arc>) { + fn fake_upstream() -> (SocketAddr, Arc>) { fake_upstream_replying(b"ok") } /// [`fake_upstream`], replying `body` to every request. - fn fake_upstream_replying( + fn fake_upstream_replying(body: &'static [u8]) -> (SocketAddr, Arc>) { + fake_upstream_answering(body, &[]) + } + + /// [`fake_upstream_replying`], with these headers. + fn fake_upstream_answering( body: &'static [u8], - ) -> (SocketAddr, Arc>) { - let seen = Arc::new(std::sync::Mutex::new((String::new(), String::new()))); + headers: &'static [(&'static str, &'static str)], + ) -> (SocketAddr, Arc>) { + let seen = Arc::new(std::sync::Mutex::new(Seen::default())); let listener = std::net::TcpListener::bind(("127.0.0.1", 0)).unwrap(); listener.set_nonblocking(true).unwrap(); let addr = listener.local_addr().unwrap(); @@ -976,10 +1397,12 @@ mod tests { .unwrap_or("") .to_string(); let path = req.uri().path_and_query().unwrap().to_string(); - *s.lock().unwrap() = (auth, path); - Ok::<_, Infallible>(Response::new(Full::new(Bytes::from_static( - body, - )))) + *s.lock().unwrap() = (auth, path, req.headers().clone()); + let mut resp = Response::new(Full::new(Bytes::from_static(body))); + for (k, v) in headers { + resp.headers_mut().insert(*k, v.parse().unwrap()); + } + Ok::<_, Infallible>(resp) } }); let _ = http1::Builder::new().serve_connection(io, svc).await; @@ -1019,7 +1442,7 @@ mod tests { assert!(r.status().is_success()); assert_eq!(r.text().await.unwrap(), "ok"); - let (auth, path) = seen.lock().unwrap().clone(); + let (auth, path, _) = seen.lock().unwrap().clone(); assert_eq!( path, "/v2/app/manifests/latest", "path must be forwarded as resolved" @@ -1124,7 +1547,7 @@ mod tests { .unwrap(); assert!(r.status().is_success()); - let (auth, _) = seen.lock().unwrap().clone(); + let (auth, ..) = seen.lock().unwrap().clone(); assert_eq!( auth, "Bearer vkr_x", "the proxy must inject the bearer token" @@ -1174,9 +1597,9 @@ mod tests { proxy: SocketAddr, method: &str, target: &str, - seen: &std::sync::Mutex<(String, String)>, + seen: &std::sync::Mutex, ) -> (u16, String, String) { - *seen.lock().unwrap() = (String::new(), String::new()); + *seen.lock().unwrap() = Seen::default(); let r = reqwest::Client::new() .request( reqwest::Method::from_bytes(method.as_bytes()).unwrap(), @@ -1186,7 +1609,7 @@ mod tests { .send() .await .unwrap(); - let (auth, path) = seen.lock().unwrap().clone(); + let (auth, path, _) = seen.lock().unwrap().clone(); (r.status().as_u16(), auth, path) } @@ -1372,7 +1795,7 @@ mod tests { let status = raw_status(proxy, "GET", "/dav/files/job/task-cache/../../other").await; assert_eq!(status, 200); assert_eq!(seen.lock().unwrap().1, "/dav/files/other"); - *seen.lock().unwrap() = (String::new(), String::new()); + *seen.lock().unwrap() = Seen::default(); let status = raw_status(proxy, "GET", "/v2/job/task-cache/..%2F..%2Flock/x").await; assert_eq!(status, 403); assert_eq!(seen.lock().unwrap().1, "", "nothing forwarded"); @@ -1436,6 +1859,422 @@ mod tests { } } + /// The areas the S3 tests map: a prefix of one, two and three components. + fn s3_areas() -> [Area; 3] { + [ + area("one", "solo"), + area("two", "duo/b"), + area("three", "tri/b/c"), + ] + } + + /// A job's token reaches the S3 view, every method it serves, injected in place of the + /// guest's own `Authorization`, SigV4 included; the node's credential does not, nor a job + /// without a token. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn a_job_token_reaches_s3_too() { + let _ = rustls::crypto::ring::default_provider().install_default(); + let (up_addr, seen) = fake_upstream(); + let proxy = spawn_blocking(ProxyCfg::for_job(®istry(up_addr), Some("vkt_job")).unwrap()) + .unwrap(); + for (method, target) in [ + ("GET", "/s3"), + ("GET", "/s3/"), + ("HEAD", "/s3/b"), + ("GET", "/s3/b?list-type=2&prefix=x/"), + ("PUT", "/s3/b/k"), + ("POST", "/s3/b/k?uploads"), + ("DELETE", "/s3/b/k"), + ("OPTIONS", "/s3/b/k"), + ] { + let (status, auth, path) = status_and_auth(proxy, method, target, &seen).await; + assert_eq!(status, 200, "{method} {target}"); + assert_eq!(auth, "Bearer vkt_job", "{method} {target}"); + assert_eq!(path, target, "{method} {target}"); + } + *seen.lock().unwrap() = Seen::default(); + let r = reqwest::Client::new() + .put(format!("http://{proxy}/s3/b/k")) + .header( + "authorization", + "AWS4-HMAC-SHA256 Credential=dummy/20261105/auto/s3/aws4_request, \ + SignedHeaders=host;x-amz-date, Signature=00", + ) + .header("x-amz-content-sha256", "STREAMING-AWS4-HMAC-SHA256-PAYLOAD") + .header("x-amz-security-token", "dummy") + .body("0;chunk-signature=00\r\n\r\n") + .send() + .await + .unwrap(); + assert_eq!(r.status(), 200); + let (auth, _, headers) = seen.lock().unwrap().clone(); + assert_eq!(auth, "Bearer vkt_job"); + assert_eq!(headers.get_all("authorization").iter().count(), 1); + assert!(!headers.contains_key("x-amz-security-token")); + assert_eq!( + headers["x-amz-content-sha256"], + "STREAMING-AWS4-HMAC-SHA256-PAYLOAD" + ); + for (method, target) in [("PROPFIND", "/s3/b/k"), ("PATCH", "/s3/b"), ("COPY", "/s3")] { + let (status, _, path) = status_and_auth(proxy, method, target, &seen).await; + assert_eq!(status, 405, "{method} {target}"); + assert_eq!(path, "", "{method} {target} must not be forwarded"); + } + let node = spawn_blocking(basic_cfg(up_addr)).unwrap(); + let deny = spawn_blocking(ProxyCfg::for_job(®istry(up_addr), None).unwrap()).unwrap(); + for proxy in [node, deny] { + for target in ["/s3", "/s3/b/k"] { + let (status, _, path) = status_and_auth(proxy, "GET", target, &seen).await; + assert_eq!(status, 403, "{target}"); + assert_eq!(path, "", "{target} must not be forwarded"); + } + } + } + + /// `/s3/job//` is the key below the area's prefix; `/s3/job/` alone is + /// the area as a bucket, its listings kept to the area by their query; anything else + /// under `job/` is refused unforwarded. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn a_job_reaches_s3_areas_by_alias() { + let _ = rustls::crypto::ring::default_provider().install_default(); + let (up_addr, seen) = fake_upstream(); + let cfg = ProxyCfg::for_job(®istry(up_addr), Some("vkt_job")) + .unwrap() + .with_areas(&s3_areas()); + let proxy = spawn_blocking(cfg).unwrap(); + for (method, target, want) in [ + ("GET", "/s3/job/one/a/b", "/s3/solo/a/b"), + ( + "PUT", + "/s3/job/two/.sccache_check", + "/s3/duo/b/.sccache_check", + ), + ( + "PUT", + "/s3/job/three/k?partNumber=1&uploadId=u", + "/s3/tri/b/c/k?partNumber=1&uploadId=u", + ), + ("GET", "/s3/job/one", "/s3/solo"), + ( + "GET", + "/s3/job/one/?list-type=2&prefix=a", + "/s3/solo?list-type=2&prefix=a", + ), + ("PUT", "/s3/job/one", "/s3/solo"), + ("DELETE", "/s3/job/one/", "/s3/solo"), + ("POST", "/s3/job/one?delete", "/s3/solo?delete"), + ("GET", "/s3/job/two", "/s3/duo?prefix=b%2F"), + ("GET", "/s3/job/three", "/s3/tri?prefix=b%2Fc%2F"), + ("GET", "/s3/job/three/", "/s3/tri?prefix=b%2Fc%2F"), + ( + "GET", + "/s3/job/three?list-type=2&prefix=d/&delimiter=/&start-after=d/a", + "/s3/tri?list-type=2&prefix=b%2Fc%2Fd/&delimiter=/&start-after=b%2Fc%2Fd/a", + ), + ( + "GET", + "/s3/job/three?marker=k", + "/s3/tri?marker=b%2Fc%2Fk&prefix=b%2Fc%2F", + ), + ( + "GET", + "/s3/job/three?uploads&key-marker=k", + "/s3/tri?uploads&key-marker=b%2Fc%2Fk&prefix=b%2Fc%2F", + ), + ("GET", "/s3/job/three?location", "/s3/tri?location"), + ("HEAD", "/s3/job/three", "/s3/tri"), + ("GET", "/s3/jobs/k", "/s3/jobs/k"), + ("GET", "/s3/tri/b/c/k", "/s3/tri/b/c/k"), + ] { + let (status, auth, path) = status_and_auth(proxy, method, target, &seen).await; + assert_eq!(status, 200, "{method} {target}"); + assert_eq!(auth, "Bearer vkt_job", "{method} {target}"); + assert_eq!(path, want, "{method} {target}"); + } + for (method, target) in [ + ("GET", "/s3/job"), + ("GET", "/s3/job/"), + ("GET", "/s3/job?list-type=2&prefix=one/"), + ("GET", "/s3/job/other/k"), + ("PUT", "/s3/job/other"), + ("PUT", "/s3/job/three"), + ("DELETE", "/s3/job/two/"), + ("POST", "/s3/job/three?delete"), + ("GET", "/s3/job/three?list-type=2&x=1"), + ("GET", "/s3/job/two?uploads&delimiter=/"), + ("GET", "/s3/job/two?prefix=a&acl"), + ] { + let (status, _, path) = status_and_auth(proxy, method, target, &seen).await; + assert_eq!(status, 403, "{method} {target}"); + assert_eq!(path, "", "{method} {target} must not be forwarded"); + } + } + + /// An S3 alias opens no way out either: `..` resolves before mapping and an encoded + /// separator is refused. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn an_s3_alias_does_not_escape_its_scope() { + let _ = rustls::crypto::ring::default_provider().install_default(); + let (up_addr, seen) = fake_upstream(); + let cfg = ProxyCfg::for_job(®istry(up_addr), Some("vkt_job")) + .unwrap() + .with_areas(&s3_areas()); + let proxy = spawn_blocking(cfg).unwrap(); + for (target, status, want) in [ + ("/s3/job/three/../../other", 200, "/s3/other"), + ("/s3/job/../", 200, "/s3/"), + ("/s3/job/three/../../../settings/keys", 403, ""), + ("/s3/job/three/%2e%2e/%2e%2e/%2e%2e/browse", 403, ""), + ("/s3/job/three/..%2F..%2F..%2Fv2", 403, ""), + ("/s3/job/three/..%5Cx", 403, ""), + ] { + *seen.lock().unwrap() = Seen::default(); + assert_eq!(raw_status(proxy, "GET", target).await, status, "{target}"); + assert_eq!(seen.lock().unwrap().1, want, "{target}"); + } + } + + /// A listing's keys and markers get the key prefix, a continuation token in the key it + /// names, and back again; a query that is not a listing's is kept. + #[test] + fn listing_queries_take_the_key_prefix() { + use vk_hub_proto::token::to_base64url; + let [_, _, three] = s3_areas(); + let names = S3Names::new(&three); + assert_eq!(names.bucket, "tri"); + assert_eq!(names.keys, "b/c/"); + let token = to_base64url(b"d/k"); + let real = to_base64url(b"b/c/d/k"); + assert_eq!( + listing_query( + &names, + &format!("list-type=2&continuation-token={token}&max-keys=2") + ), + Some(format!( + "list-type=2&continuation-token={real}&max-keys=2&prefix=b%2Fc%2F" + )) + ); + assert_eq!( + names.token_out(real.as_bytes()), + Some(token.clone().into_bytes()) + ); + assert_eq!(names.token_in(&token), real); + // One the registry did not give, or naming a key outside the area, is kept. + assert_eq!(names.token_in("!!"), "!!"); + assert_eq!(names.token_in(""), ""); + assert_eq!(names.token_out(to_base64url(b"x/k").as_bytes()), None); + assert_eq!(names.token_out(b"!!"), None); + assert_eq!( + listing_query(&names, "prefix&start-after").as_deref(), + Some("prefix=b%2Fc%2F&start-after=b%2Fc%2F") + ); + // A name is taken decoded, as the registry takes it. + assert_eq!( + listing_query(&names, "list-type=2&prefi%78=d/").as_deref(), + Some("list-type=2&prefix=b%2Fc%2Fd/") + ); + assert_eq!( + listing_query(&names, "upload%73").as_deref(), + Some("upload%73&prefix=b%2Fc%2F") + ); + // A configuration's query is kept; a listing's with anything else is refused. + for kept in ["versioning", "acl", "acl&x-id=GetBucketAcl"] { + assert_eq!(listing_query(&names, kept).as_deref(), Some(kept)); + } + for refused in [ + "list-type=2&tagging", + "uploads&list-type=2", + "uploads&delimiter=/", + "prefix=a&acl", + "list-type=2&x", + "list-type=2&%78", + ] { + assert_eq!(listing_query(&names, refused), None, "{refused}"); + } + // A one-component prefix is the bucket alone. + let one = S3Names::new(&area("one", "solo")); + assert_eq!((one.bucket.as_str(), one.keys.as_str()), ("solo", "")); + } + + /// A copy's source names its area as a path does, escaped or not. + #[test] + fn copy_sources_map_their_alias() { + let areas = s3_areas(); + for (source, want) in [ + ("/job/three/k", "/tri/b/c/k"), + ("job/two/a/k", "/duo/b/a/k"), + ("job%2Fone%2Fk", "/solo/k"), + ("%2Fjob%2Fthree%2Fk", "/tri/b/c/k"), + ] { + assert_eq!( + map_copy_source(&areas, source), + Ok(Some(want.to_string())), + "{source}" + ); + } + assert_eq!(map_copy_source(&areas, "/b/k"), Ok(None)); + assert_eq!(map_copy_source(&areas, "jobs/k"), Ok(None)); + assert_eq!(map_copy_source(&areas, "/job/other/k"), Err(())); + assert_eq!(map_copy_source(&areas, "job"), Err(())); + } + + /// An S3 document's names in the area go back to the alias, wherever the chunks split; a + /// key outside the area, and other elements, are left alone. + #[test] + fn s3_documents_map_back_to_the_alias() { + use vk_hub_proto::token::to_base64url; + let (tok_in, tok_out) = (to_base64url(b"b/c/d/"), to_base64url(b"d/")); + let body = format!( + "\ + trib/c/2/\ + true{tok_in}\ + {tok_in}\ + b/c/a\ + b/c/a&b1\ + b/cd/x2\ + b/c/d/\ + b/c/tritrio\ + /s3/tri/b/c/k/s3/tri\ + /s3/trio" + ); + let want = format!( + "\ + three2/\ + true{tok_out}\ + {tok_out}\ + a\ + a&b1\ + b/cd/x2\ + d/\ + threetrio\ + /s3/job/three/k/s3/job/three\ + /s3/trio" + ); + let names = S3Names::new(&area("three", "tri/b/c")); + for size in [1, 2, 3, 7, 25, body.len()] { + let mut map = S3Map::new(names.clone()); + let mut out = Vec::new(); + for chunk in body.as_bytes().chunks(size) { + out.extend_from_slice(&map.push(chunk)); + assert!(map.carry.len() <= MAX_ELEMENT + size, "chunks of {size}"); + } + out.extend_from_slice(&map.finish()); + assert_eq!(String::from_utf8(out).unwrap(), want, "chunks of {size}"); + } + // An element past the bound streams on unmapped rather than waiting. + let long = format!("b/c/{}", "x".repeat(MAX_ELEMENT)); + let mut map = S3Map::new(names); + let mut out = map.push(long.as_bytes()).to_vec(); + assert!(map.carry.is_empty()); + out.extend_from_slice(&map.push(b"")); + out.extend_from_slice(&map.finish()); + assert_eq!(out, format!("{long}").into_bytes()); + } + + /// An S3 document reached through an area, and a bucket's `Location`, come back under the + /// alias; an object's bytes, and an answer reached by the real path, are left alone. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn only_an_s3_document_through_the_alias_is_mapped_back() { + const LISTING: &str = "trib/c/\ + b/c/k"; + let _ = rustls::crypto::ring::default_provider().install_default(); + let start = |content_type: &'static str| { + let (up_addr, _) = fake_upstream_answering( + LISTING.as_bytes(), + if content_type == "application/xml" { + &[("content-type", "application/xml"), ("location", "/s3/tri")] + } else { + &[ + ("content-type", "application/octet-stream"), + ("location", "/s3/tri"), + ] + }, + ); + let cfg = ProxyCfg::for_job(®istry(up_addr), Some("vkt_job")) + .unwrap() + .with_areas(&s3_areas()); + spawn_blocking(cfg).unwrap() + }; + let get = |proxy: SocketAddr, target: &'static str| async move { + let r = reqwest::Client::new() + .get(format!("http://{proxy}{target}")) + .send() + .await + .unwrap(); + let location = r.headers()["location"].to_str().unwrap().to_string(); + (location, r.text().await.unwrap()) + }; + let xml = start("application/xml"); + assert_eq!( + get(xml, "/s3/job/three?list-type=2").await, + ( + "/s3/job/three".to_string(), + "three\ + k" + .to_string() + ) + ); + assert_eq!( + get(xml, "/s3/tri?list-type=2").await, + ("/s3/tri".to_string(), LISTING.to_string()) + ); + let bytes = start("application/octet-stream"); + assert_eq!( + get(bytes, "/s3/job/three/k").await, + ("/s3/job/three".to_string(), LISTING.to_string()) + ); + } + + /// An answer whose body is passed on as it came keeps its length, a HEAD's included, which + /// is how a client learns an object's size; a rewritten document drops the stale one. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn an_answer_keeps_its_length_unless_rewritten() { + let _ = rustls::crypto::ring::default_provider().install_default(); + let (up_addr, _) = fake_upstream_answering( + b"tri", + &[ + ("content-type", "application/xml"), + ("accept-ranges", "bytes"), + ("content-range", "bytes 0-52/100"), + ], + ); + let cfg = ProxyCfg::for_job(®istry(up_addr), Some("vkt_job")) + .unwrap() + .with_areas(&s3_areas()); + let proxy = spawn_blocking(cfg).unwrap(); + let client = reqwest::Client::new(); + for (method, target) in [ + (reqwest::Method::HEAD, "/s3/tri/b/c/k"), + (reqwest::Method::GET, "/s3/tri/b/c/k"), + (reqwest::Method::HEAD, "/s3/job/three/k"), + (reqwest::Method::HEAD, "/s3/job/three"), + ] { + let r = client + .request(method.clone(), format!("http://{proxy}{target}")) + .send() + .await + .unwrap(); + let h = r.headers(); + assert_eq!(h.get("content-length").unwrap(), "53", "{method} {target}"); + assert_eq!( + h.get("accept-ranges").unwrap(), + "bytes", + "{method} {target}" + ); + assert_eq!(h.get("content-range").unwrap(), "bytes 0-52/100"); + } + let r = client + .get(format!("http://{proxy}/s3/job/three?list-type=2")) + .send() + .await + .unwrap(); + assert!(!r.headers().contains_key("content-length")); + assert_eq!( + r.text().await.unwrap(), + "three" + ); + } + /// A job meant to have a token and given none gets a proxy that lends nothing. #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn a_job_without_a_token_gets_nothing() { @@ -1489,6 +2328,11 @@ mod tests { "/lock/acquire?name=p:k", "/dav/files/p/x", "/v2/p/tags/list", + "/s3", + "/s3?x-id=ListBuckets", + "/s3/", + "/s3/b/k", + "/s3/job/../b", ] { assert!(ok(fine).is_some(), "{fine}"); } @@ -1499,9 +2343,17 @@ mod tests { "/dav/files/../../x", "/dav/files/../repos/x", "/lock/%2e%2e/browse", + "/s3x", + "/s3../x", + "/s3/../settings/keys", + "/s3/%2e%2e/browse", + "/s3/b/..%2f..%2fv2", + "/s3/job/%2e%2e/%2e%2e/dav/repos", ] { assert_eq!(ok(escape), None, "{escape}"); } + // Nor the node's credential the S3 view. + assert_eq!(upstream_url(&up, "/s3/b/k"), None); } #[test] @@ -1827,21 +2679,56 @@ mod tests { let get = Method::GET; let put = Method::PUT; let post = Method::POST; - assert!(repeatable(&get, "/v2/app/blobs/sha256:ab")); - assert!(repeatable(&Method::HEAD, "/v2/app/manifests/latest")); - assert!(repeatable(&put, "/v2/app/manifests/latest")); - assert!(repeatable(&put, "/v2/team/app/manifests/sha256:ab")); - assert!(repeatable(&put, "/dav/files/p/x")); - assert!(repeatable(&post, "/v2/app/blobs/uploads/")); - assert!(!repeatable(&put, "/v2/app/blobs/uploads/u1")); - assert!(!repeatable(&Method::PATCH, "/v2/app/blobs/uploads/u1")); - assert!(!repeatable(&Method::DELETE, "/v2/app/manifests/sha256:ab")); - assert!(!repeatable(&post, "/lock/acquire")); - assert!(!repeatable(&get, "/lock/acquire")); + let none = hyper::HeaderMap::new(); + assert!(repeatable(&get, "/v2/app/blobs/sha256:ab", &none)); + assert!(repeatable(&Method::HEAD, "/v2/app/manifests/latest", &none)); + assert!(repeatable(&put, "/v2/app/manifests/latest", &none)); + assert!(repeatable(&put, "/v2/team/app/manifests/sha256:ab", &none)); + assert!(repeatable(&put, "/dav/files/p/x", &none)); + assert!(repeatable(&put, "/s3/b/k", &none)); + assert!(repeatable(&get, "/s3/b", &none)); + assert!(!repeatable(&put, "/s3/b", &none)); + assert!(!repeatable(&put, "/s3/b/", &none)); + assert!(!repeatable(&post, "/s3/b/k?uploadId=u", &none)); + assert!(!repeatable(&Method::DELETE, "/s3/b/k", &none)); + assert!(repeatable(&post, "/v2/app/blobs/uploads/", &none)); + assert!(!repeatable(&put, "/v2/app/blobs/uploads/u1", &none)); + assert!(!repeatable( + &Method::PATCH, + "/v2/app/blobs/uploads/u1", + &none + )); + assert!(!repeatable( + &Method::DELETE, + "/v2/app/manifests/sha256:ab", + &none + )); + assert!(!repeatable(&post, "/lock/acquire", &none)); + assert!(!repeatable(&get, "/lock/acquire", &none)); assert!(!repeatable( &Method::from_bytes(b"MKCOL").unwrap(), - "/dav/files/p/x" + "/dav/files/p/x", + &none )); + // A write under a precondition its first attempt may have spent is sent once. + for name in [ + "if-none-match", + "if-match", + "if-unmodified-since", + "x-amz-copy-source-if-match", + "x-amz-copy-source-if-none-match", + "x-amz-copy-source-if-modified-since", + "x-amz-copy-source-if-unmodified-since", + ] { + let mut conditional = hyper::HeaderMap::new(); + conditional.insert( + hyper::header::HeaderName::from_static(name), + hyper::header::HeaderValue::from_static("*"), + ); + assert!(!repeatable(&put, "/s3/b/k", &conditional), "{name}"); + assert!(!repeatable(&put, "/dav/files/p/x", &conditional), "{name}"); + assert!(repeatable(&get, "/s3/b/k", &conditional), "{name}"); + } assert!(digest_addressed("/v2/app/blobs/sha256:ab")); assert!(digest_addressed("/v2/team/app/manifests/sha256:ab")); assert!(!digest_addressed("/v2/app/manifests/latest")); diff --git a/vk-registry/Cargo.toml b/vk-registry/Cargo.toml index 225338b2..1082973f 100644 --- a/vk-registry/Cargo.toml +++ b/vk-registry/Cargo.toml @@ -35,6 +35,16 @@ rustls.workspace = true tokio-rustls = { version = "0.26", default-features = false, features = ["ring", "logging", "tls12"] } oci-client.workspace = true sha2.workspace = true +# The digests S3 clients send or expect beside SHA-256: an object's MD5 (its ETag), and the +# SHA-1, CRC32, CRC32C, CRC64NVME and XXHASH an upload may be checked against. +md5.workspace = true +sha1.workspace = true +crc32fast.workspace = true +crc.workspace = true +xxhash-rust.workspace = true +# SigV4's HMAC chain for /s3/; zeroize wipes the keys it derives. +hmac.workspace = true +zeroize.workspace = true zstd.workspace = true libc.workspace = true # Private admin-socket binding, shared with vk-core. diff --git a/vk-registry/DESIGN.md b/vk-registry/DESIGN.md index 367422c4..4484ffd9 100644 --- a/vk-registry/DESIGN.md +++ b/vk-registry/DESIGN.md @@ -556,7 +556,317 @@ an object under a name a collection already holds. Idleness is the tag's mtime, `PUT` sets and a `GET` refreshes once it is an hour old, so a read-only pipeline's hits keep an entry alive as a writer's do. There is no per-directory policy and no size cap; a replaced or deleted object frees no disk until the next `gc`. A staging file a crashed `PUT` -left in `uploads/` goes with the idle uploads. +left in `uploads/` goes with the idle uploads, and so does an S3 multipart upload nothing has +been written into for the grace window: its directory under `uploads/s3/` is idle by the +newest mtime among its files (the directory's own does not move when a part is replaced), +and goes whole, parts and all. Completing an upload marks it used, so the sweep does not +take it while its parts are read, and leaves its directory holding only the record of what +it made, swept in turn once idle, so a completion sent again until then is answered again. + +## S3 view (`/s3/`) + +The `files/` area is also served as S3, path-style, under `/s3/`, through the same listener, +TLS and credentials: a bucket is a top-level `files/` directory and an object key the path +below it, so `PUT /s3/sccache/a/b/abcdef` writes the same tag `/dav/files/sccache/a/b/abcdef` +names. Nothing new is stored, and the WebDAV view, `/v2/`, `/browse`, `vk registry get` and +the gc see the same entries. S3 is enabled by default. Set `s3 = false` to disable it: `/s3/` +requests then return the 404 any unknown path returns, after the usual authentication, and +the server logs this setting at startup. + +```text +/s3/ the buckets: top-level directories of files/ +/s3/ files/ +/s3// the object files// names, as in /dav/files/ +``` + +Clients take `endpoint = https:///s3` and any region: `sccache` with +`SCCACHE_BUCKET= SCCACHE_ENDPOINT=https:///s3 SCCACHE_REGION=auto` +(`SCCACHE_S3_KEY_PREFIX` names a directory inside the bucket). + +**Virtual-hosted style** is opt-in, for clients that cannot be made to use paths: with +`s3_host = "s3.example.com"`, a request sent to `.s3.example.com` names its bucket by +its host and its key by its whole path, and is served as `/s3//`, so `/` there +is the bucket and `/a/b` its object `a/b`, even where the path is one another route would +take. The `Host` port is ignored. The signature is checked over the request as sent, host and +path included, so nothing about credentials changes. It needs a wildcard DNS record +`*.s3.example.com` pointing at the registry and, with TLS, a certificate covering that +wildcard (one label deep: a bucket name with a dot is not covered). A host whose label cannot +be a bucket, `s3_host` itself, and every other host are served as before, and path-style +`/s3/` keeps working on any of them. The setting is a lowercase host name with no port; the +server refuses anything else at startup. The registry's own host name must not be below +`s3_host`, since every `