CAT-D3: distribution's mirror copies an outside image by its pinned digest into upstream/<host>/<path>

Until now mirror only answered a copy the store already held, so mesh-build-image had nothing to call for an upstream
base (design 51: an upstream image is copied by mirror with its pinned digest). The copy takes an anonymous pull
token, checks every manifest's and blob's bytes against its digest, copies an index's every platform first, and
fetches only the blobs the store lacks. tags was already served.

Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig

modules/distribution/cmd/distribution-tools/main.go

@@ -208,7 +208,7 @@ func tools(v *holder.Verbs, windows kept[holder.Window]) []stdio.Tool {
 			Run: func(args map[string]any) (any, error) {
 				return v.Tags(holder.ImageSeat, stdio.CallerOf(args), str(args, "name"))
 			}},
-		stdio.Tool{Name: holder.ImageSeat + "." + holder.VerbMirror, Description: "Where the store holds a copy of an outside image, by its pinned digest (mesh-build-image alone).",
+		stdio.Tool{Name: holder.ImageSeat + "." + holder.VerbMirror, Description: "Copy an outside image, pinned by its digest, into the store as upstream/<host>/<path> (every platform, only the parts it lacks) and answer its reference; says published (mesh-build-image alone).",
 			Input: obj(map[string]any{"source": s("<host>/<path>"), "digest": s("sha256:…")}, "source", "digest"),
 			Run: func(args map[string]any) (any, error) {
 				r, err := arg[holder.MirrorRequest](args)
@@ -286,7 +286,7 @@ func main() {
 		os.Exit(1)
 	}
 	base, _ := url.Parse("http://127.0.0.1:5001")
-	store := &holder.OCI{Base: base, Registry: os.Getenv("MESH_REGISTRY_REACH")}
+	store := &holder.OCI{Base: base, Registry: os.Getenv("MESH_REGISTRY_REACH"), Upstream: &holder.Upstream{}}
 	h := &holder.Holder{Callers: callers, Objects: objects{}, Builder: builderVersions{}, Store: store, Say: say, Finding: finding,
 		Transfers: kept[holder.Transfer]{state: holder.TransfersState, key: func(t holder.Transfer) string { return t.Transfer }}}
 	v := &holder.Verbs{Holder: h, Catalog: store, Holds: kept[holder.Hold]{state: "holds", key: holder.Hold.Key}}

modules/distribution/internal/holder/holder_test.go

@@ -139,6 +139,14 @@ func newBed(builderOK bool) *bed {
 	return b
 }
 
+func keys[V any](m map[string]V) []string {
+	var out []string
+	for k := range m {
+		out = append(out, k)
+	}
+	return out
+}
+
 func digestOf(b []byte) string { s := sha256.Sum256(b); return "sha256:" + hex.EncodeToString(s[:]) }
 
 const builder = "ace.mesh-build-go"
@@ -356,6 +364,54 @@ func TestTheHolderOfTheTwoImageSeats(t *testing.T) {
 			}
 		}
 	})
+	t.Run("mirror copies an outside image by its pinned digest, every platform, only the parts the store lacks", func(t *testing.T) {
+		up, store := newFakeRegistry(), newFakeRegistry()
+		ups, sts := httptest.NewServer(up), httptest.NewServer(store)
+		defer ups.Close()
+		defer sts.Close()
+		up.token, up.realm = "anon", ups.URL+"/token"
+		host := strings.TrimPrefix(ups.URL, "http://")
+		repo := "library/nats"
+		layer, config := []byte(strings.Repeat("l", 5000)), []byte(`{"os":"linux"}`)
+		up.blobs[repo+"@"+digestOf(layer)], up.blobs[repo+"@"+digestOf(config)] = layer, config
+		img, _ := json.Marshal(map[string]any{"schemaVersion": 2, "mediaType": mediaManifest,
+			"config": map[string]any{"mediaType": "application/vnd.oci.image.config.v1+json", "digest": digestOf(config), "size": len(config)},
+			"layers": []map[string]any{{"mediaType": mediaLayer, "digest": digestOf(layer), "size": len(layer)}}})
+		up.manifests[repo+":"+digestOf(img)], up.types[repo+":"+digestOf(img)] = img, mediaManifest
+		idx, _ := json.Marshal(map[string]any{"schemaVersion": 2, "mediaType": "application/vnd.oci.image.index.v1+json",
+			"manifests": []map[string]any{{"mediaType": mediaManifest, "digest": digestOf(img), "size": len(img)}}})
+		up.manifests[repo+":"+digestOf(idx)] = idx
+		base, _ := url.Parse(sts.URL)
+		o := &OCI{Base: base, Registry: "registry.mesh:5000", Upstream: &Upstream{Scheme: "http"}}
+		name := "upstream/" + host + "/" + repo
+		store.blobs[name+"@"+digestOf(config)] = config // held already: not fetched again
+		ref, err := o.Mirror(host+"/"+repo+":2", digestOf(idx))
+		if err != nil {
+			t.Fatal(err)
+		}
+		if want := "registry.mesh:5000/" + name + "@" + digestOf(idx); ref != want {
+			t.Errorf("reference %s, want %s", ref, want)
+		}
+		if store.manifests[name+":"+digestOf(idx)] == nil || store.manifests[name+":"+digestOf(img)] == nil || store.blobs[name+"@"+digestOf(layer)] == nil {
+			t.Errorf("the store holds %v and %v, want the index, its image and the layer", keys(store.manifests), keys(store.blobs))
+		}
+		if !reflect.DeepEqual(up.gets, []string{digestOf(layer)}) {
+			t.Errorf("blobs fetched %v, want only the layer the store lacked", up.gets)
+		}
+		up.manifests[repo+":"+digestOf(idx)] = []byte(`{"tampered":true}`)
+		if _, err := o.Mirror(host+"/"+repo, digestOf(idx)); err != nil {
+			t.Errorf("a copy held whole is answered at once, without asking the upstream: %v", err)
+		}
+		delete(store.manifests, name+":"+digestOf(idx))
+		if _, err := o.Mirror(host+"/"+repo, digestOf(idx)); err == nil || !strings.Contains(err.Error(), "does not hash") {
+			t.Errorf("a manifest whose bytes are not its digest: %v", err)
+		}
+		for in, want := range map[string]string{"nats:2": "upstream/docker.io/library/nats", "ghcr.io/o/p@sha256:ab": "upstream/ghcr.io/o/p", "localhost:5000/x": "upstream/localhost:5000/x"} {
+			if s, _ := ParseSource(in); s.Name() != want {
+				t.Errorf("ParseSource(%q) = %s, want %s", in, s.Name(), want)
+			}
+		}
+	})
 	t.Run("the windows", func(t *testing.T) {
 		nodes := []NodeRead{
 			{Node: "ace", Serves: true, Stated: []string{NewNamesCapability}, NamesArtifact: true, BuildModules: map[string]bool{"mesh-build-go": true}},

modules/distribution/internal/holder/mirror.go

@@ -0,0 +1,299 @@
+package holder
+
+import (
+	"crypto/sha256"
+	"encoding/hex"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"net/url"
+	"strconv"
+	"strings"
+)
+
+// Upstream reads an outside registry by the OCI distribution protocol, anonymously: on a 401 it takes the bearer
+// token the registry's WWW-Authenticate names, scoped to a pull of the one repository, and asks again (the form
+// Docker Hub, ghcr.io and quay.io answer). Scheme is "https" but in tests.
+type Upstream struct {
+	Client *http.Client
+	Scheme string
+}
+
+func (u *Upstream) client() *http.Client {
+	if u != nil && u.Client != nil {
+		return u.Client
+	}
+	return http.DefaultClient
+}
+
+func (u *Upstream) scheme() string {
+	if u != nil && u.Scheme != "" {
+		return u.Scheme
+	}
+	return "https"
+}
+
+// accepted are the manifest forms a mirror copies.
+var accepted = []string{
+	"application/vnd.oci.image.index.v1+json", mediaManifest,
+	"application/vnd.docker.distribution.manifest.list.v2+json", "application/vnd.docker.distribution.manifest.v2+json",
+}
+
+// Source is an outside image's place: its registry host (as reached) and its repository path, from `<host>/<path>`
+// with a tag or digest dropped. A name with no host is Docker Hub's, and a one-part path there is under library/.
+type Source struct {
+	Host, Path string
+}
+
+// ParseSource reads an outside image's name.
+func ParseSource(s string) (Source, error) {
+	s = strings.TrimSpace(s)
+	if i := strings.Index(s, "@"); i >= 0 {
+		s = s[:i]
+	}
+	if i := strings.LastIndex(s, ":"); i > strings.LastIndex(s, "/") {
+		s = s[:i]
+	}
+	if s == "" || strings.Contains(s, "..") {
+		return Source{}, fmt.Errorf("mirror: %q is not an image's name", s)
+	}
+	host, path, ok := strings.Cut(s, "/")
+	if !ok || !(strings.ContainsAny(host, ".:") || host == "localhost") {
+		host, path = "docker.io", s
+	}
+	if host == "docker.io" && !strings.Contains(path, "/") {
+		path = "library/" + path
+	}
+	return Source{Host: host, Path: path}, nil
+}
+
+// Name is the copy's name in the store, `upstream/<host>/<path>` (design 51).
+func (s Source) Name() string { return "upstream/" + s.Host + "/" + s.Path }
+
+func (s Source) reach() string {
+	if s.Host == "docker.io" {
+		return "registry-1.docker.io"
+	}
+	return s.Host
+}
+
+// get asks the upstream for path, taking an anonymous token once when the upstream asks for one.
+func (u *Upstream) get(src Source, path string, accept []string) (*http.Response, error) {
+	target := u.scheme() + "://" + src.reach() + "/v2/" + src.Path + path
+	token := ""
+	for try := 0; try < 2; try++ {
+		req, err := http.NewRequest(http.MethodGet, target, nil)
+		if err != nil {
+			return nil, err
+		}
+		for _, a := range accept {
+			req.Header.Add("Accept", a)
+		}
+		if token != "" {
+			req.Header.Set("Authorization", "Bearer "+token)
+		}
+		resp, err := u.client().Do(req)
+		if err != nil {
+			return nil, err
+		}
+		if resp.StatusCode != http.StatusUnauthorized || token != "" {
+			return resp, nil
+		}
+		challenge := resp.Header.Get("WWW-Authenticate")
+		resp.Body.Close()
+		if token, err = u.token(challenge, src); err != nil {
+			return nil, err
+		}
+	}
+	return nil, fmt.Errorf("mirror: %s refused the anonymous token", src.reach())
+}
+
+// token takes an anonymous pull token from a Bearer challenge.
+func (u *Upstream) token(challenge string, src Source) (string, error) {
+	rest, ok := strings.CutPrefix(challenge, "Bearer ")
+	if !ok {
+		return "", fmt.Errorf("mirror: %s asks a credential the mesh does not hold (%q); only anonymous reads are copied", src.reach(), challenge)
+	}
+	params := map[string]string{}
+	for _, kv := range strings.Split(rest, ",") {
+		k, v, _ := strings.Cut(strings.TrimSpace(kv), "=")
+		params[k] = strings.Trim(v, `"`)
+	}
+	realm, err := url.Parse(params["realm"])
+	if err != nil || params["realm"] == "" {
+		return "", fmt.Errorf("mirror: %s names no token realm", src.reach())
+	}
+	q := realm.Query()
+	if params["service"] != "" {
+		q.Set("service", params["service"])
+	}
+	q.Set("scope", "repository:"+src.Path+":pull")
+	realm.RawQuery = q.Encode()
+	resp, err := u.client().Get(realm.String())
+	if err != nil {
+		return "", err
+	}
+	defer resp.Body.Close()
+	var t struct {
+		Token       string `json:"token"`
+		AccessToken string `json:"access_token"`
+	}
+	if resp.StatusCode != http.StatusOK || json.NewDecoder(io.LimitReader(resp.Body, 1<<20)).Decode(&t) != nil {
+		return "", fmt.Errorf("mirror: %s gave no anonymous token (%d)", realm.Host, resp.StatusCode)
+	}
+	if t.Token == "" {
+		t.Token = t.AccessToken
+	}
+	return t.Token, nil
+}
+
+// manifest reads one manifest by its digest and checks its bytes against it.
+func (u *Upstream) manifest(src Source, digest string) ([]byte, string, error) {
+	resp, err := u.get(src, "/manifests/"+digest, accepted)
+	if err != nil {
+		return nil, "", err
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusOK {
+		return nil, "", fmt.Errorf("mirror: %s/%s@%s answered %d", src.Host, src.Path, digest, resp.StatusCode)
+	}
+	b, err := io.ReadAll(io.LimitReader(resp.Body, 16<<20))
+	if err != nil {
+		return nil, "", err
+	}
+	if "sha256:"+sum(b) != digest {
+		return nil, "", fmt.Errorf("mirror: the manifest %s answered does not hash to its digest", digest)
+	}
+	mt := resp.Header.Get("Content-Type")
+	var probe struct {
+		MediaType string `json:"mediaType"`
+	}
+	if json.Unmarshal(b, &probe) == nil && probe.MediaType != "" {
+		mt = probe.MediaType
+	}
+	return b, mt, nil
+}
+
+// copyBlob streams one blob from the upstream into the store, its digest checked as it passes; a blob held is
+// not fetched.
+func (o *OCI) copyBlob(up *Upstream, src Source, name string, desc descriptor) error {
+	if has, err := o.Has(name, desc.Digest); err != nil {
+		return err
+	} else if has {
+		return nil
+	}
+	resp, err := up.get(src, "/blobs/"+desc.Digest, nil)
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusOK {
+		return fmt.Errorf("mirror: the blob %s answered %d", desc.Digest, resp.StatusCode)
+	}
+	h := sha256.New()
+	body := io.TeeReader(io.LimitReader(resp.Body, desc.Size+1), h)
+	if err := o.putBlobStream(name, desc.Digest, desc.Size, body); err != nil {
+		return err
+	}
+	if got := "sha256:" + hex.EncodeToString(h.Sum(nil)); got != desc.Digest {
+		return fmt.Errorf("mirror: the blob %s answered hashes to %s", desc.Digest, got)
+	}
+	return nil
+}
+
+// putBlobStream stores a blob of a known size from a reader, monolithically.
+func (o *OCI) putBlobStream(name, digest string, size int64, body io.Reader) error {
+	resp, b, err := o.do(http.MethodPost, "/v2/"+name+"/blobs/uploads/", "", nil)
+	if err != nil {
+		return err
+	}
+	if resp.StatusCode != http.StatusAccepted {
+		return fmt.Errorf("the store did not open an upload of %s (%d): %s", digest, resp.StatusCode, b)
+	}
+	loc, err := url.Parse(resp.Header.Get("Location"))
+	if err != nil {
+		return err
+	}
+	q := loc.Query()
+	q.Set("digest", digest)
+	loc.RawQuery = q.Encode()
+	req, err := http.NewRequest(http.MethodPut, o.url(loc.Path+"?"+loc.RawQuery), body)
+	if err != nil {
+		return err
+	}
+	req.ContentLength = size
+	req.Header.Set("Content-Type", "application/octet-stream")
+	req.Header.Set("Content-Length", strconv.FormatInt(size, 10))
+	put, err := o.client().Do(req)
+	if err != nil {
+		return err
+	}
+	defer put.Body.Close()
+	if put.StatusCode != http.StatusCreated {
+		b, _ := io.ReadAll(io.LimitReader(put.Body, 4096))
+		return fmt.Errorf("the store did not take %s (%d): %s", digest, put.StatusCode, b)
+	}
+	return nil
+}
+
+type descriptor struct {
+	MediaType string `json:"mediaType"`
+	Digest    string `json:"digest"`
+	Size      int64  `json:"size"`
+}
+
+// copyManifest copies one manifest and all it names: an index's every child first (the store refuses an index whose
+// children it lacks), an image's config and layers; then the manifest itself under its digest.
+func (o *OCI) copyManifest(up *Upstream, src Source, name, digest string, depth int) error {
+	if depth > 2 {
+		return fmt.Errorf("mirror: %s nests indexes deeper than an image does", digest)
+	}
+	if d, err := o.Digest(name, digest); err == nil && d == digest {
+		return nil // held whole already
+	}
+	b, mt, err := up.manifest(src, digest)
+	if err != nil {
+		return err
+	}
+	var m struct {
+		Manifests []descriptor `json:"manifests"`
+		Config    descriptor   `json:"config"`
+		Layers    []descriptor `json:"layers"`
+	}
+	if err := json.Unmarshal(b, &m); err != nil {
+		return fmt.Errorf("mirror: the manifest %s does not read: %w", digest, err)
+	}
+	for _, c := range m.Manifests {
+		if err := o.copyManifest(up, src, name, c.Digest, depth+1); err != nil {
+			return err
+		}
+	}
+	if m.Config.Digest != "" {
+		if err := o.copyBlob(up, src, name, m.Config); err != nil {
+			return err
+		}
+	}
+	for _, l := range m.Layers {
+		if err := o.copyBlob(up, src, name, l); err != nil {
+			return err
+		}
+	}
+	_, err = o.putManifest(name, digest, mt, b)
+	return err
+}
+
+// Mirror copies an outside image, pinned by its digest, into the store as `upstream/<host>/<path>` (design 51: an
+// upstream image is copied by `mirror` with its pinned digest), every platform of an index included, only the parts
+// the store lacks; it answers the copy's reference. A copy held already is answered at once.
+func (o *OCI) Mirror(source, digest string) (string, error) {
+	src, err := ParseSource(source)
+	if err != nil {
+		return "", err
+	}
+	name := src.Name()
+	if err := o.copyManifest(o.Upstream, src, name, digest, 0); err != nil {
+		return "", err
+	}
+	return o.Registry + "/" + name + "@" + digest, nil
+}

modules/distribution/internal/holder/mirror_fakes_test.go

@@ -0,0 +1,87 @@
+package holder
+
+import (
+	"io"
+	"net/http"
+	"strings"
+	"sync"
+)
+
+// fakeRegistry is an OCI registry in memory, enough for a mirror's copy: blobs, uploads and manifests by
+// reference. With token set it answers 401 with a Bearer challenge until a request carries it.
+type fakeRegistry struct {
+	mu        sync.Mutex
+	blobs     map[string][]byte // name@digest
+	manifests map[string][]byte // name:ref
+	types     map[string]string
+	token     string
+	realm     string
+	gets      []string
+}
+
+func newFakeRegistry() *fakeRegistry {
+	return &fakeRegistry{blobs: map[string][]byte{}, manifests: map[string][]byte{}, types: map[string]string{}}
+}
+
+func (f *fakeRegistry) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	if r.URL.Path == "/token" {
+		_, _ = io.WriteString(w, `{"token":"`+f.token+`"}`)
+		return
+	}
+	if f.token != "" && r.Header.Get("Authorization") != "Bearer "+f.token {
+		w.Header().Set("WWW-Authenticate", `Bearer realm="`+f.realm+`",service="fake"`)
+		w.WriteHeader(http.StatusUnauthorized)
+		return
+	}
+	p := strings.TrimPrefix(r.URL.Path, "/v2/")
+	switch {
+	case strings.Contains(p, "/blobs/uploads/"):
+		name := p[:strings.Index(p, "/blobs/uploads/")]
+		if r.Method == http.MethodPost {
+			w.Header().Set("Location", "/v2/"+name+"/blobs/uploads/u1")
+			w.WriteHeader(http.StatusAccepted)
+			return
+		}
+		b, _ := io.ReadAll(r.Body)
+		d := r.URL.Query().Get("digest")
+		if digestOf(b) != d {
+			w.WriteHeader(http.StatusBadRequest)
+			return
+		}
+		f.blobs[name+"@"+d] = b
+		w.WriteHeader(http.StatusCreated)
+	case strings.Contains(p, "/blobs/"):
+		i := strings.Index(p, "/blobs/")
+		b, ok := f.blobs[p[:i]+"@"+p[i+7:]]
+		if !ok {
+			w.WriteHeader(http.StatusNotFound)
+			return
+		}
+		if r.Method == http.MethodGet {
+			f.gets = append(f.gets, p[i+7:])
+		}
+		_, _ = w.Write(b)
+	case strings.Contains(p, "/manifests/"):
+		i := strings.Index(p, "/manifests/")
+		key := p[:i] + ":" + p[i+11:]
+		if r.Method == http.MethodPut {
+			b, _ := io.ReadAll(r.Body)
+			f.manifests[key] = b
+			f.types[key] = r.Header.Get("Content-Type")
+			w.WriteHeader(http.StatusCreated)
+			return
+		}
+		b, ok := f.manifests[key]
+		if !ok {
+			w.WriteHeader(http.StatusNotFound)
+			return
+		}
+		w.Header().Set("Content-Type", f.types[key])
+		w.Header().Set("Docker-Content-Digest", digestOf(b))
+		_, _ = w.Write(b)
+	default:
+		w.WriteHeader(http.StatusNotFound)
+	}
+}

modules/distribution/internal/holder/oci.go

@@ -18,6 +18,8 @@ type OCI struct {
 	Base     *url.URL
 	Registry string
 	Client   *http.Client
+	// Upstream reads the outside registries `mirror` copies from.
+	Upstream *Upstream
 }
 
 const (
@@ -241,22 +243,3 @@ func (o *OCI) RemoveDigest(name, digest string) error {
 	}
 	return nil
 }
-
-// Mirror is not done by the store itself: mesh-build-image sends an upstream image's parts through publish of kind
-// upstream, named `upstream/<host>/<path>`; mirror answers where a held copy is, and refuses one the store lacks so
-// the build module publishes it.
-func (o *OCI) Mirror(source, digest string) (string, error) {
-	name := "upstream/" + source
-	if i := strings.LastIndexAny(name, ":@"); i > strings.LastIndex(name, "/") {
-		name = name[:i]
-	}
-	resp, _, err := o.do(http.MethodHead, "/v2/"+name+"/manifests/"+digest, "", nil, "application/vnd.oci.image.index.v1+json",
-		mediaManifest, "application/vnd.docker.distribution.manifest.list.v2+json", "application/vnd.docker.distribution.manifest.v2+json")
-	if err != nil {
-		return "", err
-	}
-	if resp.StatusCode != http.StatusOK {
-		return "", fmt.Errorf("the store holds no copy of %s@%s: publish it with kind upstream under %s", source, digest, name)
-	}
-	return o.Registry + "/" + name + "@" + digest, nil
-}

modules/distribution/internal/holder/verbs.go

@@ -217,11 +217,15 @@ func (v *Verbs) Mirror(seat, caller string, r MirrorRequest) (string, error) {
 	if r.Source == "" || !strings.HasPrefix(r.Digest, "sha256:") {
 		return "", errors.New("mirror: refused: an upstream image is copied only by its pinned digest")
 	}
+	src, err := ParseSource(r.Source)
+	if err != nil {
+		return "", err
+	}
 	ref, err := v.Catalog.Mirror(r.Source, r.Digest)
 	if err != nil {
 		return "", fmt.Errorf("mirror: %s@%s was not copied: %w", r.Source, r.Digest, err)
 	}
-	return ref, v.Say(seat, EventPublished, Event{Name: "upstream/" + r.Source, Kind: KindUpstream, Digest: r.Digest, Holder: Self, Reference: ref, By: caller})
+	return ref, v.Say(seat, EventPublished, Event{Name: src.Name(), Kind: KindUpstream, Digest: r.Digest, Holder: Self, Reference: ref, By: caller})
 }
 
 // HeldNow lists the holds, for registry_holds.