diff --git a/README.md b/README.md index e8ecec05..bcfd3276 100644 --- a/README.md +++ b/README.md @@ -925,7 +925,8 @@ git config --global url."http://localhost:8080/git/github/".insteadOf https://gi ``` Each repository is cached as a bare clone and refreshed whenever its branches -or tags change upstream. See [configuration](docs/configuration.md#git-repositories) +or tags change upstream. Git LFS objects are cached in the artifact cache; +git-lfs uses the proxy for them with no further setup. See [configuration](docs/configuration.md#git-repositories) for setting the rewrite through environment variables, size limits, private repositories and outage behavior. diff --git a/docs/architecture.md b/docs/architecture.md index adaaf7b2..e7ce8c8f 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -308,6 +308,7 @@ HTTP protocol handlers for each registry type. **GitHandler:** - Serves `/git/{name}/{repo}` from `internal/gitcache`, read-only over smart HTTP - Validates repository paths and the `repos` allowlist, and maps cache errors to statuses +- Serves the Git LFS API under `{repo}.git/info/lfs/`: forwards download batches upstream and points each object at `objects/{oid}/{size}`, which is served from the artifact cache (ecosystem `git`, name `{name}/{repo}`, version the object ID); a miss uses the download URL kept from the batch or asks the upstream batch API for a fresh one, recorded without its signed query, copies the object from another repository when its content is already stored, and checks the object against its sha256 ID; a batch the upstream can't answer is answered from the cache ### `internal/gitcache` diff --git a/docs/configuration.md b/docs/configuration.md index 1049f2b4..269d5bdb 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -837,7 +837,42 @@ those yourself. The proxy locks `cache_dir` while it runs, through a `.gitcache-lock` file in it, so a second process given the same directory, such as another replica on a shared volume, fails to start; give each its own. -Pushing, the dumb HTTP protocol and Git LFS are not supported. The settings +Git LFS objects are cached in the artifact cache, so they use the configured +storage backend and retention rather than `cache_dir`. git-lfs finds the LFS +API under the rewritten remote, at `/git/{name}/{repo}.git/info/lfs`, with no +further configuration. A batch request is passed to the upstream, which +decides which objects the repository has, and each download is pointed at +the proxy. An object the cache doesn't have is fetched from the download URL +the batch returned, which the proxy keeps in memory until shortly before it +expires, or else from one a fresh batch request returns. That URL is often +on another host than the Git server, such as GitHub's or Hugging Face's +object storage; the proxy fetches from whatever host the upstream names, +with the usual address checks, and `upstream.auth` credentials only go to +the hosts they are configured for. A download that isn't the object's size +is cut off. If the object's content is already stored under another +repository, such as a fork, it is copied from there instead of downloaded, +once the upstream has confirmed this repository has it; the copy takes its +own space. + +Large LFS objects may need longer timeouts. The proxy downloads an object it +doesn't have before sending any of it, within `artifact_timeout`, and the +whole response, download included, must fit within `write_timeout`; both +default to 5 minutes. git-lfs gives up on a response after 30 seconds without +data (`lfs.activitytimeout`) and retries. A retry waits for the download still +running, so the object is served once it is cached, but a download that +outlasts git-lfs's retries fails that clone or pull until it finishes. +Download URLs are often signed; only the URL without its query is logged +and recorded with the object. An object is only cached once its content +matches its sha256 object ID. Objects are cached per repository, so one +repository's objects are never served for another. When the upstream can't +be reached or fails, a batch is answered with the objects cached for the +repository, and the others are reported as unavailable; a batch the upstream +refuses is refused. A cached object stays available after it is removed +upstream, until retention removes it or you delete it through the +[cache delete API](#cache-delete-api). + +Pushing, LFS uploads and locking, and the dumb HTTP protocol are not +supported. The settings under `git` can also be set with `PROXY_GIT_CACHE_DIR`, `PROXY_GIT_MAX_SIZE`, `PROXY_GIT_MAX_REPO_SIZE`, `PROXY_GIT_REFRESH_WAIT` and `PROXY_GIT_STALE_MAX_AGE`. diff --git a/internal/database/queries.go b/internal/database/queries.go index 4b039d8d..7af5d070 100644 --- a/internal/database/queries.go +++ b/internal/database/queries.go @@ -265,6 +265,34 @@ func (db *DB) GetCachedArtifact(packagePURL, versionPURL, filename string) (*Cac return row.artifact(versionPURL, filename), nil } +// StoredContent is a stored object with a given content hash. +type StoredContent struct { + StoragePath string `db:"storage_path"` + Size int64 `db:"size"` + ContentType sql.NullString `db:"content_type"` +} + +// FindStoredContent returns a cached artifact whose content has the sha256 +// hash, under any name, or nil if none is stored. Pushed artifacts are left +// out. +func (db *DB) FindStoredContent(hash string) (*StoredContent, error) { + var found StoredContent + query := db.Rebind(` + SELECT storage_path, size, content_type FROM artifacts + WHERE content_hash = ? AND storage_path IS NOT NULL AND fetched_at IS NOT NULL + AND size IS NOT NULL AND source <> 'local' + LIMIT 1 + `) + err := db.Get(&found, query, hash) + if err == sql.ErrNoRows { + return nil, nil + } + if err != nil { + return nil, err + } + return &found, nil +} + type cachedArtifactRow struct { Ecosystem string `db:"ecosystem"` StoragePath string `db:"storage_path"` diff --git a/internal/database/schema.go b/internal/database/schema.go index b2fef631..46a21eca 100644 --- a/internal/database/schema.go +++ b/internal/database/schema.go @@ -79,6 +79,7 @@ CREATE TABLE IF NOT EXISTS artifacts ( ); CREATE UNIQUE INDEX IF NOT EXISTS idx_artifacts_version_filename ON artifacts(version_purl, filename); CREATE INDEX IF NOT EXISTS idx_artifacts_storage_path ON artifacts(storage_path); +CREATE INDEX IF NOT EXISTS idx_artifacts_content_hash ON artifacts(content_hash); CREATE INDEX IF NOT EXISTS idx_artifacts_last_accessed ON artifacts(last_accessed_at); CREATE TABLE IF NOT EXISTS pending_deletes ( @@ -220,6 +221,7 @@ CREATE TABLE IF NOT EXISTS artifacts ( ); CREATE UNIQUE INDEX IF NOT EXISTS idx_artifacts_version_filename ON artifacts(version_purl, filename); CREATE INDEX IF NOT EXISTS idx_artifacts_storage_path ON artifacts(storage_path); +CREATE INDEX IF NOT EXISTS idx_artifacts_content_hash ON artifacts(content_hash); CREATE INDEX IF NOT EXISTS idx_artifacts_last_accessed ON artifacts(last_accessed_at); CREATE TABLE IF NOT EXISTS pending_deletes ( @@ -320,6 +322,7 @@ CREATE TABLE IF NOT EXISTS artifacts ( ); CREATE UNIQUE INDEX IF NOT EXISTS idx_artifacts_version_filename ON artifacts(version_purl, filename); CREATE INDEX IF NOT EXISTS idx_artifacts_storage_path ON artifacts(storage_path); +CREATE INDEX IF NOT EXISTS idx_artifacts_content_hash ON artifacts(content_hash); CREATE INDEX IF NOT EXISTS idx_artifacts_last_accessed ON artifacts(last_accessed_at); ` @@ -343,6 +346,7 @@ CREATE TABLE IF NOT EXISTS artifacts ( ); CREATE UNIQUE INDEX IF NOT EXISTS idx_artifacts_version_filename ON artifacts(version_purl, filename); CREATE INDEX IF NOT EXISTS idx_artifacts_storage_path ON artifacts(storage_path); +CREATE INDEX IF NOT EXISTS idx_artifacts_content_hash ON artifacts(content_hash); CREATE INDEX IF NOT EXISTS idx_artifacts_last_accessed ON artifacts(last_accessed_at); ` @@ -457,6 +461,7 @@ var migrations = []migration{ {"014_add_metadata_last_accessed", migrateAddMetadataLastAccessed}, {"015_add_job_runs", migrateAddJobRuns}, {"016_add_fetch_leases", migrateAddFetchLeases}, + {"017_add_artifact_content_hash_index", migrateAddArtifactContentHashIndex}, {"018_add_upstream_holds", migrateAddUpstreamHolds}, } @@ -820,6 +825,16 @@ func migrateAddPendingDeletesQueuedAtIndex(db *DB) error { return nil } +// migrateAddArtifactContentHashIndex indexes artifacts by content, so an +// artifact already stored under another name can be found by its hash. +func migrateAddArtifactContentHashIndex(db *DB) error { + _, err := db.Exec(`CREATE INDEX IF NOT EXISTS idx_artifacts_content_hash ON artifacts(content_hash)`) + if err != nil { + return fmt.Errorf("creating artifacts content_hash index: %w", err) + } + return nil +} + // migrateAddMetadataDates creates the table of Last-Modified dates for // rewritten metadata, which replicas sharing the database agree on. func migrateAddMetadataDates(db *DB) error { diff --git a/internal/handler/git.go b/internal/handler/git.go index 895be44a..7243d84d 100644 --- a/internal/handler/git.go +++ b/internal/handler/git.go @@ -39,6 +39,11 @@ type GitHandler struct { proxy *Proxy cache *gitcache.Cache upstreams map[string]gitUpstream + // baseURL is where clients reach the proxy, for LFS download URLs. + baseURL string + // lfsDownloads keeps upstream LFS download URLs between a batch and + // the GETs that follow it. + lfsDownloads *lfsDownloads // writeStall is gitWriteStallTimeout, shorter in tests. writeStall time.Duration } @@ -56,10 +61,11 @@ type gitUpstream struct { // NewGitHandler creates a Git handler serving upstream.git from cache. func NewGitHandler(proxy *Proxy, cache *gitcache.Cache, upstream config.UpstreamConfig) *GitHandler { h := &GitHandler{ - proxy: proxy, - cache: cache, - upstreams: make(map[string]gitUpstream, len(upstream.Git)), - writeStall: gitWriteStallTimeout, + proxy: proxy, + cache: cache, + upstreams: make(map[string]gitUpstream, len(upstream.Git)), + writeStall: gitWriteStallTimeout, + lfsDownloads: &lfsDownloads{}, } for name, git := range upstream.Git { git.URL = strings.TrimSuffix(git.URL, "/") @@ -90,6 +96,10 @@ type gitRequest struct { } func (h *GitHandler) serve(w http.ResponseWriter, r *http.Request) { + if repoPath, endpoint, ok := strings.Cut(strings.TrimPrefix(r.URL.Path, "/"), gitLFSPath); ok { + h.serveLFS(w, r, repoPath, endpoint) + return + } req, status, msg := h.parse(r) if status != 0 { http.Error(w, msg, status) @@ -168,10 +178,38 @@ func (h *GitHandler) parse(r *http.Request) (gitRequest, int, string) { return req, http.StatusNotFound, "not found" } + repo, status, msg := h.repository(path) + if status != 0 { + return req, status, msg + } + req.key = repo.key + req.repoURL = repo.url + req.fetchWants = repo.upstream.fetchWants + return req, 0, "" +} + +// gitRepository is a repository of an upstream.git host, as a request path +// names it. +type gitRepository struct { + // name is the upstream's name and repo the path under it, without .git. + name, repo string + upstream gitUpstream + // key names the repository in the cache: {name}-{URL hash}/{repo}.git, + // with any .git the client gave removed from repo first, so both + // spellings share one copy. + key string + // url is the upstream repository, spelled as the client spelled it, and + // escapedRepo is repo escaped for a URL path. + url, escapedRepo string +} + +// repository resolves {name}/{repo} to an upstream repository, or returns +// the status and message to refuse it with. +func (h *GitHandler) repository(path string) (gitRepository, int, string) { name, repoPath, _ := strings.Cut(path, "/") upstream, ok := h.upstreams[name] if !ok { - return req, http.StatusNotFound, "unknown upstream" + return gitRepository{}, http.StatusNotFound, "unknown upstream" } segments := strings.Split(repoPath, "/") last := len(segments) - 1 @@ -182,7 +220,7 @@ func (h *GitHandler) parse(r *http.Request) (gitRequest, int, string) { escaped := make([]string, len(segments)) for i, segment := range segments { if !validGitSegment(segment, i == last) { - return req, http.StatusNotFound, "invalid repository path" + return gitRepository{}, http.StatusNotFound, "invalid repository path" } escaped[i] = url.PathEscape(segment) } @@ -190,12 +228,17 @@ func (h *GitHandler) parse(r *http.Request) (gitRequest, int, string) { if !upstream.AllowsRepo(repo) { // As for a missing repository, so a prober learns nothing about // the allowlist. - return req, http.StatusNotFound, "repository not found" + return gitRepository{}, http.StatusNotFound, "repository not found" } - req.key = upstream.dir + "/" + repo + ".git" - req.repoURL = upstream.URL + "/" + strings.Join(escaped, "/") + suffix - req.fetchWants = upstream.fetchWants - return req, 0, "" + escapedRepo := strings.Join(escaped, "/") + return gitRepository{ + name: name, + repo: repo, + upstream: upstream, + key: upstream.dir + "/" + repo + ".git", + url: upstream.URL + "/" + escapedRepo + suffix, + escapedRepo: escapedRepo, + }, 0, "" } // validGitSegment checks one segment of a repository path, which becomes both diff --git a/internal/handler/git_lfs.go b/internal/handler/git_lfs.go new file mode 100644 index 00000000..d4f6fe62 --- /dev/null +++ b/internal/handler/git_lfs.go @@ -0,0 +1,559 @@ +package handler + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "net/url" + "slices" + "strconv" + "strings" + "sync" + "time" +) + +const ( + // gitLFSPath separates a repository from the Git LFS API below it, as + // git-lfs derives the API URL from a remote: {remote}.git/info/lfs. + gitLFSPath = "/info/lfs/" + gitLFSMediaType = "application/vnd.git-lfs+json" + gitLFSTransfer = "basic" + gitLFSHashAlgo = "sha256" + // gitLFSObjectMediaType is what objects are served as. + gitLFSObjectMediaType = "application/octet-stream" + + // gitLFSMaxBatchRequest bounds a client's batch request. git-lfs sends + // 100 objects per batch by default, a few kilobytes. + gitLFSMaxBatchRequest = 1 << 20 + // gitLFSMaxBatchResponse bounds an upstream batch response, which holds + // an href and headers per object. + gitLFSMaxBatchResponse = 16 << 20 + + gitLFSOIDLength = 64 +) + +// lfsBatchRequest is a Git LFS batch API request. +type lfsBatchRequest struct { + Operation string `json:"operation"` + Transfers []string `json:"transfers,omitempty"` + Ref json.RawMessage `json:"ref,omitempty"` + Objects []lfsObject `json:"objects"` + HashAlgo string `json:"hash_algo,omitempty"` +} + +type lfsObject struct { + OID string `json:"oid"` + Size int64 `json:"size"` +} + +// lfsBatchResponse is a Git LFS batch API response. +type lfsBatchResponse struct { + Transfer string `json:"transfer,omitempty"` + Objects []lfsObjectResponse `json:"objects"` + HashAlgo string `json:"hash_algo,omitempty"` +} + +type lfsObjectResponse struct { + OID string `json:"oid"` + Size int64 `json:"size"` + Authenticated bool `json:"authenticated,omitempty"` + Actions map[string]lfsAction `json:"actions,omitempty"` + Error *lfsObjectError `json:"error,omitempty"` +} + +type lfsAction struct { + Href string `json:"href"` + Header map[string]string `json:"header,omitempty"` + ExpiresIn int64 `json:"expires_in,omitempty"` + ExpiresAt string `json:"expires_at,omitempty"` +} + +type lfsObjectError struct { + Code int `json:"code"` + Message string `json:"message"` +} + +// lfsUpstreamError is a batch request the upstream answered with an error +// status. +type lfsUpstreamError struct { + status int +} + +func (e *lfsUpstreamError) Error() string { + return fmt.Sprintf("LFS batch API answered %d", e.status) +} + +// WithBaseURL sets the URL clients reach the proxy at, which LFS batch +// responses send them back to for each object. +func (h *GitHandler) WithBaseURL(baseURL string) *GitHandler { + h.baseURL = strings.TrimSuffix(baseURL, "/") + return h +} + +// serveLFS answers a request to the Git LFS API of a repository: batch +// downloads, and the objects the batch responses point at. Objects are +// stored in the artifact cache, checked against their sha256 object ID. +// Uploads and locking are refused. +func (h *GitHandler) serveLFS(w http.ResponseWriter, r *http.Request, repoPath, endpoint string) { + repo, status, msg := h.repository(repoPath) + if status != 0 { + lfsError(w, status, msg) + return + } + if endpoint == "objects/batch" { + if r.Method != http.MethodPost { + lfsError(w, http.StatusMethodNotAllowed, "method not allowed") + return + } + h.serveLFSBatch(w, r, repo) + return + } + oid, sizeText, ok := strings.Cut(strings.TrimPrefix(endpoint, "objects/"), "/") + size, err := strconv.ParseInt(sizeText, 10, 64) + if !strings.HasPrefix(endpoint, "objects/") || !ok || err != nil || size < 0 || !validLFSOID(oid) { + lfsError(w, http.StatusNotFound, "not found") + return + } + if r.Method != http.MethodGet && r.Method != http.MethodHead { + lfsError(w, http.StatusMethodNotAllowed, "method not allowed") + return + } + h.serveLFSObject(w, r, repo, lfsObject{OID: oid, Size: size}) +} + +// serveLFSBatch forwards a download batch upstream and points every object +// the upstream offers at this proxy. The upstream answers for the +// repository, so a client only gets objects it could get from there. +func (h *GitHandler) serveLFSBatch(w http.ResponseWriter, r *http.Request, repo gitRepository) { + var req lfsBatchRequest + if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, gitLFSMaxBatchRequest)).Decode(&req); err != nil { + var tooLarge *http.MaxBytesError + if errors.As(err, &tooLarge) { + lfsError(w, http.StatusRequestEntityTooLarge, "batch request too large") + return + } + lfsError(w, http.StatusBadRequest, "invalid batch request") + return + } + switch { + case req.Operation != "download": + lfsError(w, http.StatusForbidden, "pushing is not supported") + return + case req.HashAlgo != "" && req.HashAlgo != gitLFSHashAlgo: + lfsError(w, http.StatusConflict, "only sha256 objects are supported") + return + case len(req.Transfers) > 0 && !slices.Contains(req.Transfers, gitLFSTransfer): + lfsError(w, http.StatusUnprocessableEntity, "only the basic transfer is supported") + return + } + requested := make(map[string]int64, len(req.Objects)) + for _, object := range req.Objects { + if !validLFSOID(object.OID) || object.Size < 0 { + lfsError(w, http.StatusUnprocessableEntity, "invalid object") + return + } + requested[object.OID] = object.Size + } + + upstream, err := h.lfsBatch(r.Context(), repo, req.Ref, req.Objects) + if err != nil { + if resp, ok := h.lfsBatchFromCache(r, repo, req.Objects, err); ok { + writeLFSBatch(w, resp) + return + } + h.serveLFSUpstreamError(w, r, repo, err) + return + } + resp := newLFSBatchResponse() + for _, object := range upstream.Objects { + // Only objects the client asked for, at the size it gave, so a + // response can't point the client at anything else. + size, ok := requested[object.OID] + if !ok || size != object.Size { + continue + } + out := lfsObjectResponse{OID: object.OID, Size: object.Size, Error: object.Error} + if download, ok := object.Actions["download"]; ok && object.Error == nil && download.Href != "" { + out.Authenticated = true + out.Actions = map[string]lfsAction{"download": {Href: h.lfsObjectURL(repo, object.OID, object.Size)}} + // Kept so the client's GET for the object needs no second + // batch request. + h.lfsDownloads.put(repo.key, lfsObject{OID: object.OID, Size: object.Size}, download, time.Now()) + } + resp.Objects = append(resp.Objects, out) + } + writeLFSBatch(w, resp) +} + +// lfsBatchFromCache answers a batch the upstream couldn't, because it was +// unreachable or failing, from the objects cached for repo. This is what a +// GET of each object would serve anyway. A refusal by the upstream is not +// overridden, and a batch none of whose objects are cached is left to fail. +func (h *GitHandler) lfsBatchFromCache(r *http.Request, repo gitRepository, objects []lfsObject, err error) (lfsBatchResponse, bool) { + var refused *lfsUpstreamError + if r.Context().Err() != nil || (errors.As(err, &refused) && refused.status < http.StatusInternalServerError) { + return lfsBatchResponse{}, false + } + resp := newLFSBatchResponse() + seen := make(map[string]bool, len(objects)) + cached := 0 + for _, object := range objects { + if seen[object.OID] { + continue + } + seen[object.OID] = true + out := lfsObjectResponse{OID: object.OID, Size: object.Size} + if ok, _ := h.proxy.HasCachedArtifact(r.Context(), gitEcosystem, repo.lfsName(), object.OID, object.OID); ok { + out.Authenticated = true + out.Actions = map[string]lfsAction{"download": {Href: h.lfsObjectURL(repo, object.OID, object.Size)}} + cached++ + } else { + out.Error = &lfsObjectError{Code: http.StatusBadGateway, Message: "upstream unavailable and object not cached"} + } + resp.Objects = append(resp.Objects, out) + } + if cached == 0 { + return lfsBatchResponse{}, false + } + h.proxy.Logger.Warn("git: LFS batch request failed, answering from the cache", + "repository", repo.key, "cached", cached, "objects", len(resp.Objects), "error", err) + return resp, true +} + +func newLFSBatchResponse() lfsBatchResponse { + return lfsBatchResponse{Transfer: gitLFSTransfer, Objects: []lfsObjectResponse{}, HashAlgo: gitLFSHashAlgo} +} + +func writeLFSBatch(w http.ResponseWriter, resp lfsBatchResponse) { + w.Header().Set("Content-Type", gitLFSMediaType) + _ = json.NewEncoder(w).Encode(resp) +} + +// serveLFSObject serves an object from the artifact cache. On a miss it +// takes the download URL from the client's batch, or asks the upstream batch +// API for a fresh one, and caches the object from there, or from a copy +// already stored under another repository, once its content matches its +// object ID. +func (h *GitHandler) serveLFSObject(w http.ResponseWriter, r *http.Request, repo gitRepository, object lfsObject) { + cached, err := h.proxy.GetCachedArtifact(r.Context(), gitEcosystem, repo.lfsName(), object.OID, object.OID) + if err != nil { + h.proxy.serveArtifactError(w, err, "failed to read cached LFS object") + return + } + if cached == nil { + var ok bool + if cached, ok = h.fetchLFSObject(w, r, repo, object); !ok { + return + } + } + serveGenericArtifact(w, r, cached) +} + +// fetchLFSObject caches an object the cache doesn't have. When it can't, or +// the request is a HEAD, it answers the request and returns false. +func (h *GitHandler) fetchLFSObject(w http.ResponseWriter, r *http.Request, repo gitRepository, object lfsObject) (*CacheResult, bool) { + download, kept := h.lfsDownloads.get(repo.key, object, time.Now()) + if !kept { + var ok bool + if download, ok = h.lfsFreshDownload(w, r, repo, object); !ok { + return nil, false + } + } + if r.Method == http.MethodHead { + // The upstream has the object, and a HEAD needs no more than that: + // fetching it is left to a GET. + w.Header().Set("Content-Type", gitLFSObjectMediaType) + w.Header().Set("Content-Length", strconv.FormatInt(object.Size, 10)) + return nil, false + } + h.lfsDownloads.drop(repo.key, object.OID) + cached, err := h.cacheLFSObject(r.Context(), repo, object, download) + if err != nil && kept && retryLFSDownload(r.Context(), err) { + // The kept URL may have expired early or been revoked. + var ok bool + if download, ok = h.lfsFreshDownload(w, r, repo, object); !ok { + return nil, false + } + cached, err = h.cacheLFSObject(r.Context(), repo, object, download) + } + if err != nil { + h.proxy.serveArtifactError(w, err, "failed to fetch LFS object") + return nil, false + } + return cached, true +} + +// lfsFreshDownload asks the upstream batch API for object's download. When +// there is none it answers the request and returns false. +func (h *GitHandler) lfsFreshDownload(w http.ResponseWriter, r *http.Request, repo gitRepository, object lfsObject) (lfsAction, bool) { + upstream, err := h.lfsBatch(r.Context(), repo, nil, []lfsObject{object}) + if err != nil { + h.serveLFSUpstreamError(w, r, repo, err) + return lfsAction{}, false + } + download, status, msg := lfsDownload(upstream, object) + if status != 0 { + lfsError(w, status, msg) + return lfsAction{}, false + } + return download, true +} + +// cacheLFSObject caches object, which the upstream offers as download. A +// copy already stored under another name, such as by a fork of the +// repository, saves the download. +func (h *GitHandler) cacheLFSObject(ctx context.Context, repo gitRepository, object lfsObject, download lfsAction) (*CacheResult, error) { + header := make(http.Header, len(download.Header)) + for key, value := range download.Header { + header.Set(key, value) + } + // Download URLs are often signed, granting access to the object until + // they expire, so only the URL without its query is logged and + // recorded. The object must have the size the client asked for. + ctx, recorded := withSignedDownload(ctx, download.Href, object.Size) + if !h.proxy.StreamArtifacts { + stored, err := h.proxy.DB.FindStoredContent(object.OID) + if err != nil { + h.proxy.Logger.Warn("git: looking up a stored copy of an LFS object failed", "oid", object.OID, "error", err) + } + if stored != nil && stored.Size == object.Size { + cached, err := h.proxy.cacheStoredCopy(ctx, gitEcosystem, repo.lfsName(), object.OID, object.OID, + recorded, object.OID, stored) + if err == nil { + return cached, nil + } + h.proxy.Logger.Warn("git: copying a stored LFS object failed, downloading it", "oid", object.OID, "error", err) + } + } + return h.proxy.getOrFetchArtifactFromURL(ctx, gitEcosystem, repo.lfsName(), object.OID, object.OID, + recorded, header, object.OID, "") +} + +// retryLFSDownload reports whether a download from a kept URL failed in a way +// a fresh URL might not. +func retryLFSDownload(ctx context.Context, err error) bool { + return ctx.Err() == nil && + !errors.Is(err, ErrArtifactDigestMismatch) && + !errors.Is(err, errDownloadSize) && + !errors.Is(err, ErrVersionDenied) && + !errors.Is(err, ErrArtifactBlocked) +} + +// lfsDownload returns the download action the upstream gave for object, or +// the status and message to answer with when it gave none. +func lfsDownload(resp *lfsBatchResponse, object lfsObject) (lfsAction, int, string) { + for _, got := range resp.Objects { + if got.OID != object.OID { + continue + } + if got.Error != nil { + if got.Error.Code == http.StatusNotFound || got.Error.Code == http.StatusGone { + return lfsAction{}, http.StatusNotFound, "object not found" + } + return lfsAction{}, http.StatusBadGateway, "upstream refused the object" + } + download, ok := got.Actions["download"] + if !ok || !validLFSHref(download.Href) { + return lfsAction{}, http.StatusBadGateway, "upstream gave no download for the object" + } + return download, 0, "" + } + return lfsAction{}, http.StatusNotFound, "object not found" +} + +// validLFSHref reports whether href is an absolute HTTP(S) URL. +func validLFSHref(href string) bool { + u, err := url.Parse(href) + return err == nil && (u.Scheme == "https" || u.Scheme == "http") && u.Host != "" +} + +// lfsBatch asks the upstream's LFS batch API for download actions for +// objects. upstream.auth applies to it as to the repository. +func (h *GitHandler) lfsBatch(ctx context.Context, repo gitRepository, ref json.RawMessage, objects []lfsObject) (*lfsBatchResponse, error) { + body, err := json.Marshal(lfsBatchRequest{ + Operation: "download", + Transfers: []string{gitLFSTransfer}, + Ref: ref, + Objects: objects, + HashAlgo: gitLFSHashAlgo, + }) + if err != nil { + return nil, err + } + batchURL := strings.TrimSuffix(repo.url, ".git") + ".git" + gitLFSPath + "objects/batch" + req, err := http.NewRequestWithContext(ctx, http.MethodPost, batchURL, bytes.NewReader(body)) + if err != nil { + return nil, err + } + req.Header.Set("Accept", gitLFSMediaType) + req.Header.Set("Content-Type", gitLFSMediaType) + resp, err := h.proxy.HTTPClient.Do(req) + if err != nil { + return nil, err + } + defer func() { _ = resp.Body.Close() }() + if resp.StatusCode != http.StatusOK { + return nil, &lfsUpstreamError{status: resp.StatusCode} + } + var batch lfsBatchResponse + if err := json.NewDecoder(io.LimitReader(resp.Body, gitLFSMaxBatchResponse)).Decode(&batch); err != nil { + return nil, fmt.Errorf("reading LFS batch response: %w", err) + } + if batch.Transfer != "" && batch.Transfer != gitLFSTransfer { + return nil, fmt.Errorf("LFS batch response uses the %q transfer", batch.Transfer) + } + return &batch, nil +} + +// serveLFSUpstreamError answers a request whose upstream batch request +// failed. +func (h *GitHandler) serveLFSUpstreamError(w http.ResponseWriter, r *http.Request, repo gitRepository, err error) { + if r.Context().Err() != nil { + return + } + h.proxy.Logger.Warn("git: LFS batch request failed", "repository", repo.key, "error", err) + var upstream *lfsUpstreamError + switch { + case errors.As(err, &upstream) && upstream.status == http.StatusNotFound: + lfsError(w, http.StatusNotFound, "repository not found") + case errors.As(err, &upstream) && upstream.status < http.StatusInternalServerError: + lfsError(w, http.StatusBadGateway, "upstream refused the request") + default: + lfsError(w, http.StatusBadGateway, "upstream unavailable") + } +} + +// lfsName names repo's LFS objects in the artifact cache. Objects are cached +// per repository: having one cached for another repository says nothing +// about whether this one has it. +func (repo gitRepository) lfsName() string { + return repo.name + "/" + repo.repo +} + +// lfsObjectURL is where this proxy serves an object of repo. +func (h *GitHandler) lfsObjectURL(repo gitRepository, oid string, size int64) string { + return fmt.Sprintf("%s/git/%s/%s.git%sobjects/%s/%d", + h.baseURL, url.PathEscape(repo.name), repo.escapedRepo, gitLFSPath, oid, size) +} + +// validLFSOID reports whether oid is a sha256 object ID in lowercase hex. +func validLFSOID(oid string) bool { + if len(oid) != gitLFSOIDLength { + return false + } + for _, c := range oid { + if (c < '0' || c > '9') && (c < 'a' || c > 'f') { + return false + } + } + return true +} + +// lfsError answers with a Git LFS error document, which git-lfs shows. +func lfsError(w http.ResponseWriter, status int, msg string) { + w.Header().Set("Content-Type", gitLFSMediaType) + w.WriteHeader(status) + _ = json.NewEncoder(w).Encode(map[string]string{"message": msg}) +} + +const ( + // gitLFSKeptDownloads bounds how many download URLs from batch responses + // are kept for the GETs that follow them. + gitLFSKeptDownloads = 4096 + // gitLFSKeptTTL is how long a download URL is kept when the upstream + // gives no expiry, and gitLFSKeptMaxTTL how long at most. + gitLFSKeptTTL = 5 * time.Minute + gitLFSKeptMaxTTL = time.Hour + // gitLFSExpiryMargin is how long before its expiry a URL is no longer + // used, so a download started with it can finish. + gitLFSExpiryMargin = 30 * time.Second +) + +// lfsDownloads keeps the download actions of recent batch responses, so the +// GET for an object a client just got in a batch needs no second batch +// request. They stay in memory, never reach a client and expire with the +// URLs. +type lfsDownloads struct { + mu sync.Mutex + entries map[lfsDownloadKey]lfsKeptDownload +} + +type lfsDownloadKey struct { + repo, oid string +} + +type lfsKeptDownload struct { + size int64 + action lfsAction + expires time.Time +} + +// put keeps action for object of repo until shortly before it expires. +func (d *lfsDownloads) put(repo string, object lfsObject, action lfsAction, now time.Time) { + if !validLFSHref(action.Href) { + return + } + expires := now.Add(gitLFSKeptTTL) + if action.ExpiresIn > 0 || action.ExpiresAt != "" { + expires = now.Add(gitLFSKeptMaxTTL) + } + if action.ExpiresIn > 0 { + expires = minTime(expires, now.Add(time.Duration(action.ExpiresIn)*time.Second)) + } + if at, err := time.Parse(time.RFC3339, action.ExpiresAt); err == nil { + expires = minTime(expires, at) + } + expires = expires.Add(-gitLFSExpiryMargin) + if !expires.After(now) { + return + } + d.mu.Lock() + defer d.mu.Unlock() + if d.entries == nil { + d.entries = make(map[lfsDownloadKey]lfsKeptDownload) + } + if len(d.entries) >= gitLFSKeptDownloads { + for key, kept := range d.entries { + if !kept.expires.After(now) { + delete(d.entries, key) + } + } + } + if len(d.entries) >= gitLFSKeptDownloads { + // Still full: make room by dropping any one. + for key := range d.entries { + delete(d.entries, key) + break + } + } + d.entries[lfsDownloadKey{repo, object.OID}] = lfsKeptDownload{size: object.Size, action: action, expires: expires} +} + +// get returns the action kept for object of repo, if it hasn't expired. +func (d *lfsDownloads) get(repo string, object lfsObject, now time.Time) (lfsAction, bool) { + d.mu.Lock() + defer d.mu.Unlock() + kept, ok := d.entries[lfsDownloadKey{repo, object.OID}] + if !ok || kept.size != object.Size || !kept.expires.After(now) { + return lfsAction{}, false + } + return kept.action, true +} + +// drop forgets the action kept for oid of repo, once a GET has used it. +func (d *lfsDownloads) drop(repo, oid string) { + d.mu.Lock() + defer d.mu.Unlock() + delete(d.entries, lfsDownloadKey{repo, oid}) +} + +func minTime(a, b time.Time) time.Time { + if b.Before(a) { + return b + } + return a +} diff --git a/internal/handler/git_lfs_test.go b/internal/handler/git_lfs_test.go new file mode 100644 index 00000000..37bc6aa8 --- /dev/null +++ b/internal/handler/git_lfs_test.go @@ -0,0 +1,483 @@ +package handler + +import ( + "bytes" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/git-pkgs/proxy/internal/config" + "github.com/git-pkgs/registries/fetch" +) + +const lfsTestContent = "large file content" + +// lfsSignature is the signature in the upstream's download URLs, which must +// not reach logs or the database. +const lfsSignature = "s3cr3t-signature" + +// lfsUpstream is a Git host with an LFS batch API whose download hrefs point +// at a separate object store, as GitHub's do. The store only serves objects +// to requests carrying the header and signed URL the batch response gave. +type lfsUpstream struct { + git, store *httptest.Server + oid string + content atomic.Value // string the store serves for oid + batches atomic.Int32 + downloads atomic.Int32 + // batchStatus, when set, is the status the batch API fails with. + batchStatus atomic.Int32 + // signature is the signature the batch API hands out and the store + // accepts; changing it expires the URLs handed out before. + signature atomic.Value + // lastBatch is the last batch request the upstream got. + lastBatch atomic.Value + // logs is what the proxy logged. + logs bytes.Buffer +} + +func newLFSUpstream(t *testing.T) *lfsUpstream { + t.Helper() + sum := sha256.Sum256([]byte(lfsTestContent)) + u := &lfsUpstream{oid: hex.EncodeToString(sum[:])} + u.content.Store(lfsTestContent) + u.signature.Store(lfsSignature) + u.store = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/objects/"+u.oid || r.Header.Get("X-Signed") != "token" || + r.URL.Query().Get("Signature") != u.signature.Load().(string) { + http.Error(w, "forbidden", http.StatusForbidden) + return + } + u.downloads.Add(1) + _, _ = io.WriteString(w, u.content.Load().(string)) + })) + u.git = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + // acme/widget has the object, as does its fork; stranger/widget + // exists but doesn't have it. + repo, _, _ := strings.Cut(r.URL.Path, ".git/info/lfs/objects/batch") + hasObject, exists := map[string]bool{"/acme/widget": true, "/fork/widget": true, "/stranger/widget": false}[repo] + if r.Method != http.MethodPost || !exists { + http.NotFound(w, r) + return + } + u.batches.Add(1) + if status := u.batchStatus.Load(); status != 0 { + http.Error(w, "failing", int(status)) + return + } + var req lfsBatchRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + u.lastBatch.Store(req) + resp := lfsBatchResponse{Transfer: "basic"} + for _, object := range req.Objects { + if object.OID != u.oid || !hasObject { + resp.Objects = append(resp.Objects, lfsObjectResponse{OID: object.OID, Size: object.Size, + Error: &lfsObjectError{Code: http.StatusNotFound, Message: "Object does not exist"}}) + continue + } + resp.Objects = append(resp.Objects, lfsObjectResponse{OID: object.OID, Size: object.Size, + Actions: map[string]lfsAction{"download": { + Href: u.store.URL + "/objects/" + u.oid + "?Expires=1&Signature=" + u.signature.Load().(string), + Header: map[string]string{"X-Signed": "token"}, + }}}) + } + // An object nobody asked for is not passed on. + resp.Objects = append(resp.Objects, lfsObjectResponse{OID: strings.Repeat("e", 64), Size: 1, + Actions: map[string]lfsAction{"download": {Href: u.store.URL + "/other"}}}) + w.Header().Set("Content-Type", gitLFSMediaType) + _ = json.NewEncoder(w).Encode(resp) + })) + t.Cleanup(u.git.Close) + t.Cleanup(u.store.Close) + return u +} + +func newLFSHandler(t *testing.T, u *lfsUpstream, repos ...string) *GitHandler { + t.Helper() + proxy, _, store, _ := setupTestProxy(t) + store.seekable = true + proxy.Logger = slog.New(slog.NewTextHandler(&u.logs, nil)) + proxy.HTTPClient = u.git.Client() + fetcher := fetch.NewFetcher(fetch.WithHTTPClient(u.store.Client()), fetch.WithMaxRetries(0)) + proxy.Fetcher = fetcher + t.Cleanup(func() { _ = fetcher.Close() }) + return NewGitHandler(proxy, nil, config.UpstreamConfig{Git: map[string]config.GitUpstream{ + "up": {URL: u.git.URL, Repos: repos}, + }}).WithBaseURL("http://proxy.test/") +} + +func lfsRequest(h *GitHandler, method, target, body string) *httptest.ResponseRecorder { + r := httptest.NewRequest(method, target, strings.NewReader(body)) + r.Header.Set("Content-Type", gitLFSMediaType) + w := httptest.NewRecorder() + h.Routes().ServeHTTP(w, r) + return w +} + +func lfsBatchBody(operation string, objects ...lfsObject) string { + body, _ := json.Marshal(lfsBatchRequest{Operation: operation, Transfers: []string{"basic"}, + Ref: json.RawMessage(`{"name":"refs/heads/main"}`), Objects: objects}) + return string(body) +} + +func TestGitLFSBatchPointsObjectsAtTheProxy(t *testing.T) { + u := newLFSUpstream(t) + h := newLFSHandler(t, u) + missing := strings.Repeat("0", 64) + size := int64(len(lfsTestContent)) + + w := lfsRequest(h, http.MethodPost, "/up/acme/widget.git/info/lfs/objects/batch", + lfsBatchBody("download", lfsObject{OID: u.oid, Size: size}, lfsObject{OID: missing, Size: 3})) + if w.Code != http.StatusOK || w.Header().Get("Content-Type") != gitLFSMediaType { + t.Fatalf("batch: status %d, type %q, body %s", w.Code, w.Header().Get("Content-Type"), w.Body) + } + var resp lfsBatchResponse + if err := json.Unmarshal(w.Body.Bytes(), &resp); err != nil { + t.Fatal(err) + } + if len(resp.Objects) != 2 { + t.Fatalf("objects = %+v, want the two requested", resp.Objects) + } + got, notFound := resp.Objects[0], resp.Objects[1] + want := fmt.Sprintf("http://proxy.test/git/up/acme/widget.git/info/lfs/objects/%s/%d", u.oid, size) + if download := got.Actions["download"]; download.Href != want || len(download.Header) != 0 || !got.Authenticated { + t.Errorf("download = %+v, authenticated %v; want %s without the upstream's headers", download, got.Authenticated, want) + } + if notFound.Error == nil || notFound.Error.Code != http.StatusNotFound || notFound.Actions != nil { + t.Errorf("missing object = %+v, want the upstream's 404", notFound) + } + if req, _ := u.lastBatch.Load().(lfsBatchRequest); string(req.Ref) != `{"name":"refs/heads/main"}` { + t.Errorf("upstream got ref %s, want the client's", req.Ref) + } +} + +func TestGitLFSObjectIsCached(t *testing.T) { + u := newLFSUpstream(t) + h := newLFSHandler(t, u) + path := fmt.Sprintf("/up/acme/widget.git/info/lfs/objects/%s/%d", u.oid, len(lfsTestContent)) + + w := lfsRequest(h, http.MethodGet, path, "") + if w.Code != http.StatusOK || w.Body.String() != lfsTestContent { + t.Fatalf("first GET: status %d, body %q", w.Code, w.Body) + } + if u.batches.Load() != 1 || u.downloads.Load() != 1 { + t.Errorf("batches %d, downloads %d; want one of each", u.batches.Load(), u.downloads.Load()) + } + + u.git.Close() + u.store.Close() + w = lfsRequest(h, http.MethodGet, path, "") + if w.Code != http.StatusOK || w.Body.String() != lfsTestContent { + t.Errorf("GET with the upstream down: status %d, body %q; want it from the cache", w.Code, w.Body) + } +} + +func TestGitLFSObjectRefusesWrongContent(t *testing.T) { + u := newLFSUpstream(t) + u.content.Store("tampered content!!") + h := newLFSHandler(t, u) + path := fmt.Sprintf("/up/acme/widget.git/info/lfs/objects/%s/%d", u.oid, len(lfsTestContent)) + + for range 2 { + if w := lfsRequest(h, http.MethodGet, path, ""); w.Code == http.StatusOK && w.Body.String() == "tampered content!!" { + t.Fatal("served content that does not match its object ID") + } + } + if u.downloads.Load() != 2 { + t.Errorf("downloads = %d, want 2: the mismatching object must not be cached", u.downloads.Load()) + } +} + +func TestGitLFSRefusals(t *testing.T) { + u := newLFSUpstream(t) + h := newLFSHandler(t, u, "acme/*") + object := lfsObject{OID: u.oid, Size: int64(len(lfsTestContent))} + tests := []struct { + name, method, target, body string + status int + }{ + {"upload", http.MethodPost, "/up/acme/widget.git/info/lfs/objects/batch", lfsBatchBody("upload", object), http.StatusForbidden}, + {"sha512 objects", http.MethodPost, "/up/acme/widget.git/info/lfs/objects/batch", + `{"operation":"download","objects":[],"hash_algo":"sha512"}`, http.StatusConflict}, + {"other transfers", http.MethodPost, "/up/acme/widget.git/info/lfs/objects/batch", + `{"operation":"download","transfers":["ssh"],"objects":[]}`, http.StatusUnprocessableEntity}, + {"invalid oid", http.MethodPost, "/up/acme/widget.git/info/lfs/objects/batch", + lfsBatchBody("download", lfsObject{OID: "../x", Size: 1}), http.StatusUnprocessableEntity}, + {"batch GET", http.MethodGet, "/up/acme/widget.git/info/lfs/objects/batch", "", http.StatusMethodNotAllowed}, + {"repository outside repos", http.MethodPost, "/up/other/widget.git/info/lfs/objects/batch", lfsBatchBody("download", object), http.StatusNotFound}, + {"unknown repository", http.MethodPost, "/up/acme/missing.git/info/lfs/objects/batch", lfsBatchBody("download", object), http.StatusNotFound}, + {"object without size", http.MethodGet, "/up/acme/widget.git/info/lfs/objects/" + u.oid, "", http.StatusNotFound}, + {"uppercase oid", http.MethodGet, "/up/acme/widget.git/info/lfs/objects/" + strings.ToUpper(u.oid) + "/1", "", http.StatusNotFound}, + {"object PUT", http.MethodPut, "/up/acme/widget.git/info/lfs/objects/" + u.oid + "/1", "", http.StatusMethodNotAllowed}, + {"locks", http.MethodPost, "/up/acme/widget.git/info/lfs/locks/verify", "{}", http.StatusNotFound}, + {"missing object", http.MethodGet, "/up/acme/widget.git/info/lfs/objects/" + strings.Repeat("0", 64) + "/3", "", http.StatusNotFound}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + w := lfsRequest(h, tt.method, tt.target, tt.body) + if w.Code != tt.status || w.Header().Get("Content-Type") != gitLFSMediaType { + t.Errorf("status %d, type %q, body %s; want %d", w.Code, w.Header().Get("Content-Type"), w.Body, tt.status) + } + }) + } + if u.downloads.Load() != 0 { + t.Errorf("downloads = %d, want none", u.downloads.Load()) + } +} + +func TestGitLFSKeepsSignedURLsOutOfLogsAndDatabase(t *testing.T) { + u := newLFSUpstream(t) + h := newLFSHandler(t, u) + path := fmt.Sprintf("/up/acme/widget.git/info/lfs/objects/%s/%d", u.oid, len(lfsTestContent)) + + if w := lfsRequest(h, http.MethodGet, path, ""); w.Code != http.StatusOK { + t.Fatalf("GET: status %d, body %s", w.Code, w.Body) + } + _, versionPURL, err := packagePURLStrings(gitEcosystem, "up/acme/widget", u.oid) + if err != nil { + t.Fatal(err) + } + record, err := h.proxy.DB.GetArtifact(versionPURL, u.oid) + if err != nil || record == nil { + t.Fatalf("artifact record: %v, %v", record, err) + } + if want := u.store.URL + "/objects/" + u.oid; record.UpstreamURL != want { + t.Errorf("recorded upstream URL %q, want %q", record.UpstreamURL, want) + } + + // A failed download is logged without the signature too, although the + // connection error names the URL. + other := newLFSUpstream(t) + other.store.Close() + h = newLFSHandler(t, other) + if w := lfsRequest(h, http.MethodGet, path, ""); w.Code == http.StatusOK { + t.Fatalf("GET from an unreachable store: status %d", w.Code) + } + if !strings.Contains(other.logs.String(), other.store.URL+"/objects/"+other.oid) { + t.Fatalf("logs don't name the failed download:\n%s", other.logs.String()) + } + for _, logs := range []string{u.logs.String(), other.logs.String()} { + if strings.Contains(logs, lfsSignature) { + t.Errorf("logs contain the download URL's signature:\n%s", logs) + } + } +} + +func TestGitLFSBatchFromCacheWhenUpstreamFails(t *testing.T) { + u := newLFSUpstream(t) + h := newLFSHandler(t, u) + size := int64(len(lfsTestContent)) + missing := strings.Repeat("0", 64) + batch := lfsBatchBody("download", lfsObject{OID: u.oid, Size: size}, lfsObject{OID: missing, Size: 3}) + const batchPath = "/up/acme/widget.git/info/lfs/objects/batch" + + // Nothing cached yet: a failing upstream fails the batch. + u.batchStatus.Store(http.StatusServiceUnavailable) + if w := lfsRequest(h, http.MethodPost, batchPath, batch); w.Code != http.StatusBadGateway { + t.Fatalf("batch with nothing cached: status %d, body %s; want 502", w.Code, w.Body) + } + u.batchStatus.Store(0) + if w := lfsRequest(h, http.MethodGet, fmt.Sprintf("/up/acme/widget.git/info/lfs/objects/%s/%d", u.oid, size), ""); w.Code != http.StatusOK { + t.Fatalf("GET: status %d, body %s", w.Code, w.Body) + } + + // A refusal is the upstream's answer, not an outage. + u.batchStatus.Store(http.StatusForbidden) + if w := lfsRequest(h, http.MethodPost, batchPath, batch); w.Code != http.StatusBadGateway { + t.Errorf("batch the upstream refused: status %d, body %s; want 502", w.Code, w.Body) + } + + u.git.Close() + w := lfsRequest(h, http.MethodPost, batchPath, batch) + if w.Code != http.StatusOK { + t.Fatalf("batch with the upstream down: status %d, body %s", w.Code, w.Body) + } + var resp lfsBatchResponse + if err := json.Unmarshal(w.Body.Bytes(), &resp); err != nil { + t.Fatal(err) + } + if len(resp.Objects) != 2 { + t.Fatalf("objects = %+v, want the two requested", resp.Objects) + } + want := fmt.Sprintf("http://proxy.test/git/up/acme/widget.git/info/lfs/objects/%s/%d", u.oid, size) + if got := resp.Objects[0]; got.Actions["download"].Href != want || !got.Authenticated || got.Error != nil { + t.Errorf("cached object = %+v, want a download from %s", got, want) + } + if got := resp.Objects[1]; got.Actions != nil || got.Error == nil { + t.Errorf("uncached object = %+v, want an error", got) + } +} + +func TestGitLFSHeadDoesNotDownload(t *testing.T) { + u := newLFSUpstream(t) + h := newLFSHandler(t, u) + path := fmt.Sprintf("/up/acme/widget.git/info/lfs/objects/%s/%d", u.oid, len(lfsTestContent)) + + w := lfsRequest(h, http.MethodHead, path, "") + if w.Code != http.StatusOK || w.Header().Get("Content-Length") != fmt.Sprint(len(lfsTestContent)) { + t.Errorf("HEAD: status %d, Content-Length %q", w.Code, w.Header().Get("Content-Length")) + } + if u.downloads.Load() != 0 { + t.Errorf("downloads = %d, want none for a HEAD", u.downloads.Load()) + } + missing := fmt.Sprintf("/up/acme/widget.git/info/lfs/objects/%s/3", strings.Repeat("0", 64)) + if w := lfsRequest(h, http.MethodHead, missing, ""); w.Code != http.StatusNotFound { + t.Errorf("HEAD of a missing object: status %d, want 404", w.Code) + } +} + +func TestGitLFSObjectAfterBatchNeedsNoSecondBatch(t *testing.T) { + u := newLFSUpstream(t) + h := newLFSHandler(t, u) + size := int64(len(lfsTestContent)) + path := fmt.Sprintf("/up/acme/widget.git/info/lfs/objects/%s/%d", u.oid, size) + batch := lfsBatchBody("download", lfsObject{OID: u.oid, Size: size}) + + if w := lfsRequest(h, http.MethodPost, "/up/acme/widget.git/info/lfs/objects/batch", batch); w.Code != http.StatusOK { + t.Fatalf("batch: status %d, body %s", w.Code, w.Body) + } + if w := lfsRequest(h, http.MethodGet, path, ""); w.Code != http.StatusOK || w.Body.String() != lfsTestContent { + t.Fatalf("GET: status %d, body %q", w.Code, w.Body) + } + if u.batches.Load() != 1 || u.downloads.Load() != 1 { + t.Errorf("batches %d, downloads %d; want the client's batch and one download", u.batches.Load(), u.downloads.Load()) + } + + // A URL that stopped working before its time is replaced by a fresh one. + other := newLFSUpstream(t) + h = newLFSHandler(t, other) + if w := lfsRequest(h, http.MethodPost, "/up/acme/widget.git/info/lfs/objects/batch", batch); w.Code != http.StatusOK { + t.Fatalf("batch: status %d, body %s", w.Code, w.Body) + } + other.signature.Store("rotated") + if w := lfsRequest(h, http.MethodGet, path, ""); w.Code != http.StatusOK || w.Body.String() != lfsTestContent { + t.Fatalf("GET with a revoked URL: status %d, body %q", w.Code, w.Body) + } + if other.batches.Load() != 2 { + t.Errorf("batches = %d, want a second one for a fresh URL", other.batches.Load()) + } +} + +func TestGitLFSKeptDownloadsExpire(t *testing.T) { + now := time.Date(2026, 10, 25, 12, 0, 0, 0, time.UTC) + object := lfsObject{OID: strings.Repeat("a", 64), Size: 3} + href := "https://objects.example/a?sig=1" + tests := []struct { + name string + action lfsAction + at time.Duration + kept bool + }{ + {"no expiry", lfsAction{Href: href}, 4 * time.Minute, true}, + {"no expiry, later", lfsAction{Href: href}, 5 * time.Minute, false}, + {"expires_in", lfsAction{Href: href, ExpiresIn: 600}, 9 * time.Minute, true}, + {"expires_in, within the margin", lfsAction{Href: href, ExpiresIn: 600}, 10*time.Minute - 10*time.Second, false}, + {"expires_in too short to use", lfsAction{Href: href, ExpiresIn: 10}, 0, false}, + {"expires_at", lfsAction{Href: href, ExpiresAt: now.Add(20 * time.Minute).Format(time.RFC3339)}, 19 * time.Minute, true}, + {"expires_at, past", lfsAction{Href: href, ExpiresAt: now.Add(20 * time.Minute).Format(time.RFC3339)}, 20 * time.Minute, false}, + {"long expiry capped", lfsAction{Href: href, ExpiresIn: 86400}, 2 * time.Hour, false}, + {"relative href", lfsAction{Href: "/a"}, 0, false}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + var d lfsDownloads + d.put("repo", object, tt.action, now) + if _, kept := d.get("repo", object, now.Add(tt.at)); kept != tt.kept { + t.Errorf("kept = %v, want %v", kept, tt.kept) + } + }) + } + + var d lfsDownloads + d.put("repo", object, lfsAction{Href: href}, now) + if _, kept := d.get("repo", lfsObject{OID: object.OID, Size: 4}, now); kept { + t.Error("kept for another size") + } + if _, kept := d.get("other", object, now); kept { + t.Error("kept for another repository") + } + for i := range gitLFSKeptDownloads + 10 { + d.put("repo", lfsObject{OID: fmt.Sprintf("%064x", i), Size: 1}, lfsAction{Href: href}, now) + } + if len(d.entries) > gitLFSKeptDownloads { + t.Errorf("%d entries kept, want at most %d", len(d.entries), gitLFSKeptDownloads) + } +} + +func TestGitLFSObjectOfAnotherSizeIsRefused(t *testing.T) { + u := newLFSUpstream(t) + h := newLFSHandler(t, u) + path := fmt.Sprintf("/up/acme/widget.git/info/lfs/objects/%s/%d", u.oid, len(lfsTestContent)-1) + + if w := lfsRequest(h, http.MethodGet, path, ""); w.Code == http.StatusOK { + t.Fatalf("GET at the wrong size: status %d, body %q", w.Code, w.Body) + } + if u.downloads.Load() != 1 || u.batches.Load() != 1 { + t.Errorf("downloads %d, batches %d; want one of each, without a retry", u.downloads.Load(), u.batches.Load()) + } +} + +func TestExactSizeBody(t *testing.T) { + for _, tt := range []struct { + body string + size int64 + ok bool + }{ + {"abcdef", 6, true}, + {"abcdef", 5, false}, + {"abcdef", 7, false}, + {"", 0, true}, + {"a", 0, false}, + } { + got, err := io.ReadAll(&exactSizeBody{ReadCloser: io.NopCloser(strings.NewReader(tt.body)), left: tt.size}) + if (err == nil) != tt.ok || (err != nil && !errors.Is(err, errDownloadSize)) { + t.Errorf("%q at size %d: err %v, want ok %v", tt.body, tt.size, err, tt.ok) + } + if int64(len(got)) > tt.size+1 { + t.Errorf("%q at size %d: read %d bytes", tt.body, tt.size, len(got)) + } + } +} + +func TestGitLFSObjectCopiedFromAnotherRepository(t *testing.T) { + u := newLFSUpstream(t) + h := newLFSHandler(t, u) + object := func(repo string) string { + return fmt.Sprintf("/up/%s.git/info/lfs/objects/%s/%d", repo, u.oid, len(lfsTestContent)) + } + + if w := lfsRequest(h, http.MethodGet, object("acme/widget"), ""); w.Code != http.StatusOK { + t.Fatalf("GET: status %d, body %s", w.Code, w.Body) + } + w := lfsRequest(h, http.MethodGet, object("fork/widget"), "") + if w.Code != http.StatusOK || w.Body.String() != lfsTestContent { + t.Fatalf("GET from the fork: status %d, body %q", w.Code, w.Body) + } + if u.downloads.Load() != 1 || u.batches.Load() != 2 { + t.Errorf("downloads %d, batches %d; want the fork's object copied after its own batch", u.downloads.Load(), u.batches.Load()) + } + cached, err := h.proxy.GetCachedArtifact(t.Context(), gitEcosystem, "up/fork/widget", u.oid, u.oid) + if err != nil || cached == nil { + t.Fatalf("fork's object not cached: %v, %v", cached, err) + } + _ = cached.Reader.Close() + + // A repository the upstream says lacks the object doesn't get it, though + // a copy is stored. + if w := lfsRequest(h, http.MethodGet, object("stranger/widget"), ""); w.Code != http.StatusNotFound { + t.Errorf("GET from a repository without the object: status %d, body %s; want 404", w.Code, w.Body) + } +} diff --git a/internal/handler/handler.go b/internal/handler/handler.go index 3e5984ef..363314e8 100644 --- a/internal/handler/handler.go +++ b/internal/handler/handler.go @@ -2252,6 +2252,33 @@ func (p *Proxy) coalescedFetchFromURL(ctx context.Context, ecosystem, name, vers }) } +// cacheStoredCopy caches a copy of the object stored at source under a new +// name, instead of downloading the same bytes again. It shares the coalescing +// key a download from upstreamURL would use, so the two never run at once. +func (p *Proxy) cacheStoredCopy(ctx context.Context, ecosystem, name, version, filename, upstreamURL, upstreamHash string, source *database.StoredContent) (*CacheResult, error) { + pkgPURL, versionPURL, err := packagePURLStrings(ecosystem, name, version) + if err != nil { + return nil, err + } + if p.versionDenied(ecosystem, name, version) || p.Denylist.Denied(versionPURL) { + return nil, fmt.Errorf("%w: %s", ErrVersionDenied, versionPURL) + } + key := artifactCoalesceKey(versionPURL, filename, upstreamURL, upstreamHash, "") + recheck := func() (artifacts.Artifact, string, bool) { + return p.cachedArtifactRecord(pkgPURL, versionPURL, filename, upstreamHash) + } + return p.coalesceFetch(ctx, key, recheck, func(fetchCtx context.Context) (artifacts.Artifact, string, error) { + body, err := p.Storage.Open(fetchCtx, source.StoragePath) + if err != nil { + return artifacts.Artifact{}, "", fmt.Errorf("opening stored copy: %w", err) + } + p.Logger.Info("caching a stored copy", "ecosystem", ecosystem, "name", name, "version", version, + "source", source.StoragePath) + return p.storeArtifact(fetchCtx, ecosystem, name, version, filename, pkgPURL, versionPURL, upstreamURL, upstreamHash, "", + &fetch.Artifact{Body: body, Size: source.Size, ContentType: source.ContentType.String}) + }) +} + // getCachedArtifactWithUpstreamHash returns a cached artifact whose recorded // content hash matches the checksum the upstream currently declares for it. // This detects an upstream re-publishing under the same version, which the @@ -2332,9 +2359,96 @@ type observedFetcher interface { FetchObservedWithHeaders(ctx context.Context, url string, headers http.Header) (*fetch.ObservedArtifact, error) } -// fetchUpstream fetches downloadURL, reporting its Last-Modified when ctx -// asks for it. +type signedDownloadKey struct{} + +// signedDownload is a download URL carrying a signature, and the URL without +// its query that stands for it everywhere but the request itself. size, when +// not negative, is the exact size the download must have. +type signedDownload struct { + signed, recorded string + size int64 +} + +// errDownloadSize is a download larger or smaller than the size it was +// expected to have. +var errDownloadSize = errors.New("download does not have the expected size") + +// withSignedDownload returns a context under which a fetch of the returned +// URL requests signedURL instead. The returned URL, signedURL without its +// query and fragment, is what is logged, recorded with the artifact and used +// to coalesce fetches, so a signature that grants access to the object +// stays out of logs and the database. A download of other than size bytes +// is cut off with errDownloadSize, unless size is negative. +func withSignedDownload(ctx context.Context, signedURL string, size int64) (context.Context, string) { + u, err := url.Parse(signedURL) + if err != nil { + return ctx, signedURL + } + u.User = nil + u.RawQuery, u.ForceQuery = "", false + u.Fragment, u.RawFragment = "", "" + recorded := u.String() + return context.WithValue(ctx, signedDownloadKey{}, signedDownload{signed: signedURL, recorded: recorded, size: size}), recorded +} + +// redactedFetchError is a fetch error whose text names the recorded URL +// rather than the signed one. +type redactedFetchError struct { + err error + download signedDownload +} + +func (e *redactedFetchError) Error() string { + return strings.ReplaceAll(e.err.Error(), e.download.signed, e.download.recorded) +} + +func (e *redactedFetchError) Unwrap() error { return e.err } + +// fetchUpstream fetches downloadURL, or the signed URL ctx gives for it, +// reporting its Last-Modified when ctx asks for it. func (p *Proxy) fetchUpstream(ctx context.Context, downloadURL string, headers http.Header) (*fetch.Artifact, error) { + download, ok := ctx.Value(signedDownloadKey{}).(signedDownload) + if !ok || download.recorded != downloadURL { + return p.fetchUpstreamURL(ctx, downloadURL, headers) + } + artifact, err := p.fetchUpstreamURL(ctx, download.signed, headers) + if err != nil { + return nil, &redactedFetchError{err: err, download: download} + } + if download.size < 0 { + return artifact, nil + } + if artifact.Size >= 0 && artifact.Size != download.size { + _ = artifact.Body.Close() + return nil, fmt.Errorf("%w: upstream announced %d bytes, want %d", errDownloadSize, artifact.Size, download.size) + } + artifact.Body = &exactSizeBody{ReadCloser: artifact.Body, left: download.size} + return artifact, nil +} + +// exactSizeBody fails a body that turns out longer or shorter than left +// bytes, without reading more than one byte past it. +type exactSizeBody struct { + io.ReadCloser + left int64 +} + +func (b *exactSizeBody) Read(p []byte) (int, error) { + if int64(len(p)) > b.left+1 { + p = p[:b.left+1] + } + n, err := b.ReadCloser.Read(p) + b.left -= int64(n) + switch { + case b.left < 0: + return n, fmt.Errorf("%w: upstream sent more bytes", errDownloadSize) + case errors.Is(err, io.EOF) && b.left > 0: + return n, fmt.Errorf("%w: upstream sent %d bytes too few", errDownloadSize, b.left) + } + return n, err +} + +func (p *Proxy) fetchUpstreamURL(ctx context.Context, downloadURL string, headers http.Header) (*fetch.Artifact, error) { report, _ := ctx.Value(upstreamLastModifiedKey{}).(func(time.Time)) observer, ok := p.Fetcher.(observedFetcher) if report == nil || !ok { diff --git a/internal/server/server.go b/internal/server/server.go index bb99be69..12619a6a 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -530,7 +530,7 @@ func (s *Server) mountProtocolHandlers(r chi.Router, proxy *handler.Proxy) { r.Mount("/rpm", http.StripPrefix("/rpm", rpmHandler.Routes())) r.Mount("/generic", http.StripPrefix("/generic", genericHandler.Routes())) if s.git != nil { - gitHandler := handler.NewGitHandler(proxy, s.git, s.cfg.Upstream) + gitHandler := handler.NewGitHandler(proxy, s.git, s.cfg.Upstream).WithBaseURL(s.cfg.BaseURL) r.Mount("/git", http.StripPrefix("/git", gitHandler.Routes())) } }