diff --git a/internal/server/browse.go b/internal/server/browse.go index fc5e658..26ee63e 100644 --- a/internal/server/browse.go +++ b/internal/server/browse.go @@ -3,6 +3,7 @@ package server import ( "bufio" "encoding/json" + "errors" "fmt" "io" "net/http" @@ -20,10 +21,13 @@ import ( const ( contentTypePlainText = "text/plain; charset=utf-8" browseSniffSize = 512 + // browsePrefetchSize is how much of a browsed file is read before the + // response starts. Read errors within it still produce an error status. + browsePrefetchSize = 64 << 10 ) -// maxBrowseArchiveSize caps how much data openArchive will buffer for -// prefix detection. Artifacts larger than this are rejected to prevent +// maxBrowseArchiveSize caps the compressed artifact size the browse and diff +// endpoints will read. Artifacts larger than this are rejected to prevent // memory exhaustion from a single request. const maxBrowseArchiveSize = 512 << 20 // 512 MB @@ -57,45 +61,32 @@ func isMetadataSidecar(filename string) bool { // "repo-hash/"). Returns "" if there's no single root or the archive is flat. func detectSingleRootDir(reader archives.Reader) string { files, err := reader.List() - if err != nil || len(files) == 0 { + if err != nil { return "" } - var root string + var root rootDetector for _, f := range files { - parts := strings.SplitN(f.Path, "/", 2) //nolint:mnd // split into dir + rest - if len(parts) == 0 { - continue - } - dir := parts[0] - if root == "" { - root = dir - } else if dir != root { - return "" + if !root.add(f.Path) { + break } } - - if root == "" { - return "" - } - return root + "/" + return root.prefix() } -// openArchive opens a cached artifact as an archive reader, auto-detecting -// and stripping a single top-level directory prefix (like GitHub zipballs). -// For npm, the hardcoded "package/" prefix takes precedence. +// openArchive opens a cached artifact as a random-access archive reader, +// auto-detecting and stripping a single top-level directory prefix (like +// GitHub zipballs). For npm, the hardcoded "package/" prefix takes precedence. +// The whole artifact is buffered, so only the version diff uses it; listing +// and file reads stream through browseArchive instead. func openArchive(filename string, content io.Reader, ecosystem string) (archives.Reader, error) { //nolint:ireturn // wraps multiple archive implementations - limited := io.LimitReader(content, maxBrowseArchiveSize+1) - data, err := io.ReadAll(limited) + data, err := readBrowseInput(content, maxBrowseArchiveSize) if err != nil { - return nil, fmt.Errorf("reading artifact: %w", err) - } - if int64(len(data)) > maxBrowseArchiveSize { - return nil, fmt.Errorf("artifact too large for browsing (%d bytes)", len(data)) + return nil, err } if ecosystem == "npm" { - return archives.OpenBytesWithPrefix(filename, data, "package/") + return archives.OpenBytesWithPrefix(filename, data, npmPackagePrefix) } probe, err := archives.OpenBytes(filename, data) @@ -108,6 +99,35 @@ func openArchive(filename string, content io.Reader, ecosystem string) (archives return archives.OpenBytesWithPrefix(filename, data, prefix) } +// browseLimitsOrDefault returns the stream limits for listing and file reads. +func (s *Server) browseLimitsOrDefault() archives.StreamOptions { + if s.browseLimits != (archives.StreamOptions{}) { + return s.browseLimits + } + return defaultBrowseLimits +} + +// openBrowseArchive prepares a cached artifact for streaming. content is the +// already open storage reader; later passes reopen the artifact. +func (s *Server) openBrowseArchive(r *http.Request, artifact *database.Artifact, content io.Reader, ecosystem string) (*browseArchive, error) { + storagePath := artifact.StoragePath.String + reopen := func() (io.ReadCloser, error) { + return s.storage.Open(r.Context(), storagePath) + } + return newBrowseArchive(artifact.Filename, ecosystem, content, reopen, s.browseLimitsOrDefault()) +} + +// browseArchiveError reports a failure to read an archive for browsing. +func (s *Server) browseArchiveError(w http.ResponseWriter, err error, filename string) { + if isBrowseLimitError(err) { + s.logger.Warn("archive exceeds browse limits", "error", err, "filename", filename) + internalError(w, "archive exceeds browse limits") + return + } + s.logger.Error("failed to open archive", "error", err, "filename", filename) + internalError(w, "failed to open archive") +} + // BrowseListResponse contains the file listing for a directory in an archives. type BrowseListResponse struct { Path string `json:"path"` @@ -253,20 +273,16 @@ func (s *Server) browseList(w http.ResponseWriter, r *http.Request, ecosystem, n } defer func() { _ = artifactReader.Close() }() - // Open archive with auto-detected prefix stripping - archiveReader, err := openArchive(cachedArtifact.Filename, artifactReader, ecosystem) + archive, err := s.openBrowseArchive(r, cachedArtifact, artifactReader, ecosystem) if err != nil { - s.logger.Error("failed to open archive", "error", err, "filename", cachedArtifact.Filename) - internalError(w, "failed to open archive") + s.browseArchiveError(w, err, cachedArtifact.Filename) return } - defer func() { _ = archiveReader.Close() }() - // List files in the directory - files, err := archiveReader.ListDir(dirPath) + // List files in the directory, with the root prefix stripped + files, err := archive.ListDir(dirPath) if err != nil { - s.logger.Error("failed to list directory", "error", err, "path", dirPath) - internalError(w, "failed to list directory") + s.browseArchiveError(w, err, cachedArtifact.Filename) return } @@ -340,35 +356,48 @@ func (s *Server) browseFile(w http.ResponseWriter, r *http.Request, ecosystem, n } defer func() { _ = artifactReader.Close() }() - // Open archive with auto-detected prefix stripping - archiveReader, err := openArchive(cachedArtifact.Filename, artifactReader, ecosystem) + archive, err := s.openBrowseArchive(r, cachedArtifact, artifactReader, ecosystem) if err != nil { - s.logger.Error("failed to open archive", "error", err, "filename", cachedArtifact.Filename) - internalError(w, "failed to open archive") + s.browseArchiveError(w, err, cachedArtifact.Filename) return } - defer func() { _ = archiveReader.Close() }() - // Extract the file - fileReader, err := archiveReader.Extract(filePath) - if err != nil { - if strings.Contains(err.Error(), "not found") { - notFound(w, "file not found") - return - } + // Find the file and stream it straight from the archive + fileReader, err := archive.Extract(filePath) + switch { + case errors.Is(err, errBrowseNotFound): + notFound(w, "file not found") + return + case errors.Is(err, errBrowseIsDir): s.logger.Error("failed to extract file", "error", err, "path", filePath) internalError(w, "failed to extract file") return + case err != nil: + s.browseArchiveError(w, err, cachedArtifact.Filename) + return } defer func() { _ = fileReader.Close() }() + s.writeBrowseFile(w, fileReader, filePath, cachedArtifact.Filename) +} + +// writeBrowseFile sends a file streamed from an archive. The start of the file +// is read before any headers, so a truncated or over-limit entry that fails +// there still gets an error status. A failure after the response has started +// aborts it, so the client sees an incomplete download rather than a +// successful one. +func (s *Server) writeBrowseFile(w http.ResponseWriter, file io.Reader, filePath, artifactName string) { + source := &trackedReader{reader: file} + content := bufio.NewReaderSize(source, browsePrefetchSize) + head, err := content.Peek(browsePrefetchSize) + if err != nil && !errors.Is(err, io.EOF) { + s.browseReadError(w, err, filePath, artifactName) + return + } + contentType, knownPath := detectContentTypeFromPath(filePath) - var content io.Reader = fileReader if !knownPath { - bufferedFile := bufio.NewReaderSize(fileReader, browseSniffSize) - prefix, _ := bufferedFile.Peek(browseSniffSize) - contentType = detectContentTypeFromPrefix(prefix) - content = bufferedFile + contentType = detectContentTypeFromPrefix(head[:min(len(head), browseSniffSize)]) } w.Header().Set("Content-Type", contentType) w.Header().Set("Content-Security-Policy", "sandbox") @@ -377,8 +406,43 @@ func (s *Server) browseFile(w http.ResponseWriter, r *http.Request, ecosystem, n _, filename := path.Split(filePath) w.Header().Set("Content-Disposition", fmt.Sprintf("inline; filename=%q", filename)) - // Stream the file - _, _ = io.Copy(w, content) + written, err := io.Copy(w, content) + if err == nil || source.err == nil { + // A write error without a read error means the client went away. + return + } + s.logger.Error("failed to stream file from archive", "error", source.err, + "path", filePath, "filename", artifactName, "bytes", written) + // Headers are already committed: finishing normally would turn a truncated + // file into a seemingly successful download. Let net/http close the HTTP/1 + // connection or reset the HTTP/2 stream instead. + panic(http.ErrAbortHandler) +} + +// browseReadError reports a failure reading a file before its response starts. +func (s *Server) browseReadError(w http.ResponseWriter, err error, filePath, artifactName string) { + if isBrowseLimitError(err) { + s.logger.Warn("archive exceeds browse limits", "error", err, "path", filePath, "filename", artifactName) + internalError(w, "archive exceeds browse limits") + return + } + s.logger.Error("failed to read file from archive", "error", err, "path", filePath, "filename", artifactName) + internalError(w, "failed to read file") +} + +// trackedReader records the first read error other than io.EOF, so a failed +// copy can be told apart from a failed write to the client. +type trackedReader struct { + reader io.Reader + err error +} + +func (r *trackedReader) Read(p []byte) (int, error) { + n, err := r.reader.Read(p) + if err != nil && !errors.Is(err, io.EOF) && r.err == nil { + r.err = err + } + return n, err } func detectContentTypeFromPath(filename string) (string, bool) { diff --git a/internal/server/browse_archive.go b/internal/server/browse_archive.go new file mode 100644 index 0000000..4af6afe --- /dev/null +++ b/internal/server/browse_archive.go @@ -0,0 +1,405 @@ +package server + +import ( + "bufio" + "errors" + "fmt" + "io" + "path" + "strings" + + "github.com/git-pkgs/archives" + "github.com/git-pkgs/magic" +) + +// Limits applied when listing or reading a single file. Bodies are streamed +// rather than retained, so these bound decompression work per request; they +// match the limits the buffered archive reader enforces. +const ( + maxBrowseExpandedSize = 512 << 20 // 512 MB across all entries + maxBrowseEntrySize = 512 << 20 // 512 MB for any one entry + maxBrowseEntries = 100_000 +) + +// npmPackagePrefix is the directory npm wraps every tarball's contents in. +const npmPackagePrefix = "package/" + +var defaultBrowseLimits = archives.StreamOptions{ + MaxInputBytes: maxBrowseArchiveSize, + MaxEntryBytes: maxBrowseEntrySize, + MaxExpandedBytes: maxBrowseExpandedSize, + MaxEntries: maxBrowseEntries, +} + +var ( + errArtifactTooLarge = errors.New("artifact too large for browsing") + errBrowseNotFound = errors.New("file not found") + errBrowseIsDir = errors.New("path is a directory") + + // errStopWalk ends a walk early without reporting an error. + errStopWalk = errors.New("stop walk") +) + +// isBrowseLimitError reports whether err comes from one of the browse size or +// entry-count limits rather than a malformed archive. +func isBrowseLimitError(err error) bool { + return errors.Is(err, errArtifactTooLarge) || + errors.Is(err, archives.ErrInputLimit) || + errors.Is(err, archives.ErrEntrySizeLimit) || + errors.Is(err, archives.ErrDecompressLimit) || + errors.Is(err, archives.ErrEntryLimit) +} + +// browseArchive reads a cached artifact sequentially for the browse endpoints +// without holding expanded file bodies in memory. TAR based formats stream from +// storage, reopening it when a second pass is needed. ZIP and conda need random +// access, so their input is read once and shared by every pass. +type browseArchive struct { + filename string + ecosystem string + limits archives.StreamOptions + reopen func() (io.ReadCloser, error) + + // first is the caller's reader, used for the first pass only. + first io.Reader + // data holds buffered ZIP or conda input. + data []byte +} + +// newBrowseArchive prepares content for streaming. The caller keeps ownership +// of content; reopen supplies a fresh reader for any later pass. +func newBrowseArchive(filename, ecosystem string, content io.Reader, reopen func() (io.ReadCloser, error), limits archives.StreamOptions) (*browseArchive, error) { + buffered := bufio.NewReaderSize(content, browseSniffSize) + sniff, err := buffered.Peek(browseSniffSize) + if err != nil && err != io.EOF { + return nil, fmt.Errorf("reading artifact: %w", err) + } + + a := &browseArchive{filename: filename, ecosystem: ecosystem, limits: limits, reopen: reopen} + if !needsRandomAccess(filename, sniff) { + a.first = buffered + return a, nil + } + + a.data, err = readBrowseInput(buffered, limits.MaxInputBytes) + if err != nil { + return nil, err + } + return a, nil +} + +// readBrowseInput reads all of r, failing once it exceeds limit bytes. +func readBrowseInput(r io.Reader, limit int64) ([]byte, error) { + data, err := io.ReadAll(io.LimitReader(r, limit+1)) + if err != nil { + return nil, fmt.Errorf("reading artifact: %w", err) + } + if int64(len(data)) > limit { + return nil, fmt.Errorf("%w (%d bytes)", errArtifactTooLarge, len(data)) + } + return data, nil +} + +// needsRandomAccess reports whether the archives library buffers this format +// when streaming it, following the library's own detection order: a known +// extension first, then the content's leading bytes. +func needsRandomAccess(filename string, sniff []byte) bool { + name := strings.ToLower(filename) + for _, suffix := range []string{".tar.gz", ".tar.bz2", ".tar.xz", ".tar.zst"} { + if strings.HasSuffix(name, suffix) { + return false + } + } + switch path.Ext(name) { + case ".zip", ".jar", ".whl", ".nupkg", ".egg", ".vsix", ".conda": + return true + case ".tar", ".tgz", ".crate", ".gem": + return false + } + return magic.DetectPrefix(sniff).Format == "zip" +} + +// openStream starts a pass over the archive. The returned function releases +// the stream and any reader opened for it. +func (a *browseArchive) openStream() (*archives.Stream, func(), error) { + if a.data != nil { + stream, err := archives.OpenStreamBytes(a.filename, a.data, a.limits) + if err != nil { + return nil, nil, err + } + return stream, func() { _ = stream.Close() }, nil + } + + if a.first != nil { + content := a.first + a.first = nil + stream, err := archives.OpenStream(a.filename, content, a.limits) + if err != nil { + return nil, nil, err + } + return stream, func() { _ = stream.Close() }, nil + } + + content, err := a.reopen() + if err != nil { + return nil, nil, fmt.Errorf("reopening artifact: %w", err) + } + stream, err := archives.OpenStream(a.filename, content, a.limits) + if err != nil { + _ = content.Close() + return nil, nil, err + } + return stream, func() { + _ = stream.Close() + _ = content.Close() + }, nil +} + +// walk calls fn for each entry in archive order until fn returns an error or +// the archive ends. fn may read the entry body from the stream. +func (a *browseArchive) walk(fn func(entry *archives.StreamEntry) error) error { + stream, release, err := a.openStream() + if err != nil { + return err + } + defer release() + + for { + entry, err := stream.Next() + if err == io.EOF { + return nil + } + if err != nil { + return err + } + if err := fn(entry); err != nil { + if errors.Is(err, errStopWalk) { + return nil + } + return err + } + } +} + +// prefix returns the directory stripped from every path: npm's "package/", or +// a single top-level directory shared by all entries (like GitHub zipballs). +// For other ecosystems this costs one pass, which stops early once two +// top-level names differ. +func (a *browseArchive) prefix() (string, error) { + if a.ecosystem == "npm" { + return npmPackagePrefix, nil + } + + var root rootDetector + err := a.walk(func(entry *archives.StreamEntry) error { + if !root.add(entry.Path) { + return errStopWalk + } + return nil + }) + if err != nil { + return "", err + } + return root.prefix(), nil +} + +// ListDir returns the entries directly inside dirPath, after prefix stripping, +// in a single pass that keeps only entry metadata. +func (a *browseArchive) ListDir(dirPath string) ([]archives.FileInfo, error) { + var entries []archives.FileInfo + var root rootDetector + err := a.walk(func(entry *archives.StreamEntry) error { + entries = append(entries, entry.FileInfo) + root.add(entry.Path) + return nil + }) + if err != nil { + return nil, err + } + + prefix := npmPackagePrefix + if a.ecosystem != "npm" { + prefix = root.prefix() + } + return listBrowseDir(entries, prefix, dirPath), nil +} + +// Extract returns the body of the first entry at filePath, streamed from the +// archive. The caller must close the returned reader. +func (a *browseArchive) Extract(filePath string) (io.ReadCloser, error) { + prefix, err := a.prefix() + if err != nil { + return nil, err + } + target := prefix + filePath + + stream, release, err := a.openStream() + if err != nil { + return nil, err + } + for { + entry, err := stream.Next() + if err == io.EOF { + release() + return nil, fmt.Errorf("%w: %s", errBrowseNotFound, filePath) + } + if err != nil { + release() + return nil, err + } + if entry.Path != target { + continue + } + if entry.IsDir { + release() + return nil, fmt.Errorf("%w: %s", errBrowseIsDir, filePath) + } + return &streamedFile{Reader: stream, release: release}, nil + } +} + +// streamedFile is an entry body that releases its archive stream on Close. +type streamedFile struct { + io.Reader + release func() +} + +func (f *streamedFile) Close() error { + if f.release != nil { + f.release() + f.release = nil + } + return nil +} + +// rootDetector tracks whether every path seen so far lives under one +// top-level directory. +type rootDetector struct { + root string + mixed bool + seen bool +} + +// add records p and reports whether a single root is still possible. +func (d *rootDetector) add(p string) bool { + if d.mixed { + return false + } + d.seen = true + dir, _, _ := strings.Cut(p, "/") + if d.root == "" { + d.root = dir + } else if dir != d.root { + d.mixed = true + } + return !d.mixed +} + +// prefix returns the shared root with a trailing slash, or "" if paths have +// more than one root or none were seen. +func (d *rootDetector) prefix() string { + if d.mixed || !d.seen || d.root == "" { + return "" + } + return d.root + "/" +} + +// listBrowseDir lists dirPath from entry metadata the same way a reader from +// archives.OpenWithPrefix does: the prefix is joined to dirPath before +// listing, then stripped from the results. +func listBrowseDir(entries []archives.FileInfo, prefix, dirPath string) []archives.FileInfo { + files := listDir(entries, prefix+dirPath) + if prefix == "" { + return files + } + return stripBrowsePrefix(files, prefix) +} + +// listDir mirrors archives.Reader.ListDir: entries directly in dirPath are +// returned in archive order. Deeper paths add one synthesized entry for each +// immediate subdirectory. +func listDir(entries []archives.FileInfo, dirPath string) []archives.FileInfo { + dirPath = normalizeBrowseDir(dirPath) + var files []archives.FileInfo + seenDirs := make(map[string]bool) + + for _, f := range entries { + p := f.Path + + if isInBrowseDir(p, dirPath) { + if f.IsDir { + name := strings.TrimSuffix(strings.TrimPrefix(p, dirPath), "/") + if seenDirs[name] { + continue + } + seenDirs[name] = true + } + files = append(files, f) + continue + } + + if dirPath != "" && !strings.HasPrefix(p, dirPath) { + continue + } + rel := strings.TrimSuffix(strings.TrimPrefix(p, dirPath), "/") + name, _, nested := strings.Cut(rel, "/") + if nested && !seenDirs[name] { + seenDirs[name] = true + files = append(files, archives.FileInfo{ + Path: dirPath + name + "/", + Name: name, + IsDir: true, + }) + } + } + + return files +} + +func normalizeBrowseDir(dirPath string) string { + dirPath = strings.Trim(strings.TrimSpace(dirPath), "/") + if dirPath == "" { + return "" + } + return dirPath + "/" +} + +// isInBrowseDir reports whether filePath is directly in the normalized +// dirPath. An entry naming the directory itself counts as inside it. +func isInBrowseDir(filePath, dirPath string) bool { + filePath = strings.TrimSuffix(filePath, "/") + if dirPath == "" { + return !strings.Contains(filePath, "/") + } + if filePath == dirPath[:len(dirPath)-1] { + return true + } + rest, ok := strings.CutPrefix(filePath, dirPath) + return ok && !strings.Contains(rest, "/") +} + +// stripBrowsePrefix removes prefix from each path, dropping entries outside +// it and the prefix directory itself. +func stripBrowsePrefix(files []archives.FileInfo, prefix string) []archives.FileInfo { + result := make([]archives.FileInfo, 0, len(files)) + for _, f := range files { + stripped, ok := strings.CutPrefix(f.Path, prefix) + if !ok || stripped == "" || stripped == "/" { + continue + } + f.Path = stripped + f.Name = entryName(stripped) + result = append(result, f) + } + return result +} + +// entryName returns the last element of an archive path, ignoring a trailing +// slash. +func entryName(p string) string { + p = strings.TrimSuffix(p, "/") + if i := strings.LastIndexByte(p, '/'); i >= 0 { + return p[i+1:] + } + return p +} diff --git a/internal/server/browse_archive_test.go b/internal/server/browse_archive_test.go new file mode 100644 index 0000000..71491f2 --- /dev/null +++ b/internal/server/browse_archive_test.go @@ -0,0 +1,680 @@ +package server + +import ( + "archive/tar" + "archive/zip" + "bytes" + "compress/gzip" + "database/sql" + "encoding/json" + "fmt" + "io" + "net/http" + "net/http/httptest" + "net/url" + "os" + "path/filepath" + "runtime" + "strings" + "testing" + + "github.com/git-pkgs/archives" + "github.com/git-pkgs/proxy/internal/database" +) + +type archiveEntry struct { + name string + body string + dir bool +} + +func buildTar(t testing.TB, entries []archiveEntry) []byte { + t.Helper() + buf := new(bytes.Buffer) + tw := tar.NewWriter(buf) + for _, e := range entries { + header := &tar.Header{Name: e.name, Mode: 0o644, Size: int64(len(e.body)), Typeflag: tar.TypeReg} + if e.dir { + header.Typeflag = tar.TypeDir + header.Mode = 0o755 + header.Size = 0 + } + if err := tw.WriteHeader(header); err != nil { + t.Fatalf("writing tar header: %v", err) + } + if _, err := io.WriteString(tw, e.body); err != nil { + t.Fatalf("writing tar body: %v", err) + } + } + if err := tw.Close(); err != nil { + t.Fatalf("closing tar: %v", err) + } + return buf.Bytes() +} + +func gzipBytes(t testing.TB, data []byte) []byte { + t.Helper() + buf := new(bytes.Buffer) + gw := gzip.NewWriter(buf) + if _, err := gw.Write(data); err != nil { + t.Fatalf("writing gzip: %v", err) + } + if err := gw.Close(); err != nil { + t.Fatalf("closing gzip: %v", err) + } + return buf.Bytes() +} + +func buildTarGz(t testing.TB, entries []archiveEntry) []byte { + t.Helper() + return gzipBytes(t, buildTar(t, entries)) +} + +func buildZip(t testing.TB, entries []archiveEntry) []byte { + t.Helper() + buf := new(bytes.Buffer) + zw := zip.NewWriter(buf) + for _, e := range entries { + f, err := zw.Create(e.name) + if err != nil { + t.Fatalf("creating zip entry: %v", err) + } + if _, err := io.WriteString(f, e.body); err != nil { + t.Fatalf("writing zip entry: %v", err) + } + } + if err := zw.Close(); err != nil { + t.Fatalf("closing zip: %v", err) + } + return buf.Bytes() +} + +// buildGem wraps entries in a gem: an outer tar holding data.tar.gz. +func buildGem(t testing.TB, entries []archiveEntry) []byte { + t.Helper() + return buildTar(t, []archiveEntry{ + {name: "metadata.gz", body: string(gzipBytes(t, []byte("--- {}\n")))}, + {name: "data.tar.gz", body: string(buildTarGz(t, entries))}, + }) +} + +// testBrowseVersion is the version every browse fixture is cached under. +const testBrowseVersion = "1.0.0" + +// cacheBrowseArtifact stores data as the cached artifact for ecosystem/name. +func cacheBrowseArtifact(t *testing.T, ts *testServer, ecosystem, name, filename string, data []byte) { + t.Helper() + + storagePath := fmt.Sprintf("%s-%s-%s", ecosystem, name, filename) + artifactsDir := filepath.Join(ts.tempDir, "artifacts") + if err := os.MkdirAll(artifactsDir, 0o755); err != nil { + t.Fatalf("creating artifacts dir: %v", err) + } + if err := os.WriteFile(filepath.Join(artifactsDir, storagePath), data, 0o644); err != nil { + t.Fatalf("writing artifact: %v", err) + } + + pkg := &database.Package{ + PURL: fmt.Sprintf("pkg:%s/%s", ecosystem, name), + Ecosystem: ecosystem, + Name: name, + } + if err := ts.db.UpsertPackage(pkg); err != nil { + t.Fatalf("upserting package: %v", err) + } + ver := &database.Version{PURL: pkg.PURL + "@" + testBrowseVersion, PackagePURL: pkg.PURL} + if err := ts.db.UpsertVersion(ver); err != nil { + t.Fatalf("upserting version: %v", err) + } + artifact := &database.Artifact{ + VersionPURL: ver.PURL, + Filename: filename, + UpstreamURL: "https://example.com/" + filename, + StoragePath: sql.NullString{String: storagePath, Valid: true}, + } + if err := ts.db.UpsertArtifact(artifact); err != nil { + t.Fatalf("upserting artifact: %v", err) + } +} + +func browseGet(ts *testServer, target string) *httptest.ResponseRecorder { + req := httptest.NewRequest(http.MethodGet, target, nil) + w := httptest.NewRecorder() + ts.handler.ServeHTTP(w, req) + return w +} + +func browseListURL(ecosystem, name, dir string) string { + return fmt.Sprintf("/ui/api/browse/%s/%s/%s?path=%s", ecosystem, name, testBrowseVersion, url.QueryEscape(dir)) +} + +func browseFileURL(ecosystem, name, file string) string { + return fmt.Sprintf("/ui/api/browse/%s/%s/%s/file/%s", ecosystem, name, testBrowseVersion, file) +} + +func decodeListing(t *testing.T, w *httptest.ResponseRecorder) BrowseListResponse { + t.Helper() + if w.Code != http.StatusOK { + t.Fatalf("listing status = %d, body %s", w.Code, w.Body.String()) + } + var resp BrowseListResponse + if err := json.NewDecoder(w.Body).Decode(&resp); err != nil { + t.Fatalf("decoding listing: %v", err) + } + return resp +} + +func listingPaths(resp BrowseListResponse) []string { + paths := make([]string, len(resp.Files)) + for i, f := range resp.Files { + paths[i] = f.Path + } + return paths +} + +type browseFixture struct { + name string + ecosystem string + filename string + data []byte +} + +func browseFixtures(t *testing.T) []browseFixture { + t.Helper() + + rooted := []archiveEntry{ + {name: "repo-abc123/", dir: true}, + {name: "repo-abc123/README.md", body: "# rooted\n"}, + {name: "repo-abc123/src/", dir: true}, + {name: "repo-abc123/src/main.go", body: "package main\n"}, + {name: "repo-abc123/src/util/strings.go", body: "package util\n"}, + {name: "repo-abc123/docs/guide.md", body: "guide\n"}, + } + npm := []archiveEntry{ + {name: "package/package.json", body: `{"name":"x"}`}, + {name: "package/README.md", body: "# npm\n"}, + {name: "package/src/main.go", body: "package main\n"}, + {name: "package/src/util/strings.go", body: "package util\n"}, + {name: "other/ignored.txt", body: "outside the package prefix\n"}, + } + flat := []archiveEntry{ + {name: "README.md", body: "# flat\n"}, + {name: "src/main.go", body: "package main\n"}, + {name: "src/util/strings.go", body: "package util\n"}, + {name: "docs/guide.md", body: "guide\n"}, + } + + return []browseFixture{ + {"tar.gz with root dir", "cargo", "demo-1.0.0.crate", buildTarGz(t, rooted)}, + {"extensionless tar.gz", "cargo", "artifact", buildTarGz(t, rooted)}, + {"npm package prefix", "npm", "demo-1.0.0.tgz", buildTarGz(t, npm)}, + {"flat tar.gz", "cargo", "demo-1.0.0.tar.gz", buildTarGz(t, flat)}, + {"zip with root dir", "composer", "demo.zip", buildZip(t, rooted)}, + {"extensionless zip", "composer", "d2e2f014ccd6ec9fae8dbe6336a4164346a2a856", buildZip(t, rooted)}, + {"flat zip", "composer", "demo.zip", buildZip(t, flat)}, + {"gem", "gem", "demo-1.0.0.gem", buildGem(t, flat)}, + } +} + +func TestBrowseListingStripsPrefix(t *testing.T) { + for _, fx := range browseFixtures(t) { + t.Run(fx.name, func(t *testing.T) { + ts := newTestServer(t) + defer ts.close() + cacheBrowseArtifact(t, ts, fx.ecosystem, "demo", fx.filename, fx.data) + + root := listingPaths(decodeListing(t, browseGet(ts, browseListURL(fx.ecosystem, "demo", "")))) + for _, p := range root { + if strings.HasPrefix(p, "repo-abc123/") || strings.HasPrefix(p, "package/") { + t.Errorf("root listing kept archive prefix: %q", p) + } + } + if !containsPath(root, "README.md") || !containsPath(root, "src/") { + t.Errorf("root listing = %v, want README.md and src/", root) + } + + src := listingPaths(decodeListing(t, browseGet(ts, browseListURL(fx.ecosystem, "demo", "src")))) + if !containsPath(src, "src/main.go") || !containsPath(src, "src/util/") { + t.Errorf("src listing = %v, want src/main.go and src/util/", src) + } + }) + } +} + +func containsPath(paths []string, want string) bool { + for _, p := range paths { + if p == want { + return true + } + } + return false +} + +// TestBrowseListingMatchesBufferedReader checks the streamed listing returns +// exactly what the buffered archive reader returned for the same request. +func TestBrowseListingMatchesBufferedReader(t *testing.T) { + dirs := []string{"", "/", "src", "src/", "/src", "src/util", "docs", "missing", "README.md"} + + for _, fx := range browseFixtures(t) { + t.Run(fx.name, func(t *testing.T) { + ts := newTestServer(t) + defer ts.close() + cacheBrowseArtifact(t, ts, fx.ecosystem, "demo", fx.filename, fx.data) + + buffered, err := openArchive(fx.filename, bytes.NewReader(fx.data), fx.ecosystem) + if err != nil { + t.Fatalf("openArchive: %v", err) + } + defer func() { _ = buffered.Close() }() + + for _, dir := range dirs { + want, err := buffered.ListDir(dir) + if err != nil { + t.Fatalf("buffered ListDir(%q): %v", dir, err) + } + got := decodeListing(t, browseGet(ts, browseListURL(fx.ecosystem, "demo", dir))) + if len(got.Files) != len(want) { + t.Fatalf("ListDir(%q) = %v, want %d entries", dir, listingPaths(got), len(want)) + } + for i, f := range got.Files { + w := want[i] + if f.Path != w.Path || f.Name != w.Name || f.IsDir != w.IsDir || f.Size != w.Size { + t.Errorf("ListDir(%q)[%d] = %+v, want %+v", dir, i, f, w) + } + } + } + }) + } +} + +func TestBrowseFileContents(t *testing.T) { + for _, fx := range browseFixtures(t) { + t.Run(fx.name, func(t *testing.T) { + ts := newTestServer(t) + defer ts.close() + cacheBrowseArtifact(t, ts, fx.ecosystem, "demo", fx.filename, fx.data) + + buffered, err := openArchive(fx.filename, bytes.NewReader(fx.data), fx.ecosystem) + if err != nil { + t.Fatalf("openArchive: %v", err) + } + defer func() { _ = buffered.Close() }() + + for _, file := range []string{"README.md", "src/main.go", "src/util/strings.go"} { + rc, err := buffered.Extract(file) + if err != nil { + t.Fatalf("buffered Extract(%q): %v", file, err) + } + want, _ := io.ReadAll(rc) + _ = rc.Close() + + w := browseGet(ts, browseFileURL(fx.ecosystem, "demo", file)) + if w.Code != http.StatusOK { + t.Fatalf("file %s status = %d, body %s", file, w.Code, w.Body.String()) + } + if w.Body.String() != string(want) { + t.Errorf("file %s = %q, want %q", file, w.Body.String(), want) + } + wantType, _ := detectContentTypeFromPath(file) + if got := w.Header().Get("Content-Type"); got != wantType { + t.Errorf("file %s Content-Type = %q, want %q", file, got, wantType) + } + if got := w.Header().Get("Content-Disposition"); !strings.Contains(got, filepath.Base(file)) { + t.Errorf("file %s Content-Disposition = %q", file, got) + } + } + + w := browseGet(ts, browseFileURL(fx.ecosystem, "demo", "missing.txt")) + if w.Code != http.StatusNotFound { + t.Errorf("missing file status = %d, want 404", w.Code) + } + }) + } +} + +func TestBrowseFileFirstDuplicateWins(t *testing.T) { + ts := newTestServer(t) + defer ts.close() + data := buildTarGz(t, []archiveEntry{ + {name: "package/index.js", body: "first"}, + {name: "package/index.js", body: "second"}, + }) + cacheBrowseArtifact(t, ts, "npm", "dup", "dup-1.0.0.tgz", data) + + w := browseGet(ts, browseFileURL("npm", "dup", "index.js")) + if w.Code != http.StatusOK || w.Body.String() != "first" { + t.Fatalf("got %d %q, want 200 \"first\"", w.Code, w.Body.String()) + } +} + +func TestBrowseLimits(t *testing.T) { + entries := []archiveEntry{ + {name: "package/a.txt", body: strings.Repeat("a", 100)}, + {name: "package/b.txt", body: strings.Repeat("b", 100)}, + {name: "package/c.txt", body: strings.Repeat("c", 100)}, + } + tarGz := buildTarGz(t, entries) + zipped := buildZip(t, entries) + + tests := []struct { + name string + filename string + data []byte + limits archives.StreamOptions + file string + }{ + {"entry count", "limit.tgz", tarGz, archives.StreamOptions{MaxEntries: 2}, "c.txt"}, + {"entry size", "limit.tgz", tarGz, archives.StreamOptions{MaxEntryBytes: 50}, "a.txt"}, + {"expanded size", "limit.tgz", tarGz, archives.StreamOptions{MaxExpandedBytes: 250}, "c.txt"}, + {"input size", "limit.tgz", tarGz, archives.StreamOptions{MaxInputBytes: int64(len(tarGz) / 2)}, "c.txt"}, + {"zip entry count", "limit.zip", zipped, archives.StreamOptions{MaxEntries: 2}, "c.txt"}, + {"zip input size", "limit.zip", zipped, archives.StreamOptions{MaxInputBytes: int64(len(zipped) / 2)}, "a.txt"}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ts := newTestServer(t) + defer ts.close() + ts.server.browseLimits = tt.limits + cacheBrowseArtifact(t, ts, "npm", "limited", tt.filename, tt.data) + + for _, target := range []string{ + browseListURL("npm", "limited", ""), + browseFileURL("npm", "limited", tt.file), + } { + w := browseGet(ts, target) + if w.Code != http.StatusInternalServerError { + t.Fatalf("%s status = %d, want 500 (body %s)", target, w.Code, w.Body.String()) + } + if !strings.Contains(w.Body.String(), "archive exceeds browse limits") { + t.Errorf("%s body = %s, want limit error", target, w.Body.String()) + } + } + }) + } +} + +// largeBrowseArchive is a gzip tarball with many unrelated entries and one +// small file. The padding compresses well, so the fixture is small on disk +// while its expanded size dwarfs what a streamed request should allocate. +func largeBrowseArchive(t testing.TB, entries, entrySize int) []byte { + t.Helper() + buf := new(bytes.Buffer) + gw := gzip.NewWriter(buf) + tw := tar.NewWriter(gw) + padding := make([]byte, entrySize) + write := func(name string, body []byte) { + if err := tw.WriteHeader(&tar.Header{Name: name, Mode: 0o644, Size: int64(len(body))}); err != nil { + t.Fatalf("writing tar header: %v", err) + } + if _, err := tw.Write(body); err != nil { + t.Fatalf("writing tar body: %v", err) + } + } + write("repo-large/README.md", []byte("small file\n")) + for i := range entries { + write(fmt.Sprintf("repo-large/vendor/blob%05d.bin", i), padding) + } + if err := tw.Close(); err != nil { + t.Fatalf("closing tar: %v", err) + } + if err := gw.Close(); err != nil { + t.Fatalf("closing gzip: %v", err) + } + return buf.Bytes() +} + +// allocatedBytes returns the bytes allocated while fn runs. Total allocation +// bounds the peak heap growth fn can cause. +func allocatedBytes(fn func()) uint64 { + runtime.GC() + var before, after runtime.MemStats + runtime.ReadMemStats(&before) + fn() + runtime.ReadMemStats(&after) + return after.TotalAlloc - before.TotalAlloc +} + +func TestBrowseLargeArchiveMemory(t *testing.T) { + const ( + entries = 1024 + entrySize = 32 << 10 + expanded = entries * entrySize // 32 MiB + ) + data := largeBrowseArchive(t, entries, entrySize) + + ts := newTestServer(t) + defer ts.close() + cacheBrowseArtifact(t, ts, "cargo", "large", "large-1.0.0.crate", data) + + buffered := allocatedBytes(func() { + r, err := openArchive("large-1.0.0.crate", bytes.NewReader(data), "cargo") + if err != nil { + t.Fatalf("openArchive: %v", err) + } + _ = r.Close() + }) + + var listing BrowseListResponse + list := allocatedBytes(func() { + listing = decodeListing(t, browseGet(ts, browseListURL("cargo", "large", ""))) + }) + if !containsPath(listingPaths(listing), "README.md") || !containsPath(listingPaths(listing), "vendor/") { + t.Fatalf("listing = %v, want README.md and vendor/", listingPaths(listing)) + } + + var w *httptest.ResponseRecorder + file := allocatedBytes(func() { + w = browseGet(ts, browseFileURL("cargo", "large", "README.md")) + }) + if w.Code != http.StatusOK || w.Body.String() != "small file\n" { + t.Fatalf("file = %d %q, want 200 \"small file\\n\"", w.Code, w.Body.String()) + } + + t.Logf("compressed %d bytes, expanded %d bytes", len(data), expanded) + t.Logf("allocated: buffered open %d, streamed listing %d, streamed file %d", buffered, list, file) + + if buffered < expanded { + t.Fatalf("buffered open allocated %d bytes, expected at least the expanded %d", buffered, expanded) + } + limit := uint64(expanded / 4) + if list > limit { + t.Errorf("streamed listing allocated %d bytes, want under %d", list, limit) + } + if file > limit { + t.Errorf("streamed file read allocated %d bytes, want under %d", file, limit) + } +} + +func BenchmarkBrowseLargeArchive(b *testing.B) { + data := largeBrowseArchive(b, 1024, 32<<10) + artifactPath := filepath.Join(b.TempDir(), "large.crate") + if err := os.WriteFile(artifactPath, data, 0o644); err != nil { + b.Fatal(err) + } + + b.Run("buffered-list", func(b *testing.B) { benchBufferedBrowse(b, data, false) }) + b.Run("streamed-list", func(b *testing.B) { benchStreamedBrowse(b, artifactPath, false) }) + b.Run("buffered-file", func(b *testing.B) { benchBufferedBrowse(b, data, true) }) + b.Run("streamed-file", func(b *testing.B) { benchStreamedBrowse(b, artifactPath, true) }) +} + +// benchBufferedBrowse lists the root, or reads README.md, through openArchive. +func benchBufferedBrowse(b *testing.B, data []byte, readFile bool) { + b.ReportAllocs() + for b.Loop() { + r, err := openArchive("large.crate", bytes.NewReader(data), "cargo") + if err != nil { + b.Fatal(err) + } + if readFile { + drainBrowseFile(b, r.Extract) + } else if _, err := r.ListDir(""); err != nil { + b.Fatal(err) + } + _ = r.Close() + } +} + +// benchStreamedBrowse lists the root, or reads README.md, through browseArchive. +func benchStreamedBrowse(b *testing.B, artifactPath string, readFile bool) { + reopen := func() (io.ReadCloser, error) { return os.Open(artifactPath) } + b.ReportAllocs() + for b.Loop() { + content, err := reopen() + if err != nil { + b.Fatal(err) + } + a, err := newBrowseArchive("large.crate", "cargo", content, reopen, defaultBrowseLimits) + if err != nil { + b.Fatal(err) + } + if readFile { + drainBrowseFile(b, a.Extract) + } else if _, err := a.ListDir(""); err != nil { + b.Fatal(err) + } + _ = content.Close() + } +} + +func drainBrowseFile(b *testing.B, extract func(string) (io.ReadCloser, error)) { + b.Helper() + rc, err := extract("README.md") + if err != nil { + b.Fatal(err) + } + _, _ = io.Copy(io.Discard, rc) + _ = rc.Close() +} + +// truncatedTar is a TAR whose only entry declares size bytes but carries just +// body, ending without the rest of the entry or the end-of-archive marker. +func truncatedTar(t testing.TB, name string, size int64, body []byte) []byte { + t.Helper() + buf := new(bytes.Buffer) + tw := tar.NewWriter(buf) + if err := tw.WriteHeader(&tar.Header{Name: name, Mode: 0o644, Size: size}); err != nil { + t.Fatalf("writing tar header: %v", err) + } + if _, err := tw.Write(body); err != nil { + t.Fatalf("writing tar body: %v", err) + } + // Deliberately not closed: tw.Close would fail on the short entry. + return buf.Bytes() +} + +func TestBrowseFileReadErrorBeforeResponse(t *testing.T) { + complete := buildTar(t, []archiveEntry{{name: "package/data.txt", body: strings.Repeat("x", 4096)}}) + + tests := []struct { + name string + filename string + data []byte + limits archives.StreamOptions + wantBody string + }{ + { + name: "truncated entry body", + filename: "short-1.0.0.tgz", + data: gzipBytes(t, truncatedTar(t, "package/data.txt", 4096, []byte("abc"))), + wantBody: "failed to read file", + }, + { + name: "input limit inside entry", + filename: "short-1.0.0.tar", + data: complete, + limits: archives.StreamOptions{MaxInputBytes: 1024}, + wantBody: "archive exceeds browse limits", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ts := newTestServer(t) + defer ts.close() + ts.server.browseLimits = tt.limits + cacheBrowseArtifact(t, ts, "npm", "short", tt.filename, tt.data) + + w := browseGet(ts, browseFileURL("npm", "short", "data.txt")) + if w.Code != http.StatusInternalServerError { + t.Fatalf("status = %d, want 500 (body %q)", w.Code, w.Body.String()) + } + if strings.Contains(w.Body.String(), "abc") || strings.Contains(w.Body.String(), "xxx") { + t.Errorf("error response leaked partial file content: %q", w.Body.String()) + } + if !strings.Contains(w.Body.String(), tt.wantBody) { + t.Errorf("body = %q, want %q", w.Body.String(), tt.wantBody) + } + }) + } +} + +func TestBrowseFileReadErrorAfterResponseStarts(t *testing.T) { + const size = 4 * browsePrefetchSize + body := bytes.Repeat([]byte("0123456789abcdef"), size/16) + complete := buildTar(t, []archiveEntry{{name: "package/big.txt", body: string(body)}}) + + tests := []struct { + name string + filename string + data []byte + limits archives.StreamOptions + wantErr bool + }{ + { + name: "complete file", + filename: "big-1.0.0.tar", + data: complete, + }, + { + name: "truncated entry body", + filename: "big-1.0.0.tgz", + data: gzipBytes(t, truncatedTar(t, "package/big.txt", size, body[:size/2])), + wantErr: true, + }, + { + name: "input limit inside entry", + filename: "big-1.0.0.tar", + data: complete, + limits: archives.StreamOptions{MaxInputBytes: size / 2}, + wantErr: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ts := newTestServer(t) + defer ts.close() + ts.server.browseLimits = tt.limits + cacheBrowseArtifact(t, ts, "npm", "big", tt.filename, tt.data) + + srv := httptest.NewServer(ts.handler) + defer srv.Close() + + resp, err := http.Get(srv.URL + browseFileURL("npm", "big", "big.txt")) + if err != nil { + if tt.wantErr { + return + } + t.Fatalf("GET: %v", err) + } + defer func() { _ = resp.Body.Close() }() + got, readErr := io.ReadAll(resp.Body) + + if !tt.wantErr { + if resp.StatusCode != http.StatusOK || readErr != nil || !bytes.Equal(got, body) { + t.Fatalf("status %d, read error %v, %d of %d bytes", resp.StatusCode, readErr, len(got), len(body)) + } + return + } + if readErr == nil { + t.Fatalf("status %d with %d of %d bytes and no read error; want an aborted response", + resp.StatusCode, len(got), len(body)) + } + if len(got) >= len(body) { + t.Errorf("read %d bytes, want fewer than %d", len(got), len(body)) + } + }) + } +} diff --git a/internal/server/server.go b/internal/server/server.go index 7494b94..69707f7 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -64,6 +64,7 @@ import ( "strings" "time" + "github.com/git-pkgs/archives" "github.com/git-pkgs/cooldown" swaggerdoc "github.com/git-pkgs/proxy/docs/swagger" "github.com/git-pkgs/proxy/internal/accesslog" @@ -117,6 +118,9 @@ type Server struct { breakers *breakerMonitor ecoStats ecosystemStatsCache sources sourceTracker + + // browseLimits overrides defaultBrowseLimits when set; tests use it. + browseLimits archives.StreamOptions } // New creates a new Server with the given configuration.