feat(schema): typed scope entries (schema 1.1)

README.md

@@ -154,6 +154,22 @@ Key rules:
 - Scope entries are typed (`domain`, `wildcard`, `ip`, `cidr`, `url`,
   `mobile_app`, `source_repo`, `other`), canonical, sorted and deduplicated;
   out of scope wins over in scope.
+- Schema 1.1 (additive; `openctem.programfeed/v1` is unchanged, every new
+  field is optional and omitted when the platform did not publish it, never
+  guessed). A scope entry may carry `asset_type` (the platform's own asset
+  type, normalised: `domain`, `wildcard`, `url`, `api`, `ip`, `cidr`,
+  `android_app`, `ios_app`, `mobile_app`, `source_code`, `executable`,
+  `hardware`, `smart_contract`, `ai_model`, `other`; `type` stays the coarse
+  kind), `ports` (ports or ranges such as `8000-8100`), `protocol`
+  (`tcp`/`udp`), `path_prefix` (url targets), `environment`
+  (`production`/`staging`/`other`), `instructions` (the platform's testing
+  text, at most 1000 bytes), `requires` (headers, test accounts), alongside
+  `eligible_for_bounty` and `max_severity`. The aggregate datasets currently
+  publish `asset_type`, `instructions`, eligibility and severity (and the
+  URL path becomes `path_prefix`); the other fields are in the schema for
+  sources that publish them. `notes` is now only the collector's own remark.
+  Readers that validate with `additionalProperties: false` must adopt the
+  1.1 schema.
 - `confidence`: `published` (the program's own scope), `published_by_platform`
   (scope the hosting platform published, read through a public dataset; the
   record's `provenance` names the dataset, commit, original platform and

cmd/programfeed/main.go

@@ -148,16 +148,24 @@ func cmdBuild(ctx context.Context, args []string) error {
 			return fmt.Errorf("policy: %w", err)
 		}
 	}
-	if *localOnly && *prevDir != "" {
-		return errors.New("build: a local-only bundle is rebuilt from scratch; --prev is not allowed with --local-only")
+	if !*localOnly && *prevDir != "" && bundle.IsLocalOnly(*prevDir) {
+		return errors.New("build: --prev is a local-only bundle; a publishable build needs a signed release")
 	}
 	srcs, err := sources(*names, *seedFile, pol.PublishablePlatforms(), *localOnly)
 	if err != nil {
 		return err
 	}
 	now := time.Now().UTC()
 	var prev *bundle.Snapshot
-	if *prevDir != "" {
+	if *prevDir != "" && *localOnly {
+		// Continue the previous local bundle's sequence (snapshot-only,
+		// unsigned): the platform importer refuses non-increasing sequences.
+		var err error
+		if prev, err = bundle.ReadLocal(*prevDir); err != nil {
+			return fmt.Errorf("previous local bundle: %w", err)
+		}
+		logf("previous local bundle: sequence %d, %d programs", prev.Sequence, len(prev.Programs))
+	} else if *prevDir != "" {
 		v, err := bundle.VerifyDir(*prevDir, bundle.VerifyOptions{PinnedRoot: *root, Now: now, AllowExpired: true, SkipRecordChk: true})
 		if err != nil {
 			return fmt.Errorf("previous release: %w", err)
@@ -183,7 +191,11 @@ func cmdBuild(ctx context.Context, args []string) error {
 		next = pol.Filter(next, removed)
 		logf("publish policy: %d of %d programs publishable", len(next.Programs), full)
 	}
-	snap, delta, _, err := bundle.Write(*out, prev, next, bundle.WriteOptions{Now: now,
+	writePrev := prev
+	if *localOnly {
+		writePrev = nil // snapshot-only: no delta is written for a local bundle
+	}
+	snap, delta, _, err := bundle.Write(*out, writePrev, next, bundle.WriteOptions{Now: now,
 		Collector: bundle.Collector{Version: version, Commit: os.Getenv("GITHUB_SHA"), LocalOnly: *localOnly}})
 	if err != nil {
 		return err

internal/bundle/bundle_test.go

@@ -272,3 +272,42 @@ func resignJSON(t *testing.T, dir, file, payloadType string, priv ed25519.Privat
 	out, _ := dsse.Sign(priv, payloadType, payload)
 	_ = os.WriteFile(filepath.Join(dir, file), out, 0o644)
 }
+
+func TestReadLocalContinuesSequence(t *testing.T) {
+	dir := t.TempDir()
+	if _, _, _, err := Write(dir, nil, snapshot(1, "b.openctem.io"), WriteOptions{Now: time.Now(), Collector: Collector{LocalOnly: true}}); err != nil {
+		t.Fatal(err)
+	}
+	if !IsLocalOnly(dir) {
+		t.Fatal("IsLocalOnly")
+	}
+	prev, err := ReadLocal(dir)
+	if err != nil || prev.Sequence != 1 || len(prev.Programs) != 2 {
+		t.Fatalf("ReadLocal: %v %+v", err, prev)
+	}
+	// A tampered record file is refused.
+	if err := os.WriteFile(filepath.Join(dir, FileName(KindSnapshot, recordFiles[0])), []byte("x"), 0o644); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := ReadLocal(dir); err == nil {
+		t.Fatal("tampered local bundle accepted")
+	}
+}
+
+func TestReadLocalRefusesSignedAndNonLocal(t *testing.T) {
+	k := newKeys(t)
+	signed := build(t, k, nil, snapshot(1, "b.openctem.io"))
+	if _, err := ReadLocal(signed); err == nil {
+		t.Fatal("signed release accepted as a local prev")
+	}
+	if IsLocalOnly(signed) {
+		t.Fatal("signed release reported local-only")
+	}
+	plain := t.TempDir()
+	if _, _, _, err := Write(plain, nil, snapshot(1, "b.openctem.io"), WriteOptions{Now: time.Now()}); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := ReadLocal(plain); err == nil {
+		t.Fatal("non-local unsigned bundle accepted")
+	}
+}

internal/bundle/verify.go

@@ -5,6 +5,7 @@ import (
 	"compress/gzip"
 	"crypto/sha256"
 	"encoding/hex"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"io"
@@ -275,3 +276,55 @@ func (l *limited) Read(p []byte) (int, error) {
 	l.left -= int64(n)
 	return n, err
 }
+
+// ReadLocal reads the previous UNSIGNED local-only bundle in dir (written by
+// a --local-only build) so the next local build can continue its sequence.
+// It checks schema, tag, file sizes and hashes and every record, and refuses
+// anything that is not a local-only bundle (a signed release has DSSE
+// envelopes in place of the plain pointer and manifest).
+func ReadLocal(dir string) (*Snapshot, error) {
+	var l Latest
+	raw, err := readCapped(filepath.Join(dir, "latest.json"), 1<<20)
+	if err != nil {
+		return nil, err
+	}
+	if json.Unmarshal(raw, &l) != nil || l.Schema != LatestSchema {
+		return nil, errors.New("not an unsigned local-only bundle (signed releases cannot seed a local build)")
+	}
+	if l.Tag != Tag(l.Sequence) || l.Snapshot != fmt.Sprintf(ManifestFile, KindSnapshot) {
+		return nil, fmt.Errorf("latest: tag %q or snapshot %q does not match sequence %d", l.Tag, l.Snapshot, l.Sequence)
+	}
+	raw, err = readCapped(filepath.Join(dir, "snapshot.manifest.json"), 1<<20)
+	if err != nil {
+		return nil, err
+	}
+	var m Manifest
+	if json.Unmarshal(raw, &m) != nil || m.Schema != ManifestSchema || m.Kind != KindSnapshot {
+		return nil, errors.New("snapshot manifest: not an unsigned manifest")
+	}
+	if !m.Collector.LocalOnly {
+		return nil, errors.New("previous bundle is not local-only; a local build continues only a local-only bundle")
+	}
+	if m.Sequence != l.Sequence || len(m.Files) != len(recordFiles) {
+		return nil, errors.New("snapshot manifest does not match the pointer")
+	}
+	for i, f := range m.Files {
+		if f.Name != FileName(KindSnapshot, recordFiles[i]) {
+			return nil, fmt.Errorf("unexpected file %q", f.Name)
+		}
+		if err := checkFile(filepath.Join(dir, f.Name), f); err != nil {
+			return nil, err
+		}
+	}
+	return ReadRecords(dir, m)
+}
+
+// IsLocalOnly reports whether dir holds an unsigned local-only bundle.
+func IsLocalOnly(dir string) bool {
+	raw, err := os.ReadFile(filepath.Join(dir, "snapshot.manifest.json"))
+	if err != nil {
+		return false
+	}
+	var m Manifest
+	return json.Unmarshal(raw, &m) == nil && m.Collector.LocalOnly
+}

internal/fetch/fetch.go

@@ -49,7 +49,10 @@ type Options struct {
 	UserAgent   string
 	CacheDir    string        // conditional requests; empty disables the cache
 	MinInterval time.Duration // between two requests to one host
-	Timeout     time.Duration // per request
+	Timeout     time.Duration // response-header timeout, and the default whole-request deadline
+	// IdleTimeout fails a response body that delivers no byte for this long
+	// (default 20s), so a large download may take long but never stalls.
+	IdleTimeout time.Duration
 	// AllowPrivate lets tests talk to httptest servers on loopback. Never
 	// set in the collector.
 	AllowPrivate bool
@@ -84,6 +87,9 @@ func New(opt Options) *Client {
 	if opt.Timeout == 0 {
 		opt.Timeout = 30 * time.Second
 	}
+	if opt.IdleTimeout == 0 {
+		opt.IdleTimeout = 20 * time.Second
+	}
 	c := &Client{opt: opt, next: map[string]time.Time{}, robots: map[string]*robotsRules{}}
 	dialer := &net.Dialer{Timeout: 10 * time.Second, Control: func(_, address string, _ syscall.RawConn) error {
 		if opt.AllowPrivate {
@@ -109,7 +115,7 @@ func New(opt Options) *Client {
 		MaxIdleConnsPerHost:   2,
 		IdleConnTimeout:       30 * time.Second,
 	}
-	c.http = &http.Client{Transport: transport, Timeout: opt.Timeout,
+	c.http = &http.Client{Transport: transport,
 		CheckRedirect: func(req *http.Request, via []*http.Request) error {
 			if len(via) >= 3 {
 				return errors.New("more than 3 redirects")
@@ -129,6 +135,30 @@ func New(opt Options) *Client {
 	return c
 }
 
+type deadlineKey struct{}
+
+// WithDeadline raises (or lowers) the whole-request deadline of fetches made
+// with the returned context, for sources that download large files. The
+// header, connect and idle-body timeouts still apply.
+func WithDeadline(ctx context.Context, d time.Duration) context.Context {
+	return context.WithValue(ctx, deadlineKey{}, d)
+}
+
+// idleReader resets the idle timer whenever bytes arrive.
+type idleReader struct {
+	r     io.Reader
+	timer *time.Timer
+	idle  time.Duration
+}
+
+func (i *idleReader) Read(p []byte) (int, error) {
+	n, err := i.r.Read(p)
+	if n > 0 {
+		i.timer.Reset(i.idle)
+	}
+	return n, err
+}
+
 // Get fetches rawURL (at most maxBytes) after checking robots.txt.
 func (c *Client) Get(ctx context.Context, rawURL string, maxBytes int64) (*Response, error) {
 	u, err := url.Parse(rawURL)
@@ -164,6 +194,16 @@ func (c *Client) get(ctx context.Context, u *url.URL, maxBytes int64, useCache b
 	if err := c.wait(ctx, u.Host); err != nil {
 		return nil, err
 	}
+	total := c.opt.Timeout
+	if d, ok := ctx.Value(deadlineKey{}).(time.Duration); ok && d > 0 {
+		total = d
+	}
+	ctx, cancelTotal := context.WithTimeout(ctx, total)
+	defer cancelTotal()
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel()
+	idle := time.AfterFunc(c.opt.IdleTimeout, cancel) // a stalled body fails
+	defer idle.Stop()
 	req, err := http.NewRequestWithContext(ctx, http.MethodGet, u.String(), nil)
 	if err != nil {
 		return nil, err
@@ -203,7 +243,7 @@ func (c *Client) get(ctx context.Context, u *url.URL, maxBytes int64, useCache b
 	case resp.StatusCode != http.StatusOK:
 		return nil, &StatusError{URL: u.String(), Code: resp.StatusCode}
 	}
-	body, err := io.ReadAll(io.LimitReader(resp.Body, maxBytes+1))
+	body, err := io.ReadAll(io.LimitReader(&idleReader{r: resp.Body, timer: idle, idle: c.opt.IdleTimeout}, maxBytes+1))
 	if err != nil {
 		return nil, err
 	}

internal/fetch/fetch_test.go

@@ -152,3 +152,51 @@ func TestRedirectEverythingTerminates(t *testing.T) {
 		t.Fatalf("%v %+v", err, r)
 	}
 }
+
+func slowServer(chunks int, gap time.Duration, stallAfter int) *httptest.Server {
+	return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path == "/robots.txt" {
+			http.NotFound(w, r)
+			return
+		}
+		fl := w.(http.Flusher)
+		for i := 0; i < chunks; i++ {
+			if stallAfter >= 0 && i == stallAfter {
+				select {
+				case <-r.Context().Done():
+				case <-time.After(5 * time.Second):
+				}
+				return
+			}
+			_, _ = w.Write([]byte(strings.Repeat("x", 1024)))
+			fl.Flush()
+			time.Sleep(gap)
+		}
+	}))
+}
+
+func TestSlowButSteadyBodyPasses(t *testing.T) {
+	srv := slowServer(20, 25*time.Millisecond, -1) // ~500ms total, over Timeout
+	defer srv.Close()
+	c := New(Options{UserAgent: "t", MinInterval: time.Millisecond, AllowPrivate: true,
+		Timeout: 200 * time.Millisecond, IdleTimeout: 300 * time.Millisecond})
+	if _, err := c.Get(context.Background(), srv.URL+"/big", 1<<20); err == nil {
+		t.Fatal("the default deadline must still bound the request")
+	}
+	resp, err := c.Get(WithDeadline(context.Background(), 5*time.Second), srv.URL+"/big", 1<<20)
+	if err != nil || len(resp.Body) != 20*1024 {
+		t.Fatalf("err=%v", err)
+	}
+}
+
+func TestStalledBodyFails(t *testing.T) {
+	srv := slowServer(20, time.Millisecond, 3)
+	defer srv.Close()
+	c := New(Options{UserAgent: "t", MinInterval: time.Millisecond, AllowPrivate: true,
+		Timeout: time.Second, IdleTimeout: 150 * time.Millisecond})
+	start := time.Now()
+	_, err := c.Get(WithDeadline(context.Background(), time.Minute), srv.URL+"/big", 1<<20)
+	if err == nil || time.Since(start) > 3*time.Second {
+		t.Fatalf("a stall must fail fast: err=%v after %v", err, time.Since(start))
+	}
+}