diff --git a/modules/distribution/cmd/distribution-tools/main.go b/modules/distribution/cmd/distribution-tools/main.go index 8b548cad6..86b562851 100644 --- a/modules/distribution/cmd/distribution-tools/main.go +++ b/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// (every platform, only the parts it lacks) and answer its reference; says published (mesh-build-image alone).", Input: obj(map[string]any{"source": s("/"), "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}} diff --git a/modules/distribution/internal/holder/holder_test.go b/modules/distribution/internal/holder/holder_test.go index 97835e87d..812a74d3c 100644 --- a/modules/distribution/internal/holder/holder_test.go +++ b/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}}, diff --git a/modules/distribution/internal/holder/mirror.go b/modules/distribution/internal/holder/mirror.go new file mode 100644 index 000000000..720dfca62 --- /dev/null +++ b/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 `/` +// 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//` (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//` (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 +} diff --git a/modules/distribution/internal/holder/mirror_fakes_test.go b/modules/distribution/internal/holder/mirror_fakes_test.go new file mode 100644 index 000000000..5f0f5189e --- /dev/null +++ b/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) + } +} diff --git a/modules/distribution/internal/holder/oci.go b/modules/distribution/internal/holder/oci.go index 10bfa5f2b..9753d1346 100644 --- a/modules/distribution/internal/holder/oci.go +++ b/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//`; 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 -} diff --git a/modules/distribution/internal/holder/verbs.go b/modules/distribution/internal/holder/verbs.go index 4406e7952..f8cd3f4dd 100644 --- a/modules/distribution/internal/holder/verbs.go +++ b/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.