mesh-catalog/modules/mesh-build-image/cmd/mesh-build-image/copy.go

package main

// The copy step (novox/hq ADR 0498 §4; issue 1587): an upstream artifact is copied registry to registry into the
// artifact store, whole (an index with every manifest it lists, every blob), by the digest its manifest pins. The copy
// runs no code of a change, so it needs neither buildah nor an isolation; it is made over the registry API, which
// keeps every digest as it is (a pull and a push through buildah may rewrite a manifest). An upstream without a
// digest is not copied: `not pinned:`. A copy that does not end with the pinned digest in the store is `not stored:`.

import (
	"bytes"
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"regexp"
	"strings"
	"time"

	"git.novox.be/novox/mesh-sdk/go/buildstep"
	"git.novox.be/novox/mesh-sdk/go/delivery"
)

// CopyTool is the copy step's tool.
const CopyTool = "build_image_copy"

// CopyResult is what a passed copy step says: the pinned digest, now in the store, and the reference to it in the
// form an image's is (<registry>/<module>/<artifact>@sha256:<digest>).
type CopyResult struct {
	Artifact  string `json:"artifact"`
	Digest    string `json:"digest"`
	Reference string `json:"reference"`
	Pushed    bool   `json:"pushed"`
}

var (
	nameOK       = regexp.MustCompile(`^[a-z0-9][a-z0-9._-]*(/[a-z0-9][a-z0-9._-]*)*$`)
	pinnedOK     = regexp.MustCompile(`@(sha256:[0-9a-f]{64})$`)
	manifestKind = strings.Join([]string{
		"application/vnd.oci.image.index.v1+json", "application/vnd.docker.distribution.manifest.list.v2+json",
		"application/vnd.oci.image.manifest.v1+json", "application/vnd.docker.distribution.manifest.v2+json"}, ", ")
)

func (m *Module) copyTool() StepTool {
	return StepTool{
		Name: CopyTool,
		Description: "Copy an upstream image, registry to registry, into the artifact store by the digest the manifest pins " +
			"(novox/hq ADR 0498 §4). Answers a step id at once; the step ends on its key with the digest in the store and the " +
			"reference to it. A reference without a digest fails `not pinned:`. Needs no workspace and runs no code of a change.",
		Input: map[string]any{
			"type": "object",
			"properties": map[string]any{
				"module":   map[string]any{"type": "string", "description": "the module that names the upstream artifact"},
				"artifact": map[string]any{"type": "string", "description": "the upstream artifact's name in the module's manifest"},
				"from":     map[string]any{"type": "string", "description": "the image reference the manifest gives, pinned by digest, such as nats:2.11.17-alpine@sha256:..."},
			},
			"required": []string{"module", "artifact", "from"},
		},
		Long: true,
		Run:  m.copy,
	}
}

// upstreamRef is an image reference split the way a container runtime does: no host means the public hub.
type upstreamRef struct {
	base, repository, digest string
}

func parseUpstream(from string) (upstreamRef, error) {
	pin := pinnedOK.FindStringSubmatch(from)
	if pin == nil {
		return upstreamRef{}, fmt.Errorf("%q names no digest", from)
	}
	name := strings.TrimSuffix(from, "@"+pin[1])
	if colon := strings.LastIndex(name, ":"); colon > strings.LastIndex(name, "/") {
		name = name[:colon] // the tag is only what it was called; the digest names the image
	}
	if name == "" {
		return upstreamRef{}, fmt.Errorf("%q is not an image reference", from)
	}
	host, repository := "docker.io", name
	if slash := strings.Index(name, "/"); slash >= 0 && (strings.ContainsAny(name[:slash], ".:") || name[:slash] == "localhost") {
		host, repository = name[:slash], name[slash+1:]
	}
	if host == "docker.io" {
		host = "registry-1.docker.io"
		if !strings.Contains(repository, "/") {
			repository = "library/" + repository
		}
	}
	scheme := "https://"
	if strings.HasPrefix(host, "localhost") || strings.HasPrefix(host, "127.") {
		scheme = "http://"
	}
	return upstreamRef{base: scheme + host, repository: repository, digest: pin[1]}, nil
}

func (m *Module) copy(ctx context.Context, args map[string]any) (any, error) {
	module, _ := args["module"].(string)
	name, _ := args["artifact"].(string)
	from, _ := args["from"].(string)
	if !nameOK.MatchString(module) || !nameOK.MatchString(name) {
		return nil, fmt.Errorf("%q / %q: a module and an artifact are named in lower case letters, digits, dots, dashes and underscores", module, name)
	}
	src, err := parseUpstream(from)
	if err != nil {
		return nil, fmt.Errorf("not pinned: %s: %v; an upstream artifact is copied by the digest its manifest pins", name, err)
	}
	addr, err := m.registry()
	if err != nil {
		return nil, fmt.Errorf("not stored: %s: %w", name, err)
	}
	repository := module + "/" + name
	c := &copier{m: m, src: src, dstBase: "http://" + addr, dstRepo: repository, done: map[string]bool{}}
	if err := m.waitForRegistry(ctx, addr, "copying "+from, func() error { return c.copyManifest(ctx, src.digest) }); err != nil {
		return nil, fmt.Errorf("not stored: %s: %w", name, err)
	}
	if err := c.verify(ctx); err != nil {
		return nil, fmt.Errorf("not stored: %s: %w", name, err)
	}
	return CopyResult{Artifact: name, Digest: src.digest, Reference: addr + "/" + repository + "@" + src.digest, Pushed: true}, nil
}

// copier copies one image from its upstream registry into the store's repository.
type copier struct {
	m       *Module
	src     upstreamRef
	dstBase string
	dstRepo string
	token   string
	done    map[string]bool
}

func (c *copier) client() *http.Client {
	if c.m.HTTP != nil {
		return c.m.HTTP
	}
	return &http.Client{Timeout: 30 * time.Minute}
}

// get reads from the upstream registry, answering one bearer challenge with an anonymous token.
func (c *copier) get(ctx context.Context, path, accept string) (*http.Response, error) {
	for attempt := 0; attempt < 2; attempt++ {
		req, err := http.NewRequestWithContext(ctx, http.MethodGet, c.src.base+"/v2/"+c.src.repository+path, nil)
		if err != nil {
			return nil, err
		}
		if accept != "" {
			req.Header.Set("Accept", accept)
		}
		if c.token != "" {
			req.Header.Set("Authorization", "Bearer "+c.token)
		}
		resp, err := c.client().Do(req)
		if err != nil {
			return nil, err
		}
		if resp.StatusCode == http.StatusUnauthorized && attempt == 0 {
			challenge := resp.Header.Get("Www-Authenticate")
			resp.Body.Close()
			if c.token, err = c.fetchToken(ctx, challenge); err != nil {
				return nil, err
			}
			continue
		}
		if resp.StatusCode != http.StatusOK {
			resp.Body.Close()
			return nil, fmt.Errorf("upstream answered %s for %s%s", resp.Status, c.src.repository, path)
		}
		return resp, nil
	}
	return nil, fmt.Errorf("upstream refused %s%s after a token", c.src.repository, path)
}

var challengeParam = regexp.MustCompile(`(\w+)="([^"]*)"`)

func (c *copier) fetchToken(ctx context.Context, challenge string) (string, error) {
	if !strings.HasPrefix(strings.ToLower(challenge), "bearer ") {
		return "", fmt.Errorf("upstream asks for credentials (%q) and this copy holds none", challenge)
	}
	p := map[string]string{}
	for _, kv := range challengeParam.FindAllStringSubmatch(challenge, -1) {
		p[kv[1]] = kv[2]
	}
	realm := p["realm"]
	if realm == "" {
		return "", fmt.Errorf("upstream's challenge %q names no realm", challenge)
	}
	q := url.Values{"service": {p["service"]}, "scope": {"repository:" + c.src.repository + ":pull"}}
	req, err := http.NewRequestWithContext(ctx, http.MethodGet, realm+"?"+q.Encode(), nil)
	if err != nil {
		return "", err
	}
	resp, err := c.client().Do(req)
	if err != nil {
		return "", err
	}
	defer resp.Body.Close()
	var tok struct {
		Token  string `json:"token"`
		Access string `json:"access_token"`
	}
	if resp.StatusCode != http.StatusOK || json.NewDecoder(resp.Body).Decode(&tok) != nil || (tok.Token == "" && tok.Access == "") {
		return "", fmt.Errorf("upstream's token service answered %s", resp.Status)
	}
	if tok.Token != "" {
		return tok.Token, nil
	}
	return tok.Access, nil
}

// copyManifest copies the manifest of digest and everything it names, children before the manifest itself.
func (c *copier) copyManifest(ctx context.Context, digest string) error {
	if c.done[digest] {
		return nil
	}
	resp, err := c.get(ctx, "/manifests/"+digest, manifestKind)
	if err != nil {
		return err
	}
	body, err := io.ReadAll(io.LimitReader(resp.Body, 16<<20))
	resp.Body.Close()
	if err != nil {
		return err
	}
	sum := sha256.Sum256(body)
	if got := "sha256:" + hex.EncodeToString(sum[:]); got != digest {
		return fmt.Errorf("upstream served %s for the manifest pinned as %s", got, digest)
	}
	var doc struct {
		MediaType string                    `json:"mediaType"`
		Config    struct{ Digest string }   `json:"config"`
		Layers    []struct{ Digest string } `json:"layers"`
		Manifests []struct{ Digest string } `json:"manifests"`
	}
	if err := json.Unmarshal(body, &doc); err != nil {
		return fmt.Errorf("the manifest %s is not JSON: %w", digest, err)
	}
	for _, child := range doc.Manifests {
		if err := c.copyManifest(ctx, child.Digest); err != nil {
			return err
		}
	}
	blobs := []string{}
	if doc.Config.Digest != "" {
		blobs = append(blobs, doc.Config.Digest)
	}
	for _, l := range doc.Layers {
		blobs = append(blobs, l.Digest)
	}
	for _, b := range blobs {
		if err := c.copyBlob(ctx, b); err != nil {
			return err
		}
	}
	mediaType := resp.Header.Get("Content-Type")
	if mediaType == "" {
		mediaType = doc.MediaType
	}
	if err := c.putManifest(ctx, digest, mediaType, body); err != nil {
		return err
	}
	c.done[digest] = true
	return nil
}

func (c *copier) copyBlob(ctx context.Context, digest string) error {
	if !digestOK.MatchString(digest) {
		return fmt.Errorf("%q is not a blob digest", digest)
	}
	if c.done[digest] {
		return nil
	}
	head, err := http.NewRequestWithContext(ctx, http.MethodHead, c.dstBase+"/v2/"+c.dstRepo+"/blobs/"+digest, nil)
	if err != nil {
		return err
	}
	if r, err := c.client().Do(head); err != nil {
		return err
	} else {
		r.Body.Close()
		if r.StatusCode == http.StatusOK {
			c.done[digest] = true
			return nil
		}
	}
	post, err := http.NewRequestWithContext(ctx, http.MethodPost, c.dstBase+"/v2/"+c.dstRepo+"/blobs/uploads/", nil)
	if err != nil {
		return err
	}
	pr, err := c.client().Do(post)
	if err != nil {
		return err
	}
	pr.Body.Close()
	loc, err := pr.Location()
	if pr.StatusCode != http.StatusAccepted || err != nil {
		return fmt.Errorf("the store answered %s to a blob upload of %s", pr.Status, digest)
	}
	q := loc.Query()
	q.Set("digest", digest)
	loc.RawQuery = q.Encode()
	blob, err := c.get(ctx, "/blobs/"+digest, "")
	if err != nil {
		return err
	}
	defer blob.Body.Close()
	put, err := http.NewRequestWithContext(ctx, http.MethodPut, loc.String(), blob.Body)
	if err != nil {
		return err
	}
	put.ContentLength = blob.ContentLength
	put.Header.Set("Content-Type", "application/octet-stream")
	resp, err := c.client().Do(put)
	if err != nil {
		return err
	}
	resp.Body.Close()
	if resp.StatusCode != http.StatusCreated {
		return fmt.Errorf("the store answered %s to the blob %s", resp.Status, digest)
	}
	c.done[digest] = true
	return nil
}

func (c *copier) putManifest(ctx context.Context, digest, mediaType string, body []byte) error {
	req, err := http.NewRequestWithContext(ctx, http.MethodPut, c.dstBase+"/v2/"+c.dstRepo+"/manifests/"+digest, bytes.NewReader(body))
	if err != nil {
		return err
	}
	req.Header.Set("Content-Type", mediaType)
	resp, err := c.client().Do(req)
	if err != nil {
		return err
	}
	resp.Body.Close()
	if resp.StatusCode != http.StatusCreated && resp.StatusCode != http.StatusOK {
		return fmt.Errorf("the store answered %s to the manifest %s", resp.Status, digest)
	}
	return nil
}

// verify reads back what the store serves for the pinned digest: only the store can say.
func (c *copier) verify(ctx context.Context) error {
	req, err := http.NewRequestWithContext(ctx, http.MethodHead, c.dstBase+"/v2/"+c.dstRepo+"/manifests/"+c.src.digest, nil)
	if err != nil {
		return err
	}
	req.Header.Set("Accept", manifestKind)
	resp, err := c.client().Do(req)
	if err != nil {
		return err
	}
	resp.Body.Close()
	if got := resp.Header.Get("Docker-Content-Digest"); resp.StatusCode != http.StatusOK || got != c.src.digest {
		return fmt.Errorf("the store answered %s with digest %q for the pinned %s", resp.Status, got, c.src.digest)
	}
	return nil
}

// Plain is a long step that needs nothing of the machine: no guard. The copy runs no buildah.
func (m *Module) Plain(tool string, run func(ctx context.Context, args map[string]any) (any, error)) func(map[string]any) (any, error) {
	return func(args map[string]any) (any, error) {
		id, err := buildstep.Start(m.Ctx, m.Steps, m.Node, tool, func(ctx context.Context) (any, error) { return run(ctx, args) })
		if err != nil {
			return nil, err
		}
		key, _ := delivery.StepKey(m.Node, id)
		return map[string]any{"step": id, "key": key, "state": delivery.StepRunning}, nil
	}
}