Copy an upstream artifact into the artifact store by its pinned digest (issue 1587, S11-S13, ADR 0498 §4)

This commit is contained in:
2026-10-22 17:57:05 +02:00
parent 49caf17ff5
commit 72effa1fb1
5 changed files with 603 additions and 2 deletions
+4
View File
@@ -30,6 +30,10 @@ The Dockerfile's steps may use the network (ADR 0487 §6); nothing a step can re
`<module>/<artifact>:<tag of its own>` (the first 12 hex digits of the digest unless the call gives one), waiting out
a registry held still, and the key carries the digest the registry serves and the reference pinned by it. Without
`publish` (a pull request) nothing is pushed.
- `build_image_copy` copies an upstream artifact (`module`, `artifact`, `from`) registry to registry into the artifact store
over the registry API, the whole index and every blob, by the digest the manifest pins (ADR 0498 §4), and verifies the
store serves that digest. Answers the digest and the reference `<registry>/<module>/<artifact>@sha256:<digest>`. A
`from` without a digest fails `not pinned:`. It needs no workspace and no buildah.
- It refuses, naming why, an image whose build context is a whole repository (`context`) or whose `build.on` names
another module's artifact (`unsupportedImage` in `build.go`): those forms belong to the images build-agent builds
until it is removed at R4g.
@@ -0,0 +1,374 @@
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
}
}
@@ -0,0 +1,218 @@
package main
import (
"crypto/sha256"
"encoding/hex"
"encoding/json"
"io"
"net/http"
"net/http/httptest"
"strings"
"sync"
"testing"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
func digestOf(b []byte) string { s := sha256.Sum256(b); return "sha256:" + hex.EncodeToString(s[:]) }
// memRegistry is a registry in memory: blobs and manifests by digest, per repository.
type memRegistry struct {
mu sync.Mutex
blobs map[string][]byte // repo|digest
manifests map[string][]byte // repo|digest
types map[string]string
puts []string
srv *httptest.Server
}
func newMem(t *testing.T) *memRegistry {
r := &memRegistry{blobs: map[string][]byte{}, manifests: map[string][]byte{}, types: map[string]string{}}
r.srv = httptest.NewServer(http.HandlerFunc(r.serve))
t.Cleanup(r.srv.Close)
return r
}
func (r *memRegistry) addr() string { return r.srv.Listener.Addr().String() }
func (r *memRegistry) serve(w http.ResponseWriter, req *http.Request) {
r.mu.Lock()
defer r.mu.Unlock()
path := strings.TrimPrefix(req.URL.Path, "/v2/")
if i := strings.Index(path, "/blobs/uploads/"); i >= 0 {
if req.Method == http.MethodPost {
w.Header().Set("Location", "/v2/"+path[:i]+"/blobs/uploads/u1")
w.WriteHeader(http.StatusAccepted)
return
}
body, _ := io.ReadAll(req.Body)
d := req.URL.Query().Get("digest")
if digestOf(body) != d {
w.WriteHeader(http.StatusBadRequest)
return
}
r.blobs[path[:i]+"|"+d] = body
r.puts = append(r.puts, "blob "+d)
w.WriteHeader(http.StatusCreated)
return
}
for _, kind := range []string{"blobs", "manifests"} {
if i := strings.Index(path, "/"+kind+"/"); i >= 0 {
key := path[:i] + "|" + path[i+len(kind)+2:]
store := r.blobs
if kind == "manifests" {
store = r.manifests
}
if req.Method == http.MethodPut {
body, _ := io.ReadAll(req.Body)
if digestOf(body) != path[i+len(kind)+2:] {
w.WriteHeader(http.StatusBadRequest)
return
}
r.manifests[key] = body
r.types[key] = req.Header.Get("Content-Type")
r.puts = append(r.puts, "manifest "+path[i+len(kind)+2:])
w.WriteHeader(http.StatusCreated)
return
}
body, ok := store[key]
if !ok {
http.NotFound(w, req)
return
}
w.Header().Set("Docker-Content-Digest", path[i+len(kind)+2:])
w.Header().Set("Content-Type", r.types[key])
if req.Method == http.MethodHead {
return
}
w.Write(body)
return
}
}
http.NotFound(w, req)
}
// upstreamImage puts a two-platform index (two manifests, each with a config and a layer) into r and answers the index digest.
func upstreamImage(r *memRegistry, repo string) string {
var children []map[string]string
for _, tag := range []string{"amd64", "arm64"} {
cfg, layer := []byte(`{"arch":"`+tag+`"}`), []byte("layer-"+tag)
r.blobs[repo+"|"+digestOf(cfg)], r.blobs[repo+"|"+digestOf(layer)] = cfg, layer
mf, _ := json.Marshal(map[string]any{"schemaVersion": 2, "mediaType": "application/vnd.oci.image.manifest.v1+json",
"config": map[string]string{"digest": digestOf(cfg)}, "layers": []map[string]string{{"digest": digestOf(layer)}}})
r.manifests[repo+"|"+digestOf(mf)] = mf
r.types[repo+"|"+digestOf(mf)] = "application/vnd.oci.image.manifest.v1+json"
children = append(children, map[string]string{"digest": digestOf(mf)})
}
idx, _ := json.Marshal(map[string]any{"schemaVersion": 2, "mediaType": "application/vnd.oci.image.index.v1+json", "manifests": children})
d := digestOf(idx)
r.manifests[repo+"|"+d] = idx
r.types[repo+"|"+d] = "application/vnd.oci.image.index.v1+json"
return d
}
func runCopy(t *testing.T, m *Module, states *keptStates, args map[string]any) (CopyResult, delivery.Step) {
t.Helper()
if _, err := tool(t, m, CopyTool)(args); err != nil {
t.Fatal(err)
}
st := states.waitEnd(t)
var res CopyResult
if st.State == delivery.StepPassed {
if err := json.Unmarshal(st.Result, &res); err != nil {
t.Fatal(err)
}
}
return res, st
}
// ADR 0498 §4 / AC: the copy leaves the pinned digest in the store and answers it, with the whole index (every
// manifest it lists, every blob), and runs no buildah.
func TestTheCopyLeavesThePinnedDigestInTheStoreAndAnswersIt(t *testing.T) {
up, store := newMem(t), newMem(t)
digest := upstreamImage(up, "library/nats")
f := newFake(t)
states := newStates()
m := moduleWith(t, f, states)
m.HTTP = nil
reg := store.addr()
writeFile(t, m.RegistryFile, reg+"\n")
from := up.addr() + "/library/nats:2.11.17-alpine@" + digest
res, st := runCopy(t, m, states, map[string]any{"module": "nats", "artifact": "server", "from": from})
if st.State != delivery.StepPassed {
t.Fatalf("step = %+v", st)
}
if res.Digest != digest || res.Reference != reg+"/nats/server@"+digest || !res.Pushed || res.Artifact != "server" {
t.Errorf("result = %+v", res)
}
if _, ok := store.manifests["nats/server|"+digest]; !ok {
t.Fatal("the index is not in the store")
}
if len(store.manifests) != 3 || len(store.blobs) != 4 {
t.Errorf("store holds %d manifests and %d blobs, want 3 and 4", len(store.manifests), len(store.blobs))
}
if len(f.lines()) != 0 {
t.Errorf("the copy ran commands: %v", f.lines())
}
// A second copy writes nothing new but the manifests.
store.puts = nil
if _, st := runCopy(t, m, states, map[string]any{"module": "nats", "artifact": "server", "from": from}); st.State != delivery.StepPassed {
t.Fatalf("again: %+v", st)
}
for _, p := range store.puts {
if strings.HasPrefix(p, "blob ") {
t.Errorf("a blob the store holds was sent again: %s", p)
}
}
}
// ADR 0498 §4 / AC: an upstream without a digest fails `not pinned:`, and nothing is read from the upstream.
func TestAnUpstreamWithoutADigestFailsNotPinned(t *testing.T) {
up, store := newMem(t), newMem(t)
upstreamImage(up, "library/nats")
f := newFake(t)
states := newStates()
m := moduleWith(t, f, states)
writeFile(t, m.RegistryFile, store.addr()+"\n")
_, st := runCopy(t, m, states, map[string]any{"module": "nats", "artifact": "server", "from": up.addr() + "/library/nats:2.11.17-alpine"})
if st.State != delivery.StepFailed || !strings.Contains(st.Reason, "not pinned: server") {
t.Fatalf("step = %+v", st)
}
if len(store.puts) != 0 {
t.Errorf("the store was written: %v", store.puts)
}
}
// A manifest the upstream serves under another digest than the pin is not copied; the failure is `not stored:`.
func TestAnUpstreamServingAnotherManifestThanPinnedIsNotStored(t *testing.T) {
up, store := newMem(t), newMem(t)
d := upstreamImage(up, "library/nats")
up.manifests["library/nats|"+d] = []byte(`{"tampered":true}`)
f := newFake(t)
states := newStates()
m := moduleWith(t, f, states)
writeFile(t, m.RegistryFile, store.addr()+"\n")
_, st := runCopy(t, m, states, map[string]any{"module": "nats", "artifact": "server", "from": up.addr() + "/library/nats@" + d})
if st.State != delivery.StepFailed || !strings.HasPrefix(st.Reason, "not stored: server:") {
t.Fatalf("step = %+v", st)
}
if len(store.manifests) != 0 {
t.Errorf("the store holds %v", store.manifests)
}
}
func TestAShortNameIsQualifiedAsTheHubsAndAHostIsKept(t *testing.T) {
d := "sha256:" + strings.Repeat("a", 64)
for from, want := range map[string]upstreamRef{
"alpine@" + d: {"https://registry-1.docker.io", "library/alpine", d},
"nats:2.11.17-alpine@" + d: {"https://registry-1.docker.io", "library/nats", d},
"grafana/grafana:11@" + d: {"https://registry-1.docker.io", "grafana/grafana", d},
"ghcr.io/org/app:1@" + d: {"https://ghcr.io", "org/app", d},
"localhost:5000/x/y@" + d: {"http://localhost:5000", "x/y", d},
} {
got, err := parseUpstream(from)
if err != nil || got != want {
t.Errorf("%s = %+v, %v; want %+v", from, got, err, want)
}
}
}
@@ -25,14 +25,18 @@ type StepTool struct {
}
// StepTools are the module's step tools.
func (m *Module) StepTools() []StepTool { return []StepTool{m.fetchTool(), m.buildTool()} }
func (m *Module) StepTools() []StepTool {
return []StepTool{m.fetchTool(), m.buildTool(), m.copyTool()}
}
// Tools is every tool the module serves: its step tools behind the guard, its read tools, its status and its health.
func (m *Module) Tools(steps []StepTool) []stdio.Tool {
tools := []stdio.Tool{}
for _, s := range steps {
t := stdio.Tool{Name: s.Name, Description: s.Description, Input: s.Input}
if s.Long {
if s.Name == CopyTool {
t.Run = m.Plain(s.Name, s.Run) // a copy needs no buildah, so the buildah guard does not stand in front of it
} else if s.Long {
t.Run = m.Long(s.Name, s.Run)
} else {
t.Run = m.Short(s.Run)
+1
View File
@@ -14,6 +14,7 @@
"tools": [
"build_image_fetch",
"build_image_build",
"build_image_copy",
"build_image_step",
"build_image_steps",
"build_image_status",