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:<hex>) were rejected
403 (reproducible with the aws CLI too). Route BlobKey and ManifestKey
digests through DigestToken so the object key carries no ':' separator.
This commit is contained in:
Josh Siviour
2026-06-18 10:51:26 +10:00
committed by GitHub
parent 996eb370f3
commit 332087fccc
22 changed files with 3181 additions and 12 deletions
+31
View File
@@ -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)
}
+55
View File
@@ -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)
}
+10 -5
View File
@@ -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)
}
})
}
+873
View File
@@ -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
}
@@ -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")
}
+376
View File
@@ -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()
@@ -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")
}
}
+4
View File
@@ -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{
+24
View File
@@ -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.<noun>.<verb>` shape (handler `handleECR<Noun><Verb>`):
```
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.
+495
View File
@@ -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")
}
+80
View File
@@ -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)
}
+144
View File
@@ -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
}
+116
View File
@@ -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"))
}
+162
View File
@@ -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)
}
+152
View File
@@ -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
}
+151
View File
@@ -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)
}
+100 -4
View File
@@ -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:<hex>") 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:<hex>"). 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))
}
+18 -3
View File
@@ -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)
}
}
}
+108
View File
@@ -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: `<Error><Code>BucketAlreadyOwnedByYou</Code><Message>owned</Message></Error>`,
}
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: `<Error><Code>InternalError</Code><Message>boom</Message></Error>`,
}
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"))
}
+55
View File
@@ -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()
+17
View File
@@ -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()
+17
View File
@@ -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 {