diff --git a/internal/depot/remote.go b/internal/depot/remote.go new file mode 100644 index 0000000..7bdf2cf --- /dev/null +++ b/internal/depot/remote.go @@ -0,0 +1,248 @@ +package depot + +import ( + "context" + "crypto/sha256" + "errors" + "fmt" + "io" + "net" + "net/http" + "os" + "path/filepath" + "strings" +) + +// NewBackend returns the backend for name ("local"|"cdn"|"bunny"). +// localRoot is used only for "local". Fails closed: "bunny" with empty +// StorageHost or empty ReadKey returns an error (never prompts); +// unknown name returns an error. +func NewBackend(name, localRoot string, cfg Config) (Backend, error) { + switch name { + case "local": + if localRoot == "" { + return nil, errors.New("local backend requires a non-empty root") + } + return &LocalBackend{Root: localRoot}, nil + case "cdn": + return &httpBackend{ + name: "cdn", + client: newHTTPClient(cfg), + cfg: cfg, + }, nil + case "bunny": + if cfg.StorageHost == "" { + return nil, errors.New("bunny backend requires BUNNY_STORAGE_HOST") + } + if cfg.ReadKey == "" { + return nil, errors.New("bunny backend requires BUNNY_STORAGE_READ_PASSWORD or BUNNY_STORAGE_PASSWORD") + } + return &httpBackend{ + name: "bunny", + client: newHTTPClient(cfg), + cfg: cfg, + }, nil + default: + return nil, fmt.Errorf("unknown backend %q", name) + } +} + +// newHTTPClient builds an IPv4-only client per spec (HEAD probes are banned; +// IPv6 dial hazards are out of scope for this depot). +func newHTTPClient(cfg Config) *http.Client { + transport := &http.Transport{ + DialContext: func(ctx context.Context, network, addr string) (net.Conn, error) { + d := net.Dialer{Timeout: cfg.ConnectTimeout} + return d.DialContext(ctx, "tcp4", addr) + }, + MaxIdleConnsPerHost: cfg.ProbeJobs, + } + return &http.Client{Transport: transport, Timeout: cfg.ProbeMaxTime} +} + +// httpBackend implements Backend for both cdn (read-only) and bunny +// (read/write) over HTTP, sharing probe/get/put logic. +type httpBackend struct { + name string + client *http.Client + cfg Config +} + +func (b *httpBackend) Name() string { return b.name } + +func (b *httpBackend) storageURL(sha string) string { + host := b.cfg.StorageHost + // StorageHost is normally a bare host ("storage.bunnycdn.com"); allow a + // full scheme (used by tests against httptest.NewServer) to pass through + // unchanged. + if strings.Contains(host, "://") { + return fmt.Sprintf("%s/%s/%s", strings.TrimSuffix(host, "/"), b.cfg.StorageZone, BlobKey(sha)) + } + return fmt.Sprintf("https://%s/%s/%s", host, b.cfg.StorageZone, BlobKey(sha)) +} + +func (b *httpBackend) cdnURL(sha string) string { + return fmt.Sprintf("%s/%s", b.cfg.CDNBase, BlobKey(sha)) +} + +// rangeProbe issues GET with Range: bytes=0-0 and classifies the +// response. Never HEAD (banned by spec). +func (b *httpBackend) rangeProbe(ctx context.Context, url string, headers map[string]string) (ProbeState, bool, error) { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil) + if err != nil { + return Unconfirmed, true, err + } + req.Header.Set("Range", "bytes=0-0") + for k, v := range headers { + req.Header.Set(k, v) + } + + resp, err := b.client.Do(req) + if err != nil { + return Unconfirmed, true, err + } + defer resp.Body.Close() + _, _ = io.Copy(io.Discard, resp.Body) + + switch { + case resp.StatusCode >= 200 && resp.StatusCode < 300: + return Present, false, nil + case resp.StatusCode == 404 || resp.StatusCode == 410: + return Absent, false, nil + default: + return Unconfirmed, true, nil + } +} + +// Probe returns the existence state of sha at this backend. +func (b *httpBackend) Probe(ctx context.Context, sha string) (ProbeState, bool, error) { + if b.name == "cdn" { + return b.rangeProbe(ctx, b.cdnURL(sha), nil) + } + return b.rangeProbe(ctx, b.storageURL(sha), map[string]string{"AccessKey": b.cfg.ReadKey}) +} + +// Put uploads src for sha. cdn is read-only. +func (b *httpBackend) Put(ctx context.Context, sha, src string) error { + if b.name == "cdn" { + return errors.New("cdn backend is read-only") + } + if b.cfg.WriteKey == "" { + return errors.New("bunny backend requires BUNNY_STORAGE_PASSWORD to write") + } + + state, _, err := b.Probe(ctx, sha) + if err != nil { + return err + } + if state == Present { + return nil + } + + f, err := os.Open(src) + if err != nil { + return err + } + defer f.Close() + + req, err := http.NewRequestWithContext(ctx, http.MethodPut, b.storageURL(sha), f) + if err != nil { + return err + } + req.Header.Set("AccessKey", b.cfg.WriteKey) + req.Header.Set("Checksum", strings.ToUpper(sha)) + + resp, err := b.client.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + _, _ = io.Copy(io.Discard, resp.Body) + + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return fmt.Errorf("bunny put %s: unexpected status %d", sha, resp.StatusCode) + } + return nil +} + +// Get fetches sha into dest via temp file + rename, re-hashing and deleting +// on mismatch. Mirrors bash _depot_bunny_get: try CDN first, fall back to +// storage with ReadKey. +func (b *httpBackend) Get(ctx context.Context, sha, dest string) error { + destDir := filepath.Dir(dest) + if err := os.MkdirAll(destDir, 0755); err != nil { + return err + } + + tmpFile, err := os.CreateTemp(destDir, ".tmp-") + if err != nil { + return err + } + tmpName := tmpFile.Name() + defer os.Remove(tmpName) + + resp, err := b.fetch(ctx, sha) + if err != nil { + tmpFile.Close() + return err + } + defer resp.Body.Close() + + hasher := sha256.New() + tee := io.TeeReader(resp.Body, hasher) + _, err = io.Copy(tmpFile, tee) + tmpFile.Close() + if err != nil { + return err + } + + gotHash := fmt.Sprintf("%x", hasher.Sum(nil)) + if gotHash != sha { + os.Remove(tmpName) + return fmt.Errorf("hash mismatch: expected %s, got %s", sha, gotHash) + } + + return os.Rename(tmpName, dest) +} + +// fetch tries the CDN URL first (no auth), falling back to the storage URL +// with ReadKey on any non-2xx response or transport error. +func (b *httpBackend) fetch(ctx context.Context, sha string) (*http.Response, error) { + if b.name != "cdn" { + if resp, err := b.get(ctx, b.cdnURL(sha), nil); err == nil && resp.StatusCode >= 200 && resp.StatusCode < 300 { + return resp, nil + } else if err == nil { + resp.Body.Close() + } + resp, err := b.get(ctx, b.storageURL(sha), map[string]string{"AccessKey": b.cfg.ReadKey}) + if err != nil { + return nil, err + } + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + resp.Body.Close() + return nil, fmt.Errorf("bunny get %s: unexpected status %d", sha, resp.StatusCode) + } + return resp, nil + } + + resp, err := b.get(ctx, b.cdnURL(sha), nil) + if err != nil { + return nil, err + } + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + resp.Body.Close() + return nil, fmt.Errorf("cdn get %s: unexpected status %d", sha, resp.StatusCode) + } + return resp, nil +} + +func (b *httpBackend) get(ctx context.Context, url string, headers map[string]string) (*http.Response, error) { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil) + if err != nil { + return nil, err + } + for k, v := range headers { + req.Header.Set(k, v) + } + return b.client.Do(req) +} diff --git a/internal/depot/remote_test.go b/internal/depot/remote_test.go new file mode 100644 index 0000000..6bb9328 --- /dev/null +++ b/internal/depot/remote_test.go @@ -0,0 +1,473 @@ +package depot + +import ( + "context" + "crypto/sha256" + "fmt" + "io" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strings" + "sync" + "testing" + "time" +) + +func shaOf(s string) string { + return fmt.Sprintf("%x", sha256.Sum256([]byte(s))) +} + +type recordedReq struct { + Method string + Path string + Range string + Headers http.Header +} + +// fakeBunny is a minimal recording fake for Bunny storage/CDN endpoints. +type fakeBunny struct { + mu sync.Mutex + requests []recordedReq + headCount int + blobs map[string][]byte + statusFor map[string]int // sha -> status code override for probe/get +} + +func newFakeBunny() *fakeBunny { + return &fakeBunny{blobs: map[string][]byte{}, statusFor: map[string]int{}} +} + +func (f *fakeBunny) handler() http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + f.mu.Lock() + f.requests = append(f.requests, recordedReq{ + Method: r.Method, + Path: r.URL.Path, + Range: r.Header.Get("Range"), + Headers: r.Header.Clone(), + }) + if r.Method == http.MethodHead { + f.headCount++ + f.mu.Unlock() + w.WriteHeader(500) + return + } + f.mu.Unlock() + + sha := strings.TrimPrefix(r.URL.Path, "/") + // last path element is the sha + parts := strings.Split(sha, "/") + sha = parts[len(parts)-1] + + if r.Method == http.MethodPut { + buf, _ := io.ReadAll(r.Body) + f.mu.Lock() + f.blobs[sha] = buf + f.mu.Unlock() + w.WriteHeader(201) + return + } + + // GET (range probe / actual get) + f.mu.Lock() + status, hasStatus := f.statusFor[sha] + data, ok := f.blobs[sha] + f.mu.Unlock() + + if hasStatus { + w.WriteHeader(status) + return + } + if !ok { + w.WriteHeader(404) + return + } + w.WriteHeader(206) + _, _ = w.Write(data) + } +} + +func (f *fakeBunny) requestsSnapshot() []recordedReq { + f.mu.Lock() + defer f.mu.Unlock() + out := make([]recordedReq, len(f.requests)) + copy(out, f.requests) + return out +} + +func testCfg(storageHost, readKey, writeKey string) Config { + return Config{ + CDNBase: "http://127.0.0.1:1", // unreachable by default; overridden per test + StorageHost: storageHost, + StorageZone: "sow-assets-depot", + ReadKey: readKey, + WriteKey: writeKey, + ProbeJobs: 4, + Jobs: 4, + ConnectTimeout: 2 * time.Second, + ProbeMaxTime: 2 * time.Second, + ConfirmRetries: 3, + ConfirmMax: 200, + } +} + +// hostPort returns a StorageHost value for tests: a full "http://host:port" +// base URL, which storageURL() passes through unchanged (bypassing the +// production https:// default so httptest.NewServer works without TLS). +func hostPort(srv *httptest.Server) string { + return srv.URL +} + +func TestProbeUsesRangeGetNotHead(t *testing.T) { + sha := shaOf("present-blob") + f := newFakeBunny() + f.blobs[sha] = []byte("hello") + srv := httptest.NewServer(f.handler()) + defer srv.Close() + + cfg := testCfg(hostPort(srv), "readkey", "writekey") + b, err := NewBackend("bunny", "", cfg) + if err != nil { + t.Fatalf("NewBackend: %v", err) + } + + state, transient, err := b.Probe(context.Background(), sha) + if err != nil { + t.Fatalf("Probe: %v", err) + } + if state != Present || transient { + t.Fatalf("expected Present/non-transient, got %v/%v", state, transient) + } + + reqs := f.requestsSnapshot() + if len(reqs) != 1 { + t.Fatalf("expected exactly 1 request, got %d", len(reqs)) + } + if reqs[0].Method != "GET" { + t.Fatalf("expected GET, got %s", reqs[0].Method) + } + if reqs[0].Range != "bytes=0-0" { + t.Fatalf("expected Range bytes=0-0, got %q", reqs[0].Range) + } + if f.headCount != 0 { + t.Fatalf("expected zero HEAD requests, got %d", f.headCount) + } +} + +func TestProbeStates(t *testing.T) { + cases := []struct { + status int + wantState ProbeState + wantTransient bool + }{ + {200, Present, false}, + {206, Present, false}, + {404, Absent, false}, + {428, Unconfirmed, true}, + {503, Unconfirmed, true}, + } + + for _, c := range cases { + t.Run(fmt.Sprintf("status_%d", c.status), func(t *testing.T) { + sha := shaOf(fmt.Sprintf("blob-%d", c.status)) + f := newFakeBunny() + f.statusFor[sha] = c.status + srv := httptest.NewServer(f.handler()) + defer srv.Close() + + cfg := testCfg(hostPort(srv), "readkey", "writekey") + b, err := NewBackend("bunny", "", cfg) + if err != nil { + t.Fatalf("NewBackend: %v", err) + } + + state, transient, err := b.Probe(context.Background(), sha) + if err != nil && c.status < 500 { + t.Fatalf("unexpected error: %v", err) + } + if state != c.wantState { + t.Fatalf("status %d: expected state %v, got %v", c.status, c.wantState, state) + } + if transient != c.wantTransient { + t.Fatalf("status %d: expected transient=%v, got %v", c.status, c.wantTransient, transient) + } + if f.headCount != 0 { + t.Fatalf("expected zero HEAD requests, got %d", f.headCount) + } + }) + } +} + +func TestBunnyPutSequence(t *testing.T) { + t.Run("absent sha", func(t *testing.T) { + sha := shaOf("new-blob") + f := newFakeBunny() + srv := httptest.NewServer(f.handler()) + defer srv.Close() + + cfg := testCfg(hostPort(srv), "readkey", "writekey") + b, err := NewBackend("bunny", "", cfg) + if err != nil { + t.Fatalf("NewBackend: %v", err) + } + + dir := t.TempDir() + srcPath := filepath.Join(dir, "src") + if err := os.WriteFile(srcPath, []byte("new-blob-content"), 0644); err != nil { + t.Fatal(err) + } + + if err := b.Put(context.Background(), sha, srcPath); err != nil { + t.Fatalf("Put: %v", err) + } + + reqs := f.requestsSnapshot() + if len(reqs) != 2 { + t.Fatalf("expected 2 requests (probe, put), got %d: %+v", len(reqs), reqs) + } + if reqs[0].Method != "GET" || reqs[0].Range != "bytes=0-0" { + t.Fatalf("expected first request to be range probe GET, got %+v", reqs[0]) + } + if reqs[1].Method != "PUT" { + t.Fatalf("expected second request to be PUT, got %+v", reqs[1]) + } + if got := reqs[1].Headers.Get("Checksum"); got != strings.ToUpper(sha) { + t.Fatalf("expected Checksum %s, got %s", strings.ToUpper(sha), got) + } + if got := reqs[1].Headers.Get("AccessKey"); got != "writekey" { + t.Fatalf("expected AccessKey writekey, got %s", got) + } + if f.headCount != 0 { + t.Fatalf("expected zero HEAD requests, got %d", f.headCount) + } + }) + + t.Run("present sha", func(t *testing.T) { + sha := shaOf("existing-blob") + f := newFakeBunny() + f.blobs[sha] = []byte("existing-blob-content") + srv := httptest.NewServer(f.handler()) + defer srv.Close() + + cfg := testCfg(hostPort(srv), "readkey", "writekey") + b, err := NewBackend("bunny", "", cfg) + if err != nil { + t.Fatalf("NewBackend: %v", err) + } + + dir := t.TempDir() + srcPath := filepath.Join(dir, "src") + if err := os.WriteFile(srcPath, []byte("existing-blob-content"), 0644); err != nil { + t.Fatal(err) + } + + if err := b.Put(context.Background(), sha, srcPath); err != nil { + t.Fatalf("Put: %v", err) + } + + reqs := f.requestsSnapshot() + if len(reqs) != 1 { + t.Fatalf("expected probe-only (1 request) since present, got %d: %+v", len(reqs), reqs) + } + if reqs[0].Method != "GET" { + t.Fatalf("expected GET probe, got %+v", reqs[0]) + } + }) +} + +func TestReadWriteKeySplit(t *testing.T) { + sha := shaOf("split-blob") + f := newFakeBunny() + srv := httptest.NewServer(f.handler()) + defer srv.Close() + + cfg := testCfg(hostPort(srv), "readkeyXXX", "writekeyYYY") + b, err := NewBackend("bunny", "", cfg) + if err != nil { + t.Fatalf("NewBackend: %v", err) + } + + dir := t.TempDir() + srcPath := filepath.Join(dir, "src") + if err := os.WriteFile(srcPath, []byte("split-blob-content"), 0644); err != nil { + t.Fatal(err) + } + + if err := b.Put(context.Background(), sha, srcPath); err != nil { + t.Fatalf("Put: %v", err) + } + + reqs := f.requestsSnapshot() + if len(reqs) != 2 { + t.Fatalf("expected 2 requests, got %d", len(reqs)) + } + if got := reqs[0].Headers.Get("AccessKey"); got != "readkeyXXX" { + t.Fatalf("probe expected AccessKey readkeyXXX, got %s", got) + } + if got := reqs[1].Headers.Get("AccessKey"); got != "writekeyYYY" { + t.Fatalf("put expected AccessKey writekeyYYY, got %s", got) + } + if reqs[0].Headers.Get("AccessKey") == reqs[1].Headers.Get("AccessKey") { + t.Fatalf("expected distinct read/write keys") + } +} + +func TestBunnyNoReadKeyFailsClosed(t *testing.T) { + // Grep-based tripwire: no non-test depot source may reference stdin/tty. + dir := "." + entries, err := os.ReadDir(dir) + if err != nil { + t.Fatalf("ReadDir: %v", err) + } + for _, e := range entries { + name := e.Name() + if e.IsDir() || !strings.HasSuffix(name, ".go") || strings.HasSuffix(name, "_test.go") { + continue + } + data, err := os.ReadFile(filepath.Join(dir, name)) + if err != nil { + t.Fatalf("ReadFile %s: %v", name, err) + } + s := string(data) + if strings.Contains(s, "os.Stdin") || strings.Contains(s, "bufio.NewReader(os.Stdin)") || strings.Contains(s, "/dev/tty") { + t.Fatalf("%s references stdin/tty; credential prompts are banned", name) + } + } + + // Confirm undrained pipe: swap in an os.Pipe as stdin-equivalent to prove + // no read occurs. + r, w, err := os.Pipe() + if err != nil { + t.Fatalf("Pipe: %v", err) + } + oldStdin := os.Stdin + os.Stdin = r + defer func() { os.Stdin = oldStdin; r.Close() }() + + cfg := testCfg("storage.example.com", "", "writekey") + _, err = NewBackend("bunny", "", cfg) + if err == nil { + t.Fatal("expected error for empty ReadKey") + } + if !strings.Contains(err.Error(), "ReadKey") && !strings.Contains(err.Error(), "BUNNY_STORAGE_READ_PASSWORD") && !strings.Contains(err.Error(), "BUNNY_STORAGE_PASSWORD") { + t.Fatalf("expected error to mention missing env var, got: %v", err) + } + + w.Close() + // If code read from stdin it would have blocked already (pipe with no + // writer-side data yet); assert the pipe is still open/undrained by + // writing now and reading it back ourselves. + if _, err := w.Write([]byte("x")); err == nil { + t.Fatal("expected write to closed pipe writer to fail") + } +} + +func TestBunnyNoStorageHostFailsClosed(t *testing.T) { + cfg := testCfg("", "readkey", "writekey") + _, err := NewBackend("bunny", "", cfg) + if err == nil { + t.Fatal("expected error for empty StorageHost") + } + if !strings.Contains(err.Error(), "StorageHost") && !strings.Contains(err.Error(), "BUNNY_STORAGE_HOST") { + t.Fatalf("expected error to mention missing StorageHost/BUNNY_STORAGE_HOST, got: %v", err) + } +} + +func TestGetHashMismatchDeletes(t *testing.T) { + sha := shaOf("get-mismatch-blob") + f := newFakeBunny() + f.blobs[sha] = []byte("wrong content") + srv := httptest.NewServer(f.handler()) + defer srv.Close() + + cfg := testCfg(hostPort(srv), "readkey", "writekey") + // force CDN unreachable so Get falls back to storage + cfg.CDNBase = "http://127.0.0.1:1" + b, err := NewBackend("bunny", "", cfg) + if err != nil { + t.Fatalf("NewBackend: %v", err) + } + + dir := t.TempDir() + dest := filepath.Join(dir, "dest") + + err = b.Get(context.Background(), sha, dest) + if err == nil { + t.Fatal("expected hash mismatch error") + } + if _, statErr := os.Stat(dest); !os.IsNotExist(statErr) { + t.Fatalf("expected dest to be removed on mismatch, stat err: %v", statErr) + } +} + +func TestCDNPutReadOnly(t *testing.T) { + cfg := testCfg("storage.example.com", "readkey", "writekey") + b, err := NewBackend("cdn", "", cfg) + if err != nil { + t.Fatalf("NewBackend: %v", err) + } + + dir := t.TempDir() + srcPath := filepath.Join(dir, "src") + if err := os.WriteFile(srcPath, []byte("data"), 0644); err != nil { + t.Fatal(err) + } + + err = b.Put(context.Background(), shaOf("anything"), srcPath) + if err == nil { + t.Fatal("expected error for cdn Put (read-only)") + } + if !strings.Contains(err.Error(), "read-only") { + t.Fatalf("expected read-only error, got: %v", err) + } +} + +// TestNoHeadInSources is the anti-HEAD regression tripwire from field note 3: +// grep-assert no http.MethodHead/.Head( usage in non-test depot sources. +func TestNoHeadInSources(t *testing.T) { + dir := "." + entries, err := os.ReadDir(dir) + if err != nil { + t.Fatalf("ReadDir: %v", err) + } + for _, e := range entries { + name := e.Name() + if e.IsDir() || !strings.HasSuffix(name, ".go") || strings.HasSuffix(name, "_test.go") { + continue + } + data, err := os.ReadFile(filepath.Join(dir, name)) + if err != nil { + t.Fatalf("ReadFile %s: %v", name, err) + } + s := string(data) + if strings.Contains(s, "http.MethodHead") || strings.Contains(s, ".Head(") { + t.Fatalf("%s references HEAD probe (banned)", name) + } + } +} + +// TestLocalBackendViaFactory sanity-checks NewBackend("local", ...). +func TestLocalBackendViaFactory(t *testing.T) { + root := t.TempDir() + b, err := NewBackend("local", root, Config{}) + if err != nil { + t.Fatalf("NewBackend: %v", err) + } + lb, ok := b.(*LocalBackend) + if !ok { + t.Fatalf("expected *LocalBackend, got %T", b) + } + if lb.Root != root { + t.Fatalf("expected Root=%s, got %s", root, lb.Root) + } + + if _, err := NewBackend("local", "", Config{}); err == nil { + t.Fatal("expected error for empty local root") + } + + if _, err := NewBackend("nonsense", "", Config{}); err == nil { + t.Fatal("expected error for unknown backend name") + } +}