From 332087fcccc8951516e2c1a3ba809e19f0330bb4 Mon Sep 17 00:00:00 2001 From: Josh Siviour Date: Thu, 18 Jun 2026 10:51:26 +1000 Subject: [PATCH] feat: OCI Distribution v2 blob + manifest storage (#375) * feat: implement OCI Distribution v2 blob and manifest storage Replace the /v2/* 501 stub with a real path-parsing dispatcher and content-addressable storage for the ECR registry surface: - Blob upload (monolithic + chunked PATCH), HEAD/GET, cross-repo mount, and cancel, with server-side sha256 verification on finalize. - Manifest PUT with type-aware shallow validation (image manifest blob existence; image index child-manifest existence + mediaType match), byte-for-byte GET with stored Content-Type, HEAD, and tag delete. - Tag list and account-scoped catalog. Blob and manifest bytes stream straight to predastore; repo, tag, manifest and in-progress-upload metadata live in a per-account JetStream KV bucket with optimistic-concurrency CAS on uploads. Chunks are buffered to a temp upload object and reassembled on finalize, since predastore exposes no append primitive through the shared object-store interface. Adds HeadObject to the objectstore abstraction. Refs: mulgadc/mulga#73 * refactor(ecr): route registry metadata through NATS daemon handlers The gateway no longer talks JetStream KV directly. The daemon owns the per-account KV via a MetaService bound to ecr.* request/reply subjects; the gateway is a NATS client (NATSMetaStore). Blob and manifest bytes still stream gateway-to-predastore directly. Also fixes four registry review findings: - finishUpload: clean up the temp object and upload record on a blob store failure so the uuid is reusable and no partial blob lingers. - patchUpload: commit the appended bytes and advanced hash state under a single KV CAS keyed per write, so a racing PATCH is rejected (409) rather than corrupting the assembled bytes; the verified digest always matches the stored bytes. - cancelUpload: delete both the temp bytes and the record, 204 on success, 404 on unknown uuid, no dead branch. - getManifest: return MANIFEST_INVALID (not MANIFEST_UNKNOWN) for a 406 Accept media-type mismatch. Refs: mulgadc/mulga#73 * fix(ecr): provision account predastore bucket on demand OCI blob/manifest writes target ecr-{accountID}, which nothing created, so every push failed with a backend 403. Add idempotent EnsureBucket to the object store (HeadBucket -> CreateBucket, swallowing already-owned) and ensure it once per process at the top of the registry ServeHTTP, so ECR storage appears transparently like AWS-managed ECR. * fix(ecr): sanitize ':' from predastore blob/manifest object keys predastore mis-canonicalizes SigV4 for object keys containing ':', so writes/reads of content-addressable keys (sha256:) were rejected 403 (reproducible with the aws CLI too). Route BlobKey and ManifestKey digests through DigestToken so the object key carries no ':' separator. --- spinifex/daemon/daemon.go | 31 + spinifex/daemon/daemon_handlers_ecr.go | 55 ++ spinifex/gateway/ecr.go | 15 +- spinifex/gateway/ecr/registry.go | 873 ++++++++++++++++++ .../gateway/ecr/registry_ensurebucket_test.go | 46 + spinifex/gateway/ecr/registry_test.go | 376 ++++++++ spinifex/gateway/ecr/registry_upload_test.go | 147 +++ spinifex/gateway/gateway.go | 4 + spinifex/handlers/NATS_HANDLER_NAMING.md | 24 + spinifex/handlers/ecr/meta.go | 495 ++++++++++ spinifex/handlers/ecr/meta_kv_test.go | 80 ++ spinifex/handlers/ecr/meta_nats.go | 144 +++ spinifex/handlers/ecr/meta_test.go | 116 +++ spinifex/handlers/ecr/service.go | 162 ++++ spinifex/handlers/ecr/service_impl.go | 152 +++ spinifex/handlers/ecr/service_test.go | 151 +++ spinifex/handlers/ecr/storage.go | 104 ++- spinifex/handlers/ecr/storage_test.go | 21 +- spinifex/objectstore/ensurebucket_test.go | 108 +++ spinifex/objectstore/objectstore.go | 55 ++ spinifex/objectstore/objectstore_test.go | 17 + spinifex/services/awsgw/awsgw.go | 17 + 22 files changed, 3181 insertions(+), 12 deletions(-) create mode 100644 spinifex/daemon/daemon_handlers_ecr.go create mode 100644 spinifex/gateway/ecr/registry.go create mode 100644 spinifex/gateway/ecr/registry_ensurebucket_test.go create mode 100644 spinifex/gateway/ecr/registry_test.go create mode 100644 spinifex/gateway/ecr/registry_upload_test.go create mode 100644 spinifex/handlers/ecr/meta.go create mode 100644 spinifex/handlers/ecr/meta_kv_test.go create mode 100644 spinifex/handlers/ecr/meta_nats.go create mode 100644 spinifex/handlers/ecr/meta_test.go create mode 100644 spinifex/handlers/ecr/service.go create mode 100644 spinifex/handlers/ecr/service_impl.go create mode 100644 spinifex/handlers/ecr/service_test.go create mode 100644 spinifex/objectstore/ensurebucket_test.go diff --git a/spinifex/daemon/daemon.go b/spinifex/daemon/daemon.go index 2b25738c3..7f2ce29ae 100644 --- a/spinifex/daemon/daemon.go +++ b/spinifex/daemon/daemon.go @@ -48,6 +48,7 @@ import ( handlers_ec2_tags "github.com/mulgadc/spinifex/spinifex/handlers/ec2/tags" handlers_ec2_volume "github.com/mulgadc/spinifex/spinifex/handlers/ec2/volume" handlers_ec2_vpc "github.com/mulgadc/spinifex/spinifex/handlers/ec2/vpc" + handlers_ecr "github.com/mulgadc/spinifex/spinifex/handlers/ecr" handlers_eks "github.com/mulgadc/spinifex/spinifex/handlers/eks" handlers_elbv2 "github.com/mulgadc/spinifex/spinifex/handlers/elbv2" handlers_imds "github.com/mulgadc/spinifex/spinifex/handlers/imds" @@ -150,6 +151,7 @@ type Daemon struct { elbv2Service *handlers_elbv2.ELBv2ServiceImpl eksService *handlers_eks.EKSServiceImpl acmService *handlers_acm.ACMServiceImpl + ecrMetaService *handlers_ecr.MetaServiceImpl routeTableService *handlers_ec2_routetable.RouteTableServiceImpl natGatewayService *handlers_ec2_natgw.NatGatewayServiceImpl externalIPAM *handlers_ec2_vpc.ExternalIPAM @@ -956,6 +958,26 @@ func (d *Daemon) subscribeAll() error { ) } + // ECR gateway → daemon subscriptions. The daemon owns the per-account + // JetStream KV metadata; blob/manifest bytes never traverse these subjects. + if d.ecrMetaService != nil { + subs = append(subs, + natsSub{handlers_ecr.SubjectRepoCreate, d.handleECRRepoCreate, "spinifex-workers"}, + natsSub{handlers_ecr.SubjectRepoDescribe, d.handleECRRepoDescribe, "spinifex-workers"}, + natsSub{handlers_ecr.SubjectRepoList, d.handleECRRepoList, "spinifex-workers"}, + natsSub{handlers_ecr.SubjectTagPut, d.handleECRTagPut, "spinifex-workers"}, + natsSub{handlers_ecr.SubjectTagGet, d.handleECRTagGet, "spinifex-workers"}, + natsSub{handlers_ecr.SubjectTagList, d.handleECRTagList, "spinifex-workers"}, + natsSub{handlers_ecr.SubjectTagDelete, d.handleECRTagDelete, "spinifex-workers"}, + natsSub{handlers_ecr.SubjectManifestPut, d.handleECRManifestPut, "spinifex-workers"}, + natsSub{handlers_ecr.SubjectManifestDescribe, d.handleECRManifestDescribe, "spinifex-workers"}, + natsSub{handlers_ecr.SubjectUploadCreate, d.handleECRUploadCreate, "spinifex-workers"}, + natsSub{handlers_ecr.SubjectUploadGet, d.handleECRUploadGet, "spinifex-workers"}, + natsSub{handlers_ecr.SubjectUploadUpdate, d.handleECRUploadUpdate, "spinifex-workers"}, + natsSub{handlers_ecr.SubjectUploadDelete, d.handleECRUploadDelete, "spinifex-workers"}, + ) + } + // EIP operations require external IPAM (pool mode). Only subscribe when available; // without a subscriber the gateway returns a NATS timeout → clean error to the client. if d.eipService != nil { @@ -1360,6 +1382,15 @@ func (d *Daemon) startCluster() error { return fmt.Errorf("failed to initialize ACM service: %w", err) } + // ECR metadata service: owns per-account JetStream KV for repos, tags, + // manifest records and upload-state CAS. Disabled (gateway returns NATS + // timeouts) when JetStream is unavailable. + if js, jsErr := d.natsConn.JetStream(); jsErr != nil { + slog.Warn("ECR metadata service disabled: JetStream unavailable", "err", jsErr) + } else { + d.ecrMetaService = handlers_ecr.NewKVMetaService(js) + } + if err := d.eksService.SpawnRegisteredReconcilers(); err != nil { slog.Warn("EKS: SpawnRegisteredReconcilers failed", "err", err) } diff --git a/spinifex/daemon/daemon_handlers_ecr.go b/spinifex/daemon/daemon_handlers_ecr.go new file mode 100644 index 000000000..f99a3259f --- /dev/null +++ b/spinifex/daemon/daemon_handlers_ecr.go @@ -0,0 +1,55 @@ +package daemon + +import "github.com/nats-io/nats.go" + +func (d *Daemon) handleECRRepoCreate(msg *nats.Msg) { + handleNATSRequest(msg, d.ecrMetaService.RepoCreate) +} + +func (d *Daemon) handleECRRepoDescribe(msg *nats.Msg) { + handleNATSRequest(msg, d.ecrMetaService.RepoDescribe) +} + +func (d *Daemon) handleECRRepoList(msg *nats.Msg) { + handleNATSRequest(msg, d.ecrMetaService.RepoList) +} + +func (d *Daemon) handleECRTagPut(msg *nats.Msg) { + handleNATSRequest(msg, d.ecrMetaService.TagPut) +} + +func (d *Daemon) handleECRTagGet(msg *nats.Msg) { + handleNATSRequest(msg, d.ecrMetaService.TagGet) +} + +func (d *Daemon) handleECRTagList(msg *nats.Msg) { + handleNATSRequest(msg, d.ecrMetaService.TagList) +} + +func (d *Daemon) handleECRTagDelete(msg *nats.Msg) { + handleNATSRequest(msg, d.ecrMetaService.TagDelete) +} + +func (d *Daemon) handleECRManifestPut(msg *nats.Msg) { + handleNATSRequest(msg, d.ecrMetaService.ManifestPut) +} + +func (d *Daemon) handleECRManifestDescribe(msg *nats.Msg) { + handleNATSRequest(msg, d.ecrMetaService.ManifestDescribe) +} + +func (d *Daemon) handleECRUploadCreate(msg *nats.Msg) { + handleNATSRequest(msg, d.ecrMetaService.UploadCreate) +} + +func (d *Daemon) handleECRUploadGet(msg *nats.Msg) { + handleNATSRequest(msg, d.ecrMetaService.UploadGet) +} + +func (d *Daemon) handleECRUploadUpdate(msg *nats.Msg) { + handleNATSRequest(msg, d.ecrMetaService.UploadUpdate) +} + +func (d *Daemon) handleECRUploadDelete(msg *nats.Msg) { + handleNATSRequest(msg, d.ecrMetaService.UploadDelete) +} diff --git a/spinifex/gateway/ecr.go b/spinifex/gateway/ecr.go index 2e2b135cf..bde6c145e 100644 --- a/spinifex/gateway/ecr.go +++ b/spinifex/gateway/ecr.go @@ -12,14 +12,19 @@ import ( // hostMatch host-routing middleware. // // OCI repository names may contain slashes (e.g. team/app), so the blob, -// manifest and tag paths can't be expressed as fixed chi segments. Everything -// under /v2 therefore routes through a single catch-all 501 stub until a -// path-parsing dispatcher and storage exist. GET /v2/ is live so the registry -// version-check probe succeeds. +// manifest and tag paths can't be expressed as fixed chi segments. The +// Registry parses the path manually. GET /v2/ is always live so the registry +// version-check probe succeeds even before any repository exists. When no +// Registry is wired (e.g. unit tests of unrelated routes), the surface falls +// back to the 501 stub. func (gw *GatewayConfig) mountOCIRegistry(r chi.Router) { r.Route("/v2", func(v2 chi.Router) { v2.Use(gw.hostMatch) v2.Get("/", gateway_ecr.APIVersion) - v2.HandleFunc("/*", gateway_ecr.NotImplemented) + if gw.ECRRegistry != nil { + v2.HandleFunc("/*", gw.ECRRegistry.ServeHTTP) + } else { + v2.HandleFunc("/*", gateway_ecr.NotImplemented) + } }) } diff --git a/spinifex/gateway/ecr/registry.go b/spinifex/gateway/ecr/registry.go new file mode 100644 index 000000000..f31d58f2b --- /dev/null +++ b/spinifex/gateway/ecr/registry.go @@ -0,0 +1,873 @@ +package gateway_ecr + +import ( + "crypto/sha256" + "encoding" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "hash" + "io" + "log/slog" + "net/http" + "strconv" + "strings" + "sync" + "time" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/service/s3" + "github.com/google/uuid" + "github.com/mulgadc/spinifex/spinifex/handlers/ecr" + "github.com/mulgadc/spinifex/spinifex/objectstore" +) + +// OCI media types recognised by the manifest validator. +const ( + mediaTypeDockerManifest = "application/vnd.docker.distribution.manifest.v2+json" + mediaTypeDockerManifestList = "application/vnd.docker.distribution.manifest.list.v2+json" + mediaTypeOCIManifest = "application/vnd.oci.image.manifest.v1+json" + mediaTypeOCIIndex = "application/vnd.oci.image.index.v1+json" + defaultManifestContentType = mediaTypeDockerManifest +) + +// maxManifestBytes caps an accepted manifest document. Manifests are small JSON +// blobs; anything larger is rejected before buffering. +const maxManifestBytes = 4 << 20 + +// Registry serves the OCI Distribution Spec v2 surface (/v2/*). Blob and +// manifest bytes stream straight to an account-scoped predastore bucket via +// Store. Metadata (repos, tags, manifest records, in-progress upload-state CAS) +// goes through Meta, which the gateway reaches over NATS request/reply to the +// daemon that owns the per-account JetStream KV. +type Registry struct { + Store objectstore.ObjectStore + Meta ecr.MetaStore + AccountID string + + // bucketMu guards bucketReady, which caches that the account's predastore + // bucket has been provisioned. Only success is cached: a failed ensure is + // retried on the next request. + bucketMu sync.Mutex + bucketReady bool +} + +// NewRegistry wires a Registry to its predastore object store and the metadata +// store (a NATS client in production) for the given account. +func NewRegistry(store objectstore.ObjectStore, meta ecr.MetaStore, accountID string) *Registry { + return &Registry{Store: store, Meta: meta, AccountID: accountID} +} + +// ServeHTTP dispatches a /v2/* request by manually parsing the path: OCI repo +// names contain slashes, so the {name} segment is everything between "/v2/" and +// the trailing "/blobs", "/manifests" or "/tags" marker. +func (reg *Registry) ServeHTTP(w http.ResponseWriter, r *http.Request) { + if err := reg.ensureBucket(); err != nil { + reg.internal(w, "ensure bucket", err) + return + } + + path := strings.TrimPrefix(r.URL.Path, "/v2/") + + if path == "_catalog" { + reg.handleCatalog(w, r) + return + } + + name, kind, ref, ok := splitV2Path(path) + if !ok { + WriteError(w, http.StatusNotFound, "NAME_UNKNOWN", "unrecognized registry path") + return + } + + if err := ecr.ValidateRepoName(name); err != nil { + WriteError(w, http.StatusBadRequest, "NAME_INVALID", err.Error()) + return + } + + switch kind { + case "blobs": + reg.routeBlobs(w, r, name, ref) + case "manifests": + reg.routeManifests(w, r, name, ref) + case "tags": + reg.routeTags(w, r, name, ref) + default: + WriteError(w, http.StatusNotFound, "NAME_UNKNOWN", "unrecognized registry path") + } +} + +// splitV2Path parses "{name}/{kind}/{ref...}" where name may contain slashes. +// kind is one of blobs/manifests/tags. ref is the remainder (digest, reference, +// "uploads", "uploads/{uuid}", or "list"). ok is false if no marker is found. +func splitV2Path(path string) (name, kind, ref string, ok bool) { + for _, marker := range []string{"/blobs/", "/manifests/", "/tags/"} { + if before, after, found := strings.Cut(path, marker); found { + kind = strings.Trim(marker, "/") + return before, kind, after, before != "" + } + } + return "", "", "", false +} + +// ---- blobs ---- + +func (reg *Registry) routeBlobs(w http.ResponseWriter, r *http.Request, name, ref string) { + switch { + case ref == "uploads/" || ref == "uploads": + if r.Method == http.MethodPost { + reg.startUpload(w, r, name) + return + } + case strings.HasPrefix(ref, "uploads/"): + uploadID := strings.TrimPrefix(ref, "uploads/") + switch r.Method { + case http.MethodPatch: + reg.patchUpload(w, r, name, uploadID) + return + case http.MethodPut: + reg.finishUpload(w, r, name, uploadID) + return + case http.MethodDelete: + reg.cancelUpload(w, name, uploadID) + return + } + default: + // ref is a digest. + switch r.Method { + case http.MethodHead: + reg.headBlob(w, name, ref) + return + case http.MethodGet: + reg.getBlob(w, name, ref) + return + } + } + WriteError(w, http.StatusNotFound, "BLOB_UNKNOWN", "unsupported blob operation") +} + +func (reg *Registry) headBlob(w http.ResponseWriter, _, digest string) { + if !ecr.ValidateDigest(digest) { + WriteError(w, http.StatusBadRequest, "DIGEST_INVALID", "malformed digest") + return + } + out, err := reg.Store.HeadObject(&s3.HeadObjectInput{ + Bucket: aws.String(reg.bucket()), + Key: aws.String(ecr.BlobKey(digest)), + }) + if err != nil { + if objectstore.IsNoSuchKeyError(err) { + WriteError(w, http.StatusNotFound, "BLOB_UNKNOWN", "blob not found") + return + } + reg.internal(w, "head blob", err) + return + } + w.Header().Set("Content-Length", strconv.FormatInt(aws.Int64Value(out.ContentLength), 10)) + w.Header().Set("Docker-Content-Digest", digest) + w.WriteHeader(http.StatusOK) +} + +func (reg *Registry) getBlob(w http.ResponseWriter, _, digest string) { + if !ecr.ValidateDigest(digest) { + WriteError(w, http.StatusBadRequest, "DIGEST_INVALID", "malformed digest") + return + } + out, err := reg.Store.GetObject(&s3.GetObjectInput{ + Bucket: aws.String(reg.bucket()), + Key: aws.String(ecr.BlobKey(digest)), + }) + if err != nil { + if objectstore.IsNoSuchKeyError(err) { + WriteError(w, http.StatusNotFound, "BLOB_UNKNOWN", "blob not found") + return + } + reg.internal(w, "get blob", err) + return + } + defer func() { _ = out.Body.Close() }() + if out.ContentLength != nil { + w.Header().Set("Content-Length", strconv.FormatInt(aws.Int64Value(out.ContentLength), 10)) + } + w.Header().Set("Docker-Content-Digest", digest) + w.Header().Set("Content-Type", "application/octet-stream") + w.WriteHeader(http.StatusOK) + if _, err := io.Copy(w, out.Body); err != nil { + slog.Error("ECR/OCI: blob stream failed", "err", err) + } +} + +// startUpload begins a blob upload. A ?mount&from cross-repo mount short-circuits +// to 201 when the digest already lives in the account pool; otherwise it opens a +// new chunked upload (202) with a fresh running sha256 state. +func (reg *Registry) startUpload(w http.ResponseWriter, r *http.Request, name string) { + if err := reg.ensureRepo(name); err != nil { + reg.internal(w, "ensure repo", err) + return + } + + q := r.URL.Query() + if mountDigest := q.Get("mount"); mountDigest != "" { + if ecr.ValidateDigest(mountDigest) && reg.blobExists(mountDigest) { + reg.writeBlobCreated(w, name, mountDigest) + return + } + // Mount miss: fall through to a normal upload start per the spec. + } + + uploadID := uuid.NewString() + h := sha256.New() + state, err := marshalHash(h) + if err != nil { + reg.internal(w, "marshal hash", err) + return + } + now := time.Now().UTC() + if _, err := reg.Meta.PutUpload(reg.AccountID, uploadID, ecr.UploadState{ + RepoName: name, + StartedAt: now, + LastActivity: now, + SHA256MarshaledState: state, + }); err != nil { + reg.internal(w, "create upload", err) + return + } + + w.Header().Set("Location", uploadPath(name, uploadID)) + w.Header().Set("Range", "0-0") + w.Header().Set("Docker-Upload-UUID", uploadID) + w.WriteHeader(http.StatusAccepted) +} + +// patchUpload appends a chunk to an in-progress upload. predastore exposes no +// append primitive, so each PATCH reads the bytes committed by the prior +// revision, concatenates the chunk, and writes the result under a fresh unique +// key. The running sha256 advances from the stored marshaled state. The new +// byte key and hash state are committed together under a single KV CAS at the +// revision read at the top: a concurrent PATCH that loses the CAS gets 409 and +// its freshly-written object is simply orphaned, never referenced. Because the +// winning record names the exact object it hashed, the digest finally verified +// always equals the bytes finally stored. +func (reg *Registry) patchUpload(w http.ResponseWriter, r *http.Request, name, uploadID string) { + st, rev, err := reg.Meta.GetUpload(reg.AccountID, uploadID) + if err != nil { + if errors.Is(err, ecr.ErrNotFound) { + WriteError(w, http.StatusNotFound, "BLOB_UPLOAD_UNKNOWN", "upload not found") + return + } + reg.internal(w, "get upload", err) + return + } + + chunk, err := io.ReadAll(r.Body) + if err != nil { + reg.internal(w, "read chunk", err) + return + } + + prior := reg.readUploadBytesAt(st.BytesKey) + assembled := append(prior, chunk...) + newKey := ecr.UploadChunkKey(uploadID, uuid.NewString()) + if err := reg.putUploadBytesAt(newKey, assembled); err != nil { + reg.internal(w, "store chunk", err) + return + } + + h, err := unmarshalHash(st.SHA256MarshaledState) + if err != nil { + reg.internal(w, "restore hash", err) + return + } + h.Write(chunk) + marshaled, err := marshalHash(h) + if err != nil { + reg.internal(w, "marshal hash", err) + return + } + + priorKey := st.BytesKey + st.CommittedBytes += int64(len(chunk)) + st.SHA256MarshaledState = marshaled + st.LastActivity = time.Now().UTC() + st.BytesKey = newKey + if _, err := reg.Meta.UpdateUpload(reg.AccountID, uploadID, st, rev); err != nil { + // Lost the CAS or a real failure: drop the object this attempt wrote so + // it doesn't linger, then surface 409 (the client/registry retries). + _ = reg.deleteUploadBytesAt(newKey) + if errors.Is(err, ecr.ErrConflict) { + WriteError(w, http.StatusConflict, "BLOB_UPLOAD_INVALID", "concurrent upload modification") + return + } + reg.internal(w, "update upload", err) + return + } + // CAS won: the superseded object is no longer referenced. + if priorKey != "" { + _ = reg.deleteUploadBytesAt(priorKey) + } + + w.Header().Set("Location", uploadPath(name, uploadID)) + w.Header().Set("Range", fmt.Sprintf("0-%d", st.CommittedBytes-1)) + w.Header().Set("Docker-Upload-UUID", uploadID) + w.WriteHeader(http.StatusAccepted) +} + +// finishUpload finalizes an upload. It supports both monolithic PUT (full body +// here) and chunked completion (body empty, bytes already PATCHed). The server +// recomputes sha256 over all bytes and rejects a mismatch with DIGEST_INVALID. +func (reg *Registry) finishUpload(w http.ResponseWriter, r *http.Request, name, uploadID string) { + digest := r.URL.Query().Get("digest") + if digest == "" || !ecr.ValidateDigest(digest) { + WriteError(w, http.StatusBadRequest, "DIGEST_INVALID", "missing or malformed digest") + return + } + + st, _, err := reg.Meta.GetUpload(reg.AccountID, uploadID) + if err != nil { + if errors.Is(err, ecr.ErrNotFound) { + WriteError(w, http.StatusNotFound, "BLOB_UPLOAD_UNKNOWN", "upload not found") + return + } + reg.internal(w, "get upload", err) + return + } + + finalChunk, err := io.ReadAll(r.Body) + if err != nil { + reg.internal(w, "read body", err) + return + } + + h, err := unmarshalHash(st.SHA256MarshaledState) + if err != nil { + reg.internal(w, "restore hash", err) + return + } + + var assembled []byte + if len(finalChunk) > 0 { + assembled = append(reg.readUploadBytesAt(st.BytesKey), finalChunk...) + h.Write(finalChunk) + st.CommittedBytes += int64(len(finalChunk)) + } else { + assembled = reg.readUploadBytesAt(st.BytesKey) + } + + computed := "sha256:" + hex.EncodeToString(h.Sum(nil)) + if computed != digest { + reg.cleanupUpload(uploadID) + WriteError(w, http.StatusBadRequest, "DIGEST_INVALID", "computed digest does not match expected digest") + return + } + + // Concurrent-push short-circuit: if the pool already holds this blob, skip + // the store and just drop the temp upload. + if !reg.blobExists(digest) { + if _, err := reg.Store.PutObject(&s3.PutObjectInput{ + Bucket: aws.String(reg.bucket()), + Key: aws.String(ecr.BlobKey(digest)), + Body: aws.ReadSeekCloser(strings.NewReader(string(assembled))), + }); err != nil { + // Finalize failed: drop the temp object and upload record so the + // uuid is reusable and no partial blob is left assembled. + reg.cleanupUpload(uploadID) + reg.internal(w, "store blob", err) + return + } + } + + reg.cleanupUpload(uploadID) + reg.writeBlobCreated(w, name, digest) +} + +// cancelUpload aborts an in-progress upload, deleting both the committed temp +// bytes and the upload record. An unknown uuid is a 404; a successful delete is +// 204 (idempotent — a second cancel sees the record gone and returns 404). +func (reg *Registry) cancelUpload(w http.ResponseWriter, _, uploadID string) { + st, _, getErr := reg.Meta.GetUpload(reg.AccountID, uploadID) + err := reg.Meta.DeleteUpload(reg.AccountID, uploadID) + switch { + case errors.Is(err, ecr.ErrNotFound): + WriteError(w, http.StatusNotFound, "BLOB_UPLOAD_UNKNOWN", "upload not found") + return + case err != nil: + reg.internal(w, "cancel upload", err) + return + } + if getErr == nil && st.BytesKey != "" { + if delErr := reg.deleteUploadBytesAt(st.BytesKey); delErr != nil { + slog.Error("ECR/OCI: cancel upload temp cleanup failed", "uploadID", uploadID, "err", delErr) + } + } + w.WriteHeader(http.StatusNoContent) +} + +// cleanupUpload drops the upload record and its committed temp bytes. Used on +// finalize success and on finalize failure so an aborted upload leaves neither +// an orphaned record (which would block the uuid) nor partial blob bytes. +func (reg *Registry) cleanupUpload(uploadID string) { + st, _, getErr := reg.Meta.GetUpload(reg.AccountID, uploadID) + if err := reg.Meta.DeleteUpload(reg.AccountID, uploadID); err != nil && !errors.Is(err, ecr.ErrNotFound) { + slog.Error("ECR/OCI: upload record cleanup failed", "uploadID", uploadID, "err", err) + } + if getErr == nil && st.BytesKey != "" { + if err := reg.deleteUploadBytesAt(st.BytesKey); err != nil { + slog.Error("ECR/OCI: upload temp cleanup failed", "uploadID", uploadID, "err", err) + } + } +} + +func (reg *Registry) writeBlobCreated(w http.ResponseWriter, name, digest string) { + w.Header().Set("Location", fmt.Sprintf("/v2/%s/blobs/%s", name, digest)) + w.Header().Set("Docker-Content-Digest", digest) + w.WriteHeader(http.StatusCreated) +} + +// ---- manifests ---- + +func (reg *Registry) routeManifests(w http.ResponseWriter, r *http.Request, name, reference string) { + switch r.Method { + case http.MethodHead: + reg.headManifest(w, r, name, reference) + case http.MethodGet: + reg.getManifest(w, r, name, reference) + case http.MethodPut: + reg.putManifest(w, r, name, reference) + case http.MethodDelete: + reg.deleteManifest(w, name, reference) + default: + WriteError(w, http.StatusMethodNotAllowed, "UNSUPPORTED", "unsupported manifest method") + } +} + +// resolveManifestDigest maps a reference (tag or digest) to a stored digest. +func (reg *Registry) resolveManifestDigest(name, reference string) (string, bool) { + if ecr.ValidateDigest(reference) { + return reference, true + } + digest, err := reg.Meta.GetTag(reg.AccountID, name, reference) + if err != nil { + return "", false + } + return digest, true +} + +func (reg *Registry) headManifest(w http.ResponseWriter, _ *http.Request, name, reference string) { + digest, ok := reg.resolveManifestDigest(name, reference) + if !ok { + WriteError(w, http.StatusNotFound, "MANIFEST_UNKNOWN", "manifest unknown") + return + } + meta, err := reg.Meta.GetManifestMeta(reg.AccountID, name, digest) + if err != nil { + WriteError(w, http.StatusNotFound, "MANIFEST_UNKNOWN", "manifest unknown") + return + } + w.Header().Set("Docker-Content-Digest", digest) + w.Header().Set("Content-Type", meta.MediaType) + w.Header().Set("Content-Length", strconv.FormatInt(meta.Size, 10)) + w.WriteHeader(http.StatusOK) +} + +func (reg *Registry) getManifest(w http.ResponseWriter, r *http.Request, name, reference string) { + digest, ok := reg.resolveManifestDigest(name, reference) + if !ok { + WriteError(w, http.StatusNotFound, "MANIFEST_UNKNOWN", "manifest unknown") + return + } + meta, err := reg.Meta.GetManifestMeta(reg.AccountID, name, digest) + if err != nil { + WriteError(w, http.StatusNotFound, "MANIFEST_UNKNOWN", "manifest unknown") + return + } + if !acceptsType(r.Header.Get("Accept"), meta.MediaType) { + WriteError(w, http.StatusNotAcceptable, "MANIFEST_INVALID", "no acceptable manifest media type") + return + } + out, err := reg.Store.GetObject(&s3.GetObjectInput{ + Bucket: aws.String(reg.bucket()), + Key: aws.String(ecr.ManifestKey(name, digest)), + }) + if err != nil { + if objectstore.IsNoSuchKeyError(err) { + WriteError(w, http.StatusNotFound, "MANIFEST_UNKNOWN", "manifest unknown") + return + } + reg.internal(w, "get manifest", err) + return + } + defer func() { _ = out.Body.Close() }() + body, err := io.ReadAll(out.Body) + if err != nil { + reg.internal(w, "read manifest", err) + return + } + w.Header().Set("Docker-Content-Digest", digest) + w.Header().Set("Content-Type", meta.MediaType) + w.Header().Set("Content-Length", strconv.Itoa(len(body))) + w.WriteHeader(http.StatusOK) + if _, err := w.Write(body); err != nil { + slog.Error("ECR/OCI: manifest write failed", "err", err) + } +} + +// putManifest validates, stores and tags a manifest. Image manifests are checked +// to ensure every referenced blob exists in the pool; image indexes are checked +// to ensure every referenced child manifest exists with a matching mediaType. +func (reg *Registry) putManifest(w http.ResponseWriter, r *http.Request, name, reference string) { + if err := reg.ensureRepo(name); err != nil { + reg.internal(w, "ensure repo", err) + return + } + + body, err := io.ReadAll(io.LimitReader(r.Body, maxManifestBytes+1)) + if err != nil { + reg.internal(w, "read manifest", err) + return + } + if len(body) > maxManifestBytes { + WriteError(w, http.StatusRequestEntityTooLarge, "MANIFEST_INVALID", "manifest too large") + return + } + + contentType := r.Header.Get("Content-Type") + if contentType == "" { + contentType = detectManifestType(body) + } + + digest := "sha256:" + hex.EncodeToString(sha256Sum(body)) + if ecr.ValidateDigest(reference) && reference != digest { + WriteError(w, http.StatusBadRequest, "DIGEST_INVALID", "reference digest does not match content") + return + } + + children, code, vErr := reg.validateManifest(name, contentType, body) + if vErr != nil { + WriteError(w, http.StatusBadRequest, code, vErr.Error()) + return + } + + if _, err := reg.Store.PutObject(&s3.PutObjectInput{ + Bucket: aws.String(reg.bucket()), + Key: aws.String(ecr.ManifestKey(name, digest)), + Body: aws.ReadSeekCloser(strings.NewReader(string(body))), + ContentType: aws.String(contentType), + }); err != nil { + reg.internal(w, "store manifest", err) + return + } + + if err := reg.Meta.PutManifestMeta(reg.AccountID, name, ecr.ManifestMeta{ + Digest: digest, + MediaType: contentType, + Size: int64(len(body)), + PushedAt: time.Now().UTC(), + ChildDigests: children, + }); err != nil { + reg.internal(w, "store manifest meta", err) + return + } + + if !ecr.ValidateDigest(reference) { + if err := reg.Meta.PutTag(reg.AccountID, name, reference, digest); err != nil { + reg.internal(w, "store tag", err) + return + } + } + + w.Header().Set("Docker-Content-Digest", digest) + w.Header().Set("Location", fmt.Sprintf("/v2/%s/manifests/%s", name, digest)) + w.WriteHeader(http.StatusCreated) +} + +func (reg *Registry) deleteManifest(w http.ResponseWriter, name, reference string) { + // Tag pointer deletion only; digest GC is out of scope. + if ecr.ValidateDigest(reference) { + w.WriteHeader(http.StatusAccepted) + return + } + err := reg.Meta.DeleteTag(reg.AccountID, name, reference) + if err != nil { + if errors.Is(err, ecr.ErrNotFound) { + WriteError(w, http.StatusNotFound, "MANIFEST_UNKNOWN", "manifest unknown") + return + } + reg.internal(w, "delete tag", err) + return + } + w.WriteHeader(http.StatusAccepted) +} + +// manifestDoc is the subset of an OCI image manifest the validator inspects. +type manifestDoc struct { + MediaType string `json:"mediaType"` + Config struct { + Digest string `json:"digest"` + } `json:"config"` + Layers []struct { + Digest string `json:"digest"` + } `json:"layers"` + Manifests []struct { + Digest string `json:"digest"` + MediaType string `json:"mediaType"` + } `json:"manifests"` +} + +// validateManifest performs type-aware shallow validation, returning the child +// digests recorded in the manifest metadata. For an image manifest every config +// and layer blob must exist; for an index every child manifest must exist with a +// matching stored mediaType. +func (reg *Registry) validateManifest(name, contentType string, body []byte) ([]string, string, error) { + var doc manifestDoc + if err := json.Unmarshal(body, &doc); err != nil { + return nil, "MANIFEST_INVALID", fmt.Errorf("manifest is not valid JSON: %w", err) + } + + switch contentType { + case mediaTypeDockerManifestList, mediaTypeOCIIndex: + var children []string + for _, m := range doc.Manifests { + if !ecr.ValidateDigest(m.Digest) { + return nil, "MANIFEST_INVALID", errors.New("index references malformed digest") + } + meta, err := reg.Meta.GetManifestMeta(reg.AccountID, name, m.Digest) + if err != nil { + return nil, "MANIFEST_BLOB_UNKNOWN", fmt.Errorf("referenced manifest %s not found", m.Digest) + } + if m.MediaType != "" && meta.MediaType != m.MediaType { + return nil, "MANIFEST_INVALID", fmt.Errorf("referenced manifest %s media type mismatch", m.Digest) + } + children = append(children, m.Digest) + } + return children, "", nil + default: + var children []string + refs := make([]string, 0, len(doc.Layers)+1) + if doc.Config.Digest != "" { + refs = append(refs, doc.Config.Digest) + } + for _, l := range doc.Layers { + refs = append(refs, l.Digest) + } + for _, d := range refs { + if !ecr.ValidateDigest(d) { + return nil, "MANIFEST_INVALID", errors.New("manifest references malformed digest") + } + if !reg.blobExists(d) { + return nil, "MANIFEST_BLOB_UNKNOWN", fmt.Errorf("referenced blob %s not found", d) + } + children = append(children, d) + } + return children, "", nil + } +} + +// ---- tags ---- + +func (reg *Registry) routeTags(w http.ResponseWriter, r *http.Request, name, ref string) { + if r.Method != http.MethodGet || ref != "list" { + WriteError(w, http.StatusNotFound, "UNSUPPORTED", "unsupported tags operation") + return + } + if _, err := reg.Meta.GetRepo(reg.AccountID, name); err != nil { + WriteError(w, http.StatusNotFound, "NAME_UNKNOWN", "repository unknown") + return + } + tags, err := reg.Meta.ListTags(reg.AccountID, name) + if err != nil { + reg.internal(w, "list tags", err) + return + } + if tags == nil { + tags = []string{} + } + reg.writeJSON(w, http.StatusOK, struct { + Name string `json:"name"` + Tags []string `json:"tags"` + }{Name: name, Tags: tags}) +} + +// ---- catalog ---- + +func (reg *Registry) handleCatalog(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + WriteError(w, http.StatusMethodNotAllowed, "UNSUPPORTED", "unsupported catalog method") + return + } + repos, err := reg.Meta.ListRepos(reg.AccountID) + if err != nil { + reg.internal(w, "list repos", err) + return + } + if repos == nil { + repos = []string{} + } + reg.writeJSON(w, http.StatusOK, struct { + Repositories []string `json:"repositories"` + }{Repositories: repos}) +} + +// ---- helpers ---- + +func (reg *Registry) bucket() string { return ecr.AccountBucket(reg.AccountID) } + +// ensureBucket provisions the account's predastore bucket on first use. ECR +// storage is account-managed: the bucket appears transparently rather than +// being created by an operator. The result is cached once provisioned; a +// backend failure is left uncached so the next request retries. +func (reg *Registry) ensureBucket() error { + reg.bucketMu.Lock() + defer reg.bucketMu.Unlock() + if reg.bucketReady { + return nil + } + if err := reg.Store.EnsureBucket(reg.bucket()); err != nil { + return err + } + reg.bucketReady = true + return nil +} + +func (reg *Registry) ensureRepo(name string) error { + if _, err := reg.Meta.GetRepo(reg.AccountID, name); err == nil { + return nil + } else if !errors.Is(err, ecr.ErrNotFound) { + return err + } + return reg.Meta.PutRepo(reg.AccountID, ecr.RepoMeta{Name: name, CreatedAt: time.Now().UTC()}) +} + +func (reg *Registry) blobExists(digest string) bool { + _, err := reg.Store.HeadObject(&s3.HeadObjectInput{ + Bucket: aws.String(reg.bucket()), + Key: aws.String(ecr.BlobKey(digest)), + }) + return err == nil +} + +// readUploadBytesAt returns the bytes at key, or nil for an empty key or a miss +// (a fresh upload has no committed bytes yet). +func (reg *Registry) readUploadBytesAt(key string) []byte { + if key == "" { + return nil + } + out, err := reg.Store.GetObject(&s3.GetObjectInput{ + Bucket: aws.String(reg.bucket()), + Key: aws.String(key), + }) + if err != nil { + return nil + } + defer func() { _ = out.Body.Close() }() + data, err := io.ReadAll(out.Body) + if err != nil { + return nil + } + return data +} + +func (reg *Registry) putUploadBytesAt(key string, data []byte) error { + _, err := reg.Store.PutObject(&s3.PutObjectInput{ + Bucket: aws.String(reg.bucket()), + Key: aws.String(key), + Body: aws.ReadSeekCloser(strings.NewReader(string(data))), + }) + return err +} + +func (reg *Registry) deleteUploadBytesAt(key string) error { + if key == "" { + return nil + } + _, err := reg.Store.DeleteObject(&s3.DeleteObjectInput{ + Bucket: aws.String(reg.bucket()), + Key: aws.String(key), + }) + return err +} + +func (reg *Registry) writeJSON(w http.ResponseWriter, status int, v any) { + body, err := json.Marshal(v) + if err != nil { + reg.internal(w, "marshal json", err) + return + } + w.Header().Set("Content-Type", OCIContentType) + w.WriteHeader(status) + if _, err := w.Write(body); err != nil { + slog.Error("ECR/OCI: json write failed", "err", err) + } +} + +func (reg *Registry) internal(w http.ResponseWriter, op string, err error) { + slog.Error("ECR/OCI: "+op+" failed", "err", err) + WriteError(w, http.StatusInternalServerError, "UNKNOWN", "internal registry error") +} + +func uploadPath(name, uploadID string) string { + return fmt.Sprintf("/v2/%s/blobs/uploads/%s", name, uploadID) +} + +func sha256Sum(b []byte) []byte { + sum := sha256.Sum256(b) + return sum[:] +} + +// marshalHash snapshots a running sha256 state via its BinaryMarshaler so a +// chunked upload resumes without re-reading committed bytes. +func marshalHash(h hash.Hash) ([]byte, error) { + m, ok := h.(encoding.BinaryMarshaler) + if !ok { + return nil, errors.New("sha256 hash does not support marshaling") + } + return m.MarshalBinary() +} + +func unmarshalHash(state []byte) (hash.Hash, error) { + h := sha256.New() + u, ok := h.(encoding.BinaryUnmarshaler) + if !ok { + return nil, errors.New("sha256 hash does not support unmarshaling") + } + if err := u.UnmarshalBinary(state); err != nil { + return nil, err + } + return h, nil +} + +// acceptsType reports whether an Accept header permits the stored media type. An +// empty Accept (e.g. a bare docker pull) accepts anything. +func acceptsType(accept, mediaType string) bool { + if strings.TrimSpace(accept) == "" { + return true + } + for part := range strings.SplitSeq(accept, ",") { + t := strings.TrimSpace(part) + if i := strings.Index(t, ";"); i >= 0 { + t = strings.TrimSpace(t[:i]) + } + if t == "*/*" || t == "application/*" || t == mediaType { + return true + } + } + return false +} + +// detectManifestType infers a media type from the document body when the client +// omits Content-Type, distinguishing an index from a single image manifest. +func detectManifestType(body []byte) string { + var probe struct { + MediaType string `json:"mediaType"` + Manifests []any `json:"manifests"` + } + if err := json.Unmarshal(body, &probe); err == nil { + if probe.MediaType != "" { + return probe.MediaType + } + if len(probe.Manifests) > 0 { + return mediaTypeOCIIndex + } + } + return defaultManifestContentType +} diff --git a/spinifex/gateway/ecr/registry_ensurebucket_test.go b/spinifex/gateway/ecr/registry_ensurebucket_test.go new file mode 100644 index 000000000..4c2c87d1b --- /dev/null +++ b/spinifex/gateway/ecr/registry_ensurebucket_test.go @@ -0,0 +1,46 @@ +package gateway_ecr + +import ( + "errors" + "net/http" + "testing" + + "github.com/mulgadc/spinifex/spinifex/handlers/ecr" + "github.com/mulgadc/spinifex/spinifex/objectstore" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// countingStore wraps MemoryObjectStore to observe and optionally fail the +// bucket-ensure path. +type countingStore struct { + *objectstore.MemoryObjectStore + + ensureCalls int + ensureErr error +} + +func (c *countingStore) EnsureBucket(bucket string) error { + c.ensureCalls++ + if c.ensureErr != nil { + return c.ensureErr + } + return c.MemoryObjectStore.EnsureBucket(bucket) +} + +func TestRegistry_EnsureBucketFailureReturns500(t *testing.T) { + store := &countingStore{MemoryObjectStore: objectstore.NewMemoryObjectStore(), ensureErr: errors.New("predastore down")} + reg := NewRegistry(store, ecr.NewMemoryMetaStore(), testAccount) + + w := do(reg, http.MethodGet, "/v2/myrepo/tags/list", nil, nil) + assert.Equal(t, http.StatusInternalServerError, w.Code) +} + +func TestRegistry_EnsureBucketCachedAfterSuccess(t *testing.T) { + store := &countingStore{MemoryObjectStore: objectstore.NewMemoryObjectStore()} + reg := NewRegistry(store, ecr.NewMemoryMetaStore(), testAccount) + + do(reg, http.MethodGet, "/v2/myrepo/tags/list", nil, nil) + do(reg, http.MethodGet, "/v2/other/tags/list", nil, nil) + require.Equal(t, 1, store.ensureCalls, "bucket ensure must be cached after first success") +} diff --git a/spinifex/gateway/ecr/registry_test.go b/spinifex/gateway/ecr/registry_test.go new file mode 100644 index 000000000..e4dd37c04 --- /dev/null +++ b/spinifex/gateway/ecr/registry_test.go @@ -0,0 +1,376 @@ +package gateway_ecr + +import ( + "crypto/sha256" + "encoding/hex" + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "strconv" + "strings" + "testing" + + "github.com/mulgadc/spinifex/spinifex/handlers/ecr" + "github.com/mulgadc/spinifex/spinifex/objectstore" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +const testAccount = "000000000000" + +func newTestRegistry() *Registry { + return NewRegistry(objectstore.NewMemoryObjectStore(), ecr.NewMemoryMetaStore(), testAccount) +} + +func digestOf(b []byte) string { + sum := sha256.Sum256(b) + return "sha256:" + hex.EncodeToString(sum[:]) +} + +func do(reg *Registry, method, path string, body []byte, hdr map[string]string) *httptest.ResponseRecorder { + var r *http.Request + if body != nil { + r = httptest.NewRequest(method, path, strings.NewReader(string(body))) + } else { + r = httptest.NewRequest(method, path, nil) + } + for k, v := range hdr { + r.Header.Set(k, v) + } + w := httptest.NewRecorder() + reg.ServeHTTP(w, r) + return w +} + +// pushBlob runs the monolithic upload-start + PUT?digest round-trip. +func pushBlob(t *testing.T, reg *Registry, repo string, data []byte) string { + t.Helper() + w := do(reg, http.MethodPost, "/v2/"+repo+"/blobs/uploads/", nil, nil) + require.Equal(t, http.StatusAccepted, w.Code) + loc := w.Header().Get("Location") + require.NotEmpty(t, loc) + + dg := digestOf(data) + w = do(reg, http.MethodPut, loc+"?digest="+dg, data, nil) + require.Equal(t, http.StatusCreated, w.Code, "body: %s", w.Body.String()) + assert.Equal(t, dg, w.Header().Get("Docker-Content-Digest")) + return dg +} + +func TestBlobUpload_Monolithic_And_Head_Get(t *testing.T) { + reg := newTestRegistry() + data := []byte("hello layer bytes") + dg := pushBlob(t, reg, "team/app", data) + + w := do(reg, http.MethodHead, "/v2/team/app/blobs/"+dg, nil, nil) + assert.Equal(t, http.StatusOK, w.Code) + assert.Equal(t, strconv.Itoa(len(data)), w.Header().Get("Content-Length")) + assert.Equal(t, dg, w.Header().Get("Docker-Content-Digest")) + + w = do(reg, http.MethodGet, "/v2/team/app/blobs/"+dg, nil, nil) + assert.Equal(t, http.StatusOK, w.Code) + assert.Equal(t, data, w.Body.Bytes()) +} + +func TestBlobHead_NotFound(t *testing.T) { + reg := newTestRegistry() + w := do(reg, http.MethodHead, "/v2/team/app/blobs/"+digestOf([]byte("absent")), nil, nil) + assert.Equal(t, http.StatusNotFound, w.Code) +} + +func TestBlobGet_MalformedDigest(t *testing.T) { + reg := newTestRegistry() + w := do(reg, http.MethodGet, "/v2/team/app/blobs/sha256:zzz", nil, nil) + assert.Equal(t, http.StatusBadRequest, w.Code) + assertCode(t, w, "DIGEST_INVALID") +} + +func TestBlobUpload_Chunked(t *testing.T) { + reg := newTestRegistry() + c1 := []byte("first-chunk-") + c2 := []byte("second-chunk") + full := append(append([]byte{}, c1...), c2...) + dg := digestOf(full) + + w := do(reg, http.MethodPost, "/v2/r/repo/blobs/uploads/", nil, nil) + require.Equal(t, http.StatusAccepted, w.Code) + loc := w.Header().Get("Location") + + w = do(reg, http.MethodPatch, loc, c1, nil) + require.Equal(t, http.StatusAccepted, w.Code) + assert.Equal(t, fmt.Sprintf("0-%d", len(c1)-1), w.Header().Get("Range")) + + w = do(reg, http.MethodPatch, loc, c2, nil) + require.Equal(t, http.StatusAccepted, w.Code) + + w = do(reg, http.MethodPut, loc+"?digest="+dg, nil, nil) + require.Equal(t, http.StatusCreated, w.Code, "body: %s", w.Body.String()) + + w = do(reg, http.MethodGet, "/v2/r/repo/blobs/"+dg, nil, nil) + require.Equal(t, http.StatusOK, w.Code) + assert.Equal(t, full, w.Body.Bytes()) +} + +func TestBlobUpload_DigestMismatch(t *testing.T) { + reg := newTestRegistry() + w := do(reg, http.MethodPost, "/v2/r/repo/blobs/uploads/", nil, nil) + loc := w.Header().Get("Location") + w = do(reg, http.MethodPut, loc+"?digest="+digestOf([]byte("wrong")), []byte("actual"), nil) + assert.Equal(t, http.StatusBadRequest, w.Code) + assertCode(t, w, "DIGEST_INVALID") +} + +func TestBlobUpload_Cancel(t *testing.T) { + reg := newTestRegistry() + w := do(reg, http.MethodPost, "/v2/r/repo/blobs/uploads/", nil, nil) + loc := w.Header().Get("Location") + + w = do(reg, http.MethodDelete, loc, nil, nil) + assert.Equal(t, http.StatusNoContent, w.Code) + + // Second cancel is unknown. + w = do(reg, http.MethodDelete, loc, nil, nil) + assert.Equal(t, http.StatusNotFound, w.Code) +} + +func TestBlobMount_HitAndMiss(t *testing.T) { + reg := newTestRegistry() + data := []byte("shared-layer") + dg := pushBlob(t, reg, "team/source", data) + + // Mount hit: blob already in account pool -> 201 without upload. + w := do(reg, http.MethodPost, + "/v2/team/dest/blobs/uploads/?mount="+dg+"&from=team/source", nil, nil) + assert.Equal(t, http.StatusCreated, w.Code) + assert.Equal(t, dg, w.Header().Get("Docker-Content-Digest")) + + // Mount miss: unknown digest -> falls back to upload start (202). + w = do(reg, http.MethodPost, + "/v2/team/dest/blobs/uploads/?mount="+digestOf([]byte("nope"))+"&from=team/source", nil, nil) + assert.Equal(t, http.StatusAccepted, w.Code) +} + +func TestManifest_PutGetHead_ImageManifest(t *testing.T) { + reg := newTestRegistry() + repo := "team/app" + + cfg := []byte(`{"config":true}`) + layer := []byte("layer-bytes-xyz") + cfgDg := pushBlob(t, reg, repo, cfg) + layerDg := pushBlob(t, reg, repo, layer) + + manifest := fmt.Appendf(nil, + `{"schemaVersion":2,"mediaType":"%s","config":{"digest":"%s"},"layers":[{"digest":"%s"}]}`, + mediaTypeDockerManifest, cfgDg, layerDg) + mdg := digestOf(manifest) + + w := do(reg, http.MethodPut, "/v2/"+repo+"/manifests/v1", manifest, + map[string]string{"Content-Type": mediaTypeDockerManifest}) + require.Equal(t, http.StatusCreated, w.Code, "body: %s", w.Body.String()) + assert.Equal(t, mdg, w.Header().Get("Docker-Content-Digest")) + + // HEAD by tag. + w = do(reg, http.MethodHead, "/v2/"+repo+"/manifests/v1", nil, nil) + assert.Equal(t, http.StatusOK, w.Code) + assert.Equal(t, mdg, w.Header().Get("Docker-Content-Digest")) + + // GET by digest returns bytes + stored content type. + w = do(reg, http.MethodGet, "/v2/"+repo+"/manifests/"+mdg, nil, nil) + assert.Equal(t, http.StatusOK, w.Code) + assert.Equal(t, manifest, w.Body.Bytes()) + assert.Equal(t, mediaTypeDockerManifest, w.Header().Get("Content-Type")) +} + +func TestManifest_Put_MissingBlob(t *testing.T) { + reg := newTestRegistry() + repo := "team/app" + manifest := fmt.Appendf(nil, + `{"mediaType":"%s","config":{"digest":"%s"},"layers":[]}`, + mediaTypeDockerManifest, digestOf([]byte("nope"))) + w := do(reg, http.MethodPut, "/v2/"+repo+"/manifests/v1", manifest, + map[string]string{"Content-Type": mediaTypeDockerManifest}) + assert.Equal(t, http.StatusBadRequest, w.Code) + assertCode(t, w, "MANIFEST_BLOB_UNKNOWN") +} + +func TestManifest_Get_AcceptMismatch(t *testing.T) { + reg := newTestRegistry() + repo := "team/app" + cfg := pushBlob(t, reg, repo, []byte("c")) + manifest := fmt.Appendf(nil, + `{"mediaType":"%s","config":{"digest":"%s"},"layers":[]}`, mediaTypeDockerManifest, cfg) + do(reg, http.MethodPut, "/v2/"+repo+"/manifests/v1", manifest, + map[string]string{"Content-Type": mediaTypeDockerManifest}) + + w := do(reg, http.MethodGet, "/v2/"+repo+"/manifests/v1", nil, + map[string]string{"Accept": mediaTypeOCIIndex}) + assert.Equal(t, http.StatusNotAcceptable, w.Code) +} + +func TestManifest_Get_Unknown(t *testing.T) { + reg := newTestRegistry() + w := do(reg, http.MethodGet, "/v2/team/app/manifests/missing", nil, nil) + assert.Equal(t, http.StatusNotFound, w.Code) + assertCode(t, w, "MANIFEST_UNKNOWN") +} + +func TestManifest_ImageIndex_ChildValidation(t *testing.T) { + reg := newTestRegistry() + repo := "team/app" + + // Push a child image manifest first. + cfg := pushBlob(t, reg, repo, []byte("cfg")) + child := fmt.Appendf(nil, + `{"mediaType":"%s","config":{"digest":"%s"},"layers":[]}`, mediaTypeOCIManifest, cfg) + childDg := digestOf(child) + w := do(reg, http.MethodPut, "/v2/"+repo+"/manifests/"+childDg, child, + map[string]string{"Content-Type": mediaTypeOCIManifest}) + require.Equal(t, http.StatusCreated, w.Code) + + // Index referencing the child with matching mediaType. + index := fmt.Appendf(nil, + `{"mediaType":"%s","manifests":[{"digest":"%s","mediaType":"%s"}]}`, + mediaTypeOCIIndex, childDg, mediaTypeOCIManifest) + w = do(reg, http.MethodPut, "/v2/"+repo+"/manifests/latest", index, + map[string]string{"Content-Type": mediaTypeOCIIndex}) + require.Equal(t, http.StatusCreated, w.Code, "body: %s", w.Body.String()) + + // Index referencing a missing child -> MANIFEST_BLOB_UNKNOWN. + bad := fmt.Appendf(nil, + `{"mediaType":"%s","manifests":[{"digest":"%s","mediaType":"%s"}]}`, + mediaTypeOCIIndex, digestOf([]byte("absent")), mediaTypeOCIManifest) + w = do(reg, http.MethodPut, "/v2/"+repo+"/manifests/bad", bad, + map[string]string{"Content-Type": mediaTypeOCIIndex}) + assert.Equal(t, http.StatusBadRequest, w.Code) + assertCode(t, w, "MANIFEST_BLOB_UNKNOWN") + + // Index with mediaType mismatch -> MANIFEST_INVALID. + mismatch := fmt.Appendf(nil, + `{"mediaType":"%s","manifests":[{"digest":"%s","mediaType":"%s"}]}`, + mediaTypeOCIIndex, childDg, mediaTypeDockerManifest) + w = do(reg, http.MethodPut, "/v2/"+repo+"/manifests/mm", mismatch, + map[string]string{"Content-Type": mediaTypeOCIIndex}) + assert.Equal(t, http.StatusBadRequest, w.Code) + assertCode(t, w, "MANIFEST_INVALID") +} + +func TestManifest_Delete_Tag(t *testing.T) { + reg := newTestRegistry() + repo := "team/app" + cfg := pushBlob(t, reg, repo, []byte("c")) + manifest := fmt.Appendf(nil, + `{"mediaType":"%s","config":{"digest":"%s"},"layers":[]}`, mediaTypeDockerManifest, cfg) + do(reg, http.MethodPut, "/v2/"+repo+"/manifests/v1", manifest, + map[string]string{"Content-Type": mediaTypeDockerManifest}) + + w := do(reg, http.MethodDelete, "/v2/"+repo+"/manifests/v1", nil, nil) + assert.Equal(t, http.StatusAccepted, w.Code) + + w = do(reg, http.MethodDelete, "/v2/"+repo+"/manifests/v1", nil, nil) + assert.Equal(t, http.StatusNotFound, w.Code) +} + +func TestTagsList(t *testing.T) { + reg := newTestRegistry() + repo := "team/app" + cfg := pushBlob(t, reg, repo, []byte("c")) + manifest := fmt.Appendf(nil, + `{"mediaType":"%s","config":{"digest":"%s"},"layers":[]}`, mediaTypeDockerManifest, cfg) + for _, tag := range []string{"v1", "v2"} { + do(reg, http.MethodPut, "/v2/"+repo+"/manifests/"+tag, manifest, + map[string]string{"Content-Type": mediaTypeDockerManifest}) + } + + w := do(reg, http.MethodGet, "/v2/"+repo+"/tags/list", nil, nil) + require.Equal(t, http.StatusOK, w.Code) + var resp struct { + Name string `json:"name"` + Tags []string `json:"tags"` + } + require.NoError(t, json.Unmarshal(w.Body.Bytes(), &resp)) + assert.Equal(t, repo, resp.Name) + assert.ElementsMatch(t, []string{"v1", "v2"}, resp.Tags) +} + +func TestTagsList_UnknownRepo(t *testing.T) { + reg := newTestRegistry() + w := do(reg, http.MethodGet, "/v2/team/ghost/tags/list", nil, nil) + assert.Equal(t, http.StatusNotFound, w.Code) +} + +func TestCatalog(t *testing.T) { + reg := newTestRegistry() + pushBlob(t, reg, "team/a", []byte("a")) + pushBlob(t, reg, "team/b", []byte("b")) + + w := do(reg, http.MethodGet, "/v2/_catalog", nil, nil) + require.Equal(t, http.StatusOK, w.Code) + var resp struct { + Repositories []string `json:"repositories"` + } + require.NoError(t, json.Unmarshal(w.Body.Bytes(), &resp)) + assert.ElementsMatch(t, []string{"team/a", "team/b"}, resp.Repositories) +} + +func TestDispatch_InvalidRepoName(t *testing.T) { + reg := newTestRegistry() + w := do(reg, http.MethodGet, "/v2/Team/App/tags/list", nil, nil) + assert.Equal(t, http.StatusBadRequest, w.Code) + assertCode(t, w, "NAME_INVALID") +} + +func TestDispatch_UnknownPath(t *testing.T) { + reg := newTestRegistry() + w := do(reg, http.MethodGet, "/v2/team/app/frobnicate", nil, nil) + assert.Equal(t, http.StatusNotFound, w.Code) +} + +func TestSplitV2Path(t *testing.T) { + name, kind, ref, ok := splitV2Path("team/app/blobs/sha256:abc") + require.True(t, ok) + assert.Equal(t, "team/app", name) + assert.Equal(t, "blobs", kind) + assert.Equal(t, "sha256:abc", ref) + + name, kind, ref, ok = splitV2Path("a/b/c/manifests/latest") + require.True(t, ok) + assert.Equal(t, "a/b/c", name) + assert.Equal(t, "manifests", kind) + assert.Equal(t, "latest", ref) + + _, _, _, ok = splitV2Path("noslashmarker") + assert.False(t, ok) +} + +func TestAcceptsType(t *testing.T) { + assert.True(t, acceptsType("", mediaTypeDockerManifest)) + assert.True(t, acceptsType("*/*", mediaTypeDockerManifest)) + assert.True(t, acceptsType("application/*", mediaTypeDockerManifest)) + assert.True(t, acceptsType(mediaTypeOCIIndex+", "+mediaTypeDockerManifest, mediaTypeDockerManifest)) + assert.False(t, acceptsType(mediaTypeOCIIndex, mediaTypeDockerManifest)) +} + +func TestDetectManifestType(t *testing.T) { + assert.Equal(t, mediaTypeOCIIndex, + detectManifestType([]byte(`{"manifests":[{"digest":"x"}]}`))) + assert.Equal(t, mediaTypeDockerManifest, + detectManifestType([]byte(`{"config":{"digest":"x"}}`))) + assert.Equal(t, mediaTypeOCIManifest, + detectManifestType([]byte(`{"mediaType":"`+mediaTypeOCIManifest+`"}`))) +} + +func assertCode(t *testing.T, w *httptest.ResponseRecorder, code string) { + t.Helper() + var body struct { + Errors []struct { + Code string `json:"code"` + } `json:"errors"` + } + require.NoError(t, json.Unmarshal(w.Body.Bytes(), &body)) + require.NotEmpty(t, body.Errors) + assert.Equal(t, code, body.Errors[0].Code) +} + +// memoryObjectStore satisfies the ObjectStore interface for completeness check. +var _ objectstore.ObjectStore = objectstore.NewMemoryObjectStore() diff --git a/spinifex/gateway/ecr/registry_upload_test.go b/spinifex/gateway/ecr/registry_upload_test.go new file mode 100644 index 000000000..1d64199c4 --- /dev/null +++ b/spinifex/gateway/ecr/registry_upload_test.go @@ -0,0 +1,147 @@ +package gateway_ecr + +import ( + "errors" + "net/http" + "strings" + "sync" + "testing" + + "github.com/aws/aws-sdk-go/service/s3" + "github.com/mulgadc/spinifex/spinifex/handlers/ecr" + "github.com/mulgadc/spinifex/spinifex/objectstore" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// blobPutFailStore wraps a MemoryObjectStore and fails PutObject for blob-pool +// keys, simulating a predastore write error during upload finalize. +type blobPutFailStore struct { + *objectstore.MemoryObjectStore +} + +func (s *blobPutFailStore) PutObject(input *s3.PutObjectInput) (*s3.PutObjectOutput, error) { + if input.Key != nil && strings.HasPrefix(*input.Key, "blobs/") { + return nil, errors.New("predastore unavailable") + } + return s.MemoryObjectStore.PutObject(input) +} + +// TestFinishUpload_StoreFailure_CleansUp proves that when the blob PutObject +// fails on finalize, neither the temp upload object nor the upload KV record is +// left orphaned (so the uuid is reusable and no partial blob lingers). +func TestFinishUpload_StoreFailure_CleansUp(t *testing.T) { + mem := objectstore.NewMemoryObjectStore() + store := &blobPutFailStore{MemoryObjectStore: mem} + meta := ecr.NewMemoryMetaStore() + reg := NewRegistry(store, meta, testAccount) + + w := do(reg, http.MethodPost, "/v2/team/app/blobs/uploads/", nil, nil) + require.Equal(t, http.StatusAccepted, w.Code) + loc := w.Header().Get("Location") + + // PATCH a chunk so a temp upload object exists, then finalize (empty body) + // where the blob PutObject fails — both the temp object and the upload + // record must be cleaned up. + data := []byte("a blob that will fail to store") + w = do(reg, http.MethodPatch, loc, data, nil) + require.Equal(t, http.StatusAccepted, w.Code) + + dg := digestOf(data) + w = do(reg, http.MethodPut, loc+"?digest="+dg, nil, nil) + require.Equal(t, http.StatusInternalServerError, w.Code) + + uploadID := strings.TrimPrefix(loc, "/v2/team/app/blobs/uploads/") + _, _, err := meta.GetUpload(testAccount, uploadID) + assert.ErrorIs(t, err, ecr.ErrNotFound, "upload record must be cleaned up after finalize failure") + + listed, err := mem.ListObjectsV2(&s3.ListObjectsV2Input{ + Bucket: awsString(ecr.AccountBucket(testAccount)), + Prefix: awsString("uploads/"), + }) + require.NoError(t, err) + assert.Empty(t, listed.Contents, "no temp upload bytes may linger after finalize failure") +} + +func awsString(s string) *string { return &s } + +// TestUploadCAS_SameRevisionSingleWinner proves the serialization point: two +// updates issued from the same observed revision cannot both commit. One wins, +// the other gets ErrConflict — so a racing PATCH is rejected, never merged. +func TestUploadCAS_SameRevisionSingleWinner(t *testing.T) { + meta := ecr.NewMemoryMetaStore() + rev, err := meta.PutUpload(testAccount, "u-race", ecr.UploadState{RepoName: "r"}) + require.NoError(t, err) + + stA := ecr.UploadState{RepoName: "r", CommittedBytes: 8, BytesKey: "uploads/u-race/a"} + stB := ecr.UploadState{RepoName: "r", CommittedBytes: 8, BytesKey: "uploads/u-race/b"} + + var wg sync.WaitGroup + errs := make([]error, 2) + for i, st := range []ecr.UploadState{stA, stB} { + wg.Add(1) + go func(idx int, s ecr.UploadState) { + defer wg.Done() + _, errs[idx] = meta.UpdateUpload(testAccount, "u-race", s, rev) + }(i, st) + } + wg.Wait() + + wins, conflicts := 0, 0 + for _, e := range errs { + switch { + case e == nil: + wins++ + case errors.Is(e, ecr.ErrConflict): + conflicts++ + default: + t.Fatalf("unexpected error: %v", e) + } + } + assert.Equal(t, 1, wins, "exactly one CAS update may commit") + assert.Equal(t, 1, conflicts, "the racing update must conflict, not silently merge") + + // The committed record points at exactly one byte key — never a mix. + final, _, err := meta.GetUpload(testAccount, "u-race") + require.NoError(t, err) + assert.Contains(t, []string{"uploads/u-race/a", "uploads/u-race/b"}, final.BytesKey) +} + +// TestPatchUpload_ConcurrentNoCorruption fires overlapping chunked PATCHes and +// finalizes. Whichever PATCHes commit, the bytes finally stored under the blob +// digest must equal the bytes the hash was computed over — i.e. the recorded +// hash state and the recorded byte object never desynchronise. +func TestPatchUpload_ConcurrentNoCorruption(t *testing.T) { + for iter := range 25 { + reg := newTestRegistry() + repo := "team/race" + + w := do(reg, http.MethodPost, "/v2/"+repo+"/blobs/uploads/", nil, nil) + require.Equal(t, http.StatusAccepted, w.Code) + loc := w.Header().Get("Location") + + var wg sync.WaitGroup + for _, chunk := range [][]byte{[]byte("AAAA"), []byte("BBBB"), []byte("CCCC")} { + wg.Add(1) + go func(body []byte) { + defer wg.Done() + do(reg, http.MethodPatch, loc, body, nil) + }(chunk) + } + wg.Wait() + + // Read the committed bytes the server recorded, finalize against their + // digest, and require the stored blob to match exactly. + st, _, err := reg.Meta.GetUpload(testAccount, strings.TrimPrefix(loc, "/v2/"+repo+"/blobs/uploads/")) + require.NoError(t, err) + committed := reg.readUploadBytesAt(st.BytesKey) + dg := digestOf(committed) + + w = do(reg, http.MethodPut, loc+"?digest="+dg, nil, nil) + require.Equal(t, http.StatusCreated, w.Code, "iter %d body: %s", iter, w.Body.String()) + + w = do(reg, http.MethodGet, "/v2/"+repo+"/blobs/"+dg, nil, nil) + require.Equal(t, http.StatusOK, w.Code) + assert.Equal(t, committed, w.Body.Bytes(), "stored bytes must equal the digest-verified bytes") + } +} diff --git a/spinifex/gateway/gateway.go b/spinifex/gateway/gateway.go index 035a29d0f..4adffeb0d 100644 --- a/spinifex/gateway/gateway.go +++ b/spinifex/gateway/gateway.go @@ -19,6 +19,7 @@ import ( "github.com/mulgadc/predastore/ratelimit" "github.com/mulgadc/spinifex/spinifex/awsec2query" "github.com/mulgadc/spinifex/spinifex/awserrors" + gateway_ecr "github.com/mulgadc/spinifex/spinifex/gateway/ecr" "github.com/mulgadc/spinifex/spinifex/gateway/policy" handlers_iam "github.com/mulgadc/spinifex/spinifex/handlers/iam" handlers_sts "github.com/mulgadc/spinifex/spinifex/handlers/sts" @@ -77,6 +78,9 @@ type GatewayConfig struct { Throttler *ratelimit.Throttler // Per-account+action API request throttler Version string // Build-time version string (set from cmd.Version) Commit string // Build-time commit hash (set from cmd.Commit) + // ECRRegistry serves the OCI Distribution v2 (/v2/*) surface. Nil falls back + // to the 501 stub (e.g. in unit tests of unrelated routes). + ECRRegistry *gateway_ecr.Registry } var supportedServices = map[string]bool{ diff --git a/spinifex/handlers/NATS_HANDLER_NAMING.md b/spinifex/handlers/NATS_HANDLER_NAMING.md index 6a1db4740..3a75d4802 100644 --- a/spinifex/handlers/NATS_HANDLER_NAMING.md +++ b/spinifex/handlers/NATS_HANDLER_NAMING.md @@ -198,3 +198,27 @@ EKS reconcilers also publish on existing namespaces when interacting with other services (no `eks.` prefix): `ec2.RunInstances`, `ec2.CreateNetworkInterface`, `elbv2.CreateLoadBalancer`, `elbv2.RegisterTargets`, `route53.ChangeResourceRecordSets`, etc. + +## ECR (OCI Distribution registry metadata) + +ECR is not a SigV4 AWS-action surface but an OCI Distribution v2 (`/v2/*`) +registry. Blob and manifest *bytes* stream straight from the gateway to +predastore and never traverse NATS. Only the *metadata* — repo/tag/manifest +records and in-progress upload-state CAS — is owned by the daemon, which holds +the per-account JetStream KV. The gateway is a request/reply client; the daemon +serves these subjects under the `spinifex-workers` queue group. + +Subjects use a `ecr..` shape (handler `handleECR`): + +``` +ecr.repo.create, ecr.repo.describe, ecr.repo.list +ecr.tag.put, ecr.tag.get, ecr.tag.list, ecr.tag.delete +ecr.manifest.put, ecr.manifest.describe +ecr.upload.create, ecr.upload.get, ecr.upload.update, ecr.upload.delete +``` + +`ecr.upload.update` is the serialization point for chunked blob uploads: it is a +JetStream KV compare-and-swap, so a concurrent PATCH that loses the revision +race is rejected rather than silently merged. Absent records and CAS conflicts +travel back as response flags (`found`, `conflict`), not transport errors, so +they round-trip a reply envelope that otherwise only carries AWS error codes. diff --git a/spinifex/handlers/ecr/meta.go b/spinifex/handlers/ecr/meta.go new file mode 100644 index 000000000..b1c7bc397 --- /dev/null +++ b/spinifex/handlers/ecr/meta.go @@ -0,0 +1,495 @@ +package ecr + +import ( + "encoding/json" + "errors" + "fmt" + "sort" + "strings" + "sync" + "time" + + "github.com/mulgadc/spinifex/spinifex/utils" + "github.com/nats-io/nats.go" +) + +// ErrNotFound is returned by MetaStore reads when the requested record is absent. +var ErrNotFound = errors.New("ecr: metadata record not found") + +// ErrConflict signals an optimistic-concurrency clash on a CAS update. Callers +// retry the read-modify-write loop on this error. +var ErrConflict = errors.New("ecr: metadata update conflict") + +// RepoMeta is the per-repository metadata record. +type RepoMeta struct { + Name string `json:"name"` + CreatedAt time.Time `json:"createdAt"` +} + +// ManifestMeta records a stored manifest's properties for shallow validation +// of image indexes and Docker-Content-Digest responses. +type ManifestMeta struct { + Digest string `json:"digest"` + MediaType string `json:"mediaType"` + Size int64 `json:"size"` + PushedAt time.Time `json:"pushedAt"` + ChildDigests []string `json:"childDigests,omitempty"` +} + +// UploadState tracks an in-progress chunked blob upload. The sha256 hash state +// is carried as a BinaryMarshaler-marshaled snapshot so PATCH chunks resume the +// running digest without re-reading committed bytes. +type UploadState struct { + RepoName string `json:"repoName"` + StartedAt time.Time `json:"startedAt"` + LastActivity time.Time `json:"lastActivity"` + CommittedBytes int64 `json:"committedBytes"` + SHA256MarshaledState []byte `json:"sha256MarshaledState"` + ExpectedDigest string `json:"expectedDigest,omitempty"` + // BytesKey addresses the object holding all bytes committed so far. Each + // PATCH writes a fresh key and records it under CAS, so the committed hash + // and the committed bytes always refer to the same object. + BytesKey string `json:"bytesKey,omitempty"` +} + +// MetaStore is the per-account metadata surface backing the OCI registry. Reads +// return ErrNotFound for missing records; CAS updates return ErrConflict on a +// concurrent revision clash so the caller can retry. +type MetaStore interface { + PutRepo(accountID string, meta RepoMeta) error + GetRepo(accountID, repo string) (RepoMeta, error) + ListRepos(accountID string) ([]string, error) + + PutTag(accountID, repo, tag, digest string) error + GetTag(accountID, repo, tag string) (string, error) + DeleteTag(accountID, repo, tag string) error + ListTags(accountID, repo string) ([]string, error) + + PutManifestMeta(accountID, repo string, meta ManifestMeta) error + GetManifestMeta(accountID, repo, digest string) (ManifestMeta, error) + + PutUpload(accountID, uploadID string, state UploadState) (uint64, error) + GetUpload(accountID, uploadID string) (UploadState, uint64, error) + UpdateUpload(accountID, uploadID string, state UploadState, rev uint64) (uint64, error) + DeleteUpload(accountID, uploadID string) error +} + +// KVMetaStore is the JetStream-KV-backed MetaStore. Per-account buckets are +// created lazily on first write and cached per process. +type KVMetaStore struct { + js nats.JetStreamContext + mu sync.Mutex + buckets map[string]nats.KeyValue +} + +var _ MetaStore = (*KVMetaStore)(nil) + +// NewKVMetaStore constructs a KVMetaStore over the supplied JetStream context. +func NewKVMetaStore(js nats.JetStreamContext) *KVMetaStore { + return &KVMetaStore{js: js, buckets: make(map[string]nats.KeyValue)} +} + +func (s *KVMetaStore) bucket(accountID string) (nats.KeyValue, error) { + s.mu.Lock() + defer s.mu.Unlock() + if kv, ok := s.buckets[accountID]; ok { + return kv, nil + } + name := KVAccountBucket(accountID) + kv, err := utils.GetOrCreateKVBucket(s.js, name, KVBucketAccountHistory) + if err != nil { + return nil, fmt.Errorf("ecr: open account bucket %s: %w", name, err) + } + s.buckets[accountID] = kv + return kv, nil +} + +func (s *KVMetaStore) PutRepo(accountID string, meta RepoMeta) error { + kv, err := s.bucket(accountID) + if err != nil { + return err + } + data, err := json.Marshal(meta) + if err != nil { + return err + } + _, err = kv.Put(KVRepoMetaKey(meta.Name), data) + return err +} + +func (s *KVMetaStore) GetRepo(accountID, repo string) (RepoMeta, error) { + kv, err := s.bucket(accountID) + if err != nil { + return RepoMeta{}, err + } + entry, err := kv.Get(KVRepoMetaKey(repo)) + if err != nil { + return RepoMeta{}, mapKVErr(err) + } + var m RepoMeta + if err := json.Unmarshal(entry.Value(), &m); err != nil { + return RepoMeta{}, err + } + return m, nil +} + +func (s *KVMetaStore) ListRepos(accountID string) ([]string, error) { + kv, err := s.bucket(accountID) + if err != nil { + return nil, err + } + keys, err := kv.Keys() + if err != nil { + if errors.Is(err, nats.ErrNoKeysFound) { + return nil, nil + } + return nil, err + } + var repos []string + for _, k := range keys { + if !strings.HasPrefix(k, KVReposPrefix) || !strings.HasSuffix(k, "/meta") { + continue + } + name := trimSuffixMeta(k) + if name != "" { + repos = append(repos, name) + } + } + sort.Strings(repos) + return repos, nil +} + +func (s *KVMetaStore) PutTag(accountID, repo, tag, digest string) error { + kv, err := s.bucket(accountID) + if err != nil { + return err + } + _, err = kv.Put(KVTagKey(repo, tag), []byte(digest)) + return err +} + +func (s *KVMetaStore) GetTag(accountID, repo, tag string) (string, error) { + kv, err := s.bucket(accountID) + if err != nil { + return "", err + } + entry, err := kv.Get(KVTagKey(repo, tag)) + if err != nil { + return "", mapKVErr(err) + } + return string(entry.Value()), nil +} + +func (s *KVMetaStore) DeleteTag(accountID, repo, tag string) error { + kv, err := s.bucket(accountID) + if err != nil { + return err + } + // JetStream KV Delete is silently idempotent, so existence is checked first + // to surface a real not-found to the caller. + if _, err := kv.Get(KVTagKey(repo, tag)); err != nil { + return mapKVErr(err) + } + if err := kv.Delete(KVTagKey(repo, tag)); err != nil { + return mapKVErr(err) + } + return nil +} + +func (s *KVMetaStore) ListTags(accountID, repo string) ([]string, error) { + kv, err := s.bucket(accountID) + if err != nil { + return nil, err + } + keys, err := kv.Keys() + if err != nil { + if errors.Is(err, nats.ErrNoKeysFound) { + return nil, nil + } + return nil, err + } + prefix := KVTagsPrefix(repo) + var tags []string + for _, k := range keys { + if tag, ok := strings.CutPrefix(k, prefix); ok { + tags = append(tags, tag) + } + } + sort.Strings(tags) + return tags, nil +} + +func (s *KVMetaStore) PutManifestMeta(accountID, repo string, meta ManifestMeta) error { + kv, err := s.bucket(accountID) + if err != nil { + return err + } + data, err := json.Marshal(meta) + if err != nil { + return err + } + _, err = kv.Put(KVManifestKey(repo, meta.Digest), data) + return err +} + +func (s *KVMetaStore) GetManifestMeta(accountID, repo, digest string) (ManifestMeta, error) { + kv, err := s.bucket(accountID) + if err != nil { + return ManifestMeta{}, err + } + entry, err := kv.Get(KVManifestKey(repo, digest)) + if err != nil { + return ManifestMeta{}, mapKVErr(err) + } + var m ManifestMeta + if err := json.Unmarshal(entry.Value(), &m); err != nil { + return ManifestMeta{}, err + } + return m, nil +} + +func (s *KVMetaStore) PutUpload(accountID, uploadID string, state UploadState) (uint64, error) { + kv, err := s.bucket(accountID) + if err != nil { + return 0, err + } + data, err := json.Marshal(state) + if err != nil { + return 0, err + } + return kv.Put(KVUploadKey(uploadID), data) +} + +func (s *KVMetaStore) GetUpload(accountID, uploadID string) (UploadState, uint64, error) { + kv, err := s.bucket(accountID) + if err != nil { + return UploadState{}, 0, err + } + entry, err := kv.Get(KVUploadKey(uploadID)) + if err != nil { + return UploadState{}, 0, mapKVErr(err) + } + var st UploadState + if err := json.Unmarshal(entry.Value(), &st); err != nil { + return UploadState{}, 0, err + } + return st, entry.Revision(), nil +} + +func (s *KVMetaStore) UpdateUpload(accountID, uploadID string, state UploadState, rev uint64) (uint64, error) { + kv, err := s.bucket(accountID) + if err != nil { + return 0, err + } + data, err := json.Marshal(state) + if err != nil { + return 0, err + } + newRev, err := kv.Update(KVUploadKey(uploadID), data, rev) + if err != nil { + if errors.Is(err, nats.ErrKeyExists) { + return 0, ErrConflict + } + return 0, err + } + return newRev, nil +} + +func (s *KVMetaStore) DeleteUpload(accountID, uploadID string) error { + kv, err := s.bucket(accountID) + if err != nil { + return err + } + // JetStream KV Delete is silently idempotent, so existence is checked first + // to surface a real not-found to the caller. + if _, err := kv.Get(KVUploadKey(uploadID)); err != nil { + return mapKVErr(err) + } + if err := kv.Delete(KVUploadKey(uploadID)); err != nil { + return mapKVErr(err) + } + return nil +} + +// MemoryMetaStore is an in-memory MetaStore used by tests and single-process +// dev runs. It is safe for concurrent use. +type MemoryMetaStore struct { + mu sync.Mutex + repos map[string]map[string]RepoMeta // account -> repo -> meta + tags map[string]map[string]string // account -> repo|tag -> digest + manifests map[string]map[string]ManifestMeta // account -> repo|digest -> meta + uploads map[string]map[string]uploadRev // account -> id -> state+rev +} + +type uploadRev struct { + state UploadState + rev uint64 +} + +var _ MetaStore = (*MemoryMetaStore)(nil) + +// NewMemoryMetaStore constructs an empty in-memory MetaStore. +func NewMemoryMetaStore() *MemoryMetaStore { + return &MemoryMetaStore{ + repos: make(map[string]map[string]RepoMeta), + tags: make(map[string]map[string]string), + manifests: make(map[string]map[string]ManifestMeta), + uploads: make(map[string]map[string]uploadRev), + } +} + +func (m *MemoryMetaStore) PutRepo(accountID string, meta RepoMeta) error { + m.mu.Lock() + defer m.mu.Unlock() + if m.repos[accountID] == nil { + m.repos[accountID] = make(map[string]RepoMeta) + } + m.repos[accountID][meta.Name] = meta + return nil +} + +func (m *MemoryMetaStore) GetRepo(accountID, repo string) (RepoMeta, error) { + m.mu.Lock() + defer m.mu.Unlock() + r, ok := m.repos[accountID][repo] + if !ok { + return RepoMeta{}, ErrNotFound + } + return r, nil +} + +func (m *MemoryMetaStore) ListRepos(accountID string) ([]string, error) { + m.mu.Lock() + defer m.mu.Unlock() + var out []string + for name := range m.repos[accountID] { + out = append(out, name) + } + sort.Strings(out) + return out, nil +} + +func (m *MemoryMetaStore) PutTag(accountID, repo, tag, digest string) error { + m.mu.Lock() + defer m.mu.Unlock() + if m.tags[accountID] == nil { + m.tags[accountID] = make(map[string]string) + } + m.tags[accountID][repo+"|"+tag] = digest + return nil +} + +func (m *MemoryMetaStore) GetTag(accountID, repo, tag string) (string, error) { + m.mu.Lock() + defer m.mu.Unlock() + d, ok := m.tags[accountID][repo+"|"+tag] + if !ok { + return "", ErrNotFound + } + return d, nil +} + +func (m *MemoryMetaStore) DeleteTag(accountID, repo, tag string) error { + m.mu.Lock() + defer m.mu.Unlock() + key := repo + "|" + tag + if _, ok := m.tags[accountID][key]; !ok { + return ErrNotFound + } + delete(m.tags[accountID], key) + return nil +} + +func (m *MemoryMetaStore) ListTags(accountID, repo string) ([]string, error) { + m.mu.Lock() + defer m.mu.Unlock() + var out []string + for k := range m.tags[accountID] { + if r, t, ok := strings.Cut(k, "|"); ok && r == repo { + out = append(out, t) + } + } + sort.Strings(out) + return out, nil +} + +func (m *MemoryMetaStore) PutManifestMeta(accountID, repo string, meta ManifestMeta) error { + m.mu.Lock() + defer m.mu.Unlock() + if m.manifests[accountID] == nil { + m.manifests[accountID] = make(map[string]ManifestMeta) + } + m.manifests[accountID][repo+"|"+meta.Digest] = meta + return nil +} + +func (m *MemoryMetaStore) GetManifestMeta(accountID, repo, digest string) (ManifestMeta, error) { + m.mu.Lock() + defer m.mu.Unlock() + meta, ok := m.manifests[accountID][repo+"|"+digest] + if !ok { + return ManifestMeta{}, ErrNotFound + } + return meta, nil +} + +func (m *MemoryMetaStore) PutUpload(accountID, uploadID string, state UploadState) (uint64, error) { + m.mu.Lock() + defer m.mu.Unlock() + if m.uploads[accountID] == nil { + m.uploads[accountID] = make(map[string]uploadRev) + } + rev := m.uploads[accountID][uploadID].rev + 1 + m.uploads[accountID][uploadID] = uploadRev{state: state, rev: rev} + return rev, nil +} + +func (m *MemoryMetaStore) GetUpload(accountID, uploadID string) (UploadState, uint64, error) { + m.mu.Lock() + defer m.mu.Unlock() + u, ok := m.uploads[accountID][uploadID] + if !ok { + return UploadState{}, 0, ErrNotFound + } + return u.state, u.rev, nil +} + +func (m *MemoryMetaStore) UpdateUpload(accountID, uploadID string, state UploadState, rev uint64) (uint64, error) { + m.mu.Lock() + defer m.mu.Unlock() + u, ok := m.uploads[accountID][uploadID] + if !ok { + return 0, ErrNotFound + } + if u.rev != rev { + return 0, ErrConflict + } + u.state = state + u.rev = rev + 1 + m.uploads[accountID][uploadID] = u + return u.rev, nil +} + +func (m *MemoryMetaStore) DeleteUpload(accountID, uploadID string) error { + m.mu.Lock() + defer m.mu.Unlock() + if _, ok := m.uploads[accountID][uploadID]; !ok { + return ErrNotFound + } + delete(m.uploads[accountID], uploadID) + return nil +} + +// mapKVErr normalizes JetStream KV errors to the MetaStore error vocabulary. +func mapKVErr(err error) error { + if errors.Is(err, nats.ErrKeyNotFound) { + return ErrNotFound + } + return err +} + +// trimSuffixMeta extracts the repo name from a "repos/{name}/meta" key. +func trimSuffixMeta(key string) string { + name := strings.TrimPrefix(key, KVReposPrefix) + return strings.TrimSuffix(name, "/meta") +} diff --git a/spinifex/handlers/ecr/meta_kv_test.go b/spinifex/handlers/ecr/meta_kv_test.go new file mode 100644 index 000000000..699f23774 --- /dev/null +++ b/spinifex/handlers/ecr/meta_kv_test.go @@ -0,0 +1,80 @@ +package ecr + +import ( + "testing" + "time" + + "github.com/mulgadc/spinifex/spinifex/testutil" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// TestKVMetaStore_FullCycle exercises the JetStream-backed MetaStore against an +// embedded JetStream server, covering the production CAS path. +func TestKVMetaStore_FullCycle(t *testing.T) { + _, _, js := testutil.StartTestJetStream(t) + store := NewKVMetaStore(js) + const acct = "000000000000" + + // Repo create + list (also lazily creates the bucket). + require.NoError(t, store.PutRepo(acct, RepoMeta{Name: "team/app", CreatedAt: time.Now()})) + require.NoError(t, store.PutRepo(acct, RepoMeta{Name: "team/web", CreatedAt: time.Now()})) + got, err := store.GetRepo(acct, "team/app") + require.NoError(t, err) + assert.Equal(t, "team/app", got.Name) + _, err = store.GetRepo(acct, "team/ghost") + assert.ErrorIs(t, err, ErrNotFound) + + repos, err := store.ListRepos(acct) + require.NoError(t, err) + assert.Equal(t, []string{"team/app", "team/web"}, repos) + + // Tags. + require.NoError(t, store.PutTag(acct, "team/app", "v1", "sha256:aaa")) + require.NoError(t, store.PutTag(acct, "team/app", "v2", "sha256:bbb")) + d, err := store.GetTag(acct, "team/app", "v1") + require.NoError(t, err) + assert.Equal(t, "sha256:aaa", d) + tags, err := store.ListTags(acct, "team/app") + require.NoError(t, err) + assert.Equal(t, []string{"v1", "v2"}, tags) + require.NoError(t, store.DeleteTag(acct, "team/app", "v1")) + _, err = store.GetTag(acct, "team/app", "v1") + assert.ErrorIs(t, err, ErrNotFound) + assert.ErrorIs(t, store.DeleteTag(acct, "team/app", "missing"), ErrNotFound) + + // Manifest meta. + require.NoError(t, store.PutManifestMeta(acct, "team/app", ManifestMeta{ + Digest: "sha256:ccc", MediaType: "application/json", Size: 12, + })) + mm, err := store.GetManifestMeta(acct, "team/app", "sha256:ccc") + require.NoError(t, err) + assert.Equal(t, int64(12), mm.Size) + _, err = store.GetManifestMeta(acct, "team/app", "sha256:zzz") + assert.ErrorIs(t, err, ErrNotFound) + + // Upload CAS lifecycle. + rev, err := store.PutUpload(acct, "u1", UploadState{RepoName: "team/app"}) + require.NoError(t, err) + st, gotRev, err := store.GetUpload(acct, "u1") + require.NoError(t, err) + assert.Equal(t, rev, gotRev) + + _, err = store.UpdateUpload(acct, "u1", st, gotRev+99) + assert.ErrorIs(t, err, ErrConflict) + + st.CommittedBytes = 5 + newRev, err := store.UpdateUpload(acct, "u1", st, gotRev) + require.NoError(t, err) + assert.Greater(t, newRev, gotRev) + + require.NoError(t, store.DeleteUpload(acct, "u1")) + _, _, err = store.GetUpload(acct, "u1") + assert.ErrorIs(t, err, ErrNotFound) + assert.ErrorIs(t, store.DeleteUpload(acct, "u1"), ErrNotFound) + + // Empty-bucket listing returns no repos for a fresh account. + empty, err := store.ListRepos("999999999999") + require.NoError(t, err) + assert.Empty(t, empty) +} diff --git a/spinifex/handlers/ecr/meta_nats.go b/spinifex/handlers/ecr/meta_nats.go new file mode 100644 index 000000000..63dbb4662 --- /dev/null +++ b/spinifex/handlers/ecr/meta_nats.go @@ -0,0 +1,144 @@ +package ecr + +import ( + "time" + + "github.com/mulgadc/spinifex/spinifex/utils" + "github.com/nats-io/nats.go" +) + +const metaRequestTimeout = 30 * time.Second + +// NATSMetaStore is the gateway-side MetaStore. It forwards every metadata +// operation to the daemon over NATS request/reply; the daemon owns the KV. The +// Found/Conflict response flags are mapped back to ErrNotFound/ErrConflict so +// callers see the same MetaStore contract as the in-process stores. +type NATSMetaStore struct { + conn *nats.Conn +} + +var _ MetaStore = (*NATSMetaStore)(nil) + +// NewNATSMetaStore builds a NATS-backed MetaStore client. +func NewNATSMetaStore(conn *nats.Conn) *NATSMetaStore { + return &NATSMetaStore{conn: conn} +} + +func (s *NATSMetaStore) PutRepo(accountID string, meta RepoMeta) error { + _, err := utils.NATSRequest[RepoCreateResponse](s.conn, SubjectRepoCreate, &RepoCreateRequest{Meta: meta}, metaRequestTimeout, accountID) + return err +} + +func (s *NATSMetaStore) GetRepo(accountID, repo string) (RepoMeta, error) { + resp, err := utils.NATSRequest[RepoDescribeResponse](s.conn, SubjectRepoDescribe, &RepoDescribeRequest{Repo: repo}, metaRequestTimeout, accountID) + if err != nil { + return RepoMeta{}, err + } + if !resp.Found { + return RepoMeta{}, ErrNotFound + } + return resp.Meta, nil +} + +func (s *NATSMetaStore) ListRepos(accountID string) ([]string, error) { + resp, err := utils.NATSRequest[RepoListResponse](s.conn, SubjectRepoList, &RepoListRequest{}, metaRequestTimeout, accountID) + if err != nil { + return nil, err + } + return resp.Repos, nil +} + +func (s *NATSMetaStore) PutTag(accountID, repo, tag, digest string) error { + _, err := utils.NATSRequest[TagPutResponse](s.conn, SubjectTagPut, &TagPutRequest{Repo: repo, Tag: tag, Digest: digest}, metaRequestTimeout, accountID) + return err +} + +func (s *NATSMetaStore) GetTag(accountID, repo, tag string) (string, error) { + resp, err := utils.NATSRequest[TagGetResponse](s.conn, SubjectTagGet, &TagGetRequest{Repo: repo, Tag: tag}, metaRequestTimeout, accountID) + if err != nil { + return "", err + } + if !resp.Found { + return "", ErrNotFound + } + return resp.Digest, nil +} + +func (s *NATSMetaStore) DeleteTag(accountID, repo, tag string) error { + resp, err := utils.NATSRequest[TagDeleteResponse](s.conn, SubjectTagDelete, &TagDeleteRequest{Repo: repo, Tag: tag}, metaRequestTimeout, accountID) + if err != nil { + return err + } + if !resp.Found { + return ErrNotFound + } + return nil +} + +func (s *NATSMetaStore) ListTags(accountID, repo string) ([]string, error) { + resp, err := utils.NATSRequest[TagListResponse](s.conn, SubjectTagList, &TagListRequest{Repo: repo}, metaRequestTimeout, accountID) + if err != nil { + return nil, err + } + return resp.Tags, nil +} + +func (s *NATSMetaStore) PutManifestMeta(accountID, repo string, meta ManifestMeta) error { + _, err := utils.NATSRequest[ManifestPutResponse](s.conn, SubjectManifestPut, &ManifestPutRequest{Repo: repo, Meta: meta}, metaRequestTimeout, accountID) + return err +} + +func (s *NATSMetaStore) GetManifestMeta(accountID, repo, digest string) (ManifestMeta, error) { + resp, err := utils.NATSRequest[ManifestDescribeResponse](s.conn, SubjectManifestDescribe, &ManifestDescribeRequest{Repo: repo, Digest: digest}, metaRequestTimeout, accountID) + if err != nil { + return ManifestMeta{}, err + } + if !resp.Found { + return ManifestMeta{}, ErrNotFound + } + return resp.Meta, nil +} + +func (s *NATSMetaStore) PutUpload(accountID, uploadID string, state UploadState) (uint64, error) { + resp, err := utils.NATSRequest[UploadCreateResponse](s.conn, SubjectUploadCreate, &UploadCreateRequest{UploadID: uploadID, State: state}, metaRequestTimeout, accountID) + if err != nil { + return 0, err + } + return resp.Revision, nil +} + +func (s *NATSMetaStore) GetUpload(accountID, uploadID string) (UploadState, uint64, error) { + resp, err := utils.NATSRequest[UploadGetResponse](s.conn, SubjectUploadGet, &UploadGetRequest{UploadID: uploadID}, metaRequestTimeout, accountID) + if err != nil { + return UploadState{}, 0, err + } + if !resp.Found { + return UploadState{}, 0, ErrNotFound + } + return resp.State, resp.Revision, nil +} + +func (s *NATSMetaStore) UpdateUpload(accountID, uploadID string, state UploadState, rev uint64) (uint64, error) { + resp, err := utils.NATSRequest[UploadUpdateResponse](s.conn, SubjectUploadUpdate, &UploadUpdateRequest{UploadID: uploadID, State: state, Revision: rev}, metaRequestTimeout, accountID) + if err != nil { + return 0, err + } + if !resp.Found { + return 0, ErrNotFound + } + if resp.Conflict { + return 0, ErrConflict + } + return resp.Revision, nil +} + +func (s *NATSMetaStore) DeleteUpload(accountID, uploadID string) error { + resp, err := utils.NATSRequest[UploadDeleteResponse](s.conn, SubjectUploadDelete, &UploadDeleteRequest{UploadID: uploadID}, metaRequestTimeout, accountID) + if err != nil { + return err + } + if !resp.Found { + return ErrNotFound + } + return nil +} diff --git a/spinifex/handlers/ecr/meta_test.go b/spinifex/handlers/ecr/meta_test.go new file mode 100644 index 000000000..ea6a1143d --- /dev/null +++ b/spinifex/handlers/ecr/meta_test.go @@ -0,0 +1,116 @@ +package ecr + +import ( + "strings" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestValidateRepoName(t *testing.T) { + valid := []string{"ab", "team/app", "a/b/c", "my-repo.name_1", "x0/y1/z2"} + for _, n := range valid { + assert.NoError(t, ValidateRepoName(n), "expected %q valid", n) + } + invalid := []string{"A", "Team/App", "a", "/leading", "trailing/", "a//b", strings.Repeat("a", 257)} + for _, n := range invalid { + assert.Error(t, ValidateRepoName(n), "expected %q invalid", n) + } +} + +func TestValidateDigest(t *testing.T) { + good := "sha256:" + strings.Repeat("a", 64) + assert.True(t, ValidateDigest(good)) + assert.False(t, ValidateDigest("sha256:abc")) + assert.False(t, ValidateDigest("md5:"+strings.Repeat("a", 64))) + assert.False(t, ValidateDigest("sha256:"+strings.Repeat("Z", 64))) +} + +func TestKVKeyHelpers(t *testing.T) { + assert.Equal(t, "ecr-account-000000000000", KVAccountBucket("000000000000")) + assert.Equal(t, "repos/team/app/meta", KVRepoMetaKey("team/app")) + assert.Equal(t, "repos/team/app/tags/", KVTagsPrefix("team/app")) + assert.Equal(t, "repos/team/app/tags/v1", KVTagKey("team/app", "v1")) + assert.Equal(t, "uploads/u1", KVUploadKey("u1")) + assert.True(t, strings.HasPrefix(KVManifestKey("r", "sha256:abc"), "repos/r/manifests/sha256-abc")) + assert.Equal(t, "sha256-abc", DigestToken("sha256:abc")) +} + +func TestMemoryMetaStore_RepoRoundtrip(t *testing.T) { + m := NewMemoryMetaStore() + const acct = "000000000000" + + _, err := m.GetRepo(acct, "team/app") + assert.ErrorIs(t, err, ErrNotFound) + + require.NoError(t, m.PutRepo(acct, RepoMeta{Name: "team/app", CreatedAt: time.Now()})) + require.NoError(t, m.PutRepo(acct, RepoMeta{Name: "team/web", CreatedAt: time.Now()})) + + got, err := m.GetRepo(acct, "team/app") + require.NoError(t, err) + assert.Equal(t, "team/app", got.Name) + + repos, err := m.ListRepos(acct) + require.NoError(t, err) + assert.Equal(t, []string{"team/app", "team/web"}, repos) +} + +func TestMemoryMetaStore_TagsAndManifests(t *testing.T) { + m := NewMemoryMetaStore() + const acct = "000000000000" + + require.NoError(t, m.PutTag(acct, "r", "v1", "sha256:aaa")) + require.NoError(t, m.PutTag(acct, "r", "v2", "sha256:bbb")) + d, err := m.GetTag(acct, "r", "v1") + require.NoError(t, err) + assert.Equal(t, "sha256:aaa", d) + + tags, err := m.ListTags(acct, "r") + require.NoError(t, err) + assert.Equal(t, []string{"v1", "v2"}, tags) + + require.NoError(t, m.DeleteTag(acct, "r", "v1")) + _, err = m.GetTag(acct, "r", "v1") + assert.ErrorIs(t, err, ErrNotFound) + assert.ErrorIs(t, m.DeleteTag(acct, "r", "missing"), ErrNotFound) + + require.NoError(t, m.PutManifestMeta(acct, "r", ManifestMeta{Digest: "sha256:ccc", MediaType: "x", Size: 7})) + mm, err := m.GetManifestMeta(acct, "r", "sha256:ccc") + require.NoError(t, err) + assert.Equal(t, int64(7), mm.Size) + _, err = m.GetManifestMeta(acct, "r", "sha256:zzz") + assert.ErrorIs(t, err, ErrNotFound) +} + +func TestMemoryMetaStore_UploadCAS(t *testing.T) { + m := NewMemoryMetaStore() + const acct = "000000000000" + + rev, err := m.PutUpload(acct, "u1", UploadState{RepoName: "r"}) + require.NoError(t, err) + + st, gotRev, err := m.GetUpload(acct, "u1") + require.NoError(t, err) + assert.Equal(t, rev, gotRev) + assert.Equal(t, "r", st.RepoName) + + // Stale revision is rejected. + _, err = m.UpdateUpload(acct, "u1", st, rev+99) + assert.ErrorIs(t, err, ErrConflict) + + st.CommittedBytes = 10 + newRev, err := m.UpdateUpload(acct, "u1", st, gotRev) + require.NoError(t, err) + assert.Greater(t, newRev, gotRev) + + require.NoError(t, m.DeleteUpload(acct, "u1")) + _, _, err = m.GetUpload(acct, "u1") + assert.ErrorIs(t, err, ErrNotFound) + assert.ErrorIs(t, m.DeleteUpload(acct, "u1"), ErrNotFound) +} + +func TestTrimSuffixMeta(t *testing.T) { + assert.Equal(t, "team/app", trimSuffixMeta("repos/team/app/meta")) +} diff --git a/spinifex/handlers/ecr/service.go b/spinifex/handlers/ecr/service.go new file mode 100644 index 000000000..b94d9c03e --- /dev/null +++ b/spinifex/handlers/ecr/service.go @@ -0,0 +1,162 @@ +package ecr + +// NATS subjects for the daemon-served ECR metadata surface. The daemon owns the +// JetStream KV; the gateway is a request/reply client. Blob and manifest bytes +// do NOT travel these subjects — only metadata records and upload-state CAS. +const ( + SubjectRepoCreate = "ecr.repo.create" + SubjectRepoDescribe = "ecr.repo.describe" + SubjectRepoList = "ecr.repo.list" + + SubjectTagPut = "ecr.tag.put" + SubjectTagGet = "ecr.tag.get" + SubjectTagList = "ecr.tag.list" + SubjectTagDelete = "ecr.tag.delete" + + SubjectManifestPut = "ecr.manifest.put" + SubjectManifestDescribe = "ecr.manifest.describe" + + SubjectUploadCreate = "ecr.upload.create" + SubjectUploadGet = "ecr.upload.get" + SubjectUploadUpdate = "ecr.upload.update" + SubjectUploadDelete = "ecr.upload.delete" +) + +// Request/response envelopes for the metadata surface. Absent records and CAS +// conflicts are reported via the Found/Conflict flags rather than transport +// errors, so they round-trip a NATS reply that only carries AWS error codes. + +type RepoCreateRequest struct { + Meta RepoMeta `json:"meta"` +} + +type RepoCreateResponse struct{} + +type RepoDescribeRequest struct { + Repo string `json:"repo"` +} + +type RepoDescribeResponse struct { + Found bool `json:"found"` + Meta RepoMeta `json:"meta"` +} + +type RepoListRequest struct{} + +type RepoListResponse struct { + Repos []string `json:"repos"` +} + +type TagPutRequest struct { + Repo string `json:"repo"` + Tag string `json:"tag"` + Digest string `json:"digest"` +} + +type TagPutResponse struct{} + +type TagGetRequest struct { + Repo string `json:"repo"` + Tag string `json:"tag"` +} + +type TagGetResponse struct { + Found bool `json:"found"` + Digest string `json:"digest"` +} + +type TagListRequest struct { + Repo string `json:"repo"` +} + +type TagListResponse struct { + Tags []string `json:"tags"` +} + +type TagDeleteRequest struct { + Repo string `json:"repo"` + Tag string `json:"tag"` +} + +type TagDeleteResponse struct { + Found bool `json:"found"` +} + +type ManifestPutRequest struct { + Repo string `json:"repo"` + Meta ManifestMeta `json:"meta"` +} + +type ManifestPutResponse struct{} + +type ManifestDescribeRequest struct { + Repo string `json:"repo"` + Digest string `json:"digest"` +} + +type ManifestDescribeResponse struct { + Found bool `json:"found"` + Meta ManifestMeta `json:"meta"` +} + +type UploadCreateRequest struct { + UploadID string `json:"uploadID"` + State UploadState `json:"state"` +} + +type UploadCreateResponse struct { + Revision uint64 `json:"revision"` +} + +type UploadGetRequest struct { + UploadID string `json:"uploadID"` +} + +type UploadGetResponse struct { + Found bool `json:"found"` + State UploadState `json:"state"` + Revision uint64 `json:"revision"` +} + +type UploadUpdateRequest struct { + UploadID string `json:"uploadID"` + State UploadState `json:"state"` + Revision uint64 `json:"revision"` +} + +type UploadUpdateResponse struct { + Found bool `json:"found"` + Conflict bool `json:"conflict"` + Revision uint64 `json:"revision"` +} + +type UploadDeleteRequest struct { + UploadID string `json:"uploadID"` +} + +type UploadDeleteResponse struct { + Found bool `json:"found"` +} + +// MetaService is the daemon-side metadata surface. Each method takes the +// account ID (carried in the NATS header by the gateway) and returns a typed +// response. Absence and CAS conflicts are encoded in the response, not as a +// transport error, so they survive the AWS-error-code-only NATS reply. +type MetaService interface { + RepoCreate(req *RepoCreateRequest, accountID string) (*RepoCreateResponse, error) + RepoDescribe(req *RepoDescribeRequest, accountID string) (*RepoDescribeResponse, error) + RepoList(req *RepoListRequest, accountID string) (*RepoListResponse, error) + + TagPut(req *TagPutRequest, accountID string) (*TagPutResponse, error) + TagGet(req *TagGetRequest, accountID string) (*TagGetResponse, error) + TagList(req *TagListRequest, accountID string) (*TagListResponse, error) + TagDelete(req *TagDeleteRequest, accountID string) (*TagDeleteResponse, error) + + ManifestPut(req *ManifestPutRequest, accountID string) (*ManifestPutResponse, error) + ManifestDescribe(req *ManifestDescribeRequest, accountID string) (*ManifestDescribeResponse, error) + + UploadCreate(req *UploadCreateRequest, accountID string) (*UploadCreateResponse, error) + UploadGet(req *UploadGetRequest, accountID string) (*UploadGetResponse, error) + UploadUpdate(req *UploadUpdateRequest, accountID string) (*UploadUpdateResponse, error) + UploadDelete(req *UploadDeleteRequest, accountID string) (*UploadDeleteResponse, error) +} diff --git a/spinifex/handlers/ecr/service_impl.go b/spinifex/handlers/ecr/service_impl.go new file mode 100644 index 000000000..a1efad5e3 --- /dev/null +++ b/spinifex/handlers/ecr/service_impl.go @@ -0,0 +1,152 @@ +package ecr + +import ( + "errors" + + "github.com/nats-io/nats.go" +) + +// MetaServiceImpl is the daemon-side MetaService. It owns the backing MetaStore +// (JetStream KV in production) and translates ErrNotFound/ErrConflict into the +// response flags so the gateway client can reconstruct them across NATS. +type MetaServiceImpl struct { + store MetaStore +} + +var _ MetaService = (*MetaServiceImpl)(nil) + +// NewMetaServiceImpl builds a MetaService over an explicit MetaStore (used by +// daemon-side tests with a MemoryMetaStore). +func NewMetaServiceImpl(store MetaStore) *MetaServiceImpl { + return &MetaServiceImpl{store: store} +} + +// NewKVMetaService builds a MetaService backed by per-account JetStream KV. +func NewKVMetaService(js nats.JetStreamContext) *MetaServiceImpl { + return &MetaServiceImpl{store: NewKVMetaStore(js)} +} + +func (s *MetaServiceImpl) RepoCreate(req *RepoCreateRequest, accountID string) (*RepoCreateResponse, error) { + if err := s.store.PutRepo(accountID, req.Meta); err != nil { + return nil, err + } + return &RepoCreateResponse{}, nil +} + +func (s *MetaServiceImpl) RepoDescribe(req *RepoDescribeRequest, accountID string) (*RepoDescribeResponse, error) { + meta, err := s.store.GetRepo(accountID, req.Repo) + if errors.Is(err, ErrNotFound) { + return &RepoDescribeResponse{Found: false}, nil + } + if err != nil { + return nil, err + } + return &RepoDescribeResponse{Found: true, Meta: meta}, nil +} + +func (s *MetaServiceImpl) RepoList(_ *RepoListRequest, accountID string) (*RepoListResponse, error) { + repos, err := s.store.ListRepos(accountID) + if err != nil { + return nil, err + } + return &RepoListResponse{Repos: repos}, nil +} + +func (s *MetaServiceImpl) TagPut(req *TagPutRequest, accountID string) (*TagPutResponse, error) { + if err := s.store.PutTag(accountID, req.Repo, req.Tag, req.Digest); err != nil { + return nil, err + } + return &TagPutResponse{}, nil +} + +func (s *MetaServiceImpl) TagGet(req *TagGetRequest, accountID string) (*TagGetResponse, error) { + digest, err := s.store.GetTag(accountID, req.Repo, req.Tag) + if errors.Is(err, ErrNotFound) { + return &TagGetResponse{Found: false}, nil + } + if err != nil { + return nil, err + } + return &TagGetResponse{Found: true, Digest: digest}, nil +} + +func (s *MetaServiceImpl) TagList(req *TagListRequest, accountID string) (*TagListResponse, error) { + tags, err := s.store.ListTags(accountID, req.Repo) + if err != nil { + return nil, err + } + return &TagListResponse{Tags: tags}, nil +} + +func (s *MetaServiceImpl) TagDelete(req *TagDeleteRequest, accountID string) (*TagDeleteResponse, error) { + err := s.store.DeleteTag(accountID, req.Repo, req.Tag) + if errors.Is(err, ErrNotFound) { + return &TagDeleteResponse{Found: false}, nil + } + if err != nil { + return nil, err + } + return &TagDeleteResponse{Found: true}, nil +} + +func (s *MetaServiceImpl) ManifestPut(req *ManifestPutRequest, accountID string) (*ManifestPutResponse, error) { + if err := s.store.PutManifestMeta(accountID, req.Repo, req.Meta); err != nil { + return nil, err + } + return &ManifestPutResponse{}, nil +} + +func (s *MetaServiceImpl) ManifestDescribe(req *ManifestDescribeRequest, accountID string) (*ManifestDescribeResponse, error) { + meta, err := s.store.GetManifestMeta(accountID, req.Repo, req.Digest) + if errors.Is(err, ErrNotFound) { + return &ManifestDescribeResponse{Found: false}, nil + } + if err != nil { + return nil, err + } + return &ManifestDescribeResponse{Found: true, Meta: meta}, nil +} + +func (s *MetaServiceImpl) UploadCreate(req *UploadCreateRequest, accountID string) (*UploadCreateResponse, error) { + rev, err := s.store.PutUpload(accountID, req.UploadID, req.State) + if err != nil { + return nil, err + } + return &UploadCreateResponse{Revision: rev}, nil +} + +func (s *MetaServiceImpl) UploadGet(req *UploadGetRequest, accountID string) (*UploadGetResponse, error) { + st, rev, err := s.store.GetUpload(accountID, req.UploadID) + if errors.Is(err, ErrNotFound) { + return &UploadGetResponse{Found: false}, nil + } + if err != nil { + return nil, err + } + return &UploadGetResponse{Found: true, State: st, Revision: rev}, nil +} + +func (s *MetaServiceImpl) UploadUpdate(req *UploadUpdateRequest, accountID string) (*UploadUpdateResponse, error) { + rev, err := s.store.UpdateUpload(accountID, req.UploadID, req.State, req.Revision) + if errors.Is(err, ErrNotFound) { + return &UploadUpdateResponse{Found: false}, nil + } + if errors.Is(err, ErrConflict) { + return &UploadUpdateResponse{Found: true, Conflict: true}, nil + } + if err != nil { + return nil, err + } + return &UploadUpdateResponse{Found: true, Revision: rev}, nil +} + +func (s *MetaServiceImpl) UploadDelete(req *UploadDeleteRequest, accountID string) (*UploadDeleteResponse, error) { + err := s.store.DeleteUpload(accountID, req.UploadID) + if errors.Is(err, ErrNotFound) { + return &UploadDeleteResponse{Found: false}, nil + } + if err != nil { + return nil, err + } + return &UploadDeleteResponse{Found: true}, nil +} diff --git a/spinifex/handlers/ecr/service_test.go b/spinifex/handlers/ecr/service_test.go new file mode 100644 index 000000000..e38a3a783 --- /dev/null +++ b/spinifex/handlers/ecr/service_test.go @@ -0,0 +1,151 @@ +package ecr + +import ( + "encoding/json" + "testing" + "time" + + "github.com/mulgadc/spinifex/spinifex/testutil" + "github.com/mulgadc/spinifex/spinifex/utils" + "github.com/nats-io/nats.go" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +const svcTestAccount = "000000000000" + +// TestMetaServiceImpl_FlagMapping checks the daemon-side service maps store +// ErrNotFound/ErrConflict into the response flags rather than transport errors, +// so they survive the AWS-error-code-only NATS reply. +func TestMetaServiceImpl_FlagMapping(t *testing.T) { + svc := NewMetaServiceImpl(NewMemoryMetaStore()) + + // Absent repo -> Found:false, nil error. + dr, err := svc.RepoDescribe(&RepoDescribeRequest{Repo: "ghost"}, svcTestAccount) + require.NoError(t, err) + assert.False(t, dr.Found) + + _, err = svc.RepoCreate(&RepoCreateRequest{Meta: RepoMeta{Name: "team/app", CreatedAt: time.Now()}}, svcTestAccount) + require.NoError(t, err) + dr, err = svc.RepoDescribe(&RepoDescribeRequest{Repo: "team/app"}, svcTestAccount) + require.NoError(t, err) + assert.True(t, dr.Found) + + // Absent tag/manifest/upload all report Found:false. + tg, err := svc.TagGet(&TagGetRequest{Repo: "team/app", Tag: "v1"}, svcTestAccount) + require.NoError(t, err) + assert.False(t, tg.Found) + td, err := svc.TagDelete(&TagDeleteRequest{Repo: "team/app", Tag: "v1"}, svcTestAccount) + require.NoError(t, err) + assert.False(t, td.Found) + ug, err := svc.UploadGet(&UploadGetRequest{UploadID: "nope"}, svcTestAccount) + require.NoError(t, err) + assert.False(t, ug.Found) + + // Upload CAS conflict -> Conflict:true, Found:true. + cr, err := svc.UploadCreate(&UploadCreateRequest{UploadID: "u1", State: UploadState{RepoName: "team/app"}}, svcTestAccount) + require.NoError(t, err) + uu, err := svc.UploadUpdate(&UploadUpdateRequest{UploadID: "u1", State: UploadState{CommittedBytes: 5}, Revision: cr.Revision + 99}, svcTestAccount) + require.NoError(t, err) + assert.True(t, uu.Found) + assert.True(t, uu.Conflict) +} + +// serveMeta wires a MetaService method to a NATS subject, mirroring the daemon's +// handleNATSRequest (unmarshal -> service(accountID) -> JSON or error payload). +func serveMeta[I any, O any](t *testing.T, nc *nats.Conn, subject string, fn func(*I, string) (*O, error)) { + t.Helper() + sub, err := nc.Subscribe(subject, func(msg *nats.Msg) { + accountID := utils.AccountIDFromMsg(msg) + in := new(I) + if errResp := utils.UnmarshalJsonPayload(in, msg.Data); errResp != nil { + _ = msg.Respond(errResp) + return + } + out, err := fn(in, accountID) + if err != nil { + _ = msg.Respond(utils.GenerateErrorPayload("ServerInternal")) + return + } + data, _ := json.Marshal(out) + _ = msg.Respond(data) + }) + require.NoError(t, err) + t.Cleanup(func() { _ = sub.Unsubscribe() }) +} + +// TestNATSMetaStore_RoundTrip drives the gateway client (NATSMetaStore) against +// a KV-backed daemon service over an embedded NATS/JetStream server, proving the +// full request/reply path including ErrNotFound/ErrConflict reconstruction. +func TestNATSMetaStore_RoundTrip(t *testing.T) { + _, nc, js := testutil.StartTestJetStream(t) + svc := NewKVMetaService(js) + + serveMeta(t, nc, SubjectRepoCreate, svc.RepoCreate) + serveMeta(t, nc, SubjectRepoDescribe, svc.RepoDescribe) + serveMeta(t, nc, SubjectRepoList, svc.RepoList) + serveMeta(t, nc, SubjectTagPut, svc.TagPut) + serveMeta(t, nc, SubjectTagGet, svc.TagGet) + serveMeta(t, nc, SubjectTagList, svc.TagList) + serveMeta(t, nc, SubjectTagDelete, svc.TagDelete) + serveMeta(t, nc, SubjectManifestPut, svc.ManifestPut) + serveMeta(t, nc, SubjectManifestDescribe, svc.ManifestDescribe) + serveMeta(t, nc, SubjectUploadCreate, svc.UploadCreate) + serveMeta(t, nc, SubjectUploadGet, svc.UploadGet) + serveMeta(t, nc, SubjectUploadUpdate, svc.UploadUpdate) + serveMeta(t, nc, SubjectUploadDelete, svc.UploadDelete) + + var client MetaStore = NewNATSMetaStore(nc) + const acct = svcTestAccount + + // Repo not-found round-trips as ErrNotFound. + _, err := client.GetRepo(acct, "ghost") + assert.ErrorIs(t, err, ErrNotFound) + + require.NoError(t, client.PutRepo(acct, RepoMeta{Name: "team/app", CreatedAt: time.Now()})) + got, err := client.GetRepo(acct, "team/app") + require.NoError(t, err) + assert.Equal(t, "team/app", got.Name) + repos, err := client.ListRepos(acct) + require.NoError(t, err) + assert.Equal(t, []string{"team/app"}, repos) + + // Tags. + require.NoError(t, client.PutTag(acct, "team/app", "v1", "sha256:aaa")) + d, err := client.GetTag(acct, "team/app", "v1") + require.NoError(t, err) + assert.Equal(t, "sha256:aaa", d) + tags, err := client.ListTags(acct, "team/app") + require.NoError(t, err) + assert.Equal(t, []string{"v1"}, tags) + require.NoError(t, client.DeleteTag(acct, "team/app", "v1")) + assert.ErrorIs(t, client.DeleteTag(acct, "team/app", "v1"), ErrNotFound) + + // Manifest meta. + require.NoError(t, client.PutManifestMeta(acct, "team/app", ManifestMeta{Digest: "sha256:ccc", MediaType: "x", Size: 9})) + mm, err := client.GetManifestMeta(acct, "team/app", "sha256:ccc") + require.NoError(t, err) + assert.Equal(t, int64(9), mm.Size) + _, err = client.GetManifestMeta(acct, "team/app", "sha256:zzz") + assert.ErrorIs(t, err, ErrNotFound) + + // Upload CAS lifecycle round-trip, including the conflict mapping. + rev, err := client.PutUpload(acct, "u1", UploadState{RepoName: "team/app"}) + require.NoError(t, err) + st, gotRev, err := client.GetUpload(acct, "u1") + require.NoError(t, err) + assert.Equal(t, rev, gotRev) + + _, err = client.UpdateUpload(acct, "u1", st, gotRev+99) + assert.ErrorIs(t, err, ErrConflict) + + st.CommittedBytes = 5 + newRev, err := client.UpdateUpload(acct, "u1", st, gotRev) + require.NoError(t, err) + assert.Greater(t, newRev, gotRev) + + require.NoError(t, client.DeleteUpload(acct, "u1")) + _, _, err = client.GetUpload(acct, "u1") + assert.ErrorIs(t, err, ErrNotFound) + assert.ErrorIs(t, client.DeleteUpload(acct, "u1"), ErrNotFound) +} diff --git a/spinifex/handlers/ecr/storage.go b/spinifex/handlers/ecr/storage.go index e67c61ade..e134b9e4b 100644 --- a/spinifex/handlers/ecr/storage.go +++ b/spinifex/handlers/ecr/storage.go @@ -5,7 +5,9 @@ package ecr import ( + "errors" "fmt" + "regexp" "strings" ) @@ -36,6 +38,10 @@ const ( blobKeyFmt = "blobs/%s/%s" // uploadKeyFmt holds in-progress chunked upload bytes. uploadKeyFmt = "uploads/%s" + // uploadChunkKeyFmt holds a single revision's assembled bytes under a unique + // per-write token, so a CAS-winning PATCH always points at exactly the bytes + // it hashed (a losing concurrent write lands on a different, orphaned token). + uploadChunkKeyFmt = "uploads/%s/%s" ) // RepoMetaKey returns the key for a repository's config object. @@ -44,8 +50,11 @@ func RepoMetaKey(repo string) string { return fmt.Sprintf(repoMetaKeyFmt, repo) // TagKey returns the key for a tag's digest pointer. func TagKey(repo, tag string) string { return fmt.Sprintf(tagKeyFmt, repo, tag) } -// ManifestKey returns the key for a manifest stored by its digest. -func ManifestKey(repo, digest string) string { return fmt.Sprintf(manifestKeyFmt, repo, digest) } +// ManifestKey returns the key for a manifest stored by its digest. The digest +// is sanitized via DigestToken so the object key carries no ':' separator. +func ManifestKey(repo, digest string) string { + return fmt.Sprintf(manifestKeyFmt, repo, DigestToken(digest)) +} // IndexKey returns the key for a repository's tag->digest index. func IndexKey(repo string) string { return fmt.Sprintf(indexKeyFmt, repo) } @@ -53,9 +62,96 @@ func IndexKey(repo string) string { return fmt.Sprintf(indexKeyFmt, repo) } // UploadKey returns the key for an in-progress upload's bytes. func UploadKey(uploadID string) string { return fmt.Sprintf(uploadKeyFmt, uploadID) } +// UploadChunkKey returns the key for one revision's assembled upload bytes, +// addressed by a unique per-write token. +func UploadChunkKey(uploadID, token string) string { + return fmt.Sprintf(uploadChunkKeyFmt, uploadID, token) +} + +// KV bucket layout for per-account ECR metadata. The bucket name is +// KVBucketAccountPrefix + accountID (e.g. "ecr-account-000000000000"), created +// lazily on the first repo create or push for that account. +const ( + KVBucketAccountPrefix = "ecr-account-" + KVBucketAccountVersion = 1 + KVBucketAccountHistory = 1 +) + +// KVAccountBucket returns the per-account JetStream KV bucket name. +func KVAccountBucket(accountID string) string { + return KVBucketAccountPrefix + accountID +} + +// KV key-path helpers for per-account metadata. +// +// repos/{name}/meta +// repos/{name}/tags/{tag} +// repos/{name}/manifests/{digest} +// uploads/{uuid} +const ( + kvRepoMetaKeyFmt = "repos/%s/meta" + kvTagsPrefixFmt = "repos/%s/tags/" + kvTagKeyFmt = "repos/%s/tags/%s" + kvManifestKeyFmt = "repos/%s/manifests/%s" + kvReposPrefix = "repos/" + kvUploadKeyFmt = "uploads/%s" +) + +// KVRepoMetaKey returns the KV key for a repository's meta record. +func KVRepoMetaKey(repo string) string { return fmt.Sprintf(kvRepoMetaKeyFmt, repo) } + +// KVTagsPrefix returns the KV key prefix enumerating a repository's tags. +func KVTagsPrefix(repo string) string { return fmt.Sprintf(kvTagsPrefixFmt, repo) } + +// KVTagKey returns the KV key for a single tag record. +func KVTagKey(repo, tag string) string { return fmt.Sprintf(kvTagKeyFmt, repo, tag) } + +// KVManifestKey returns the KV key for a manifest metadata record. The digest +// (which contains ':') is sanitized to a KV-safe token via DigestToken. +func KVManifestKey(repo, digest string) string { + return fmt.Sprintf(kvManifestKeyFmt, repo, DigestToken(digest)) +} + +// KVReposPrefix is the prefix under which every repo meta key lives. Used by +// the catalog listing to enumerate account repositories. +const KVReposPrefix = kvReposPrefix + +// KVUploadKey returns the KV key for an in-progress upload's state record. +func KVUploadKey(uploadID string) string { return fmt.Sprintf(kvUploadKeyFmt, uploadID) } + +// DigestToken maps a content digest ("sha256:") to a KV-key-safe token by +// replacing the ':' separator, which JetStream KV keys disallow. +func DigestToken(digest string) string { + return strings.ReplaceAll(digest, ":", "-") +} + +// repoNameRe is the OCI Distribution repository-name grammar. Compiled once. +var repoNameRe = regexp.MustCompile(`^(?:[a-z0-9]+(?:[._-][a-z0-9]+)*/)*[a-z0-9]+(?:[._-][a-z0-9]+)*$`) + +// ValidateRepoName enforces the OCI repository-name grammar and 2-256 length +// bound. It returns nil for a valid name and a descriptive error otherwise. +func ValidateRepoName(name string) error { + if len(name) < 2 || len(name) > 256 { + return errors.New("repository name must be 2-256 characters") + } + if !repoNameRe.MatchString(name) { + return errors.New("repository name does not match the required format") + } + return nil +} + +// digestRe matches an OCI content digest ("sha256:<64 hex>"). +var digestRe = regexp.MustCompile(`^sha256:[a-f0-9]{64}$`) + +// ValidateDigest reports whether s is a well-formed sha256 content digest. +func ValidateDigest(s string) bool { + return digestRe.MatchString(s) +} + // BlobKey returns the account-pool key for a blob digest ("sha256:"). The // two-char shard is taken from the first hex bytes after the "sha256:" prefix, -// matching the on-disk layout used by upstream OCI registries. +// matching the on-disk layout used by upstream OCI registries. The digest is +// sanitized via DigestToken so the object key carries no ':' separator. func BlobKey(digest string) string { hex := digest if _, after, found := strings.Cut(digest, ":"); found { @@ -65,5 +161,5 @@ func BlobKey(digest string) string { if len(hex) >= 2 { shard = hex[:2] } - return fmt.Sprintf(blobKeyFmt, shard, digest) + return fmt.Sprintf(blobKeyFmt, shard, DigestToken(digest)) } diff --git a/spinifex/handlers/ecr/storage_test.go b/spinifex/handlers/ecr/storage_test.go index c695cda03..59bc56b1a 100644 --- a/spinifex/handlers/ecr/storage_test.go +++ b/spinifex/handlers/ecr/storage_test.go @@ -1,6 +1,9 @@ package ecr -import "testing" +import ( + "strings" + "testing" +) func TestAccountBucket(t *testing.T) { if got := AccountBucket("123456789012"); got != "ecr-123456789012" { @@ -16,7 +19,7 @@ func TestKeyHelpers(t *testing.T) { }{ {"repo meta", RepoMetaKey("team/app"), "repos/team/app/_meta.json"}, {"tag", TagKey("team/app", "v1"), "repos/team/app/manifests/by-tag/v1"}, - {"manifest", ManifestKey("team/app", "sha256:abcd"), "repos/team/app/manifests/by-digest/sha256:abcd"}, + {"manifest", ManifestKey("team/app", "sha256:abcd"), "repos/team/app/manifests/by-digest/sha256-abcd"}, {"index", IndexKey("team/app"), "repos/team/app/manifests/index.json"}, {"upload", UploadKey("u-1"), "uploads/u-1"}, } @@ -33,7 +36,7 @@ func TestBlobKey(t *testing.T) { digest string want string }{ - {"sha256 prefixed", "sha256:abcdef0123", "blobs/ab/sha256:abcdef0123"}, + {"sha256 prefixed", "sha256:abcdef0123", "blobs/ab/sha256-abcdef0123"}, {"bare hex", "abcdef0123", "blobs/ab/abcdef0123"}, {"short digest", "a", "blobs/a/a"}, } @@ -43,3 +46,15 @@ func TestBlobKey(t *testing.T) { } } } + +// TestObjectKeysAreColonFree guards the SigV4 invariant: predastore rejects +// signed requests whose object key contains ':', so digest-derived object keys +// must be sanitized. +func TestObjectKeysAreColonFree(t *testing.T) { + digest := "sha256:" + strings.Repeat("a", 64) + for _, k := range []string{BlobKey(digest), ManifestKey("team/app", digest)} { + if strings.Contains(k, ":") { + t.Errorf("object key must not contain ':': %q", k) + } + } +} diff --git a/spinifex/objectstore/ensurebucket_test.go b/spinifex/objectstore/ensurebucket_test.go new file mode 100644 index 000000000..1bd938b5a --- /dev/null +++ b/spinifex/objectstore/ensurebucket_test.go @@ -0,0 +1,108 @@ +package objectstore + +import ( + "net/http" + "net/http/httptest" + "sync" + "testing" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/aws/credentials" + "github.com/aws/aws-sdk-go/aws/session" + "github.com/aws/aws-sdk-go/service/s3" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// fakeS3 records the HEAD/PUT calls EnsureBucket makes and replies with +// canned status/body so the create-when-absent logic can be exercised without +// a real predastore. +type fakeS3 struct { + mu sync.Mutex + heads int + puts int + headCode int + putCode int + putBody string +} + +func (f *fakeS3) ServeHTTP(w http.ResponseWriter, r *http.Request) { + f.mu.Lock() + defer f.mu.Unlock() + switch r.Method { + case http.MethodHead: + f.heads++ + w.WriteHeader(f.headCode) + case http.MethodPut: + f.puts++ + if f.putBody != "" { + w.WriteHeader(f.putCode) + _, _ = w.Write([]byte(f.putBody)) + return + } + w.WriteHeader(f.putCode) + default: + w.WriteHeader(http.StatusBadRequest) + } +} + +func newS3Store(t *testing.T, url string) *S3ObjectStore { + t.Helper() + sess := session.Must(session.NewSession(&aws.Config{ + Endpoint: aws.String(url), + Region: aws.String("ap-southeast-2"), + Credentials: credentials.NewStaticCredentials("ak", "sk", ""), + S3ForcePathStyle: aws.Bool(true), + DisableSSL: aws.Bool(true), + })) + return NewS3ObjectStore(s3.New(sess)) +} + +func TestS3ObjectStore_EnsureBucket_HeadOK(t *testing.T) { + fake := &fakeS3{headCode: http.StatusOK} + srv := httptest.NewServer(fake) + defer srv.Close() + + require.NoError(t, newS3Store(t, srv.URL).EnsureBucket("ecr-000000000001")) + assert.Equal(t, 1, fake.heads) + assert.Equal(t, 0, fake.puts, "present bucket must not be re-created") +} + +func TestS3ObjectStore_EnsureBucket_CreatesWhenAbsent(t *testing.T) { + fake := &fakeS3{headCode: http.StatusNotFound, putCode: http.StatusOK} + srv := httptest.NewServer(fake) + defer srv.Close() + + require.NoError(t, newS3Store(t, srv.URL).EnsureBucket("ecr-000000000001")) + assert.Equal(t, 1, fake.heads) + assert.Equal(t, 1, fake.puts) +} + +func TestS3ObjectStore_EnsureBucket_AlreadyOwnedIsSuccess(t *testing.T) { + fake := &fakeS3{ + headCode: http.StatusNotFound, + putCode: http.StatusConflict, + putBody: `BucketAlreadyOwnedByYouowned`, + } + srv := httptest.NewServer(fake) + defer srv.Close() + + require.NoError(t, newS3Store(t, srv.URL).EnsureBucket("ecr-000000000001")) + assert.Equal(t, 1, fake.puts) +} + +func TestS3ObjectStore_EnsureBucket_BackendErrorPropagates(t *testing.T) { + fake := &fakeS3{ + headCode: http.StatusNotFound, + putCode: http.StatusInternalServerError, + putBody: `InternalErrorboom`, + } + srv := httptest.NewServer(fake) + defer srv.Close() + + require.Error(t, newS3Store(t, srv.URL).EnsureBucket("ecr-000000000001")) +} + +func TestMemoryObjectStore_EnsureBucketNoop(t *testing.T) { + require.NoError(t, NewMemoryObjectStore().EnsureBucket("any")) +} diff --git a/spinifex/objectstore/objectstore.go b/spinifex/objectstore/objectstore.go index 561fbc00b..ac030ba4d 100644 --- a/spinifex/objectstore/objectstore.go +++ b/spinifex/objectstore/objectstore.go @@ -38,9 +38,13 @@ func IsNoSuchKeyError(err error) bool { // ObjectStore defines the interface for S3-like object storage operations. type ObjectStore interface { GetObject(input *s3.GetObjectInput) (*s3.GetObjectOutput, error) + HeadObject(input *s3.HeadObjectInput) (*s3.HeadObjectOutput, error) PutObject(input *s3.PutObjectInput) (*s3.PutObjectOutput, error) DeleteObject(input *s3.DeleteObjectInput) (*s3.DeleteObjectOutput, error) ListObjectsV2(input *s3.ListObjectsV2Input) (*s3.ListObjectsV2Output, error) + // EnsureBucket idempotently makes bucket exist. It is safe to call + // concurrently and on an already-present bucket. + EnsureBucket(bucket string) error } // NewS3ObjectStoreFromConfig creates an S3ObjectStore from Predastore connection parameters. @@ -79,6 +83,18 @@ func (s *S3ObjectStore) GetObject(input *s3.GetObjectInput) (*s3.GetObjectOutput return out, nil } +func (s *S3ObjectStore) HeadObject(input *s3.HeadObjectInput) (*s3.HeadObjectOutput, error) { + out, err := s.client.HeadObject(input) + if err != nil { + if aerr, ok := err.(awserr.Error); ok && + (aerr.Code() == s3.ErrCodeNoSuchKey || aerr.Code() == "NotFound") { + return nil, &NoSuchKeyError{Key: aws.StringValue(input.Key)} + } + return nil, err + } + return out, nil +} + func (s *S3ObjectStore) PutObject(input *s3.PutObjectInput) (*s3.PutObjectOutput, error) { return s.client.PutObject(input) } @@ -91,6 +107,26 @@ func (s *S3ObjectStore) ListObjectsV2(input *s3.ListObjectsV2Input) (*s3.ListObj return s.client.ListObjectsV2(input) } +// EnsureBucket creates bucket when absent. A successful HeadBucket short-circuits +// the create; an already-owned/existing bucket from a racing creator is treated +// as success so concurrent callers converge. +func (s *S3ObjectStore) EnsureBucket(bucket string) error { + if _, err := s.client.HeadBucket(&s3.HeadBucketInput{Bucket: aws.String(bucket)}); err == nil { + return nil + } + _, err := s.client.CreateBucket(&s3.CreateBucketInput{Bucket: aws.String(bucket)}) + if err != nil { + if aerr, ok := err.(awserr.Error); ok { + switch aerr.Code() { + case s3.ErrCodeBucketAlreadyOwnedByYou, s3.ErrCodeBucketAlreadyExists: + return nil + } + } + return err + } + return nil +} + // MemoryObjectStore implements ObjectStore using in-memory storage for testing. type MemoryObjectStore struct { objects map[string][]byte // key: bucket/key -> value: object data @@ -127,6 +163,21 @@ func (m *MemoryObjectStore) GetObject(input *s3.GetObjectInput) (*s3.GetObjectOu }, nil } +func (m *MemoryObjectStore) HeadObject(input *s3.HeadObjectInput) (*s3.HeadObjectOutput, error) { + m.mu.RLock() + defer m.mu.RUnlock() + + storageKey := makeKey(*input.Bucket, *input.Key) + data, exists := m.objects[storageKey] + if !exists { + return nil, &NoSuchKeyError{Key: *input.Key} + } + + return &s3.HeadObjectOutput{ + ContentLength: aws.Int64(int64(len(data))), + }, nil +} + func (m *MemoryObjectStore) PutObject(input *s3.PutObjectInput) (*s3.PutObjectOutput, error) { m.mu.Lock() defer m.mu.Unlock() @@ -208,6 +259,10 @@ func (m *MemoryObjectStore) ListObjectsV2(input *s3.ListObjectsV2Input) (*s3.Lis }, nil } +// EnsureBucket is a no-op: the memory store has no bucket namespace, keys are +// prefixed with the bucket name on write. +func (m *MemoryObjectStore) EnsureBucket(bucket string) error { return nil } + // Clear removes all objects from the memory store (useful for test cleanup) func (m *MemoryObjectStore) Clear() { m.mu.Lock() diff --git a/spinifex/objectstore/objectstore_test.go b/spinifex/objectstore/objectstore_test.go index 90aced2ee..9dd92eed7 100644 --- a/spinifex/objectstore/objectstore_test.go +++ b/spinifex/objectstore/objectstore_test.go @@ -56,6 +56,23 @@ func TestMemoryObjectStore_GetNonExistent(t *testing.T) { assert.Equal(t, "NoSuchKey", noSuchKey.Code()) } +func TestMemoryObjectStore_HeadObject(t *testing.T) { + store := NewMemoryObjectStore() + _, err := store.PutObject(&s3.PutObjectInput{ + Bucket: aws.String("b"), + Key: aws.String("k"), + Body: bytes.NewReader([]byte("twelve bytes")), + }) + require.NoError(t, err) + + out, err := store.HeadObject(&s3.HeadObjectInput{Bucket: aws.String("b"), Key: aws.String("k")}) + require.NoError(t, err) + assert.Equal(t, int64(12), *out.ContentLength) + + _, err = store.HeadObject(&s3.HeadObjectInput{Bucket: aws.String("b"), Key: aws.String("missing")}) + assert.True(t, IsNoSuchKeyError(err)) +} + func TestMemoryObjectStore_Delete(t *testing.T) { store := NewMemoryObjectStore() diff --git a/spinifex/services/awsgw/awsgw.go b/spinifex/services/awsgw/awsgw.go index 7f5d91399..a39e3b036 100644 --- a/spinifex/services/awsgw/awsgw.go +++ b/spinifex/services/awsgw/awsgw.go @@ -16,8 +16,11 @@ import ( "github.com/mulgadc/spinifex/spinifex/admin" "github.com/mulgadc/spinifex/spinifex/config" "github.com/mulgadc/spinifex/spinifex/gateway" + gateway_ecr "github.com/mulgadc/spinifex/spinifex/gateway/ecr" + "github.com/mulgadc/spinifex/spinifex/handlers/ecr" handlers_iam "github.com/mulgadc/spinifex/spinifex/handlers/iam" handlers_sts "github.com/mulgadc/spinifex/spinifex/handlers/sts" + "github.com/mulgadc/spinifex/spinifex/objectstore" "github.com/mulgadc/spinifex/spinifex/utils" "github.com/nats-io/nats.go" toml "github.com/pelletier/go-toml/v2" @@ -196,6 +199,19 @@ func launchService(config *config.ClusterConfig) error { slog.Warn("Failed to load throttle config, throttling disabled", "err", err) } + // OCI Distribution v2 registry: blob/manifest bytes stream straight to + // predastore from the gateway; repo/tag/manifest metadata and in-progress + // uploads are owned by the daemon and reached over NATS request/reply. + // Account scoping uses the bootstrap account until the /v2 token-auth + // bridge lands. + ecrStore := objectstore.NewS3ObjectStoreFromConfig( + admin.DialTarget(nodeConfig.Predastore.Host), + nodeConfig.Predastore.Region, + nodeConfig.Predastore.AccessKey, + nodeConfig.Predastore.SecretKey, + ) + ecrRegistry := gateway_ecr.NewRegistry(ecrStore, ecr.NewNATSMetaStore(natsConn), config.Bootstrap.AccountID) + gw := gateway.GatewayConfig{ Debug: nodeConfig.AWSGW.Debug, DisableLogging: false, @@ -209,6 +225,7 @@ func launchService(config *config.ClusterConfig) error { STSService: stsService, Version: version, Commit: commit, + ECRRegistry: ecrRegistry, } if throttleCfg.Enabled {