feat(bundle): emit bundle v2 next to v1 (RFC-070 F1)

README.md

@@ -119,6 +119,18 @@ rebuilt from scratch every time (`--prev` is refused).
 | `snapshot-changes.jsonl.gz` | the change log of this run |
 | `delta.manifest.dsse.json`, `delta-programs.jsonl.gz`, `delta-changes.jsonl.gz` | the programs that changed since the previous sequence (complete new records) and the same change log |
 | `keyset.dsse.json` | the key set the manifests are signed under |
+| `latest.v2.dsse.json` | bundle v2: signed pointer that pins the digest of each v2 manifest |
+| `snapshot.v2.manifest.dsse.json`, `delta.v2.manifest.dsse.json` | bundle v2: signed manifests listing, per record stream (`programs`, `changes`), the chunks with SHA-256, sizes, record counts and first and last record id |
+| `sha256-<hex>.jsonl.gz` | bundle v2 chunks: gzip JSON Lines of the same records as v1, named by their digest |
+
+Bundle v2 (OpenCTEM RFC-070) carries the same records as v1 in chunks of
+about 2,000 records, split by record id: the same programs always give the
+same chunks, and a changed program changes one chunk, so a consumer fetches
+only the chunks it does not have. It is signed with the same keys and key
+set as v1. v1 stays in every release for one release cycle and is then
+removed; read v2 with the `pkg/transfer/bundle` consumer of the OpenCTEM
+SDK. A local-only build writes `snapshot.v2.manifest.json` (unsigned,
+`local_only` in its meta) next to the v1 files and no pointer.
 
 A platform reads
 `https://github.com/openctemio/programfeed/releases/latest/download/latest.dsse.json`
@@ -196,7 +208,7 @@ characters. Each list is capped at 2,000 entries.
 ```sh
 go build -o programfeed ./cmd/programfeed
 gh release download --repo openctemio/programfeed --dir bundle
-./programfeed verify --dir bundle --root-key-id "$(cat keys/root-keyid.txt)"
+./programfeed verify --dir bundle --root-key-id "$(cat keys/root-keyid.txt)"   # v1, and v2 when present
 ```
 
 The envelopes, key set, sequence, expiry and caps follow the same

cmd/programfeed/main.go

@@ -242,6 +242,12 @@ func cmdVerify(args []string) error {
 	if err != nil {
 		return err
 	}
+	if bundle.HasV2(*dir) {
+		if err := bundle.VerifyV2(*dir, v.KeySet, time.Now()); err != nil {
+			return err
+		}
+		logf("verified bundle v2 (sequence %d)", v.Latest.Sequence)
+	}
 	logf("verified sequence %d (snapshot %d programs, %d targets; key set v%d)", v.Latest.Sequence,
 		v.Snapshot.Stats.Programs, v.Snapshot.Stats.Targets, v.KeySet.Version)
 	return nil

go.mod

@@ -4,6 +4,21 @@ go 1.26.0
 
 toolchain go1.26.9
 
-require golang.org/x/net v0.60.0
+require (
+	// Pseudo-version of sdk-go main (17ba39a, PR 233): replace with the first tagged release that contains pkg/transfer/bundle.
+	github.com/openctemio/sdk-go v0.18.1-0.20261010083220-17ba39a04836
+	golang.org/x/net v0.60.0
+)
 
-require golang.org/x/text v0.42.0 // indirect
+require (
+	github.com/beorn7/perks v1.0.1 // indirect
+	github.com/cespare/xxhash/v2 v2.3.0 // indirect
+	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
+	github.com/prometheus/client_golang v1.24.1 // indirect
+	github.com/prometheus/client_model v0.6.2 // indirect
+	github.com/prometheus/common v0.70.1 // indirect
+	github.com/prometheus/procfs v0.21.1 // indirect
+	golang.org/x/sys v0.48.0 // indirect
+	golang.org/x/text v0.42.0 // indirect
+	google.golang.org/protobuf v1.36.12 // indirect
+)

go.sum

@@ -1,4 +1,38 @@
+github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
+github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
+github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
+github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
+github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM=
+github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
+github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU=
+github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA=
+github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ=
+github.com/openctemio/sdk-go v0.18.1-0.20261010083220-17ba39a04836 h1:WIJMgJnUb++NsP2aEcf4lKV6bxCJB9ni7F2LwkNvXj0=
+github.com/openctemio/sdk-go v0.18.1-0.20261010083220-17ba39a04836/go.mod h1:dxDZsXsRP43BXA7X26d4nbQxDc6HhFpaWQ17Mq0nxS8=
+github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 h1:Jamvg5psRIccs7FGNTlIRMkT8wgtp5eCXdBlqhYGL6U=
+github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/prometheus/client_golang v1.24.1 h1:JnJkREXzWxUdCuPFpIWZiPispT9xVV59uiuyR2bPlnU=
+github.com/prometheus/client_golang v1.24.1/go.mod h1:F+oSRECHg4sse5ucfYpYDeIv/hu68Zo0uoHKetWnzcE=
+github.com/prometheus/client_model v0.6.2 h1:oBsgwpGs7iVziMvrGhE53c/GrLUsZdHnqNwqPLxwZyk=
+github.com/prometheus/client_model v0.6.2/go.mod h1:y3m2F6Gdpfy6Ut/GBsUqTWZqCUvMVzSfMLjcu6wAwpE=
+github.com/prometheus/common v0.70.1 h1:1HvjP4D5oL3t8RsPlwxA9onvvStjtIHYE5XuuwOi/PY=
+github.com/prometheus/common v0.70.1/go.mod h1:VdFUQDMZK3VLkurFUVhia6uys/0suUp86TJz5qbJRhc=
+github.com/prometheus/procfs v0.21.1 h1:GljZCt+zSTS+NZq88cyQ1LjZ+RCHp3uVuabBWA5+OJI=
+github.com/prometheus/procfs v0.21.1/go.mod h1:aB55Cww9pdSJVHk0hUf0inxWyyjPogFIjmHKYgMKmtY=
+github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
+github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
+go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
+go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
+go.yaml.in/yaml/v2 v2.4.4 h1:tuyd0P+2Ont/d6e2rl3be67goVK4R6deVxCUX5vyPaQ=
+go.yaml.in/yaml/v2 v2.4.4/go.mod h1:gMZqIpDtDqOfM0uNfy0SkpRhvUryYH0Z6wdMYcacYXQ=
 golang.org/x/net v0.60.0 h1:79p50tfZlm0J9YfoDsSi639qSXNGVwEzOPLCxM2FsYU=
 golang.org/x/net v0.60.0/go.mod h1:2DA/G1UfVbCpQPeWTmMPGY7Cs2PkBkwu743bVX5PIVg=
+golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo=
+golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og=
 golang.org/x/text v0.42.0 h1:JbOZXgfeCPU9gacVtYliJqOhD+zhrEqK4LfdpmlUZqI=
 golang.org/x/text v0.42.0/go.mod h1:ojzP1Z+2QtioaF8DTtO8K5q7JWVVYwZKenzujK0Zd0E=
+google.golang.org/protobuf v1.36.12 h1:pJOKDDOyeXErUroCihFAd5LQuwXBSpVnKGrj5o/fwxc=
+google.golang.org/protobuf v1.36.12/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco=
+gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

internal/bundle/bundle.go

@@ -278,6 +278,9 @@ func Write(dir string, prev, next *Snapshot, opt WriteOptions) (snap, delta *Man
 	if err := writeJSON(filepath.Join(dir, "latest.json"), latest); err != nil {
 		return nil, nil, nil, err
 	}
+	if err := WriteV2(dir, prev, next, snap, delta); err != nil {
+		return nil, nil, nil, err
+	}
 	return snap, delta, latest, nil
 }
 
@@ -412,6 +415,11 @@ func Sign(dir string, priv ed25519.PrivateKey, keyset []byte, pinnedRoot string,
 			return err
 		}
 	}
+	if _, err := os.Stat(filepath.Join(dir, unsignedSnapshotV2)); err == nil {
+		if err := signV2(dir, priv); err != nil {
+			return err
+		}
+	}
 	return os.WriteFile(filepath.Join(dir, KeySetFile), keyset, 0o644)
 }
 

internal/bundle/v2.go

@@ -0,0 +1,266 @@
+package bundle
+
+// Bundle format v2 (OpenCTEM RFC-070): the same records as v1, in
+// content-addressed gzip JSON Lines chunks split by record id, listed in a
+// signed manifest per kind and a signed pointer. It is written next to the
+// v1 files for one release and read by the SDK's bundle consumer.
+//
+// Layout note: a build writes the chunks and the UNSIGNED v2 manifests
+// (snapshot.v2.manifest.json, delta.v2.manifest.json); the signing step
+// wraps them in DSSE envelopes and writes the signed pointer, because the
+// pointer pins the digest of each signed manifest.
+
+import (
+	"bytes"
+	"crypto/ed25519"
+	"crypto/sha256"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"os"
+	"path/filepath"
+	"sort"
+	"time"
+
+	sdkbundle "github.com/openctemio/sdk-go/pkg/transfer/bundle"
+
+	"github.com/openctemio/programfeed/internal/dsse"
+	"github.com/openctemio/programfeed/internal/model"
+)
+
+// FeedV2 is the feed name of the v2 bundle (its DSSE payload types).
+const FeedV2 = "programfeed"
+
+// V2 file names of unsigned payloads (the signed ones are the SDK's
+// sdkbundle.ManifestFile and sdkbundle.PointerFile).
+const (
+	unsignedSnapshotV2 = "snapshot.v2.manifest.json"
+	unsignedDeltaV2    = "delta.v2.manifest.json"
+)
+
+// V2Meta is the manifest's meta block: what v1 keeps beside its file list.
+type V2Meta struct {
+	Sources   []Source  `json:"sources"`
+	Stats     Stats     `json:"stats"`
+	Collector Collector `json:"collector"`
+}
+
+// rawSigner "signs" by returning the payload: the writer then stores the
+// unsigned manifest, which the signing step wraps later.
+type rawSigner struct{}
+
+func (rawSigner) Sign(_ string, payload []byte) ([]byte, error) { return payload, nil }
+
+func unsignedName(kind string) string {
+	if kind == KindDelta {
+		return unsignedDeltaV2
+	}
+	return unsignedSnapshotV2
+}
+
+// recordLine is one record as one JSON line (no HTML escaping, like v1).
+func recordLine(v any) ([]byte, error) {
+	var b bytes.Buffer
+	enc := json.NewEncoder(&b)
+	enc.SetEscapeHTML(false)
+	if err := enc.Encode(v); err != nil {
+		return nil, err
+	}
+	return bytes.TrimRight(b.Bytes(), "\n"), nil
+}
+
+// changeIDs gives each change a record id that ascends with (program,
+// position in the log): "<program>#<n>". '#' sorts before every character
+// a program id holds, so ids of one program stay together and in order.
+func changeIDs(changes []model.Change) (ids []string, order []int) {
+	order = make([]int, len(changes))
+	for i := range order {
+		order[i] = i
+	}
+	sort.SliceStable(order, func(a, b int) bool { return changes[order[a]].Program < changes[order[b]].Program })
+	ids = make([]string, len(changes))
+	n, prog := 0, ""
+	for k, i := range order {
+		if changes[i].Program != prog {
+			prog, n = changes[i].Program, 0
+		}
+		ids[k] = fmt.Sprintf("%s#%06d", prog, n)
+		n++
+	}
+	return ids, order
+}
+
+func writeKindV2(dir, kind string, s *Snapshot, ids []string, base uint64, now time.Time, meta V2Meta) error {
+	w, err := sdkbundle.NewWriter(dir, sdkbundle.WriterOptions{Feed: FeedV2, Sequence: s.Sequence, Kind: kind, BaseSequence: base})
+	if err != nil {
+		return err
+	}
+	if err := w.Stream("programs"); err != nil {
+		return err
+	}
+	for _, id := range ids {
+		line, err := recordLine(s.Programs[id])
+		if err != nil {
+			return err
+		}
+		if err := w.Add(id, line); err != nil {
+			return err
+		}
+	}
+	if err := w.Stream("changes"); err != nil {
+		return err
+	}
+	cids, order := changeIDs(s.Changes)
+	for k, i := range order {
+		line, err := recordLine(s.Changes[i])
+		if err != nil {
+			return err
+		}
+		if err := w.Add(cids[k], line); err != nil {
+			return err
+		}
+	}
+	rawMeta, err := json.Marshal(meta)
+	if err != nil {
+		return err
+	}
+	_, _, err = w.Finish(rawMeta, now, now.Add(Validity), rawSigner{})
+	if err != nil {
+		return err
+	}
+	// Finish wrote the payload under the signed name; keep it as unsigned.
+	return os.Rename(filepath.Join(dir, sdkbundle.ManifestFile(kind)), filepath.Join(dir, unsignedName(kind)))
+}
+
+// WriteV2 writes the v2 chunks and unsigned manifests of next (and of the
+// delta from prev when prev is not nil) into dir, next to the v1 files.
+// snap and delta are the v1 manifests of the same build (stats, sources).
+func WriteV2(dir string, prev, next *Snapshot, snap, delta *Manifest) error {
+	meta := func(m *Manifest) V2Meta { return V2Meta{Sources: m.Sources, Stats: m.Stats, Collector: m.Collector} }
+	if err := writeKindV2(dir, KindSnapshot, next, next.SortedIDs(), 0, snap.CreatedAt, meta(snap)); err != nil {
+		return fmt.Errorf("bundle v2 snapshot: %w", err)
+	}
+	if prev != nil && delta != nil {
+		if err := writeKindV2(dir, KindDelta, next, next.ChangedIDs(), prev.Sequence, delta.CreatedAt, meta(delta)); err != nil {
+			return fmt.Errorf("bundle v2 delta: %w", err)
+		}
+	}
+	return nil
+}
+
+// signV2 signs the unsigned v2 manifests in dir and writes the signed
+// pointer. The key and key set were already checked by Sign.
+func signV2(dir string, priv ed25519.PrivateKey) error {
+	signer := sdkbundle.Ed25519Signer{Key: priv}
+	var ptr sdkbundle.Pointer
+	for _, kind := range []string{KindSnapshot, KindDelta} {
+		payload, err := os.ReadFile(filepath.Join(dir, unsignedName(kind)))
+		if errors.Is(err, os.ErrNotExist) && kind == KindDelta {
+			continue
+		}
+		if err != nil {
+			return err
+		}
+		var m sdkbundle.Manifest
+		if err := json.Unmarshal(payload, &m); err != nil {
+			return fmt.Errorf("%s: %w", unsignedName(kind), err)
+		}
+		var meta V2Meta
+		if json.Unmarshal(m.Meta, &meta) == nil && meta.Collector.LocalOnly {
+			return errors.New("refusing to sign a local-only bundle: it was built without the publish policy")
+		}
+		env, err := signer.Sign(sdkbundle.ManifestPayloadType(FeedV2), payload)
+		if err != nil {
+			return err
+		}
+		name := sdkbundle.ManifestFile(kind)
+		if err := os.WriteFile(filepath.Join(dir, name), env, 0o644); err != nil {
+			return err
+		}
+		sum := sha256.Sum256(env)
+		ref := sdkbundle.FileRef{Name: name, SHA256: hex.EncodeToString(sum[:]), Size: int64(len(env))}
+		if kind == KindSnapshot {
+			ptr = sdkbundle.Pointer{Feed: FeedV2, Sequence: m.Sequence, Snapshot: ref, CreatedAt: m.CreatedAt, ExpiresAt: m.ExpiresAt}
+		} else {
+			ptr.Delta, ptr.BaseSequence = &ref, m.BaseSequence
+		}
+		if err := os.Remove(filepath.Join(dir, unsignedName(kind))); err != nil {
+			return err
+		}
+	}
+	if ptr.Sequence == 0 {
+		return errors.New("no v2 snapshot manifest to sign")
+	}
+	return sdkbundle.WritePointer(dir, ptr, signer)
+}
+
+// HasV2 reports whether dir holds a signed v2 pointer.
+func HasV2(dir string) bool {
+	_, err := os.Stat(filepath.Join(dir, sdkbundle.PointerFile))
+	return err == nil
+}
+
+// VerifyV2 checks the v2 files of dir the way the SDK consumer does: the
+// pointer and manifests against the key set's keys, the manifests against
+// the digests the pointer pins, and every chunk against its manifest entry.
+func VerifyV2(dir string, ks *dsse.KeySet, now time.Time) error {
+	var pubs []ed25519.PublicKey
+	for _, k := range ks.Keys {
+		pub, err := k.Decode()
+		if err != nil {
+			return err
+		}
+		pubs = append(pubs, pub)
+	}
+	ver := sdkbundle.NewEd25519Verifier(pubs...)
+	read := func(name string, max int64) ([]byte, error) { return readCapped(filepath.Join(dir, name), max) }
+	raw, err := read(sdkbundle.PointerFile, MaxManifestBytes)
+	if err != nil {
+		return err
+	}
+	payload, err := ver.Verify(raw, sdkbundle.PointerPayloadType(FeedV2))
+	if err != nil {
+		return fmt.Errorf("v2 pointer: %w", err)
+	}
+	var ptr sdkbundle.Pointer
+	if err := dsse.DecodeStrict(payload, &ptr); err != nil {
+		return fmt.Errorf("v2 pointer: %w", err)
+	}
+	if err := ptr.Validate(FeedV2, now, sdkbundle.Limits{}); err != nil {
+		return err
+	}
+	refs := []sdkbundle.FileRef{ptr.Snapshot}
+	if ptr.Delta != nil {
+		refs = append(refs, *ptr.Delta)
+	}
+	for _, ref := range refs {
+		env, err := read(ref.Name, MaxManifestBytes)
+		if err != nil {
+			return err
+		}
+		if sum := sha256.Sum256(env); int64(len(env)) != ref.Size || hex.EncodeToString(sum[:]) != ref.SHA256 {
+			return fmt.Errorf("v2 %s does not match the pointer", ref.Name)
+		}
+		mp, err := ver.Verify(env, sdkbundle.ManifestPayloadType(FeedV2))
+		if err != nil {
+			return fmt.Errorf("v2 %s: %w", ref.Name, err)
+		}
+		var m sdkbundle.Manifest
+		if err := dsse.DecodeStrict(mp, &m); err != nil {
+			return fmt.Errorf("v2 %s: %w", ref.Name, err)
+		}
+		if err := m.Validate(FeedV2, now, sdkbundle.Limits{}); err != nil {
+			return err
+		}
+		if m.Sequence != ptr.Sequence {
+			return fmt.Errorf("v2 %s: sequence %d, pointer %d", ref.Name, m.Sequence, ptr.Sequence)
+		}
+		for _, c := range m.Chunks() {
+			if err := checkFile(filepath.Join(dir, sdkbundle.ChunkFile(c.Ref.SHA256)), File{Name: c.Ref.SHA256, SHA256: c.Ref.SHA256, Size: c.Ref.Size}); err != nil {
+				return fmt.Errorf("v2 chunk: %w", err)
+			}
+		}
+	}
+	return nil
+}