feat(ecr): repository policy passthrough + DescribeRepositories scoping (Sprint 2e) (#385)

Add Set/Get/DeleteRepositoryPolicy as JSON policy-doc passthrough stored at
KV repos/{name}/policy, and DescribeRepositories scoped to the caller account
with registryId override (cross-account denied pending registry-policy v2).

- MetaStore: PutRepoPolicy/GetRepoPolicy/DeleteRepoPolicy (KV + memory)
- NATS surface: ecr.policy.{put,get,delete} + daemon handlers
- gateway/ecrapi: repository-policy control-plane handlers
- gateway: handleDescribeRepositories dispatched inline for region/suffix
- awserrors: RepositoryNotFoundException, RepositoryPolicyNotFoundException
This commit is contained in:
Josh Siviour
2026-06-18 18:23:22 +10:00
committed by GitHub
parent a4a4288d7d
commit 2faad4b9e7
17 changed files with 758 additions and 9 deletions
+8
View File
@@ -453,6 +453,10 @@ var (
ErrorIAMMalformedPolicyDocument = "MalformedPolicyDocument"
ErrorAccessDenied = "AccessDenied"
// ECR-specific error codes
ErrorRepositoryNotFound = "RepositoryNotFoundException"
ErrorRepositoryPolicyNotFound = "RepositoryPolicyNotFoundException"
// ELBv2-specific error codes
ErrorELBv2LoadBalancerNotFound = "LoadBalancerNotFound"
ErrorELBv2TargetGroupNotFound = "TargetGroupNotFound"
@@ -957,6 +961,10 @@ var ErrorLookup = map[string]ErrorMessage{
ErrorIAMMalformedPolicyDocument: {HTTPCode: 400, Message: "The policy document is malformed."},
ErrorAccessDenied: {HTTPCode: 403, Message: "User is not authorized to perform this action."},
// ECR error codes
ErrorRepositoryNotFound: {HTTPCode: 400, Message: "The repository could not be found. Check the spelling of the specified repository and ensure that you are performing operations on the correct registry."},
ErrorRepositoryPolicyNotFound: {HTTPCode: 400, Message: "The repository and registry do not have an associated repository policy."},
// ELBv2 error codes
ErrorELBv2LoadBalancerNotFound: {HTTPCode: 400, Message: "One or more load balancers not found."},
ErrorELBv2TargetGroupNotFound: {HTTPCode: 400, Message: "One or more target groups not found."},
+3
View File
@@ -965,6 +965,9 @@ func (d *Daemon) subscribeAll() error {
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.SubjectPolicyPut, d.handleECRPolicyPut, "spinifex-workers"},
natsSub{handlers_ecr.SubjectPolicyGet, d.handleECRPolicyGet, "spinifex-workers"},
natsSub{handlers_ecr.SubjectPolicyDelete, d.handleECRPolicyDelete, "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"},
+12
View File
@@ -14,6 +14,18 @@ func (d *Daemon) handleECRRepoList(msg *nats.Msg) {
handleNATSRequest(msg, d.ecrMetaService.RepoList)
}
func (d *Daemon) handleECRPolicyPut(msg *nats.Msg) {
handleNATSRequest(msg, d.ecrMetaService.PolicyPut)
}
func (d *Daemon) handleECRPolicyGet(msg *nats.Msg) {
handleNATSRequest(msg, d.ecrMetaService.PolicyGet)
}
func (d *Daemon) handleECRPolicyDelete(msg *nats.Msg) {
handleNATSRequest(msg, d.ecrMetaService.PolicyDelete)
}
func (d *Daemon) handleECRTagPut(msg *nats.Msg) {
handleNATSRequest(msg, d.ecrMetaService.TagPut)
}
@@ -0,0 +1,93 @@
package gateway
import (
"encoding/json"
"errors"
"io"
"log/slog"
"net/http"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ecr"
"github.com/mulgadc/spinifex/spinifex/awserrors"
gateway_ecrapi "github.com/mulgadc/spinifex/spinifex/gateway/ecrapi"
handlers_ecr "github.com/mulgadc/spinifex/spinifex/handlers/ecr"
)
// describeRepositoriesRequest is the camelCase AWS JSON 1.1 input shape. The SDK
// input struct carries locationName tags rather than json tags, so the subset
// 2e honors is decoded explicitly.
type describeRepositoriesRequest struct {
RegistryID string `json:"registryId"`
RepositoryNames []string `json:"repositoryNames"`
}
// handleDescribeRepositories lists the caller account's repositories. Scope is
// the caller account; a registryId naming a different account is the Q8 parity
// gap pending registry-policy v2 and is denied. Pagination (maxResults/
// nextToken, Q9) is not yet implemented — the full list is returned in one page.
func (gw *GatewayConfig) handleDescribeRepositories(w http.ResponseWriter, r *http.Request) error {
accountID, _ := r.Context().Value(ctxAccountID).(string)
if accountID == "" {
slog.Error("DescribeRepositories: no account ID in auth context")
return errors.New(awserrors.ErrorServerInternal)
}
body, err := io.ReadAll(r.Body)
if err != nil {
slog.Error("DescribeRepositories: failed to read body", "err", err)
return errors.New(awserrors.ErrorInvalidParameterValue)
}
var req describeRepositoriesRequest
if len(body) > 0 {
if err := json.Unmarshal(body, &req); err != nil {
return errors.New(awserrors.ErrorInvalidParameterValue)
}
}
if req.RegistryID != "" && req.RegistryID != accountID {
return errors.New(awserrors.ErrorAccessDenied)
}
store := handlers_ecr.NewNATSMetaStore(gw.NATSConn)
names := req.RepositoryNames
if len(names) == 0 {
names, err = store.ListRepos(accountID)
if err != nil {
slog.Error("DescribeRepositories: list repos failed", "err", err)
return errors.New(awserrors.ErrorServerInternal)
}
}
repos := make([]*ecr.Repository, 0, len(names))
for _, name := range names {
meta, err := store.GetRepo(accountID, name)
if err != nil {
if errors.Is(err, handlers_ecr.ErrNotFound) {
return errors.New(awserrors.ErrorRepositoryNotFound)
}
slog.Error("DescribeRepositories: get repo failed", "repo", name, "err", err)
return errors.New(awserrors.ErrorServerInternal)
}
repos = append(repos, &ecr.Repository{
RegistryId: aws.String(accountID),
RepositoryName: aws.String(name),
RepositoryArn: aws.String(gw.ecrRepositoryArn(accountID, name)),
RepositoryUri: aws.String(gw.ecrRepositoryUri(accountID, name)),
CreatedAt: aws.Time(meta.CreatedAt),
ImageTagMutability: aws.String("MUTABLE"),
})
}
gateway_ecrapi.WriteJSONResponse(w, &ecr.DescribeRepositoriesOutput{Repositories: repos})
return nil
}
// ecrRepositoryArn builds the ECR repository ARN for an account-scoped repo.
func (gw *GatewayConfig) ecrRepositoryArn(accountID, name string) string {
return "arn:aws:ecr:" + gw.Region + ":" + accountID + ":repository/" + name
}
// ecrRepositoryUri builds the registry pull/push URI for an account-scoped repo.
func (gw *GatewayConfig) ecrRepositoryUri(accountID, name string) string {
return accountID + ".dkr.ecr." + gw.Region + "." + gw.InternalSuffix + "/" + name
}
@@ -0,0 +1,135 @@
package gateway
import (
"context"
"encoding/json"
"net/http"
"net/http/httptest"
"strings"
"testing"
"time"
handlers_ecr "github.com/mulgadc/spinifex/spinifex/handlers/ecr"
"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"
)
// serveECRMeta wires a MetaService method to a NATS subject for the gateway-side
// DescribeRepositories tests, mirroring the daemon's handleNATSRequest.
func serveECRMeta[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() })
}
func newDescribeReposGateway(t *testing.T, repos ...string) *GatewayConfig {
t.Helper()
_, nc, js := testutil.StartTestJetStream(t)
svc := handlers_ecr.NewKVMetaService(js)
serveECRMeta(t, nc, handlers_ecr.SubjectRepoCreate, svc.RepoCreate)
serveECRMeta(t, nc, handlers_ecr.SubjectRepoDescribe, svc.RepoDescribe)
serveECRMeta(t, nc, handlers_ecr.SubjectRepoList, svc.RepoList)
store := handlers_ecr.NewNATSMetaStore(nc)
for _, r := range repos {
require.NoError(t, store.PutRepo(ecrTestAccount, handlers_ecr.RepoMeta{Name: r, CreatedAt: time.Now()}))
}
return &GatewayConfig{
NATSConn: nc, Region: ecrTestRegion, InternalSuffix: ecrTestSuffix, DisableLogging: true,
}
}
func describeReposRequest(t *testing.T, gw *GatewayConfig, body string) *httptest.ResponseRecorder {
t.Helper()
req := httptest.NewRequest(http.MethodPost, "/", strings.NewReader(body))
ctx := context.WithValue(req.Context(), ctxAccountID, ecrTestAccount)
w := httptest.NewRecorder()
require.NoError(t, gw.handleDescribeRepositories(w, req.WithContext(ctx)))
return w
}
type describeReposOut struct {
Repositories []struct {
RepositoryName string `json:"repositoryName"`
RegistryID string `json:"registryId"`
RepositoryArn string `json:"repositoryArn"`
RepositoryURI string `json:"repositoryUri"`
ImageTagMutability string `json:"imageTagMutability"`
CreatedAt float64 `json:"createdAt"`
} `json:"repositories"`
}
func TestDescribeRepositories_ListsAccountScoped(t *testing.T) {
gw := newDescribeReposGateway(t, "team/app", "team/web")
w := describeReposRequest(t, gw, "{}")
require.Equal(t, http.StatusOK, w.Code)
var out describeReposOut
require.NoError(t, json.Unmarshal(w.Body.Bytes(), &out))
require.Len(t, out.Repositories, 2)
byName := map[string]int{}
for i, r := range out.Repositories {
byName[r.RepositoryName] = i
}
app := out.Repositories[byName["team/app"]]
assert.Equal(t, ecrTestAccount, app.RegistryID)
assert.Equal(t, "arn:aws:ecr:"+ecrTestRegion+":"+ecrTestAccount+":repository/team/app", app.RepositoryArn)
assert.Equal(t, ecrTestAccount+".dkr.ecr."+ecrTestRegion+"."+ecrTestSuffix+"/team/app", app.RepositoryURI)
assert.Equal(t, "MUTABLE", app.ImageTagMutability)
assert.Positive(t, app.CreatedAt)
}
func TestDescribeRepositories_NameFilter(t *testing.T) {
gw := newDescribeReposGateway(t, "team/app", "team/web")
w := describeReposRequest(t, gw, `{"repositoryNames":["team/web"]}`)
var out describeReposOut
require.NoError(t, json.Unmarshal(w.Body.Bytes(), &out))
require.Len(t, out.Repositories, 1)
assert.Equal(t, "team/web", out.Repositories[0].RepositoryName)
}
func TestDescribeRepositories_MissingNamedRepo(t *testing.T) {
gw := newDescribeReposGateway(t, "team/app")
req := httptest.NewRequest(http.MethodPost, "/", strings.NewReader(`{"repositoryNames":["team/ghost"]}`))
ctx := context.WithValue(req.Context(), ctxAccountID, ecrTestAccount)
err := gw.handleDescribeRepositories(httptest.NewRecorder(), req.WithContext(ctx))
require.Error(t, err)
assert.Equal(t, "RepositoryNotFoundException", err.Error())
}
func TestDescribeRepositories_CrossAccountDenied(t *testing.T) {
gw := newDescribeReposGateway(t, "team/app")
req := httptest.NewRequest(http.MethodPost, "/", strings.NewReader(`{"registryId":"999999999999"}`))
ctx := context.WithValue(req.Context(), ctxAccountID, ecrTestAccount)
err := gw.handleDescribeRepositories(httptest.NewRecorder(), req.WithContext(ctx))
require.Error(t, err)
assert.Equal(t, "AccessDenied", err.Error())
}
func TestECRRequest_DescribeRepositoriesDispatched(t *testing.T) {
gw := newDescribeReposGateway(t, "team/app")
req := setupECRRequest("AmazonEC2ContainerRegistry_V20150921.DescribeRepositories", "{}")
ctx := context.WithValue(req.Context(), ctxAccountID, ecrTestAccount)
w := httptest.NewRecorder()
require.NoError(t, gw.ECR_Request(w, req.WithContext(ctx)))
assert.Equal(t, http.StatusOK, w.Code)
}
+7
View File
@@ -45,6 +45,13 @@ func (gw *GatewayConfig) ECR_Request(w http.ResponseWriter, r *http.Request) err
return gw.handleGetAuthorizationToken(w, r)
}
// DescribeRepositories builds repository ARNs/URIs from the gateway region
// and internal suffix, which the relayed Handler signature does not carry,
// so it is served inline like GetAuthorizationToken.
if action == "DescribeRepositories" {
return gw.handleDescribeRepositories(w, r)
}
accountID, _ := r.Context().Value(ctxAccountID).(string)
if accountID == "" {
slog.Error("ECR_Request: no account ID in auth context")
+3 -3
View File
@@ -63,9 +63,9 @@ var Actions = map[string]Handler{
"CompleteLayerUpload": NotImplemented,
// Repository / registry policy.
"SetRepositoryPolicy": NotImplemented,
"GetRepositoryPolicy": NotImplemented,
"DeleteRepositoryPolicy": NotImplemented,
"SetRepositoryPolicy": SetRepositoryPolicy,
"GetRepositoryPolicy": GetRepositoryPolicy,
"DeleteRepositoryPolicy": DeleteRepositoryPolicy,
"GetRegistryPolicy": NotImplemented,
"PutRegistryPolicy": NotImplemented,
"DescribeRegistry": NotImplemented,
+113
View File
@@ -0,0 +1,113 @@
package gateway_ecrapi
import (
"encoding/json"
"errors"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ecr"
"github.com/mulgadc/spinifex/spinifex/awserrors"
handlers_ecr "github.com/mulgadc/spinifex/spinifex/handlers/ecr"
"github.com/nats-io/nats.go"
)
// repoPolicyRequest is the shared input shape for the three repository-policy
// actions. The AWS JSON 1.1 wire keys are camelCase locationNames; the SDK
// input structs carry no json tags, so the fields are decoded explicitly.
type repoPolicyRequest struct {
RepositoryName string `json:"repositoryName"`
RegistryID string `json:"registryId"`
PolicyText string `json:"policyText"`
}
// resolvePolicyRepo parses and validates the request, enforces the registryId
// cross-account guard, and confirms the repository exists. It returns the parsed
// request and the NATS-backed MetaStore for the follow-on policy operation.
func resolvePolicyRepo(nc *nats.Conn, accountID string, body []byte) (repoPolicyRequest, *handlers_ecr.NATSMetaStore, error) {
var req repoPolicyRequest
if len(body) > 0 {
if err := json.Unmarshal(body, &req); err != nil {
return req, nil, errors.New(awserrors.ErrorInvalidParameterValue)
}
}
if req.RepositoryName == "" || handlers_ecr.ValidateRepoName(req.RepositoryName) != nil {
return req, nil, errors.New(awserrors.ErrorInvalidParameterValue)
}
// Cross-account registry access is the Q8 parity gap pending registry-policy
// v2; a registryId naming a different account is denied.
if req.RegistryID != "" && req.RegistryID != accountID {
return req, nil, errors.New(awserrors.ErrorAccessDenied)
}
store := handlers_ecr.NewNATSMetaStore(nc)
if _, err := store.GetRepo(accountID, req.RepositoryName); err != nil {
if errors.Is(err, handlers_ecr.ErrNotFound) {
return req, nil, errors.New(awserrors.ErrorRepositoryNotFound)
}
return req, nil, err
}
return req, store, nil
}
// SetRepositoryPolicy stores the JSON IAM policy document for a repository. The
// policy is passthrough metadata in v1: it is persisted and returned but not
// evaluated for cross-account access (Q8; registry-policy evaluator is v2).
func SetRepositoryPolicy(nc *nats.Conn, accountID string, body []byte) (any, error) {
req, store, err := resolvePolicyRepo(nc, accountID, body)
if err != nil {
return nil, err
}
if req.PolicyText == "" || !json.Valid([]byte(req.PolicyText)) {
return nil, errors.New(awserrors.ErrorInvalidParameterValue)
}
if err := store.PutRepoPolicy(accountID, req.RepositoryName, []byte(req.PolicyText)); err != nil {
return nil, err
}
return &ecr.SetRepositoryPolicyOutput{
RegistryId: aws.String(accountID),
RepositoryName: aws.String(req.RepositoryName),
PolicyText: aws.String(req.PolicyText),
}, nil
}
// GetRepositoryPolicy returns the stored policy document, or
// RepositoryPolicyNotFoundException when none is set.
func GetRepositoryPolicy(nc *nats.Conn, accountID string, body []byte) (any, error) {
req, store, err := resolvePolicyRepo(nc, accountID, body)
if err != nil {
return nil, err
}
policy, err := store.GetRepoPolicy(accountID, req.RepositoryName)
if err != nil {
if errors.Is(err, handlers_ecr.ErrNotFound) {
return nil, errors.New(awserrors.ErrorRepositoryPolicyNotFound)
}
return nil, err
}
return &ecr.GetRepositoryPolicyOutput{
RegistryId: aws.String(accountID),
RepositoryName: aws.String(req.RepositoryName),
PolicyText: aws.String(string(policy)),
}, nil
}
// DeleteRepositoryPolicy removes and returns the stored policy document, or
// RepositoryPolicyNotFoundException when none is set.
func DeleteRepositoryPolicy(nc *nats.Conn, accountID string, body []byte) (any, error) {
req, store, err := resolvePolicyRepo(nc, accountID, body)
if err != nil {
return nil, err
}
policy, err := store.DeleteRepoPolicy(accountID, req.RepositoryName)
if err != nil {
if errors.Is(err, handlers_ecr.ErrNotFound) {
return nil, errors.New(awserrors.ErrorRepositoryPolicyNotFound)
}
return nil, err
}
return &ecr.DeleteRepositoryPolicyOutput{
RegistryId: aws.String(accountID),
RepositoryName: aws.String(req.RepositoryName),
PolicyText: aws.String(string(policy)),
}, nil
}
@@ -0,0 +1,125 @@
package gateway_ecrapi
import (
"encoding/json"
"testing"
"time"
"github.com/aws/aws-sdk-go/service/ecr"
"github.com/mulgadc/spinifex/spinifex/awserrors"
handlers_ecr "github.com/mulgadc/spinifex/spinifex/handlers/ecr"
"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 policyTestAccount = "000000000000"
// serveMeta wires a MetaService method to a NATS subject, mirroring the daemon.
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() })
}
// newPolicyTestConn starts an embedded JetStream server with the repo + policy
// metadata subjects served, returning the gateway NATS connection.
func newPolicyTestConn(t *testing.T) *nats.Conn {
t.Helper()
_, nc, js := testutil.StartTestJetStream(t)
svc := handlers_ecr.NewKVMetaService(js)
serveMeta(t, nc, handlers_ecr.SubjectRepoCreate, svc.RepoCreate)
serveMeta(t, nc, handlers_ecr.SubjectRepoDescribe, svc.RepoDescribe)
serveMeta(t, nc, handlers_ecr.SubjectPolicyPut, svc.PolicyPut)
serveMeta(t, nc, handlers_ecr.SubjectPolicyGet, svc.PolicyGet)
serveMeta(t, nc, handlers_ecr.SubjectPolicyDelete, svc.PolicyDelete)
return nc
}
func seedRepo(t *testing.T, nc *nats.Conn, repo string) {
t.Helper()
store := handlers_ecr.NewNATSMetaStore(nc)
require.NoError(t, store.PutRepo(policyTestAccount, handlers_ecr.RepoMeta{Name: repo, CreatedAt: time.Now()}))
}
func TestRepositoryPolicy_Lifecycle(t *testing.T) {
nc := newPolicyTestConn(t)
seedRepo(t, nc, "team/app")
const policy = `{"Version":"2012-10-17","Statement":[]}`
body := []byte(`{"repositoryName":"team/app","policyText":` + strconvQuote(policy) + `}`)
out, err := SetRepositoryPolicy(nc, policyTestAccount, body)
require.NoError(t, err)
set, ok := out.(*ecr.SetRepositoryPolicyOutput)
require.True(t, ok)
assert.Equal(t, policy, *set.PolicyText)
assert.Equal(t, "team/app", *set.RepositoryName)
assert.Equal(t, policyTestAccount, *set.RegistryId)
out, err = GetRepositoryPolicy(nc, policyTestAccount, []byte(`{"repositoryName":"team/app"}`))
require.NoError(t, err)
got, ok := out.(*ecr.GetRepositoryPolicyOutput)
require.True(t, ok)
assert.Equal(t, policy, *got.PolicyText)
out, err = DeleteRepositoryPolicy(nc, policyTestAccount, []byte(`{"repositoryName":"team/app"}`))
require.NoError(t, err)
del, ok := out.(*ecr.DeleteRepositoryPolicyOutput)
require.True(t, ok)
assert.Equal(t, policy, *del.PolicyText)
// Policy gone after delete.
_, err = GetRepositoryPolicy(nc, policyTestAccount, []byte(`{"repositoryName":"team/app"}`))
require.Error(t, err)
assert.Equal(t, awserrors.ErrorRepositoryPolicyNotFound, err.Error())
}
func TestRepositoryPolicy_Errors(t *testing.T) {
nc := newPolicyTestConn(t)
seedRepo(t, nc, "team/app")
cases := []struct {
name string
fn func(*nats.Conn, string, []byte) (any, error)
body string
expect string
}{
{"set missing repo", SetRepositoryPolicy, `{"repositoryName":"team/ghost","policyText":"{}"}`, awserrors.ErrorRepositoryNotFound},
{"set invalid json policy", SetRepositoryPolicy, `{"repositoryName":"team/app","policyText":"not-json"}`, awserrors.ErrorInvalidParameterValue},
{"set empty name", SetRepositoryPolicy, `{"policyText":"{}"}`, awserrors.ErrorInvalidParameterValue},
{"set cross-account", SetRepositoryPolicy, `{"repositoryName":"team/app","registryId":"999999999999","policyText":"{}"}`, awserrors.ErrorAccessDenied},
{"get no policy", GetRepositoryPolicy, `{"repositoryName":"team/app"}`, awserrors.ErrorRepositoryPolicyNotFound},
{"delete no policy", DeleteRepositoryPolicy, `{"repositoryName":"team/app"}`, awserrors.ErrorRepositoryPolicyNotFound},
{"get cross-account", GetRepositoryPolicy, `{"repositoryName":"team/app","registryId":"999999999999"}`, awserrors.ErrorAccessDenied},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
_, err := tc.fn(nc, policyTestAccount, []byte(tc.body))
require.Error(t, err)
assert.Equal(t, tc.expect, err.Error())
})
}
}
// strconvQuote JSON-quotes a string for inline test bodies.
func strconvQuote(s string) string {
b, _ := json.Marshal(s)
return string(b)
}
+75
View File
@@ -60,6 +60,10 @@ type MetaStore interface {
GetRepo(accountID, repo string) (RepoMeta, error)
ListRepos(accountID string) ([]string, error)
PutRepoPolicy(accountID, repo string, policyText []byte) error
GetRepoPolicy(accountID, repo string) ([]byte, error)
DeleteRepoPolicy(accountID, repo string) ([]byte, error)
PutTag(accountID, repo, tag, digest string) error
GetTag(accountID, repo, tag string) (string, error)
DeleteTag(accountID, repo, tag string) error
@@ -159,6 +163,44 @@ func (s *KVMetaStore) ListRepos(accountID string) ([]string, error) {
return repos, nil
}
func (s *KVMetaStore) PutRepoPolicy(accountID, repo string, policyText []byte) error {
kv, err := s.bucket(accountID)
if err != nil {
return err
}
_, err = kv.Put(KVRepoPolicyKey(repo), policyText)
return err
}
func (s *KVMetaStore) GetRepoPolicy(accountID, repo string) ([]byte, error) {
kv, err := s.bucket(accountID)
if err != nil {
return nil, err
}
entry, err := kv.Get(KVRepoPolicyKey(repo))
if err != nil {
return nil, mapKVErr(err)
}
return entry.Value(), nil
}
func (s *KVMetaStore) DeleteRepoPolicy(accountID, repo string) ([]byte, error) {
kv, err := s.bucket(accountID)
if err != nil {
return nil, err
}
// JetStream KV Delete is silently idempotent, so the value is read first to
// surface a real not-found and return the deleted document to the caller.
entry, err := kv.Get(KVRepoPolicyKey(repo))
if err != nil {
return nil, mapKVErr(err)
}
if err := kv.Delete(KVRepoPolicyKey(repo)); err != nil {
return nil, mapKVErr(err)
}
return entry.Value(), nil
}
func (s *KVMetaStore) PutTag(accountID, repo, tag, digest string) error {
kv, err := s.bucket(accountID)
if err != nil {
@@ -316,6 +358,7 @@ func (s *KVMetaStore) DeleteUpload(accountID, uploadID string) error {
type MemoryMetaStore struct {
mu sync.Mutex
repos map[string]map[string]RepoMeta // account -> repo -> meta
policies map[string]map[string][]byte // account -> repo -> policyText
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
@@ -332,6 +375,7 @@ var _ MetaStore = (*MemoryMetaStore)(nil)
func NewMemoryMetaStore() *MemoryMetaStore {
return &MemoryMetaStore{
repos: make(map[string]map[string]RepoMeta),
policies: make(map[string]map[string][]byte),
tags: make(map[string]map[string]string),
manifests: make(map[string]map[string]ManifestMeta),
uploads: make(map[string]map[string]uploadRev),
@@ -369,6 +413,37 @@ func (m *MemoryMetaStore) ListRepos(accountID string) ([]string, error) {
return out, nil
}
func (m *MemoryMetaStore) PutRepoPolicy(accountID, repo string, policyText []byte) error {
m.mu.Lock()
defer m.mu.Unlock()
if m.policies[accountID] == nil {
m.policies[accountID] = make(map[string][]byte)
}
m.policies[accountID][repo] = policyText
return nil
}
func (m *MemoryMetaStore) GetRepoPolicy(accountID, repo string) ([]byte, error) {
m.mu.Lock()
defer m.mu.Unlock()
p, ok := m.policies[accountID][repo]
if !ok {
return nil, ErrNotFound
}
return p, nil
}
func (m *MemoryMetaStore) DeleteRepoPolicy(accountID, repo string) ([]byte, error) {
m.mu.Lock()
defer m.mu.Unlock()
p, ok := m.policies[accountID][repo]
if !ok {
return nil, ErrNotFound
}
delete(m.policies[accountID], repo)
return p, nil
}
func (m *MemoryMetaStore) PutTag(accountID, repo, tag, digest string) error {
m.mu.Lock()
defer m.mu.Unlock()
+31
View File
@@ -78,3 +78,34 @@ func TestKVMetaStore_FullCycle(t *testing.T) {
require.NoError(t, err)
assert.Empty(t, empty)
}
// TestKVMetaStore_RepoPolicy exercises the policy passthrough against the
// JetStream-backed store and proves the policy key shares the repo bucket
// without being mistaken for a repository by ListRepos.
func TestKVMetaStore_RepoPolicy(t *testing.T) {
_, _, js := testutil.StartTestJetStream(t)
store := NewKVMetaStore(js)
const acct = "000000000000"
const policy = `{"Version":"2012-10-17","Statement":[]}`
require.NoError(t, store.PutRepo(acct, RepoMeta{Name: "team/app", CreatedAt: time.Now()}))
_, err := store.GetRepoPolicy(acct, "team/app")
assert.ErrorIs(t, err, ErrNotFound)
require.NoError(t, store.PutRepoPolicy(acct, "team/app", []byte(policy)))
got, err := store.GetRepoPolicy(acct, "team/app")
require.NoError(t, err)
assert.Equal(t, policy, string(got))
// The policy key must not appear as a repository.
repos, err := store.ListRepos(acct)
require.NoError(t, err)
assert.Equal(t, []string{"team/app"}, repos)
deleted, err := store.DeleteRepoPolicy(acct, "team/app")
require.NoError(t, err)
assert.Equal(t, policy, string(deleted))
_, err = store.DeleteRepoPolicy(acct, "team/app")
assert.ErrorIs(t, err, ErrNotFound)
}
+27
View File
@@ -48,6 +48,33 @@ func (s *NATSMetaStore) ListRepos(accountID string) ([]string, error) {
return resp.Repos, nil
}
func (s *NATSMetaStore) PutRepoPolicy(accountID, repo string, policyText []byte) error {
_, err := utils.NATSRequest[PolicyPutResponse](s.conn, SubjectPolicyPut, &PolicyPutRequest{Repo: repo, PolicyText: policyText}, metaRequestTimeout, accountID)
return err
}
func (s *NATSMetaStore) GetRepoPolicy(accountID, repo string) ([]byte, error) {
resp, err := utils.NATSRequest[PolicyGetResponse](s.conn, SubjectPolicyGet, &PolicyGetRequest{Repo: repo}, metaRequestTimeout, accountID)
if err != nil {
return nil, err
}
if !resp.Found {
return nil, ErrNotFound
}
return resp.PolicyText, nil
}
func (s *NATSMetaStore) DeleteRepoPolicy(accountID, repo string) ([]byte, error) {
resp, err := utils.NATSRequest[PolicyDeleteResponse](s.conn, SubjectPolicyDelete, &PolicyDeleteRequest{Repo: repo}, metaRequestTimeout, accountID)
if err != nil {
return nil, err
}
if !resp.Found {
return nil, ErrNotFound
}
return resp.PolicyText, 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
+22
View File
@@ -114,3 +114,25 @@ func TestMemoryMetaStore_UploadCAS(t *testing.T) {
func TestTrimSuffixMeta(t *testing.T) {
assert.Equal(t, "team/app", trimSuffixMeta("repos/team/app/meta"))
}
func TestMemoryMetaStore_RepoPolicy(t *testing.T) {
m := NewMemoryMetaStore()
const acct = "000000000000"
const policy = `{"Version":"2012-10-17"}`
_, err := m.GetRepoPolicy(acct, "team/app")
assert.ErrorIs(t, err, ErrNotFound)
_, err = m.DeleteRepoPolicy(acct, "team/app")
assert.ErrorIs(t, err, ErrNotFound)
require.NoError(t, m.PutRepoPolicy(acct, "team/app", []byte(policy)))
got, err := m.GetRepoPolicy(acct, "team/app")
require.NoError(t, err)
assert.Equal(t, policy, string(got))
deleted, err := m.DeleteRepoPolicy(acct, "team/app")
require.NoError(t, err)
assert.Equal(t, policy, string(deleted))
_, err = m.GetRepoPolicy(acct, "team/app")
assert.ErrorIs(t, err, ErrNotFound)
}
+33
View File
@@ -8,6 +8,10 @@ const (
SubjectRepoDescribe = "ecr.repo.describe"
SubjectRepoList = "ecr.repo.list"
SubjectPolicyPut = "ecr.policy.put"
SubjectPolicyGet = "ecr.policy.get"
SubjectPolicyDelete = "ecr.policy.delete"
SubjectTagPut = "ecr.tag.put"
SubjectTagGet = "ecr.tag.get"
SubjectTagList = "ecr.tag.list"
@@ -47,6 +51,31 @@ type RepoListResponse struct {
Repos []string `json:"repos"`
}
type PolicyPutRequest struct {
Repo string `json:"repo"`
PolicyText []byte `json:"policyText"`
}
type PolicyPutResponse struct{}
type PolicyGetRequest struct {
Repo string `json:"repo"`
}
type PolicyGetResponse struct {
Found bool `json:"found"`
PolicyText []byte `json:"policyText"`
}
type PolicyDeleteRequest struct {
Repo string `json:"repo"`
}
type PolicyDeleteResponse struct {
Found bool `json:"found"`
PolicyText []byte `json:"policyText"`
}
type TagPutRequest struct {
Repo string `json:"repo"`
Tag string `json:"tag"`
@@ -147,6 +176,10 @@ type MetaService interface {
RepoDescribe(req *RepoDescribeRequest, accountID string) (*RepoDescribeResponse, error)
RepoList(req *RepoListRequest, accountID string) (*RepoListResponse, error)
PolicyPut(req *PolicyPutRequest, accountID string) (*PolicyPutResponse, error)
PolicyGet(req *PolicyGetRequest, accountID string) (*PolicyGetResponse, error)
PolicyDelete(req *PolicyDeleteRequest, accountID string) (*PolicyDeleteResponse, error)
TagPut(req *TagPutRequest, accountID string) (*TagPutResponse, error)
TagGet(req *TagGetRequest, accountID string) (*TagGetResponse, error)
TagList(req *TagListRequest, accountID string) (*TagListResponse, error)
+29
View File
@@ -52,6 +52,35 @@ func (s *MetaServiceImpl) RepoList(_ *RepoListRequest, accountID string) (*RepoL
return &RepoListResponse{Repos: repos}, nil
}
func (s *MetaServiceImpl) PolicyPut(req *PolicyPutRequest, accountID string) (*PolicyPutResponse, error) {
if err := s.store.PutRepoPolicy(accountID, req.Repo, req.PolicyText); err != nil {
return nil, err
}
return &PolicyPutResponse{}, nil
}
func (s *MetaServiceImpl) PolicyGet(req *PolicyGetRequest, accountID string) (*PolicyGetResponse, error) {
policy, err := s.store.GetRepoPolicy(accountID, req.Repo)
if errors.Is(err, ErrNotFound) {
return &PolicyGetResponse{Found: false}, nil
}
if err != nil {
return nil, err
}
return &PolicyGetResponse{Found: true, PolicyText: policy}, nil
}
func (s *MetaServiceImpl) PolicyDelete(req *PolicyDeleteRequest, accountID string) (*PolicyDeleteResponse, error) {
policy, err := s.store.DeleteRepoPolicy(accountID, req.Repo)
if errors.Is(err, ErrNotFound) {
return &PolicyDeleteResponse{Found: false}, nil
}
if err != nil {
return nil, err
}
return &PolicyDeleteResponse{Found: true, PolicyText: policy}, 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
+32
View File
@@ -51,6 +51,38 @@ func TestMetaServiceImpl_FlagMapping(t *testing.T) {
assert.True(t, uu.Conflict)
}
// TestNATSMetaStore_RepoPolicy drives the policy passthrough round-trip
// (put/get/delete + not-found mapping) end to end over the embedded server.
func TestNATSMetaStore_RepoPolicy(t *testing.T) {
_, nc, js := testutil.StartTestJetStream(t)
svc := NewKVMetaService(js)
serveMeta(t, nc, SubjectPolicyPut, svc.PolicyPut)
serveMeta(t, nc, SubjectPolicyGet, svc.PolicyGet)
serveMeta(t, nc, SubjectPolicyDelete, svc.PolicyDelete)
var client MetaStore = NewNATSMetaStore(nc)
const acct = svcTestAccount
const policy = `{"Version":"2012-10-17"}`
// Unset policy round-trips as ErrNotFound for both get and delete.
_, err := client.GetRepoPolicy(acct, "team/app")
assert.ErrorIs(t, err, ErrNotFound)
_, err = client.DeleteRepoPolicy(acct, "team/app")
assert.ErrorIs(t, err, ErrNotFound)
require.NoError(t, client.PutRepoPolicy(acct, "team/app", []byte(policy)))
got, err := client.GetRepoPolicy(acct, "team/app")
require.NoError(t, err)
assert.Equal(t, policy, string(got))
deleted, err := client.DeleteRepoPolicy(acct, "team/app")
require.NoError(t, err)
assert.Equal(t, policy, string(deleted))
_, err = client.GetRepoPolicy(acct, "team/app")
assert.ErrorIs(t, err, ErrNotFound)
}
// 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)) {
+10 -6
View File
@@ -89,17 +89,21 @@ func KVAccountBucket(accountID string) string {
// 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"
kvRepoMetaKeyFmt = "repos/%s/meta"
kvRepoPolicyKeyFmt = "repos/%s/policy"
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) }
// KVRepoPolicyKey returns the KV key for a repository's policy document.
func KVRepoPolicyKey(repo string) string { return fmt.Sprintf(kvRepoPolicyKeyFmt, repo) }
// KVTagsPrefix returns the KV key prefix enumerating a repository's tags.
func KVTagsPrefix(repo string) string { return fmt.Sprintf(kvTagsPrefixFmt, repo) }