mesh-catalog/modules/distribution/internal/holder/mirror.go

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.Reference(name, digest), nil
}

// Reference is the seat's answer for name at digest: `<registry>/<name>@<digest>`.
func (o *OCI) Reference(name, digest string) string { return o.Registry + "/" + name + "@" + digest }