diff --git a/cmd/programfeed/main.go b/cmd/programfeed/main.go index 473f674..2b8edb8 100644 --- a/cmd/programfeed/main.go +++ b/cmd/programfeed/main.go @@ -148,8 +148,8 @@ 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 { @@ -157,7 +157,15 @@ func cmdBuild(ctx context.Context, args []string) error { } 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 diff --git a/internal/bundle/bundle_test.go b/internal/bundle/bundle_test.go index a2c4762..1ec5b41 100644 --- a/internal/bundle/bundle_test.go +++ b/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") + } +} diff --git a/internal/bundle/verify.go b/internal/bundle/verify.go index bd7e4dd..c848c92 100644 --- a/internal/bundle/verify.go +++ b/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 +} diff --git a/internal/fetch/fetch.go b/internal/fetch/fetch.go index bdede14..7280834 100644 --- a/internal/fetch/fetch.go +++ b/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 } diff --git a/internal/fetch/fetch_test.go b/internal/fetch/fetch_test.go index e575f47..8f79d1f 100644 --- a/internal/fetch/fetch_test.go +++ b/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)) + } +} diff --git a/internal/source/bountytargets.go b/internal/source/bountytargets.go index 50261d5..7e3b145 100644 --- a/internal/source/bountytargets.go +++ b/internal/source/bountytargets.go @@ -108,7 +108,7 @@ func (b BountyTargets) Collect(ctx context.Context, c *fetch.Client) (*Result, e continue } fileURL := b.Dataset.rawURL(commit, "data/"+plat.file) - resp, err := c.Get(ctx, fileURL, maxDatasetBytes) + resp, err := c.Get(fetch.WithDeadline(ctx, datasetDeadline), fileURL, maxDatasetBytes) if err != nil { return nil, err } diff --git a/internal/source/dataset.go b/internal/source/dataset.go index 8aeb223..8a7ae8e 100644 --- a/internal/source/dataset.go +++ b/internal/source/dataset.go @@ -8,6 +8,7 @@ import ( "net/url" "regexp" "strings" + "time" "github.com/openctemio/programfeed/internal/fetch" "github.com/openctemio/programfeed/internal/model" @@ -28,7 +29,10 @@ const ( githubAPI = "https://api.github.com" githubRaw = "https://raw.githubusercontent.com" - maxDatasetBytes = 32 << 20 + maxDatasetBytes = 32 << 20 + // datasetDeadline bounds one large dataset download (the idle-read guard + // still fails a stalled one early). + datasetDeadline = 5 * time.Minute maxDatasetRecords = 20000 maxCommitListSize = 1 << 20 ) diff --git a/internal/source/diodb.go b/internal/source/diodb.go index e9720d7..bc7c5a5 100644 --- a/internal/source/diodb.go +++ b/internal/source/diodb.go @@ -73,7 +73,7 @@ var platformHosts = map[string]string{ // Collect implements Source. func (d DioDB) Collect(ctx context.Context, c *fetch.Client) (*Result, error) { - resp, err := c.Get(ctx, d.url(), maxDioDBBytes) + resp, err := c.Get(fetch.WithDeadline(ctx, datasetDeadline), d.url(), maxDioDBBytes) if err != nil { return nil, err } diff --git a/internal/source/pdbounty.go b/internal/source/pdbounty.go index 8cc12de..218908d 100644 --- a/internal/source/pdbounty.go +++ b/internal/source/pdbounty.go @@ -58,7 +58,7 @@ func (d PDBounty) Collect(ctx context.Context, c *fetch.Client) (*Result, error) return nil, err } fileURL := d.Dataset.rawURL(commit, pdFile) - resp, err := c.Get(ctx, fileURL, maxDatasetBytes) + resp, err := c.Get(fetch.WithDeadline(ctx, datasetDeadline), fileURL, maxDatasetBytes) if err != nil { return nil, err }