From 3f0bed48f587bb29f2325cc2d658d0b26db3aeeb Mon Sep 17 00:00:00 2001 From: Joe Chen Date: Wed, 11 Nov 2026 14:41:04 +0000 Subject: [PATCH 1/3] lfs: extract the Batch API into an importable module Move the transfer logic of lfsd into a separate Go module, unknwon.dev/git-lfs-server/lfs, so that other Git servers can serve LFS with the same storage backends. The module has an lfs.Service that answers batch, upload, download, and verify requests on top of an lfs.Store for metadata and a storage.Backend for content, and leaves routing, authentication, and response encoding to the caller. lfsd now builds one lfs.Service per forge host and keeps its HTTP responses, except that a proxied upload of an already stored object must send matching content before the object is linked to another repository. Previously, any body was accepted in that case, so a writer of one repository could gain read access to any object of another repository by its OID and size. --- .github/workflows/go.yml | 17 +- .github/workflows/integration.yml | 2 + .moon/workspace.yml | 1 + Dockerfile | 1 + README.md | 6 + cmd/lfsd/config.go | 2 +- cmd/lfsd/lfs.go | 505 +++--------------- cmd/lfsd/main.go | 20 +- go.mod | 13 +- go.sum | 10 - internal/database/objects.go | 84 +-- internal/janitor/janitor.go | 2 +- lfs/go.mod | 40 ++ lfs/go.sum | 114 ++++ .../internal}/iox/exact_size_reader.go | 0 .../internal}/iox/exact_size_reader_test.go | 0 .../internal}/iox/sha256_reader.go | 0 .../internal}/iox/sha256_reader_test.go | 0 lfs/lfs.go | 123 +++++ lfs/moon.yml | 45 ++ lfs/service.go | 379 +++++++++++++ lfs/service_test.go | 373 +++++++++++++ {internal => lfs}/storage/filesystem.go | 5 + {internal => lfs}/storage/filesystem_test.go | 0 {internal => lfs}/storage/s3_presign.go | 0 {internal => lfs}/storage/s3_presign_test.go | 0 {internal => lfs}/storage/storage.go | 9 +- 27 files changed, 1241 insertions(+), 510 deletions(-) create mode 100644 lfs/go.mod create mode 100644 lfs/go.sum rename {internal => lfs/internal}/iox/exact_size_reader.go (100%) rename {internal => lfs/internal}/iox/exact_size_reader_test.go (100%) rename {internal => lfs/internal}/iox/sha256_reader.go (100%) rename {internal => lfs/internal}/iox/sha256_reader_test.go (100%) create mode 100644 lfs/lfs.go create mode 100644 lfs/moon.yml create mode 100644 lfs/service.go create mode 100644 lfs/service_test.go rename {internal => lfs}/storage/filesystem.go (96%) rename {internal => lfs}/storage/filesystem_test.go (100%) rename {internal => lfs}/storage/s3_presign.go (100%) rename {internal => lfs}/storage/s3_presign_test.go (100%) rename {internal => lfs}/storage/storage.go (97%) diff --git a/.github/workflows/go.yml b/.github/workflows/go.yml index 75fe764..91ed31d 100644 --- a/.github/workflows/go.yml +++ b/.github/workflows/go.yml @@ -7,6 +7,8 @@ on: - '**.go' - 'go.mod' - 'go.sum' + - 'lfs/go.mod' + - 'lfs/go.sum' - '.golangci.yml' - '.github/workflows/go.yml' pull_request: @@ -14,6 +16,8 @@ on: - '**.go' - 'go.mod' - 'go.sum' + - 'lfs/go.mod' + - 'lfs/go.sum' - '.golangci.yml' - '.github/workflows/go.yml' env: @@ -38,6 +42,7 @@ jobs: shell: bash run: | go mod tidy + (cd lfs && go mod tidy) STATUS=$(git status --porcelain) if [ ! -z "$STATUS" ]; then echo "Unstaged files:" @@ -50,6 +55,12 @@ jobs: with: version: latest args: --timeout=30m + - name: Run golangci-lint (lfs module) + uses: golangci/golangci-lint-action@9fae48acfc02a90574d7c304a1758ef9895495fa # v7.0.1 + with: + version: latest + working-directory: lfs + args: --timeout=30m test: name: Test @@ -68,6 +79,7 @@ jobs: - name: Run tests with coverage run: | go test -shuffle=on -v -race -coverprofile=coverage -covermode=atomic -json ./... > test-report.json + (cd lfs && go test -shuffle=on -v -race -coverprofile=coverage -covermode=atomic -json ./...) >> test-report.json go install github.com/mfridman/tparse@latest tparse -all -file=test-report.json @@ -88,4 +100,7 @@ jobs: with: go-version: ${{ matrix.go-version }} - name: Run tests with coverage - run: go test -shuffle=on -v -coverprofile=coverage -covermode=atomic ./... + shell: bash + run: | + go test -shuffle=on -v -coverprofile=coverage -covermode=atomic ./... + cd lfs && go test -shuffle=on -v -coverprofile=coverage -covermode=atomic ./... diff --git a/.github/workflows/integration.yml b/.github/workflows/integration.yml index 3f8a221..4ff901a 100644 --- a/.github/workflows/integration.yml +++ b/.github/workflows/integration.yml @@ -9,6 +9,7 @@ on: - 'integration-tests/**' - 'go.mod' - 'go.sum' + - 'lfs/**' - '.github/workflows/integration.yml' pull_request: paths: @@ -17,6 +18,7 @@ on: - 'integration-tests/**' - 'go.mod' - 'go.sum' + - 'lfs/**' - '.github/workflows/integration.yml' permissions: diff --git a/.moon/workspace.yml b/.moon/workspace.yml index e16843f..d78ebb2 100644 --- a/.moon/workspace.yml +++ b/.moon/workspace.yml @@ -2,3 +2,4 @@ $schema: 'https://moonrepo.dev/schemas/workspace.json' projects: - '.' + - 'lfs' diff --git a/Dockerfile b/Dockerfile index 1fb9a38..606f072 100644 --- a/Dockerfile +++ b/Dockerfile @@ -2,6 +2,7 @@ FROM golang:1.27.4-alpine3.24@sha256:0a5f7fe796599ff487605987432e7619ae45beeb2fba4c2ce0b907879932aebf AS builder WORKDIR /src COPY go.mod go.sum ./ +COPY lfs/go.mod lfs/go.sum ./lfs/ RUN go mod download COPY . . ARG BUILD_DATE diff --git a/README.md b/README.md index e833b39..ac1d589 100644 --- a/README.md +++ b/README.md @@ -94,6 +94,12 @@ REPO_ALLOWLIST = myorg/**, otherorg/specific-repo The allowlist is enforced before the forge token check, so a request for a disallowed repo is rejected without a round-trip to the forge. +## Using as a library + +The `unknwon.dev/git-lfs-server/lfs` module implements the Batch API and the basic transfer adapter for other Git servers to embed. It is a separate Go module so that importers do not pull in the dependencies of lfsd (PostgreSQL, Flamego, forge clients). + +An `lfs.Service` combines an `lfs.Store`, which records objects and the repositories they are linked to, with a `storage.Backend`, which holds the content. The caller keeps its own routing, authentication, authorization, and response encoding, identifies repositories with any string it likes, and calls `Batch`, `Upload`, `Download`, and `Verify` from its handlers. A `storage.Proxier` backend (e.g., `storage.FilesystemBackend`) streams content through the caller, and a `storage.Presigner` backend (e.g., `storage.S3PresignBackend`) redirects clients to the storage. `PreviousStorages` keeps objects readable after the backend for new uploads changes. + ## Local development This project uses [moon](https://moonrepo.dev/) as its task runner. Install it with Homebrew: diff --git a/cmd/lfsd/config.go b/cmd/lfsd/config.go index 352ab2c..9a22d35 100644 --- a/cmd/lfsd/config.go +++ b/cmd/lfsd/config.go @@ -10,7 +10,7 @@ import ( "unknwon.dev/git-lfs-server/internal/database" "unknwon.dev/git-lfs-server/internal/embed" "unknwon.dev/git-lfs-server/internal/forge" - "unknwon.dev/git-lfs-server/internal/storage" + "unknwon.dev/git-lfs-server/lfs/storage" ) // Host is the host of a forge, e.g., "github.com". diff --git a/cmd/lfsd/lfs.go b/cmd/lfsd/lfs.go index f35aa73..c93842a 100644 --- a/cmd/lfsd/lfs.go +++ b/cmd/lfsd/lfs.go @@ -1,103 +1,63 @@ package main import ( - "encoding/base64" - "encoding/hex" + "context" "encoding/json" "fmt" "io" - "maps" "net/http" - "regexp" - "slices" "strconv" "github.com/cockroachdb/errors" "github.com/flamego/flamego" - "unknwon.dev/git-lfs-server/internal/database" "unknwon.dev/git-lfs-server/internal/forge" - "unknwon.dev/git-lfs-server/internal/iox" "unknwon.dev/git-lfs-server/internal/logx" - "unknwon.dev/git-lfs-server/internal/ptrx" - "unknwon.dev/git-lfs-server/internal/storage" + "unknwon.dev/git-lfs-server/lfs" ) -// oidPattern matches the SHA-256 hex digest format used as the LFS object ID: -// exactly 64 lowercase hex characters. -var oidPattern = regexp.MustCompile(`^[0-9a-f]{64}$`) - const ( - operationUpload = "upload" - operationDownload = "download" - transferBasic = "basic" - mediaTypeLFS = "application/vnd.git-lfs+json" + mediaTypeLFS = "application/vnd.git-lfs+json" // maxJSONRequestBytes caps JSON request bodies (batch, verify) to prevent // authenticated callers from exhausting memory with oversized payloads. maxJSONRequestBytes = 10 << 20 ) -type batchRequest struct { - Operation string `json:"operation"` - Transfers []string `json:"transfers,omitempty"` - Objects []batchObject `json:"objects"` -} - -type batchObject struct { - OID string `json:"oid"` - Size int64 `json:"size"` -} - -type batchResponse struct { - Transfer string `json:"transfer"` - Objects []batchResponseObject `json:"objects"` -} - -type batchResponseObject struct { - OID string `json:"oid"` - Size int64 `json:"size"` - Authenticated bool `json:"authenticated,omitempty"` - Actions map[string]batchAction `json:"actions,omitempty"` - Error *batchObjectError `json:"error,omitempty"` -} - -type batchAction struct { - Href string `json:"href"` - Header map[string]string `json:"header,omitempty"` -} - -type batchObjectError struct { - Code int `json:"code"` - Message string `json:"message"` -} - -type verifyRequest struct { - OID string `json:"oid"` - Size int64 `json:"size"` +// mapService maps the *lfs.Service of the forge host to the request context. +func mapService(services map[string]*lfs.Service) flamego.Handler { + return func(c flamego.Context, logger *logx.Logger) { + service, ok := services[hostFromContext(c)] + if !ok { + logger.ErrorContext(c.Request().Context(), "No storage backend configured for host") + writeLFSError(c.ResponseWriter(), http.StatusInternalServerError, http.StatusText(http.StatusInternalServerError)) + return + } + c.Map(service) + } } -func serveBatch(db *database.DB, externalURL string, maxObjectSize int64) flamego.Handler { - return func(c flamego.Context, logger *logx.Logger, perm forge.Permission) { +func serveBatch(externalURL string) flamego.Handler { + return func(c flamego.Context, logger *logx.Logger, perm forge.Permission, service *lfs.Service) { logger = logger.Scoped("batch") ctx := c.Request().Context() body := http.MaxBytesReader(c.ResponseWriter(), c.Request().Body().ReadCloser(), maxJSONRequestBytes) defer func() { _ = body.Close() }() - var req batchRequest + var req lfs.BatchRequest if err := json.NewDecoder(body).Decode(&req); err != nil { writeLFSError(c.ResponseWriter(), http.StatusBadRequest, fmt.Sprintf("invalid JSON body: %v", err)) return } switch req.Operation { - case operationUpload: + case lfs.OperationUpload: if perm != forge.PermissionWrite { writeLFSError(c.ResponseWriter(), http.StatusForbidden, "write permission required for upload") return } - case operationDownload: + case lfs.OperationDownload: if perm != forge.PermissionRead && perm != forge.PermissionWrite { writeLFSError(c.ResponseWriter(), http.StatusForbidden, "read permission required for download") return @@ -107,431 +67,132 @@ func serveBatch(db *database.DB, externalURL string, maxObjectSize int64) flameg return } - if len(req.Transfers) > 0 && !slices.Contains(req.Transfers, transferBasic) { - writeLFSError(c.ResponseWriter(), http.StatusUnprocessableEntity, `only "basic" transfer is supported`) - return - } - - for _, o := range req.Objects { - if !oidPattern.MatchString(o.OID) { - writeLFSError(c.ResponseWriter(), http.StatusBadRequest, fmt.Sprintf("invalid oid %q", o.OID)) - return - } - } - repoName := repoNameFromContext(c) - - var authHeader map[string]string - if authz := c.Request().Header.Get("Authorization"); authz != "" { - authHeader = map[string]string{"Authorization": authz} + endpoint := lfs.Endpoint{ + Href: fmt.Sprintf("%s/%s/info/lfs/objects", externalURL, repoName), + // Every request passes the authorize middleware with Basic Auth, so + // forwarding the header authenticates the actions. + Authenticated: true, } - - oids := make([]string, len(req.Objects)) - for i, o := range req.Objects { - oids[i] = o.OID + if authz := c.Request().Header.Get("Authorization"); authz != "" { + endpoint.Header = map[string]string{"Authorization": authz} } - var ( - existing map[string]database.Object - err error - ) - switch req.Operation { - case operationDownload: - existing, err = db.GetRepoObjectsByOIDs(ctx, repoName, oids) - case operationUpload: - existing, err = db.GetObjectsByOIDs(ctx, oids) - } + resp, err := service.Batch(ctx, logger.Logger, repoName, endpoint, req) if err != nil { - logger.ErrorContext(ctx, "Failed to look up objects", "error", err) - writeLFSError(c.ResponseWriter(), http.StatusInternalServerError, "failed to process batch") + writeServiceError(ctx, logger, c.ResponseWriter(), err, "Failed to process batch", "failed to process batch") return } - out := batchResponse{ - Transfer: transferBasic, - Objects: make([]batchResponseObject, 0, len(req.Objects)), - } - for _, in := range req.Objects { - obj, ok := existing[in.OID] - respObj := batchResponseObject{OID: in.OID, Size: in.Size, Authenticated: true} - - switch req.Operation { - case operationDownload: - switch { - case !ok: - respObj.Error = &batchObjectError{Code: http.StatusNotFound, Message: "object does not exist"} - case ptrx.Deref(obj.Size, -1) != in.Size: - respObj.Error = &batchObjectError{Code: http.StatusUnprocessableEntity, Message: "OID exists with different size"} - default: - respObj.Actions = map[string]batchAction{ - "download": {Href: objectHref(externalURL, repoName, in.OID), Header: authHeader}, - } - } - case operationUpload: - switch { - case maxObjectSize > 0 && in.Size > maxObjectSize: - respObj.Error = &batchObjectError{ - Code: http.StatusRequestEntityTooLarge, - Message: fmt.Sprintf("object size %d exceeds limit %d", in.Size, maxObjectSize), - } - case ok && ptrx.Deref(obj.Size, -1) == in.Size: - // Cross-repo dedup: object is verified globally, ensure - // the calling repo is linked so future downloads succeed. - // COALESCE in LinkObject's upsert preserves the existing - // object_uri, so the placeholder is never written. - if _, err := db.LinkObject(ctx, repoName, in.OID, in.Size, "ignored"); err != nil { - logger.ErrorContext(ctx, "Failed to link repo to existing object", "oid", in.OID, "error", err) - respObj.Error = &batchObjectError{Code: http.StatusInternalServerError, Message: "failed to link object to repo"} - } - case ok && ptrx.Deref(obj.Size, -1) != in.Size: - respObj.Error = &batchObjectError{Code: http.StatusConflict, Message: "OID exists with different size"} - default: - rawOID, err := hex.DecodeString(in.OID) - if err != nil { - respObj.Error = &batchObjectError{Code: http.StatusBadRequest, Message: fmt.Sprintf("invalid oid %q", in.OID)} - break - } - uploadHeader := make(map[string]string, len(authHeader)+1) - maps.Copy(uploadHeader, authHeader) - uploadHeader["x-amz-checksum-sha256"] = base64.StdEncoding.EncodeToString(rawOID) - respObj.Actions = map[string]batchAction{ - "upload": {Href: objectHref(externalURL, repoName, in.OID), Header: uploadHeader}, - "verify": {Href: verifyHref(externalURL, repoName), Header: authHeader}, - } - } - } - out.Objects = append(out.Objects, respObj) - } - c.ResponseWriter().Header().Set("Content-Type", mediaTypeLFS) - if err := json.NewEncoder(c.ResponseWriter()).Encode(out); err != nil { + if err := json.NewEncoder(c.ResponseWriter()).Encode(resp); err != nil { logger.ErrorContext(ctx, "Failed to encode batch response", "error", err) } } } -func serveUpload(db *database.DB, storages map[string]storage.Backend, maxObjectSize int64) flamego.Handler { - return func(c flamego.Context, logger *logx.Logger, perm forge.Permission) { - logger = logger.Scoped("upload") - ctx := c.Request().Context() - - if perm != forge.PermissionWrite { - writeLFSError(c.ResponseWriter(), http.StatusForbidden, "write permission required for upload") - return - } - - oid := c.Param("oid") - if !oidPattern.MatchString(oid) { - writeLFSError(c.ResponseWriter(), http.StatusBadRequest, "invalid oid") - return - } - - size := c.Request().ContentLength - if size < 0 { - writeLFSError(c.ResponseWriter(), http.StatusLengthRequired, "Content-Length is required") - return - } - if maxObjectSize > 0 && size > maxObjectSize { - writeLFSError(c.ResponseWriter(), http.StatusRequestEntityTooLarge, fmt.Sprintf("object size %d exceeds limit %d", size, maxObjectSize)) - return - } - - backend, ok := storages[hostFromContext(c)] - if !ok { - logger.ErrorContext(ctx, "No storage backend configured for host") - writeLFSError(c.ResponseWriter(), http.StatusInternalServerError, http.StatusText(http.StatusInternalServerError)) - return - } +func serveUpload(c flamego.Context, logger *logx.Logger, perm forge.Permission, service *lfs.Service) { + logger = logger.Scoped("upload") + ctx := c.Request().Context() - switch b := backend.(type) { - case storage.Proxier: - serveUploadProxy(c, logger, db, b, oid, size) - case storage.Presigner: - serveUploadPresign(c, logger, db, b, oid, size) - default: - logger.ErrorContext(ctx, "Storage backend does not implement a known transfer mode", "type", fmt.Sprintf("%T", backend)) - writeLFSError(c.ResponseWriter(), http.StatusInternalServerError, http.StatusText(http.StatusInternalServerError)) - } + if perm != forge.PermissionWrite { + writeLFSError(c.ResponseWriter(), http.StatusForbidden, "write permission required for upload") + return } -} - -func serveUploadProxy(c flamego.Context, logger *logx.Logger, db *database.DB, backend storage.Proxier, oid string, size int64) { - ctx := c.Request().Context() body := c.Request().Body().ReadCloser() defer func() { _ = body.Close() }() - reader := iox.NewSHA256Reader(iox.NewExactSizeReader(body, size), oid) - - uri, err := backend.Put(ctx, oid, reader) + oid := c.Param("oid") + result, err := service.Upload(ctx, logger.Logger, repoNameFromContext(c), oid, c.Request().ContentLength, body) if err != nil { - var sizeErr *iox.SizeMismatchError - var hashErr *iox.SHA256MismatchError - switch { - case errors.As(err, &sizeErr): - writeLFSError(c.ResponseWriter(), http.StatusUnprocessableEntity, sizeErr.Error()) - case errors.As(err, &hashErr): - writeLFSError(c.ResponseWriter(), http.StatusUnprocessableEntity, hashErr.Error()) - default: - logger.ErrorContext(ctx, "Failed to store object", "oid", oid, "error", err) - writeLFSError(c.ResponseWriter(), http.StatusInternalServerError, http.StatusText(http.StatusInternalServerError)) - } + writeServiceError(ctx, logger.With("oid", oid), c.ResponseWriter(), err, "Failed to upload object", http.StatusText(http.StatusInternalServerError)) return } - repoName := repoNameFromContext(c) - storedURI, err := db.LinkObject(ctx, repoName, oid, size, uri) - if err != nil { - logger.ErrorContext(ctx, "Failed to link object", "oid", oid, "error", err) - writeLFSError(c.ResponseWriter(), http.StatusInternalServerError, http.StatusText(http.StatusInternalServerError)) - return - } - if storedURI != uri { - if delErr := backend.Delete(ctx, uri); delErr != nil { - logger.ErrorContext(ctx, "Failed to delete orphan object", "uri", uri, "error", delErr) + w := c.ResponseWriter() + switch { + case result.Redirect != nil: + for k, v := range result.Redirect.Header { + w.Header().Set(k, v) } + w.Header().Set("Location", result.Redirect.URL) + w.WriteHeader(http.StatusTemporaryRedirect) + case result.Existed: + w.WriteHeader(http.StatusOK) + default: + w.WriteHeader(http.StatusNoContent) } - - c.ResponseWriter().WriteHeader(http.StatusNoContent) } -func serveUploadPresign(c flamego.Context, logger *logx.Logger, db *database.DB, backend storage.Presigner, oid string, size int64) { +func serveDownload(c flamego.Context, logger *logx.Logger, perm forge.Permission, service *lfs.Service) { + logger = logger.Scoped("download") ctx := c.Request().Context() - existing, err := db.GetObjectByOID(ctx, oid) - if err != nil && !errors.Is(err, database.ErrObjectNotFound) { - logger.ErrorContext(ctx, "Failed to look up object", "oid", oid, "error", err) - writeLFSError(c.ResponseWriter(), http.StatusInternalServerError, http.StatusText(http.StatusInternalServerError)) - return - } - if existing != nil && existing.VerifiedAt != nil { - c.ResponseWriter().WriteHeader(http.StatusOK) + if perm != forge.PermissionRead && perm != forge.PermissionWrite { + writeLFSError(c.ResponseWriter(), http.StatusForbidden, "read permission required for download") return } - if err = db.InsertPendingObject(ctx, oid); err != nil { - logger.ErrorContext(ctx, "Failed to insert pending object", "oid", oid, "error", err) - writeLFSError(c.ResponseWriter(), http.StatusInternalServerError, http.StatusText(http.StatusInternalServerError)) - return - } - - url, headers, err := backend.PresignPut(ctx, oid, size) + oid := c.Param("oid") + result, err := service.Download(ctx, repoNameFromContext(c), oid) if err != nil { - logger.ErrorContext(ctx, "Failed to presign upload", "oid", oid, "error", err) - writeLFSError(c.ResponseWriter(), http.StatusInternalServerError, http.StatusText(http.StatusInternalServerError)) + writeServiceError(ctx, logger.With("oid", oid), c.ResponseWriter(), err, "Failed to download object", http.StatusText(http.StatusInternalServerError)) return } w := c.ResponseWriter() - for k, v := range headers { - w.Header().Set(k, v) - } - w.Header().Set("Location", url) - w.WriteHeader(http.StatusTemporaryRedirect) -} - -func serveDownload(db *database.DB, storages map[string]storage.Backend) flamego.Handler { - return func(c flamego.Context, logger *logx.Logger, perm forge.Permission) { - logger = logger.Scoped("download") - ctx := c.Request().Context() - - if perm != forge.PermissionRead && perm != forge.PermissionWrite { - writeLFSError(c.ResponseWriter(), http.StatusForbidden, "read permission required for download") - return - } - - oid := c.Param("oid") - if !oidPattern.MatchString(oid) { - writeLFSError(c.ResponseWriter(), http.StatusBadRequest, "invalid oid") - return - } - - backend, ok := storages[hostFromContext(c)] - if !ok { - logger.ErrorContext(ctx, "No storage backend configured for host") - writeLFSError(c.ResponseWriter(), http.StatusInternalServerError, http.StatusText(http.StatusInternalServerError)) - return - } - - repoName := repoNameFromContext(c) - object, err := db.GetRepoObjectByOID(ctx, repoName, oid) - if err != nil { - if errors.Is(err, database.ErrObjectNotFound) { - writeLFSError(c.ResponseWriter(), http.StatusNotFound, "object does not exist") - return - } - logger.ErrorContext(ctx, "Failed to look up object", "oid", oid, "error", err) - writeLFSError(c.ResponseWriter(), http.StatusInternalServerError, http.StatusText(http.StatusInternalServerError)) - return - } - // GetRepoObjectByOID filters verified rows, so Size and ObjectURI are - // guaranteed non-nil here by the schema CHECK. - objectURI := ptrx.Deref(object.ObjectURI, "") - objectSize := ptrx.Deref(object.Size, -1) - if objectURI == "" || objectSize < 0 { - logger.ErrorContext(ctx, "Verified object has nil size or uri (schema violation)", "oid", oid) - writeLFSError(c.ResponseWriter(), http.StatusInternalServerError, http.StatusText(http.StatusInternalServerError)) - return - } - - switch b := backend.(type) { - case storage.Proxier: - serveDownloadProxy(c, logger, b, oid, objectURI, objectSize) - case storage.Presigner: - serveDownloadPresign(c, logger, b, oid, objectURI) - default: - logger.ErrorContext(ctx, "Storage backend does not implement a known transfer mode", "type", fmt.Sprintf("%T", backend)) - writeLFSError(c.ResponseWriter(), http.StatusInternalServerError, http.StatusText(http.StatusInternalServerError)) - } - } -} - -func serveDownloadProxy(c flamego.Context, logger *logx.Logger, backend storage.Proxier, oid, uri string, size int64) { - ctx := c.Request().Context() - - rc, err := backend.Open(ctx, uri) - if err != nil { - if errors.Is(err, storage.ErrNotFound) { - logger.ErrorContext(ctx, "DB row exists but storage object is missing", "oid", oid, "uri", uri) - } else { - logger.ErrorContext(ctx, "Failed to open object", "oid", oid, "error", err) - } - writeLFSError(c.ResponseWriter(), http.StatusInternalServerError, http.StatusText(http.StatusInternalServerError)) + if result.Redirect != nil { + w.Header().Set("Location", result.Redirect.URL) + w.WriteHeader(http.StatusTemporaryRedirect) return } - defer func() { _ = rc.Close() }() - w := c.ResponseWriter() + defer func() { _ = result.Content.Close() }() w.Header().Set("Content-Type", "application/octet-stream") - w.Header().Set("Content-Length", strconv.FormatInt(size, 10)) + w.Header().Set("Content-Length", strconv.FormatInt(result.Size, 10)) w.WriteHeader(http.StatusOK) - if _, err = io.Copy(w, rc); err != nil { + if _, err = io.Copy(w, result.Content); err != nil { logger.ErrorContext(ctx, "Failed to stream object", "oid", oid, "error", err) } } -func serveDownloadPresign(c flamego.Context, logger *logx.Logger, backend storage.Presigner, oid, uri string) { +func serveVerify(c flamego.Context, logger *logx.Logger, perm forge.Permission, service *lfs.Service) { + logger = logger.Scoped("verify") ctx := c.Request().Context() - url, err := backend.PresignGet(ctx, uri) - if err != nil { - logger.ErrorContext(ctx, "Failed to presign download", "oid", oid, "error", err) - writeLFSError(c.ResponseWriter(), http.StatusInternalServerError, http.StatusText(http.StatusInternalServerError)) + if perm != forge.PermissionWrite { + writeLFSError(c.ResponseWriter(), http.StatusForbidden, "write permission required for verify") return } - w := c.ResponseWriter() - w.Header().Set("Location", url) - w.WriteHeader(http.StatusTemporaryRedirect) -} - -func serveVerify(db *database.DB, storages map[string]storage.Backend) flamego.Handler { - return func(c flamego.Context, logger *logx.Logger, perm forge.Permission) { - logger = logger.Scoped("verify") - ctx := c.Request().Context() - - if perm != forge.PermissionWrite { - writeLFSError(c.ResponseWriter(), http.StatusForbidden, "write permission required for verify") - return - } - - body := http.MaxBytesReader(c.ResponseWriter(), c.Request().Body().ReadCloser(), maxJSONRequestBytes) - defer func() { _ = body.Close() }() - - var req verifyRequest - if err := json.NewDecoder(body).Decode(&req); err != nil { - writeLFSError(c.ResponseWriter(), http.StatusBadRequest, fmt.Sprintf("invalid JSON body: %v", err)) - return - } - if !oidPattern.MatchString(req.OID) { - writeLFSError(c.ResponseWriter(), http.StatusBadRequest, "invalid oid") - return - } - - backend, ok := storages[hostFromContext(c)] - if !ok { - logger.ErrorContext(ctx, "No storage backend configured for host") - writeLFSError(c.ResponseWriter(), http.StatusInternalServerError, http.StatusText(http.StatusInternalServerError)) - return - } - - switch b := backend.(type) { - case storage.Proxier: - serveVerifyProxy(c, logger, db, req) - case storage.Presigner: - serveVerifyPresign(c, logger, db, b, req) - default: - logger.ErrorContext(ctx, "Storage backend does not implement a known transfer mode", "type", fmt.Sprintf("%T", backend)) - writeLFSError(c.ResponseWriter(), http.StatusInternalServerError, http.StatusText(http.StatusInternalServerError)) - } - } -} - -func serveVerifyProxy(c flamego.Context, logger *logx.Logger, db *database.DB, req verifyRequest) { - ctx := c.Request().Context() + body := http.MaxBytesReader(c.ResponseWriter(), c.Request().Body().ReadCloser(), maxJSONRequestBytes) + defer func() { _ = body.Close() }() - repoName := repoNameFromContext(c) - object, err := db.GetRepoObjectByOID(ctx, repoName, req.OID) - if err != nil { - if errors.Is(err, database.ErrObjectNotFound) { - writeLFSError(c.ResponseWriter(), http.StatusNotFound, "object does not exist") - return - } - logger.ErrorContext(ctx, "Failed to look up object", "oid", req.OID, "error", err) - writeLFSError(c.ResponseWriter(), http.StatusInternalServerError, http.StatusText(http.StatusInternalServerError)) + var req lfs.VerifyRequest + if err := json.NewDecoder(body).Decode(&req); err != nil { + writeLFSError(c.ResponseWriter(), http.StatusBadRequest, fmt.Sprintf("invalid JSON body: %v", err)) return } - if ptrx.Deref(object.Size, -1) != req.Size { - writeLFSError(c.ResponseWriter(), http.StatusUnprocessableEntity, "size mismatch") + + if err := service.Verify(ctx, logger.Logger, repoNameFromContext(c), req); err != nil { + writeServiceError(ctx, logger.With("oid", req.OID), c.ResponseWriter(), err, "Failed to verify object", http.StatusText(http.StatusInternalServerError)) return } - c.ResponseWriter().WriteHeader(http.StatusNoContent) } -func serveVerifyPresign(c flamego.Context, logger *logx.Logger, db *database.DB, backend storage.Presigner, req verifyRequest) { - ctx := c.Request().Context() - - uri := backend.URI(req.OID) - headSize, err := backend.Head(ctx, uri) - if err != nil { - if errors.Is(err, storage.ErrNotFound) { - writeLFSError(c.ResponseWriter(), http.StatusUnprocessableEntity, "object does not exist in storage") - return - } - logger.ErrorContext(ctx, "Failed to head object", "oid", req.OID, "error", err) - writeLFSError(c.ResponseWriter(), http.StatusInternalServerError, http.StatusText(http.StatusInternalServerError)) - return - } - // R2's signed Content-Length should make this impossible, but treat any - // mismatch as a hard error rather than trusting either side. - if headSize != req.Size { - writeLFSError(c.ResponseWriter(), http.StatusUnprocessableEntity, "size mismatch") +// writeServiceError responds with the client error carried by err as an +// *lfs.Error. Any other error is internal, so it is logged with logMessage and +// the response only carries internalMessage. +func writeServiceError(ctx context.Context, logger *logx.Logger, w http.ResponseWriter, err error, logMessage, internalMessage string) { + var lfsErr *lfs.Error + if errors.As(err, &lfsErr) { + writeLFSError(w, lfsErr.Code, lfsErr.Message) return } - - repoName := repoNameFromContext(c) - storedURI, err := db.VerifyObject(ctx, repoName, req.OID, req.Size, uri) - if err != nil { - logger.ErrorContext(ctx, "Failed to mark object verified", "oid", req.OID, "error", err) - writeLFSError(c.ResponseWriter(), http.StatusInternalServerError, http.StatusText(http.StatusInternalServerError)) - return - } - if storedURI != uri { - if delErr := backend.Delete(ctx, uri); delErr != nil { - logger.ErrorContext(ctx, "Failed to delete orphan object", "uri", uri, "error", delErr) - } - } - - c.ResponseWriter().WriteHeader(http.StatusNoContent) -} - -func objectHref(externalURL, repoName, oid string) string { - return fmt.Sprintf("%s/%s/info/lfs/objects/%s", externalURL, repoName, oid) -} - -func verifyHref(externalURL, repoName string) string { - return fmt.Sprintf("%s/%s/info/lfs/objects/verify", externalURL, repoName) + logger.ErrorContext(ctx, logMessage, "error", err) + writeLFSError(w, http.StatusInternalServerError, internalMessage) } // writeLFSError writes an error response in the format required by the LFS diff --git a/cmd/lfsd/main.go b/cmd/lfsd/main.go index 30e6048..b071bd9 100644 --- a/cmd/lfsd/main.go +++ b/cmd/lfsd/main.go @@ -18,6 +18,7 @@ import ( "unknwon.dev/git-lfs-server/internal/janitor" "unknwon.dev/git-lfs-server/internal/logx" "unknwon.dev/git-lfs-server/internal/strx" + "unknwon.dev/git-lfs-server/lfs" ) // Set via ldflags. @@ -60,12 +61,21 @@ func main() { f.Get("/healthz", serveHealthz(db)) + services := make(map[string]*lfs.Service, len(config.Storages)) + for host, backend := range config.Storages { + services[host] = &lfs.Service{ + Store: db, + Storage: backend, + MaxObjectSize: config.Server.MaxObjectSize, + CrossRepoDedup: true, + } + } f.Group("/{host}/{**}/info/lfs/objects", func() { - f.Post("/batch", serveBatch(db, config.Server.ExternalURL, config.Server.MaxObjectSize)) - f.Get("/{oid}", serveDownload(db, config.Storages)) - f.Put("/{oid}", serveUpload(db, config.Storages, config.Server.MaxObjectSize)) - f.Post("/verify", serveVerify(db, config.Storages)) - }, authorize(config.Forges)) + f.Post("/batch", serveBatch(config.Server.ExternalURL)) + f.Get("/{oid}", serveDownload) + f.Put("/{oid}", serveUpload) + f.Post("/verify", serveVerify) + }, authorize(config.Forges), mapService(services)) ctx, stop := signal.NotifyContext(ctx, syscall.SIGINT, syscall.SIGTERM) defer stop() diff --git a/go.mod b/go.mod index 80c2e1f..bbb2b84 100644 --- a/go.mod +++ b/go.mod @@ -5,13 +5,9 @@ go 1.27 require ( charm.land/lipgloss/v2 v2.0.3 charm.land/log/v2 v2.0.0 - github.com/aws/aws-sdk-go-v2 v1.41.7 - github.com/aws/aws-sdk-go-v2/credentials v1.19.16 - github.com/aws/aws-sdk-go-v2/service/s3 v1.100.1 github.com/cockroachdb/errors v1.13.0 github.com/flamego/cache v1.5.1 github.com/flamego/flamego v1.9.7 - github.com/johannesboyne/gofakes3 v0.0.0-20260208201424-4c385a1f6a73 github.com/lib/pq v1.12.3 github.com/sourcegraph/conc v0.3.0 github.com/sourcegraph/run v0.12.0 @@ -20,12 +16,15 @@ require ( gopkg.in/ini.v1 v1.67.2 gorm.io/driver/postgres v1.6.0 gorm.io/gorm v1.31.1 + unknwon.dev/git-lfs-server/lfs v0.0.0-00010101000000-000000000000 ) require ( bitbucket.org/creachadair/shell v0.0.7 // indirect github.com/alecthomas/participle/v2 v2.1.4 // indirect + github.com/aws/aws-sdk-go-v2 v1.41.7 // indirect github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.10 // indirect + github.com/aws/aws-sdk-go-v2/credentials v1.19.16 // indirect github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.23 // indirect github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.23 // indirect github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.24 // indirect @@ -33,6 +32,7 @@ require ( github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.9.15 // indirect github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.23 // indirect github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.23 // indirect + github.com/aws/aws-sdk-go-v2/service/s3 v1.100.1 // indirect github.com/aws/smithy-go v1.25.1 // indirect github.com/aymanbagabas/go-osc52/v2 v2.0.1 // indirect github.com/charmbracelet/colorprofile v0.4.3 // indirect @@ -75,15 +75,14 @@ require ( github.com/pmezard/go-difflib v1.0.0 // indirect github.com/rivo/uniseg v0.4.7 // indirect github.com/rogpeppe/go-internal v1.14.1 // indirect - github.com/ryszard/goskiplist v0.0.0-20150312221310-2dfbae5fcf46 // indirect github.com/xo/terminfo v0.0.0-20220910002029-abceb7e1c41e // indirect go.opentelemetry.io/otel v1.11.0 // indirect go.opentelemetry.io/otel/trace v1.11.0 // indirect - go.shabbyrobe.org/gocovmerge v0.0.0-20230507111327-fa4f82cfbf4d // indirect golang.org/x/exp v0.0.0-20260410095643-746e56fc9e2f // indirect golang.org/x/sync v0.22.0 // indirect golang.org/x/sys v0.47.0 // indirect golang.org/x/text v0.41.0 // indirect - golang.org/x/tools v0.48.0 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect ) + +replace unknwon.dev/git-lfs-server/lfs => ./lfs diff --git a/go.sum b/go.sum index 0cbb3ab..62a7ab2 100644 --- a/go.sum +++ b/go.sum @@ -16,8 +16,6 @@ github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.10 h1:gx1AwW1Iyk9Z9dD github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.10/go.mod h1:qqY157uZoqm5OXq/amuaBJyC9hgBCBQnsaWnPe905GY= github.com/aws/aws-sdk-go-v2/credentials v1.19.16 h1:r3RJBuU7X9ibt8RHbMjWE6y60QbKBiII6wSrXnapxSU= github.com/aws/aws-sdk-go-v2/credentials v1.19.16/go.mod h1:6cx7zqDENJDbBIIWX6P8s0h6hqHC8Avbjh9Dseo27ug= -github.com/aws/aws-sdk-go-v2/feature/s3/manager v1.17.75 h1:S61/E3N01oral6B3y9hZ2E1iFDqCZPPOBoBQretCnBI= -github.com/aws/aws-sdk-go-v2/feature/s3/manager v1.17.75/go.mod h1:bDMQbkI1vJbNjnvJYpPTSNYBkI/VIv18ngWb/K84tkk= github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.23 h1:GpT/TrnBYuE5gan2cZbTtvP+JlHsutdmlV2YfEyNde0= github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.23/go.mod h1:xYWD6BS9ywC5bS3sz9Xh04whO/hzK2plt2Zkyrp4JuA= github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.23 h1:bpd8vxhlQi2r1hiueOw02f/duEPTMK59Q4QMAoTTtTo= @@ -38,8 +36,6 @@ github.com/aws/smithy-go v1.25.1 h1:J8ERsGSU7d+aCmdQur5Txg6bVoYelvQJgtZehD12GkI= github.com/aws/smithy-go v1.25.1/go.mod h1:YE2RhdIuDbA5E5bTdciG9KrW3+TiEONeUWCqxX9i1Fc= github.com/aymanbagabas/go-osc52/v2 v2.0.1 h1:HwpRHbFMcZLEVr42D4p7XBqjyuxQH5SMiErDT4WkJ2k= github.com/aymanbagabas/go-osc52/v2 v2.0.1/go.mod h1:uYgXzlJ7ZpABp8OJ+exZzJJhRNQ2ASbcXHWsFqH8hp8= -github.com/cevatbarisyilmaz/ara v0.0.4 h1:SGH10hXpBJhhTlObuZzTuFn1rrdmjQImITXnZVPSodc= -github.com/cevatbarisyilmaz/ara v0.0.4/go.mod h1:BfFOxnUd6Mj6xmcvRxHN3Sr21Z1T3U2MYkYOmoQe4Ts= github.com/charmbracelet/colorprofile v0.4.3 h1:QPa1IWkYI+AOB+fE+mg/5/4HRMZcaXex9t5KX76i20Q= github.com/charmbracelet/colorprofile v0.4.3/go.mod h1:/zT4BhpD5aGFpqQQqw7a+VtHCzu+zrQtt1zhMt9mR4Q= github.com/charmbracelet/lipgloss v1.1.0 h1:vYXsiLHVkK7fp74RkV7b2kq9+zDLoEU4MZoFqR/noCY= @@ -165,8 +161,6 @@ github.com/sourcegraph/conc v0.3.0 h1:OQTbbt6P72L20UqAkXXuLOj79LfEanQ+YQFNpLA9yS github.com/sourcegraph/conc v0.3.0/go.mod h1:Sdozi7LEKbFPqYX2/J+iBAM6HpqSLTASQIKqDmF7Mt0= github.com/sourcegraph/run v0.12.0 h1:3A8w5e8HIYPfafHekvmdmmh42RHKGVhmiTZAPJclg7I= github.com/sourcegraph/run v0.12.0/go.mod h1:PwaP936BTnAJC1cqR5rSbG5kOs/EWStTK3lqvMX5GUA= -github.com/spf13/afero v1.2.1 h1:qgMbHoJbPbw579P+1zVY+6n4nIFuIchaIjzZ/I/Yq8M= -github.com/spf13/afero v1.2.1/go.mod h1:9ZxEEn6pIJ8Rxe320qSDBk6AsU0r9pR7Q4OcevTdifk= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw= github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo= @@ -184,8 +178,6 @@ github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9de github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= go.bobheadxi.dev/streamline v1.3.2 h1:EwbDggkws9Qo/fl4Zo801Z8mgf4xZBOX7/bbfTthsy4= go.bobheadxi.dev/streamline v1.3.2/go.mod h1:QSS0MlQm+3ABEr0uMYrkAGYiILwJvTIGFygVARxDFdg= -go.etcd.io/bbolt v1.3.5 h1:XAzx9gjCb0Rxj7EoqcClPD1d5ZBxZJk0jbuoPHenBt0= -go.etcd.io/bbolt v1.3.5/go.mod h1:G5EMThwa9y8QZGBClrRx5EY+Yw9kAhnjy3bSjsnlVTQ= go.opentelemetry.io/otel v1.11.0 h1:kfToEGMDq6TrVrJ9Vht84Y8y9enykSZzDDZglV0kIEk= go.opentelemetry.io/otel v1.11.0/go.mod h1:H2KtuEphyMvlhZ+F7tg9GRhAOe60moNx61Ex+WmiKkk= go.opentelemetry.io/otel/sdk v1.11.0 h1:ZnKIL9V9Ztaq+ME43IUi/eo22mNsb6a7tGfzaOWB5fo= @@ -238,8 +230,6 @@ gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntN gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q= gopkg.in/ini.v1 v1.67.2 h1:JtOSMb9OuaCZKr7h5D/h6iii14sK0hLbplTc6frx4Ss= gopkg.in/ini.v1 v1.67.2/go.mod h1:x/cyOwCgZqOkJoDIJ3c1KNHMo10+nLGAhh+kn3Zizss= -gopkg.in/mgo.v2 v2.0.0-20180705113604-9856a29383ce h1:xcEWjVhvbDy+nHP67nPDDpbYrY+ILlfndk4bRioVHaU= -gopkg.in/mgo.v2 v2.0.0-20180705113604-9856a29383ce/go.mod h1:yeKp02qBN3iKW1OzL3MGk2IdtZzaj7SFntXj72NppTA= gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/internal/database/objects.go b/internal/database/objects.go index 84200b3..b376fb1 100644 --- a/internal/database/objects.go +++ b/internal/database/objects.go @@ -10,8 +10,11 @@ import ( "gorm.io/gorm" "unknwon.dev/git-lfs-server/internal/ptrx" + "unknwon.dev/git-lfs-server/lfs" ) +var _ lfs.Store = (*DB)(nil) + type Object struct { ID int64 `gorm:"column:id;primaryKey"` OID string `gorm:"column:oid;uniqueIndex;not null"` @@ -24,48 +27,6 @@ type Object struct { var ErrObjectNotFound = errors.New("database: object not found") -// GetRepoObjectByOID fetches the verified Object linked to repoName by oid. -// Pending (unverified) rows are excluded so unverified objects can never serve -// as download sources. Returns ErrObjectNotFound when no row matches. -func (d *DB) GetRepoObjectByOID(ctx context.Context, repoName, oid string) (*Object, error) { - var object Object - err := d.db.WithContext(ctx).Raw(` -SELECT objects.* -FROM objects -JOIN repo_objects ON repo_objects.object_id = objects.id -WHERE - repo_objects.repo_name = @repoName -AND objects.oid = @oid -AND objects.verified_at IS NOT NULL`, - map[string]any{"repoName": repoName, "oid": oid}, - ).Scan(&object).Error - if err != nil { - return nil, errors.Wrap(err, "select repo object by oid") - } - if object.ID == 0 { - return nil, errors.WithStack(ErrObjectNotFound) - } - return &object, nil -} - -// GetObjectByOID fetches an Object by oid (verified or pending) without any -// repo filtering. Used by the upload handler for the dedup short-circuit and -// to detect pending rows. Returns ErrObjectNotFound when no row matches. -func (d *DB) GetObjectByOID(ctx context.Context, oid string) (*Object, error) { - var object Object - err := d.db.WithContext(ctx).Raw( - `SELECT * FROM objects WHERE oid = @oid LIMIT 1`, - map[string]any{"oid": oid}, - ).Scan(&object).Error - if err != nil { - return nil, errors.Wrap(err, "select object by oid") - } - if object.ID == 0 { - return nil, errors.WithStack(ErrObjectNotFound) - } - return &object, nil -} - // LinkObject is the storage.Proxier verify-equivalent path: bytes are already // stored, write a fully-populated (verified) objects row and link it to // repoName in one transaction. Returns the canonical object_uri (which may @@ -195,11 +156,11 @@ WHERE id = @id`, return storedURI, nil } -// GetObjectsByOIDs fetches verified Object rows for the given OIDs. Results are -// keyed by OID, silently dropping missing OIDs. -func (d *DB) GetObjectsByOIDs(ctx context.Context, oids []string) (map[string]Object, error) { +// GetObjects fetches verified objects for the given OIDs. Results are keyed by +// OID, silently dropping missing OIDs. +func (d *DB) GetObjects(ctx context.Context, oids []string) (map[string]lfs.Object, error) { if len(oids) == 0 { - return map[string]Object{}, nil + return map[string]lfs.Object{}, nil } var rows []Object err := d.db.WithContext(ctx).Raw(` @@ -210,19 +171,14 @@ WHERE oid = ANY(@oids) AND verified_at IS NOT NULL`, if err != nil { return nil, errors.Wrap(err, "select objects by OIDs") } - out := make(map[string]Object, len(rows)) - for _, r := range rows { - out[r.OID] = r - } - return out, nil + return verifiedObjectsByOID(rows), nil } -// GetRepoObjectsByOIDs fetches verified Object rows for the given OIDs that are -// linked to the repository. Results are keyed by OID, silently dropping missing -// OIDs. -func (d *DB) GetRepoObjectsByOIDs(ctx context.Context, repoName string, oids []string) (map[string]Object, error) { +// GetRepoObjects fetches verified objects for the given OIDs that are linked to +// the repository. Results are keyed by OID, silently dropping missing OIDs. +func (d *DB) GetRepoObjects(ctx context.Context, repoName string, oids []string) (map[string]lfs.Object, error) { if len(oids) == 0 { - return map[string]Object{}, nil + return map[string]lfs.Object{}, nil } var rows []Object err := d.db.WithContext(ctx).Raw(` @@ -238,11 +194,21 @@ AND objects.verified_at IS NOT NULL`, if err != nil { return nil, errors.Wrap(err, "select repo objects by OIDs") } - out := make(map[string]Object, len(rows)) + return verifiedObjectsByOID(rows), nil +} + +// verifiedObjectsByOID converts verified rows, whose size and object_uri are +// guaranteed non-null by the check_verified_consistency constraint. +func verifiedObjectsByOID(rows []Object) map[string]lfs.Object { + out := make(map[string]lfs.Object, len(rows)) for _, r := range rows { - out[r.OID] = r + out[r.OID] = lfs.Object{ + OID: r.OID, + Size: ptrx.Deref(r.Size, -1), + URI: ptrx.Deref(r.ObjectURI, ""), + } } - return out, nil + return out } // SweepOrphanObjects deletes up to limit pending objects (verified_at IS NULL) diff --git a/internal/janitor/janitor.go b/internal/janitor/janitor.go index c8db2a0..2332c5f 100644 --- a/internal/janitor/janitor.go +++ b/internal/janitor/janitor.go @@ -10,7 +10,7 @@ import ( "unknwon.dev/git-lfs-server/internal/database" "unknwon.dev/git-lfs-server/internal/logx" "unknwon.dev/git-lfs-server/internal/ptrx" - "unknwon.dev/git-lfs-server/internal/storage" + "unknwon.dev/git-lfs-server/lfs/storage" ) const ( diff --git a/lfs/go.mod b/lfs/go.mod new file mode 100644 index 0000000..ba96a41 --- /dev/null +++ b/lfs/go.mod @@ -0,0 +1,40 @@ +module unknwon.dev/git-lfs-server/lfs + +go 1.26 + +require ( + github.com/aws/aws-sdk-go-v2 v1.41.7 + github.com/aws/aws-sdk-go-v2/credentials v1.19.16 + github.com/aws/aws-sdk-go-v2/service/s3 v1.100.1 + github.com/cockroachdb/errors v1.13.0 + github.com/johannesboyne/gofakes3 v0.0.0-20260208201424-4c385a1f6a73 + github.com/stretchr/testify v1.11.1 +) + +require ( + github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.10 // indirect + github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.23 // indirect + github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.23 // indirect + github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.24 // indirect + github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.9 // indirect + github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.9.15 // indirect + github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.23 // indirect + github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.23 // indirect + github.com/aws/smithy-go v1.25.1 // indirect + github.com/cockroachdb/logtags v0.0.0-20230118201751-21c54148d20b // indirect + github.com/cockroachdb/redact v1.1.5 // indirect + github.com/davecgh/go-spew v1.1.1 // indirect + github.com/getsentry/sentry-go v0.46.0 // indirect + github.com/gogo/protobuf v1.3.2 // indirect + github.com/kr/pretty v0.3.1 // indirect + github.com/kr/text v0.2.0 // indirect + github.com/pkg/errors v0.9.1 // indirect + github.com/pmezard/go-difflib v1.0.0 // indirect + github.com/rogpeppe/go-internal v1.9.0 // indirect + github.com/ryszard/goskiplist v0.0.0-20150312221310-2dfbae5fcf46 // indirect + go.shabbyrobe.org/gocovmerge v0.0.0-20230507111327-fa4f82cfbf4d // indirect + golang.org/x/sys v0.31.0 // indirect + golang.org/x/text v0.23.0 // indirect + golang.org/x/tools v0.21.1-0.20240508182429-e35e4ccd0d2d // indirect + gopkg.in/yaml.v3 v3.0.1 // indirect +) diff --git a/lfs/go.sum b/lfs/go.sum new file mode 100644 index 0000000..a48112f --- /dev/null +++ b/lfs/go.sum @@ -0,0 +1,114 @@ +github.com/aws/aws-sdk-go-v2 v1.41.7 h1:DWpAJt66FmnnaRIOT/8ASTucrvuDPZASqhhLey6tLY8= +github.com/aws/aws-sdk-go-v2 v1.41.7/go.mod h1:4LAfZOPHNVNQEckOACQx60Y8pSRjIkNZQz1w92xpMJc= +github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.10 h1:gx1AwW1Iyk9Z9dD9F4akX5gnN3QZwUB20GGKH/I+Rho= +github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.10/go.mod h1:qqY157uZoqm5OXq/amuaBJyC9hgBCBQnsaWnPe905GY= +github.com/aws/aws-sdk-go-v2/credentials v1.19.16 h1:r3RJBuU7X9ibt8RHbMjWE6y60QbKBiII6wSrXnapxSU= +github.com/aws/aws-sdk-go-v2/credentials v1.19.16/go.mod h1:6cx7zqDENJDbBIIWX6P8s0h6hqHC8Avbjh9Dseo27ug= +github.com/aws/aws-sdk-go-v2/feature/s3/manager v1.17.75 h1:S61/E3N01oral6B3y9hZ2E1iFDqCZPPOBoBQretCnBI= +github.com/aws/aws-sdk-go-v2/feature/s3/manager v1.17.75/go.mod h1:bDMQbkI1vJbNjnvJYpPTSNYBkI/VIv18ngWb/K84tkk= +github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.23 h1:GpT/TrnBYuE5gan2cZbTtvP+JlHsutdmlV2YfEyNde0= +github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.23/go.mod h1:xYWD6BS9ywC5bS3sz9Xh04whO/hzK2plt2Zkyrp4JuA= +github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.23 h1:bpd8vxhlQi2r1hiueOw02f/duEPTMK59Q4QMAoTTtTo= +github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.23/go.mod h1:15DfR2nw+CRHIk0tqNyifu3G1YdAOy68RftkhMDDwYk= +github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.24 h1:OQqn11BtaYv1WLUowvcA30MpzIu8Ti4pcLPIIyoKZrA= +github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.24/go.mod h1:X5ZJyfwVrWA96GzPmUCWFQaEARPR7gCrpq2E92PJwAE= +github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.9 h1:FLudkZLt5ci0ozzgkVo8BJGwvqNaZbTWb3UcucAateA= +github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.9/go.mod h1:w7wZ/s9qK7c8g4al+UyoF1Sp/Z45UwMGcqIzLWVQHWk= +github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.9.15 h1:ieLCO1JxUWuxTZ1cRd0GAaeX7O6cIxnwk7tc1LsQhC4= +github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.9.15/go.mod h1:e3IzZvQ3kAWNykvE0Tr0RDZCMFInMvhku3qNpcIQXhM= +github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.23 h1:pbrxO/kuIwgEsOPLkaHu0O+m4fNgLU8B3vxQ+72jTPw= +github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.23/go.mod h1:/CMNUqoj46HpS3MNRDEDIwcgEnrtZlKRaHNaHxIFpNA= +github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.23 h1:03xatSQO4+AM1lTAbnRg5OK528EUg744nW7F73U8DKw= +github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.23/go.mod h1:M8l3mwgx5ToK7wot2sBBce/ojzgnPzZXUV445gTSyE8= +github.com/aws/aws-sdk-go-v2/service/s3 v1.100.1 h1:mxuT1xE+dI54NW3RkNjP8DUT5HXqbkiAFvfdyDFwE5c= +github.com/aws/aws-sdk-go-v2/service/s3 v1.100.1/go.mod h1:L2dcoOgS2VSgbPLvpak2NyUPsO1TBN7M45Z4H7DlRc4= +github.com/aws/smithy-go v1.25.1 h1:J8ERsGSU7d+aCmdQur5Txg6bVoYelvQJgtZehD12GkI= +github.com/aws/smithy-go v1.25.1/go.mod h1:YE2RhdIuDbA5E5bTdciG9KrW3+TiEONeUWCqxX9i1Fc= +github.com/cevatbarisyilmaz/ara v0.0.4 h1:SGH10hXpBJhhTlObuZzTuFn1rrdmjQImITXnZVPSodc= +github.com/cevatbarisyilmaz/ara v0.0.4/go.mod h1:BfFOxnUd6Mj6xmcvRxHN3Sr21Z1T3U2MYkYOmoQe4Ts= +github.com/cockroachdb/errors v1.13.0 h1:BoCcJeiP9hpBJDETkX19qi8Tb8So37srSsp3stTaDMQ= +github.com/cockroachdb/errors v1.13.0/go.mod h1:bjxt/4E5+OyuAnacpTIU9rn2mzPu1VlthvHP+xpROq0= +github.com/cockroachdb/logtags v0.0.0-20230118201751-21c54148d20b h1:r6VH0faHjZeQy818SGhaone5OnYfxFR/+AzdY3sf5aE= +github.com/cockroachdb/logtags v0.0.0-20230118201751-21c54148d20b/go.mod h1:Vz9DsVWQQhf3vs21MhPMZpMGSht7O/2vFW2xusFUVOs= +github.com/cockroachdb/redact v1.1.5 h1:u1PMllDkdFfPWaNGMyLD1+so+aq3uUItthCFqzwPJ30= +github.com/cockroachdb/redact v1.1.5/go.mod h1:BVNblN9mBWFyMyqK1k3AAiSxhvhfK2oOZZ2lK+dpvRg= +github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E= +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/getsentry/sentry-go v0.46.0 h1:mbdDaarbUdOt9X+dx6kDdntkShLEX3/+KyOsVDTPDj0= +github.com/getsentry/sentry-go v0.46.0/go.mod h1:evVbw2qotNUdYG8KxXbAdjOQWWvWIwKxpjdZZIvcIPw= +github.com/go-errors/errors v1.4.2 h1:J6MZopCL4uSllY1OfXM374weqZFFItUbrImctkmUxIA= +github.com/go-errors/errors v1.4.2/go.mod h1:sIVyrIiJhuEF+Pj9Ebtd6P/rEYROXFi3BopGUQ5a5Og= +github.com/gogo/protobuf v1.3.2 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q= +github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q= +github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI= +github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY= +github.com/johannesboyne/gofakes3 v0.0.0-20260208201424-4c385a1f6a73 h1:0xkWp+RMC2ImuKacheMHEAtrbOTMOa0kYkxyzM1Z/II= +github.com/johannesboyne/gofakes3 v0.0.0-20260208201424-4c385a1f6a73/go.mod h1:S4S9jGBVlLri0OeqrSSbCGG5vsI6he06UJyuz1WT1EE= +github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8= +github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck= +github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= +github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= +github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= +github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= +github.com/pingcap/errors v0.11.4 h1:lFuQV/oaUMGcD2tqt+01ROSmJs75VG1ToEOkZIZ4nE4= +github.com/pingcap/errors v0.11.4/go.mod h1:Oi8TUi2kEtXXLMJk9l1cGmz20kV3TaQ0usTwv5KuLY8= +github.com/pkg/diff v0.0.0-20210226163009-20ebb0f2a09e/go.mod h1:pJLUxLENpZxwdsKMEsNbx1VGcRFpLqf3715MtcvvzbA= +github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4= +github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/rogpeppe/go-internal v1.9.0 h1:73kH8U+JUqXU8lRuOHeVHaa/SZPifC7BkcraZVejAe8= +github.com/rogpeppe/go-internal v1.9.0/go.mod h1:WtVeX8xhTBvf0smdhujwtBcq4Qrzq/fJaraNFVN+nFs= +github.com/ryszard/goskiplist v0.0.0-20150312221310-2dfbae5fcf46 h1:GHRpF1pTW19a8tTFrMLUcfWwyC0pnifVo2ClaLq+hP8= +github.com/ryszard/goskiplist v0.0.0-20150312221310-2dfbae5fcf46/go.mod h1:uAQ5PCi+MFsC7HjREoAz1BU+Mq60+05gifQSsHSDG/8= +github.com/spf13/afero v1.2.1 h1:qgMbHoJbPbw579P+1zVY+6n4nIFuIchaIjzZ/I/Yq8M= +github.com/spf13/afero v1.2.1/go.mod h1:9ZxEEn6pIJ8Rxe320qSDBk6AsU0r9pR7Q4OcevTdifk= +github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= +github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= +github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= +github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= +go.etcd.io/bbolt v1.3.5 h1:XAzx9gjCb0Rxj7EoqcClPD1d5ZBxZJk0jbuoPHenBt0= +go.etcd.io/bbolt v1.3.5/go.mod h1:G5EMThwa9y8QZGBClrRx5EY+Yw9kAhnjy3bSjsnlVTQ= +go.shabbyrobe.org/gocovmerge v0.0.0-20230507111327-fa4f82cfbf4d h1:Ns9kd1Rwzw7t0BR8XMphenji4SmIoNZPn8zhYmaVKP8= +go.shabbyrobe.org/gocovmerge v0.0.0-20230507111327-fa4f82cfbf4d/go.mod h1:92Uoe3l++MlthCm+koNi0tcUCX3anayogF0Pa/sp24k= +go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= +go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE= +golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= +golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI= +golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto= +golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= +golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= +golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg= +golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= +golang.org/x/net v0.0.0-20200226121028-0de0cce0169b/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= +golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU= +golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= +golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.31.0 h1:ioabZlmFYtWhL+TRYpcnNlLwhyxaM9kWTDEmfnprqik= +golang.org/x/sys v0.31.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k= +golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= +golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= +golang.org/x/text v0.23.0 h1:D71I7dUrlY+VX0gQShAThNGHFxZ13dGLBHQLVl1mJlY= +golang.org/x/text v0.23.0/go.mod h1:/BLNzu4aZCJ1+kcD0DNRotWKage4q2rGVAg4o22unh4= +golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= +golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= +golang.org/x/tools v0.0.0-20200619180055-7c47624df98f/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE= +golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA= +golang.org/x/tools v0.21.1-0.20240508182429-e35e4ccd0d2d h1:vU5i/LfpvrRCpgM/VPfJLg5KjxD3E+hfT1SH+d9zLwg= +golang.org/x/tools v0.21.1-0.20240508182429-e35e4ccd0d2d/go.mod h1:aiJjzUbINMkxbQROHiO6hDPo2LHcIPhhQsa9DLh0yGk= +golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk= +gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q= +gopkg.in/mgo.v2 v2.0.0-20180705113604-9856a29383ce h1:xcEWjVhvbDy+nHP67nPDDpbYrY+ILlfndk4bRioVHaU= +gopkg.in/mgo.v2 v2.0.0-20180705113604-9856a29383ce/go.mod h1:yeKp02qBN3iKW1OzL3MGk2IdtZzaj7SFntXj72NppTA= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/internal/iox/exact_size_reader.go b/lfs/internal/iox/exact_size_reader.go similarity index 100% rename from internal/iox/exact_size_reader.go rename to lfs/internal/iox/exact_size_reader.go diff --git a/internal/iox/exact_size_reader_test.go b/lfs/internal/iox/exact_size_reader_test.go similarity index 100% rename from internal/iox/exact_size_reader_test.go rename to lfs/internal/iox/exact_size_reader_test.go diff --git a/internal/iox/sha256_reader.go b/lfs/internal/iox/sha256_reader.go similarity index 100% rename from internal/iox/sha256_reader.go rename to lfs/internal/iox/sha256_reader.go diff --git a/internal/iox/sha256_reader_test.go b/lfs/internal/iox/sha256_reader_test.go similarity index 100% rename from internal/iox/sha256_reader_test.go rename to lfs/internal/iox/sha256_reader_test.go diff --git a/lfs/lfs.go b/lfs/lfs.go new file mode 100644 index 0000000..548f32e --- /dev/null +++ b/lfs/lfs.go @@ -0,0 +1,123 @@ +// Package lfs implements the server side of the Git LFS Batch API and the +// basic transfer adapter on top of pluggable metadata and object storage. +// +// It leaves HTTP routing, authentication, authorization, and the encoding of +// responses to the caller, so that servers with different URL layouts, +// credentials, and databases can share the transfer logic. +// +// Spec: https://github.com/git-lfs/git-lfs/blob/main/docs/api/batch.md +package lfs + +import ( + "context" + "regexp" +) + +// oidPattern matches the SHA-256 hex digest format used as the LFS object ID: +// exactly 64 lowercase hex characters. +var oidPattern = regexp.MustCompile(`^[0-9a-f]{64}$`) + +// ValidOID reports whether oid is a valid LFS object ID. +func ValidOID(oid string) bool { + return oidPattern.MatchString(oid) +} + +const ( + OperationUpload = "upload" + OperationDownload = "download" + TransferBasic = "basic" +) + +// BatchRequest is the request payload of the batch endpoint. +type BatchRequest struct { + Operation string `json:"operation"` + Transfers []string `json:"transfers,omitempty"` + Objects []BatchObject `json:"objects"` +} + +// BatchObject is an object requested in a BatchRequest. +type BatchObject struct { + OID string `json:"oid"` + Size int64 `json:"size"` +} + +// BatchResponse is the response payload of the batch endpoint. +type BatchResponse struct { + Transfer string `json:"transfer"` + Objects []BatchResponseObject `json:"objects"` +} + +// BatchResponseObject is the result for one object of a BatchRequest. Either +// Actions or Error is set, unless the object needs no transfer. +type BatchResponseObject struct { + OID string `json:"oid"` + Size int64 `json:"size"` + Authenticated bool `json:"authenticated,omitempty"` + Actions map[string]Action `json:"actions,omitempty"` + Error *ObjectError `json:"error,omitempty"` +} + +// Action tells the client where and how to transfer an object. +type Action struct { + Href string `json:"href"` + Header map[string]string `json:"header,omitempty"` +} + +// ObjectError is the error of one object in a BatchResponse. +type ObjectError struct { + Code int `json:"code"` + Message string `json:"message"` +} + +// VerifyRequest is the request payload of the verify endpoint. +type VerifyRequest struct { + OID string `json:"oid"` + Size int64 `json:"size"` +} + +// Error is a client error that the caller should respond with, using Code as +// the HTTP status code. +type Error struct { + Code int + Message string +} + +func (e *Error) Error() string { + return e.Message +} + +// Object is a verified LFS object. +type Object struct { + OID string + Size int64 + // URI is the location of the object content, as returned by the storage + // backend that stores it. + URI string +} + +// Store is the metadata layer of LFS objects. An object is "pending" from the +// start of a presigned upload until it is verified, and only verified objects +// are served. A repository is identified by an opaque string chosen by the +// caller. +type Store interface { + // GetRepoObjects returns the verified objects among oids that are linked to + // repo, keyed by OID. Missing OIDs are omitted. + GetRepoObjects(ctx context.Context, repo string, oids []string) (map[string]Object, error) + // GetObjects returns the verified objects among oids regardless of the + // repositories they are linked to, keyed by OID. Missing OIDs are omitted. + // It is only called when Service.CrossRepoDedup is true. + GetObjects(ctx context.Context, oids []string) (map[string]Object, error) + // LinkObject records a verified object whose content is stored at uri and + // links it to repo. It returns the URI the object is recorded with, and when + // that differs from uri, the caller deletes the content at uri as no longer + // referenced. + LinkObject(ctx context.Context, repo, oid string, size int64, uri string) (storedURI string, err error) + // InsertPendingObject records a pending object for oid, unless the object is + // already recorded, so that abandoned presigned uploads can be cleaned up. + InsertPendingObject(ctx context.Context, oid string) error + // VerifyObject marks the pending object for oid as verified with content + // stored at uri and links it to repo. It returns the URI the object is + // recorded with, like LinkObject. It fails when no object is recorded for + // oid. + VerifyObject(ctx context.Context, repo, oid string, size int64, uri string) (storedURI string, err error) +} diff --git a/lfs/moon.yml b/lfs/moon.yml new file mode 100644 index 0000000..7605311 --- /dev/null +++ b/lfs/moon.yml @@ -0,0 +1,45 @@ +$schema: 'https://moonrepo.dev/schemas/project.json' + +language: 'go' +stack: 'backend' +id: 'lfs' + +fileGroups: + sources: + - '**/*.go' + tests: + - '**/*_test.go' + configs: + - 'go.mod' + - 'go.sum' + +tasks: + install: + command: 'go mod tidy' + inputs: + - '@group(configs)' + + lint: + command: 'golangci-lint run' + inputs: + - '@group(sources)' + - '@group(configs)' + deps: + - 'install' + + test: + command: 'go test ./...' + inputs: + - '@group(sources)' + - '@group(tests)' + - '@group(configs)' + deps: + - 'install' + + build: + command: 'go build -o /dev/null ./...' + inputs: + - '@group(sources)' + - '@group(configs)' + deps: + - 'install' diff --git a/lfs/service.go b/lfs/service.go new file mode 100644 index 0000000..6bfd0e5 --- /dev/null +++ b/lfs/service.go @@ -0,0 +1,379 @@ +package lfs + +import ( + "context" + "encoding/base64" + "encoding/hex" + "fmt" + "io" + "log/slog" + "maps" + "net/http" + "slices" + "strings" + + "github.com/cockroachdb/errors" + + "unknwon.dev/git-lfs-server/lfs/internal/iox" + "unknwon.dev/git-lfs-server/lfs/storage" +) + +// Service serves LFS operations for repositories. Callers authenticate and +// authorize each request before calling a method, and respond with the Code +// and Message of a returned *Error. Any other returned error is an internal +// error. +type Service struct { + Store Store + // Storage receives uploads and decides the transfer mode: content is + // proxied through the caller for a storage.Proxier, and transferred + // directly by the client through redirects for a storage.Presigner. + Storage storage.Backend + // PreviousStorages are consulted for downloads of objects stored before + // Storage was changed. An object is served by the first backend, starting + // with Storage, whose scheme prefixes the object's URI. + PreviousStorages []storage.Backend + // MaxObjectSize is the maximum size in bytes of an uploaded object. Zero or + // negative means no limit. + MaxObjectSize int64 + // CrossRepoDedup allows an upload to be satisfied by an object that is + // already stored for another repository, without the client sending the + // content again. Without it, a repository only gains an object by uploading + // its full content, so knowing an OID and its size is not enough to read the + // object through a repository the client has write access to. + CrossRepoDedup bool +} + +// Endpoint describes how the client reaches the transfer endpoints of a +// repository: objects at Href/{oid} and verifications at Href/verify. +type Endpoint struct { + Href string + // Header is included in every action, e.g., to forward the credentials of + // the batch request. + Header map[string]string + // Authenticated tells the client that Header authenticates the actions, so + // it does not look up credentials for them. It is required when Storage is a + // storage.Presigner: otherwise the Git LFS client looks up credentials for + // the host of the presigned URL it is redirected to, and fails. + Authenticated bool +} + +// Redirect sends the client to another location to transfer the content. +type Redirect struct { + URL string + // Header must be sent by the client verbatim. + Header map[string]string +} + +// Batch answers a batch request for repo. Callers must check that the actor is +// allowed to perform req.Operation before calling it. +func (s *Service) Batch(ctx context.Context, logger *slog.Logger, repo string, endpoint Endpoint, req BatchRequest) (*BatchResponse, error) { + if req.Operation != OperationUpload && req.Operation != OperationDownload { + return nil, &Error{Code: http.StatusBadRequest, Message: `operation must be "upload" or "download"`} + } + if len(req.Transfers) > 0 && !slices.Contains(req.Transfers, TransferBasic) { + return nil, &Error{Code: http.StatusUnprocessableEntity, Message: `only "basic" transfer is supported`} + } + + oids := make([]string, len(req.Objects)) + for i, o := range req.Objects { + if !ValidOID(o.OID) { + return nil, &Error{Code: http.StatusBadRequest, Message: fmt.Sprintf("invalid oid %q", o.OID)} + } + oids[i] = o.OID + } + + var ( + existing map[string]Object + err error + ) + switch req.Operation { + case OperationDownload: + existing, err = s.Store.GetRepoObjects(ctx, repo, oids) + case OperationUpload: + existing, err = s.uploadedObjects(ctx, repo, oids) + } + if err != nil { + return nil, errors.Wrap(err, "look up objects") + } + + objectHref := func(oid string) string { return endpoint.Href + "/" + oid } + resp := &BatchResponse{ + Transfer: TransferBasic, + Objects: make([]BatchResponseObject, 0, len(req.Objects)), + } + for _, in := range req.Objects { + obj, ok := existing[in.OID] + out := BatchResponseObject{OID: in.OID, Size: in.Size, Authenticated: endpoint.Authenticated} + + switch req.Operation { + case OperationDownload: + switch { + case !ok: + out.Error = &ObjectError{Code: http.StatusNotFound, Message: "object does not exist"} + case obj.Size != in.Size: + out.Error = &ObjectError{Code: http.StatusUnprocessableEntity, Message: "OID exists with different size"} + default: + out.Actions = map[string]Action{ + "download": {Href: objectHref(in.OID), Header: endpoint.Header}, + } + } + case OperationUpload: + switch { + case s.MaxObjectSize > 0 && in.Size > s.MaxObjectSize: + out.Error = &ObjectError{ + Code: http.StatusRequestEntityTooLarge, + Message: fmt.Sprintf("object size %d exceeds limit %d", in.Size, s.MaxObjectSize), + } + case ok && obj.Size == in.Size: + // No actions tell the client the object needs no upload. With + // cross-repo dedup, the object may only be linked to another + // repository so far. + if s.CrossRepoDedup { + if _, err := s.Store.LinkObject(ctx, repo, in.OID, in.Size, obj.URI); err != nil { + logger.ErrorContext(ctx, "Failed to link repo to existing object", "oid", in.OID, "error", err) + out.Error = &ObjectError{Code: http.StatusInternalServerError, Message: "failed to link object to repo"} + } + } + case ok && obj.Size != in.Size: + out.Error = &ObjectError{Code: http.StatusConflict, Message: "OID exists with different size"} + default: + rawOID, _ := hex.DecodeString(in.OID) // Validated above. + uploadHeader := make(map[string]string, len(endpoint.Header)+1) + maps.Copy(uploadHeader, endpoint.Header) + // Presigned uploads require the client to send the checksum that + // the storage verifies the content against. + uploadHeader["x-amz-checksum-sha256"] = base64.StdEncoding.EncodeToString(rawOID) + out.Actions = map[string]Action{ + "upload": {Href: objectHref(in.OID), Header: uploadHeader}, + "verify": {Href: endpoint.Href + "/verify", Header: endpoint.Header}, + } + } + } + resp.Objects = append(resp.Objects, out) + } + return resp, nil +} + +// uploadedObjects returns the verified objects among oids that an upload to +// repo does not need to send again. +func (s *Service) uploadedObjects(ctx context.Context, repo string, oids []string) (map[string]Object, error) { + if s.CrossRepoDedup { + return s.Store.GetObjects(ctx, oids) + } + return s.Store.GetRepoObjects(ctx, repo, oids) +} + +// UploadResult is the outcome of a successful Service.Upload. +type UploadResult struct { + // Redirect is set when the client must upload the content to another + // location and then call the verify endpoint. + Redirect *Redirect + // Existed reports that the object was already stored and the content was not + // read. + Existed bool +} + +// Upload handles an upload of size bytes for oid to repo. For a +// storage.Proxier, it stores the content read from body after verifying its +// size and hash. For a storage.Presigner, it does not read body. +func (s *Service) Upload(ctx context.Context, logger *slog.Logger, repo, oid string, size int64, body io.Reader) (*UploadResult, error) { + if !ValidOID(oid) { + return nil, &Error{Code: http.StatusBadRequest, Message: "invalid oid"} + } + if size < 0 { + return nil, &Error{Code: http.StatusLengthRequired, Message: "Content-Length is required"} + } + if s.MaxObjectSize > 0 && size > s.MaxObjectSize { + return nil, &Error{ + Code: http.StatusRequestEntityTooLarge, + Message: fmt.Sprintf("object size %d exceeds limit %d", size, s.MaxObjectSize), + } + } + + switch b := s.Storage.(type) { + case storage.Proxier: + return s.uploadProxy(ctx, logger, b, repo, oid, size, body) + case storage.Presigner: + return s.uploadPresign(ctx, b, repo, oid, size) + default: + return nil, errors.Newf("storage backend %T does not implement a known transfer mode", s.Storage) + } +} + +func (s *Service) uploadProxy(ctx context.Context, logger *slog.Logger, backend storage.Proxier, repo, oid string, size int64, body io.Reader) (*UploadResult, error) { + reader := iox.NewSHA256Reader(iox.NewExactSizeReader(body, size), oid) + uri, err := backend.Put(ctx, oid, reader) + if err == nil { + // Put skips reading when the content is already stored, but the client + // must still prove it has the content before the object is linked to + // repo. + _, err = io.Copy(io.Discard, reader) + } + if err != nil { + var sizeErr *iox.SizeMismatchError + var hashErr *iox.SHA256MismatchError + switch { + case errors.As(err, &sizeErr): + return nil, &Error{Code: http.StatusUnprocessableEntity, Message: sizeErr.Error()} + case errors.As(err, &hashErr): + return nil, &Error{Code: http.StatusUnprocessableEntity, Message: hashErr.Error()} + default: + return nil, errors.Wrap(err, "store object") + } + } + + storedURI, err := s.Store.LinkObject(ctx, repo, oid, size, uri) + if err != nil { + return nil, errors.Wrap(err, "link object") + } + if storedURI != uri { + if err := backend.Delete(ctx, uri); err != nil { + logger.ErrorContext(ctx, "Failed to delete orphan object", "uri", uri, "error", err) + } + } + return &UploadResult{}, nil +} + +func (s *Service) uploadPresign(ctx context.Context, backend storage.Presigner, repo, oid string, size int64) (*UploadResult, error) { + existing, err := s.uploadedObjects(ctx, repo, []string{oid}) + if err != nil { + return nil, errors.Wrap(err, "look up object") + } + if _, ok := existing[oid]; ok { + return &UploadResult{Existed: true}, nil + } + + if err = s.Store.InsertPendingObject(ctx, oid); err != nil { + return nil, errors.Wrap(err, "insert pending object") + } + url, header, err := backend.PresignPut(ctx, oid, size) + if err != nil { + return nil, errors.Wrap(err, "presign upload") + } + return &UploadResult{Redirect: &Redirect{URL: url, Header: header}}, nil +} + +// DownloadResult is the outcome of a successful Service.Download. Exactly one +// of Redirect and Content is set. +type DownloadResult struct { + // Redirect is set when the client must download the content from another + // location. + Redirect *Redirect + // Content is the object content, which the caller must close. + Content io.ReadCloser + // Size is the size of Content in bytes. + Size int64 +} + +// Download handles a download of oid from repo. +func (s *Service) Download(ctx context.Context, repo, oid string) (*DownloadResult, error) { + if !ValidOID(oid) { + return nil, &Error{Code: http.StatusBadRequest, Message: "invalid oid"} + } + + objects, err := s.Store.GetRepoObjects(ctx, repo, []string{oid}) + if err != nil { + return nil, errors.Wrap(err, "look up object") + } + object, ok := objects[oid] + if !ok { + return nil, &Error{Code: http.StatusNotFound, Message: "object does not exist"} + } + + backend := s.backendFor(object.URI) + switch b := backend.(type) { + case storage.Proxier: + rc, err := b.Open(ctx, object.URI) + if err != nil { + if errors.Is(err, storage.ErrNotFound) { + return nil, errors.Wrapf(err, "object %q is recorded but missing in storage", object.URI) + } + return nil, errors.Wrap(err, "open object") + } + return &DownloadResult{Content: rc, Size: object.Size}, nil + case storage.Presigner: + url, err := b.PresignGet(ctx, object.URI) + if err != nil { + return nil, errors.Wrap(err, "presign download") + } + return &DownloadResult{Redirect: &Redirect{URL: url}}, nil + case nil: + return nil, errors.Newf("no storage backend for uri %q", object.URI) + default: + return nil, errors.Newf("storage backend %T does not implement a known transfer mode", backend) + } +} + +// backendFor returns the backend that stores the object at uri, or nil when +// none does. +func (s *Service) backendFor(uri string) storage.Backend { + if strings.HasPrefix(uri, s.Storage.Scheme()) { + return s.Storage + } + for _, b := range s.PreviousStorages { + if strings.HasPrefix(uri, b.Scheme()) { + return b + } + } + return nil +} + +// Verify handles the verification of an upload to repo. A presigned upload +// becomes downloadable only after it is verified. +func (s *Service) Verify(ctx context.Context, logger *slog.Logger, repo string, req VerifyRequest) error { + if !ValidOID(req.OID) { + return &Error{Code: http.StatusBadRequest, Message: "invalid oid"} + } + + switch b := s.Storage.(type) { + case storage.Proxier: + return s.verifyProxy(ctx, repo, req) + case storage.Presigner: + return s.verifyPresign(ctx, logger, b, repo, req) + default: + return errors.Newf("storage backend %T does not implement a known transfer mode", s.Storage) + } +} + +func (s *Service) verifyProxy(ctx context.Context, repo string, req VerifyRequest) error { + objects, err := s.Store.GetRepoObjects(ctx, repo, []string{req.OID}) + if err != nil { + return errors.Wrap(err, "look up object") + } + object, ok := objects[req.OID] + if !ok { + return &Error{Code: http.StatusNotFound, Message: "object does not exist"} + } + if object.Size != req.Size { + return &Error{Code: http.StatusUnprocessableEntity, Message: "size mismatch"} + } + return nil +} + +func (s *Service) verifyPresign(ctx context.Context, logger *slog.Logger, backend storage.Presigner, repo string, req VerifyRequest) error { + uri := backend.URI(req.OID) + headSize, err := backend.Head(ctx, uri) + if err != nil { + if errors.Is(err, storage.ErrNotFound) { + return &Error{Code: http.StatusUnprocessableEntity, Message: "object does not exist in storage"} + } + return errors.Wrap(err, "head object") + } + // The presigned upload signs Content-Length, which should make this + // impossible, but treat any mismatch as a hard error rather than trusting + // either side. + if headSize != req.Size { + return &Error{Code: http.StatusUnprocessableEntity, Message: "size mismatch"} + } + + storedURI, err := s.Store.VerifyObject(ctx, repo, req.OID, req.Size, uri) + if err != nil { + return errors.Wrap(err, "mark object verified") + } + if storedURI != uri { + if err := backend.Delete(ctx, uri); err != nil { + logger.ErrorContext(ctx, "Failed to delete orphan object", "uri", uri, "error", err) + } + } + return nil +} diff --git a/lfs/service_test.go b/lfs/service_test.go new file mode 100644 index 0000000..bfe39f1 --- /dev/null +++ b/lfs/service_test.go @@ -0,0 +1,373 @@ +package lfs + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "path/filepath" + "strings" + "sync" + "testing" + + "github.com/cockroachdb/errors" + "github.com/johannesboyne/gofakes3" + "github.com/johannesboyne/gofakes3/backend/s3mem" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "unknwon.dev/git-lfs-server/lfs/storage" +) + +// memStore is an in-memory Store. +type memStore struct { + mu sync.Mutex + objects map[string]*memObject // Keyed by OID. + links map[string]map[string]bool // Keyed by repo, then OID. +} + +type memObject struct { + size int64 + uri string + verified bool +} + +func newMemStore() *memStore { + return &memStore{objects: map[string]*memObject{}, links: map[string]map[string]bool{}} +} + +func (s *memStore) verifiedObjects(oids []string, keep func(oid string) bool) map[string]Object { + out := map[string]Object{} + for _, oid := range oids { + if o := s.objects[oid]; o != nil && o.verified && keep(oid) { + out[oid] = Object{OID: oid, Size: o.size, URI: o.uri} + } + } + return out +} + +func (s *memStore) GetRepoObjects(_ context.Context, repo string, oids []string) (map[string]Object, error) { + s.mu.Lock() + defer s.mu.Unlock() + return s.verifiedObjects(oids, func(oid string) bool { return s.links[repo][oid] }), nil +} + +func (s *memStore) GetObjects(_ context.Context, oids []string) (map[string]Object, error) { + s.mu.Lock() + defer s.mu.Unlock() + return s.verifiedObjects(oids, func(string) bool { return true }), nil +} + +func (s *memStore) link(repo, oid string) { + if s.links[repo] == nil { + s.links[repo] = map[string]bool{} + } + s.links[repo][oid] = true +} + +func (s *memStore) LinkObject(_ context.Context, repo, oid string, size int64, uri string) (string, error) { + s.mu.Lock() + defer s.mu.Unlock() + o := s.objects[oid] + if o == nil { + o = &memObject{} + s.objects[oid] = o + } + if !o.verified { + o.size, o.uri, o.verified = size, uri, true + } + s.link(repo, oid) + return o.uri, nil +} + +func (s *memStore) InsertPendingObject(_ context.Context, oid string) error { + s.mu.Lock() + defer s.mu.Unlock() + if s.objects[oid] == nil { + s.objects[oid] = &memObject{} + } + return nil +} + +func (s *memStore) VerifyObject(_ context.Context, repo, oid string, size int64, uri string) (string, error) { + s.mu.Lock() + defer s.mu.Unlock() + o := s.objects[oid] + if o == nil { + return "", errors.New("object not found") + } + if !o.verified { + o.size, o.uri, o.verified = size, uri, true + } + s.link(repo, oid) + return o.uri, nil +} + +func sha256Hex(s string) string { + sum := sha256.Sum256([]byte(s)) + return hex.EncodeToString(sum[:]) +} + +func newFilesystemBackend(t *testing.T, name, scheme string) *storage.FilesystemBackend { + t.Helper() + root := t.TempDir() + b, err := storage.NewFilesystemBackend(name, scheme, root, filepath.Join(root, ".tmp")) + require.NoError(t, err) + return b +} + +func newS3PresignBackend(t *testing.T) *storage.S3PresignBackend { + t.Helper() + mem := s3mem.New() + require.NoError(t, mem.CreateBucket("lfs-objects")) + srv := httptest.NewServer(gofakes3.New(mem).Server()) + t.Cleanup(srv.Close) + b, err := storage.NewS3PresignBackend("s3", "s3://", "lfs-objects", "test", "test", srv.URL) + require.NoError(t, err) + return b +} + +var discardLogger = slog.New(slog.DiscardHandler) + +func requireLFSError(t *testing.T, err error, wantCode int) { + t.Helper() + var lfsErr *Error + require.True(t, errors.As(err, &lfsErr), "want *Error, got %v", err) + assert.Equal(t, wantCode, lfsErr.Code, lfsErr.Message) +} + +func TestService_Proxy(t *testing.T) { + ctx := t.Context() + content := "proxied content" + oid := sha256Hex(content) + size := int64(len(content)) + + t.Run("upload stores content and links it to the repository only", func(t *testing.T) { + s := &Service{Store: newMemStore(), Storage: newFilesystemBackend(t, "fs", "file://")} + result, err := s.Upload(ctx, discardLogger, "repo1", oid, size, strings.NewReader(content)) + require.NoError(t, err) + assert.Equal(t, &UploadResult{}, result) + + require.NoError(t, s.Verify(ctx, discardLogger, "repo1", VerifyRequest{OID: oid, Size: size})) + requireLFSError(t, s.Verify(ctx, discardLogger, "repo1", VerifyRequest{OID: oid, Size: size + 1}), http.StatusUnprocessableEntity) + requireLFSError(t, s.Verify(ctx, discardLogger, "repo2", VerifyRequest{OID: oid, Size: size}), http.StatusNotFound) + + download, err := s.Download(ctx, "repo1", oid) + require.NoError(t, err) + got, err := io.ReadAll(download.Content) + require.NoError(t, download.Content.Close()) + require.NoError(t, err) + assert.Equal(t, content, string(got)) + assert.Equal(t, size, download.Size) + + _, err = s.Download(ctx, "repo2", oid) + requireLFSError(t, err, http.StatusNotFound) + }) + + t.Run("upload of stored content still verifies the body", func(t *testing.T) { + s := &Service{Store: newMemStore(), Storage: newFilesystemBackend(t, "fs", "file://")} + _, err := s.Upload(ctx, discardLogger, "repo1", oid, size, strings.NewReader(content)) + require.NoError(t, err) + + forged := strings.Repeat("x", len(content)) + _, err = s.Upload(ctx, discardLogger, "repo2", oid, size, strings.NewReader(forged)) + requireLFSError(t, err, http.StatusUnprocessableEntity) + _, err = s.Download(ctx, "repo2", oid) + requireLFSError(t, err, http.StatusNotFound) + + _, err = s.Upload(ctx, discardLogger, "repo2", oid, size, strings.NewReader(content)) + require.NoError(t, err) + _, err = s.Download(ctx, "repo2", oid) + require.NoError(t, err) + }) + + t.Run("upload rejects invalid input", func(t *testing.T) { + s := &Service{Store: newMemStore(), Storage: newFilesystemBackend(t, "fs", "file://"), MaxObjectSize: size - 1} + _, err := s.Upload(ctx, discardLogger, "repo1", "not-an-oid", size, strings.NewReader(content)) + requireLFSError(t, err, http.StatusBadRequest) + _, err = s.Upload(ctx, discardLogger, "repo1", oid, -1, strings.NewReader(content)) + requireLFSError(t, err, http.StatusLengthRequired) + _, err = s.Upload(ctx, discardLogger, "repo1", oid, size, strings.NewReader(content)) + requireLFSError(t, err, http.StatusRequestEntityTooLarge) + + s.MaxObjectSize = 0 + _, err = s.Upload(ctx, discardLogger, "repo1", oid, size, strings.NewReader(content[1:])) + requireLFSError(t, err, http.StatusUnprocessableEntity) + }) + + t.Run("download reads objects from previous storages", func(t *testing.T) { + store := newMemStore() + previous := newFilesystemBackend(t, "old", "old-file://") + s := &Service{Store: store, Storage: previous} + _, err := s.Upload(ctx, discardLogger, "repo1", oid, size, strings.NewReader(content)) + require.NoError(t, err) + + s = &Service{Store: store, Storage: newFilesystemBackend(t, "new", "file://"), PreviousStorages: []storage.Backend{previous}} + download, err := s.Download(ctx, "repo1", oid) + require.NoError(t, err) + got, err := io.ReadAll(download.Content) + require.NoError(t, download.Content.Close()) + require.NoError(t, err) + assert.Equal(t, content, string(got)) + + s.PreviousStorages = nil + _, err = s.Download(ctx, "repo1", oid) + require.Error(t, err) + assert.False(t, errors.As(err, new(*Error)), "a recorded object without a backend is an internal error") + }) +} + +func TestService_Batch(t *testing.T) { + ctx := t.Context() + content := "batched content" + oid := sha256Hex(content) + size := int64(len(content)) + endpoint := Endpoint{Href: "https://lfs.example.com/repo/info/lfs/objects"} + + newServiceWithObject := func(t *testing.T, crossRepoDedup bool) *Service { + s := &Service{Store: newMemStore(), Storage: newFilesystemBackend(t, "fs", "file://"), CrossRepoDedup: crossRepoDedup} + _, err := s.Upload(ctx, discardLogger, "repo1", oid, size, strings.NewReader(content)) + require.NoError(t, err) + return s + } + upload := BatchRequest{Operation: OperationUpload, Objects: []BatchObject{{OID: oid, Size: size}}} + download := BatchRequest{Operation: OperationDownload, Objects: []BatchObject{{OID: oid, Size: size}}} + + t.Run("upload of an object in the repository needs no transfer", func(t *testing.T) { + s := newServiceWithObject(t, false) + resp, err := s.Batch(ctx, discardLogger, "repo1", endpoint, upload) + require.NoError(t, err) + assert.Equal(t, []BatchResponseObject{{OID: oid, Size: size}}, resp.Objects) + }) + + t.Run("upload of an object in another repository", func(t *testing.T) { + t.Run("needs a transfer without cross-repo dedup", func(t *testing.T) { + s := newServiceWithObject(t, false) + resp, err := s.Batch(ctx, discardLogger, "repo2", endpoint, upload) + require.NoError(t, err) + require.Len(t, resp.Objects, 1) + assert.Contains(t, resp.Objects[0].Actions, "upload") + + resp, err = s.Batch(ctx, discardLogger, "repo2", endpoint, download) + require.NoError(t, err) + require.NotNil(t, resp.Objects[0].Error) + assert.Equal(t, http.StatusNotFound, resp.Objects[0].Error.Code) + }) + + t.Run("links the object with cross-repo dedup", func(t *testing.T) { + s := newServiceWithObject(t, true) + resp, err := s.Batch(ctx, discardLogger, "repo2", endpoint, upload) + require.NoError(t, err) + assert.Equal(t, []BatchResponseObject{{OID: oid, Size: size}}, resp.Objects) + + resp, err = s.Batch(ctx, discardLogger, "repo2", endpoint, download) + require.NoError(t, err) + assert.Contains(t, resp.Objects[0].Actions, "download") + }) + }) + + t.Run("actions carry the endpoint", func(t *testing.T) { + s := newServiceWithObject(t, false) + endpoint := Endpoint{Href: endpoint.Href, Header: map[string]string{"Authorization": "Basic Zm9vOmJhcg=="}, Authenticated: true} + newContent := "new content" + newOID := sha256Hex(newContent) + + resp, err := s.Batch(ctx, discardLogger, "repo1", endpoint, BatchRequest{ + Operation: OperationUpload, + Objects: []BatchObject{{OID: newOID, Size: int64(len(newContent))}}, + }) + require.NoError(t, err) + assert.Equal(t, []BatchResponseObject{{ + OID: newOID, + Size: int64(len(newContent)), + Authenticated: true, + Actions: map[string]Action{ + "upload": { + Href: endpoint.Href + "/" + newOID, + Header: map[string]string{ + "Authorization": "Basic Zm9vOmJhcg==", + "x-amz-checksum-sha256": "/jJgjJ71ts9+P5RkgCU/928k9OwGePPQ8H+YRMv/lgE=", + }, + }, + "verify": {Href: endpoint.Href + "/verify", Header: endpoint.Header}, + }, + }}, resp.Objects) + + resp, err = s.Batch(ctx, discardLogger, "repo1", endpoint, download) + require.NoError(t, err) + assert.Equal(t, []BatchResponseObject{{ + OID: oid, + Size: size, + Authenticated: true, + Actions: map[string]Action{"download": {Href: endpoint.Href + "/" + oid, Header: endpoint.Header}}, + }}, resp.Objects) + }) + + t.Run("rejects invalid requests", func(t *testing.T) { + s := newServiceWithObject(t, false) + for _, req := range []BatchRequest{ + {Operation: "delete"}, + {Operation: OperationDownload, Objects: []BatchObject{{OID: "not-an-oid"}}}, + } { + _, err := s.Batch(ctx, discardLogger, "repo1", endpoint, req) + requireLFSError(t, err, http.StatusBadRequest) + } + _, err := s.Batch(ctx, discardLogger, "repo1", endpoint, BatchRequest{Operation: OperationUpload, Transfers: []string{"tus"}}) + requireLFSError(t, err, http.StatusUnprocessableEntity) + }) +} + +func TestService_Presign(t *testing.T) { + ctx := t.Context() + content := "presigned content" + oid := sha256Hex(content) + size := int64(len(content)) + + // uploadToRedirect sends the content to the presigned URL like a client. + uploadToRedirect := func(t *testing.T, redirect *Redirect) { + t.Helper() + req, err := http.NewRequestWithContext(ctx, http.MethodPut, redirect.URL, strings.NewReader(content)) + require.NoError(t, err) + for k, v := range redirect.Header { + req.Header.Set(k, v) + } + req.ContentLength = size + resp, err := http.DefaultClient.Do(req) + require.NoError(t, err) + require.NoError(t, resp.Body.Close()) + require.Equal(t, http.StatusOK, resp.StatusCode) + } + + for _, crossRepoDedup := range []bool{false, true} { + s := &Service{Store: newMemStore(), Storage: newS3PresignBackend(t), CrossRepoDedup: crossRepoDedup} + + result, err := s.Upload(ctx, discardLogger, "repo1", oid, size, nil) + require.NoError(t, err) + require.NotNil(t, result.Redirect) + requireLFSError(t, s.Verify(ctx, discardLogger, "repo1", VerifyRequest{OID: oid, Size: size}), http.StatusUnprocessableEntity) + _, err = s.Download(ctx, "repo1", oid) + requireLFSError(t, err, http.StatusNotFound) + + uploadToRedirect(t, result.Redirect) + requireLFSError(t, s.Verify(ctx, discardLogger, "repo1", VerifyRequest{OID: oid, Size: size + 1}), http.StatusUnprocessableEntity) + require.NoError(t, s.Verify(ctx, discardLogger, "repo1", VerifyRequest{OID: oid, Size: size})) + + download, err := s.Download(ctx, "repo1", oid) + require.NoError(t, err) + require.NotNil(t, download.Redirect) + assert.Contains(t, download.Redirect.URL, oid) + + result, err = s.Upload(ctx, discardLogger, "repo1", oid, size, nil) + require.NoError(t, err) + assert.Equal(t, &UploadResult{Existed: true}, result) + + // Without cross-repo dedup, another repository must upload the content + // itself, which the storage checks against the OID. + result, err = s.Upload(ctx, discardLogger, "repo2", oid, size, nil) + require.NoError(t, err) + assert.Equal(t, crossRepoDedup, result.Redirect == nil, "crossRepoDedup=%v", crossRepoDedup) + } +} diff --git a/internal/storage/filesystem.go b/lfs/storage/filesystem.go similarity index 96% rename from internal/storage/filesystem.go rename to lfs/storage/filesystem.go index 430efc8..a86e1b6 100644 --- a/internal/storage/filesystem.go +++ b/lfs/storage/filesystem.go @@ -48,6 +48,11 @@ func (b *FilesystemBackend) storagePath(oid string) string { return filepath.Join(b.root, oid[0:2], oid[2:4], oid) } +// URI returns the file URI of the object under the storage root. +func (b *FilesystemBackend) URI(oid string) string { + return b.uriFromPath(b.storagePath(oid)) +} + func (b *FilesystemBackend) Put(ctx context.Context, oid string, r io.Reader) (string, error) { if len(oid) < 4 { return "", errors.Newf("invalid oid %q", oid) diff --git a/internal/storage/filesystem_test.go b/lfs/storage/filesystem_test.go similarity index 100% rename from internal/storage/filesystem_test.go rename to lfs/storage/filesystem_test.go diff --git a/internal/storage/s3_presign.go b/lfs/storage/s3_presign.go similarity index 100% rename from internal/storage/s3_presign.go rename to lfs/storage/s3_presign.go diff --git a/internal/storage/s3_presign_test.go b/lfs/storage/s3_presign_test.go similarity index 100% rename from internal/storage/s3_presign_test.go rename to lfs/storage/s3_presign_test.go diff --git a/internal/storage/storage.go b/lfs/storage/storage.go similarity index 97% rename from internal/storage/storage.go rename to lfs/storage/storage.go index bdc8bf3..6bb95ae 100644 --- a/internal/storage/storage.go +++ b/lfs/storage/storage.go @@ -20,6 +20,10 @@ type Backend interface { Type() Type // Scheme returns the URI scheme prefix (including "://") that this backend owns. Scheme() string + // URI returns the canonical, non-expiring location of the object. Pure + // function of the OID and the backend's static config (e.g. bucket name), + // safe to compute without I/O. + URI(oid string) string // Delete removes the object at uri. Returns nil if already absent. Delete(ctx context.Context, uri string) error } @@ -40,6 +44,7 @@ type Proxier interface { // Put stores the bytes from the reader under the given OID and returns a URI // that uniquely identifies the stored object, including the backend scheme. + // It may return without reading r when the object is already stored. Put(ctx context.Context, oid string, r io.Reader) (uri string, err error) // Open returns a reader for the object at URI. It returns ErrNotFound when no // matching object exists. The caller MUST close the reader. @@ -51,10 +56,6 @@ type Proxier interface { type Presigner interface { Backend - // URI returns the canonical, non-expiring location of the object. Pure - // function of the OID and the backend's static config (e.g. bucket name), - // safe to compute without I/O. - URI(oid string) string // PresignPut returns a short-lived URL the client PUTs to, along with the // headers the client MUST send verbatim. PresignPut(ctx context.Context, oid string, size int64) (url string, headers map[string]string, err error) From 605bed21c584b633cafa5277d05f69e8d5369f41 Mon Sep 17 00:00:00 2001 From: Joe Chen Date: Wed, 11 Nov 2026 14:47:50 +0000 Subject: [PATCH 2/3] lfs: close downloaded content in proxy test Windows cannot remove the test's temp dir while the object file is open. --- lfs/service_test.go | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/lfs/service_test.go b/lfs/service_test.go index bfe39f1..830bcad 100644 --- a/lfs/service_test.go +++ b/lfs/service_test.go @@ -180,8 +180,9 @@ func TestService_Proxy(t *testing.T) { _, err = s.Upload(ctx, discardLogger, "repo2", oid, size, strings.NewReader(content)) require.NoError(t, err) - _, err = s.Download(ctx, "repo2", oid) + download, err := s.Download(ctx, "repo2", oid) require.NoError(t, err) + require.NoError(t, download.Content.Close()) }) t.Run("upload rejects invalid input", func(t *testing.T) { From f7d816405b4a3394b459013b4b44620a0ed3f3a4 Mon Sep 17 00:00:00 2001 From: Joe Chen Date: Wed, 11 Nov 2026 15:40:24 +0000 Subject: [PATCH 3/3] lfs: serve the LFS HTTP endpoints from the module Add Service.Serve, which handles the batch, upload, download, and verify requests of a repository given an Access that the caller resolved after authenticating the request. It dispatches on the method and the last path segment, so it works with any router that runs net/http handlers and any URL layout. lfsd now only maps its forge permissions to an Access and hands requests to it. Upload actions now set Content-Type to application/octet-stream so the Git LFS client does not send a type detected from the file content. --- README.md | 16 +++- cmd/lfsd/auth.go | 13 +-- cmd/lfsd/lfs.go | 197 +++-------------------------------------- cmd/lfsd/main.go | 11 +-- lfs/http.go | 209 ++++++++++++++++++++++++++++++++++++++++++++ lfs/http_test.go | 132 ++++++++++++++++++++++++++++ lfs/service.go | 11 ++- lfs/service_test.go | 1 + 8 files changed, 388 insertions(+), 202 deletions(-) create mode 100644 lfs/http.go create mode 100644 lfs/http_test.go diff --git a/README.md b/README.md index ac1d589..dcca796 100644 --- a/README.md +++ b/README.md @@ -98,7 +98,21 @@ The allowlist is enforced before the forge token check, so a request for a disal The `unknwon.dev/git-lfs-server/lfs` module implements the Batch API and the basic transfer adapter for other Git servers to embed. It is a separate Go module so that importers do not pull in the dependencies of lfsd (PostgreSQL, Flamego, forge clients). -An `lfs.Service` combines an `lfs.Store`, which records objects and the repositories they are linked to, with a `storage.Backend`, which holds the content. The caller keeps its own routing, authentication, authorization, and response encoding, identifies repositories with any string it likes, and calls `Batch`, `Upload`, `Download`, and `Verify` from its handlers. A `storage.Proxier` backend (e.g., `storage.FilesystemBackend`) streams content through the caller, and a `storage.Presigner` backend (e.g., `storage.S3PresignBackend`) redirects clients to the storage. `PreviousStorages` keeps objects readable after the backend for new uploads changes. +An `lfs.Service` combines an `lfs.Store`, which records objects and the repositories they are linked to, with a `storage.Backend`, which holds the content. A `storage.Proxier` backend (e.g., `storage.FilesystemBackend`) streams content through the server, and a `storage.Presigner` backend (e.g., `storage.S3PresignBackend`) redirects clients to the storage. `PreviousStorages` keeps objects readable after the backend for new uploads changes. + +`Service.Serve` handles the HTTP requests of the batch, upload, download, and verify endpoints. The caller routes the requests, authenticates them, and passes an `lfs.Access` with the repository, the base URL of its endpoints, and whether the actor can read and write. `Serve` works with any router that can run a `net/http` handler, e.g., Flamego, Macaron, or `http.ServeMux`: + +```go +mux.HandleFunc("/{repo}/info/lfs/objects/", func(w http.ResponseWriter, r *http.Request) { + access, ok := authorize(w, r) // Your authentication and authorization. + if !ok { + return + } + service.Serve(logger, w, r, access) +}) +``` + +The lower-level `Batch`, `Upload`, `Download`, and `Verify` methods are available for servers that need their own HTTP handling. ## Local development diff --git a/cmd/lfsd/auth.go b/cmd/lfsd/auth.go index 9d1dbeb..2181d82 100644 --- a/cmd/lfsd/auth.go +++ b/cmd/lfsd/auth.go @@ -15,6 +15,7 @@ import ( "unknwon.dev/git-lfs-server/internal/forge" "unknwon.dev/git-lfs-server/internal/logx" + "unknwon.dev/git-lfs-server/lfs" ) func authorize(forges map[string]forge.Provider) flamego.Handler { @@ -25,26 +26,26 @@ func authorize(forges map[string]forge.Provider) flamego.Handler { host := hostFromContext(c) provider, ok := forges[host] if !ok { - writeLFSError(c.ResponseWriter(), http.StatusBadRequest, "unsupported forge host") + lfs.WriteError(c.ResponseWriter(), http.StatusBadRequest, "unsupported forge host") return } username, token, ok := c.Request().BasicAuth() if !ok || token == "" { c.ResponseWriter().Header().Set("WWW-Authenticate", `Basic realm="git-lfs"`) - writeLFSError(c.ResponseWriter(), http.StatusUnauthorized, "basic auth credentials are required") + lfs.WriteError(c.ResponseWriter(), http.StatusUnauthorized, "basic auth credentials are required") return } repo := strings.ToLower(c.Param("**")) if repo == "" { - writeLFSError(c.ResponseWriter(), http.StatusBadRequest, "repository path is missing") + lfs.WriteError(c.ResponseWriter(), http.StatusBadRequest, "repository path is missing") return } if !provider.Allow(repo) { logger.InfoContext(ctx, "Repository rejected by allowlist", "host", host, "repo", repo) - writeLFSError(c.ResponseWriter(), http.StatusForbidden, "repository is not in the allowlist") + lfs.WriteError(c.ResponseWriter(), http.StatusForbidden, "repository is not in the allowlist") return } @@ -62,10 +63,10 @@ func authorize(forges map[string]forge.Provider) flamego.Handler { perm, rawTTL, err := provider.Authorize(ctx, logger.Scoped("forge"), repo, username, token) if err != nil { if errors.Is(err, forge.ErrTokenInvalid) { - writeLFSError(c.ResponseWriter(), http.StatusForbidden, "token is invalid or lacks repository access") + lfs.WriteError(c.ResponseWriter(), http.StatusForbidden, "token is invalid or lacks repository access") } else { logger.ErrorContext(ctx, "Failed to verify repository permissions", "repo", repo, "error", err) - writeLFSError(c.ResponseWriter(), http.StatusInternalServerError, http.StatusText(http.StatusInternalServerError)) + lfs.WriteError(c.ResponseWriter(), http.StatusInternalServerError, http.StatusText(http.StatusInternalServerError)) } return } diff --git a/cmd/lfsd/lfs.go b/cmd/lfsd/lfs.go index c93842a..5f2f8d3 100644 --- a/cmd/lfsd/lfs.go +++ b/cmd/lfsd/lfs.go @@ -1,14 +1,9 @@ package main import ( - "context" - "encoding/json" "fmt" - "io" "net/http" - "strconv" - "github.com/cockroachdb/errors" "github.com/flamego/flamego" "unknwon.dev/git-lfs-server/internal/forge" @@ -16,193 +11,23 @@ import ( "unknwon.dev/git-lfs-server/lfs" ) -const ( - mediaTypeLFS = "application/vnd.git-lfs+json" - - // maxJSONRequestBytes caps JSON request bodies (batch, verify) to prevent - // authenticated callers from exhausting memory with oversized payloads. - maxJSONRequestBytes = 10 << 20 -) - -// mapService maps the *lfs.Service of the forge host to the request context. -func mapService(services map[string]*lfs.Service) flamego.Handler { - return func(c flamego.Context, logger *logx.Logger) { +// serveLFS hands the request to the *lfs.Service of the forge host, with the +// permission resolved by the authorize middleware. +func serveLFS(externalURL string, services map[string]*lfs.Service) flamego.Handler { + return func(c flamego.Context, logger *logx.Logger, perm forge.Permission) { service, ok := services[hostFromContext(c)] if !ok { logger.ErrorContext(c.Request().Context(), "No storage backend configured for host") - writeLFSError(c.ResponseWriter(), http.StatusInternalServerError, http.StatusText(http.StatusInternalServerError)) - return - } - c.Map(service) - } -} - -func serveBatch(externalURL string) flamego.Handler { - return func(c flamego.Context, logger *logx.Logger, perm forge.Permission, service *lfs.Service) { - logger = logger.Scoped("batch") - ctx := c.Request().Context() - - body := http.MaxBytesReader(c.ResponseWriter(), c.Request().Body().ReadCloser(), maxJSONRequestBytes) - defer func() { _ = body.Close() }() - - var req lfs.BatchRequest - if err := json.NewDecoder(body).Decode(&req); err != nil { - writeLFSError(c.ResponseWriter(), http.StatusBadRequest, fmt.Sprintf("invalid JSON body: %v", err)) - return - } - - switch req.Operation { - case lfs.OperationUpload: - if perm != forge.PermissionWrite { - writeLFSError(c.ResponseWriter(), http.StatusForbidden, "write permission required for upload") - return - } - case lfs.OperationDownload: - if perm != forge.PermissionRead && perm != forge.PermissionWrite { - writeLFSError(c.ResponseWriter(), http.StatusForbidden, "read permission required for download") - return - } - default: - writeLFSError(c.ResponseWriter(), http.StatusBadRequest, `operation must be "upload" or "download"`) + lfs.WriteError(c.ResponseWriter(), http.StatusInternalServerError, http.StatusText(http.StatusInternalServerError)) return } repoName := repoNameFromContext(c) - endpoint := lfs.Endpoint{ - Href: fmt.Sprintf("%s/%s/info/lfs/objects", externalURL, repoName), - // Every request passes the authorize middleware with Basic Auth, so - // forwarding the header authenticates the actions. - Authenticated: true, - } - if authz := c.Request().Header.Get("Authorization"); authz != "" { - endpoint.Header = map[string]string{"Authorization": authz} - } - - resp, err := service.Batch(ctx, logger.Logger, repoName, endpoint, req) - if err != nil { - writeServiceError(ctx, logger, c.ResponseWriter(), err, "Failed to process batch", "failed to process batch") - return - } - - c.ResponseWriter().Header().Set("Content-Type", mediaTypeLFS) - if err := json.NewEncoder(c.ResponseWriter()).Encode(resp); err != nil { - logger.ErrorContext(ctx, "Failed to encode batch response", "error", err) - } - } -} - -func serveUpload(c flamego.Context, logger *logx.Logger, perm forge.Permission, service *lfs.Service) { - logger = logger.Scoped("upload") - ctx := c.Request().Context() - - if perm != forge.PermissionWrite { - writeLFSError(c.ResponseWriter(), http.StatusForbidden, "write permission required for upload") - return - } - - body := c.Request().Body().ReadCloser() - defer func() { _ = body.Close() }() - - oid := c.Param("oid") - result, err := service.Upload(ctx, logger.Logger, repoNameFromContext(c), oid, c.Request().ContentLength, body) - if err != nil { - writeServiceError(ctx, logger.With("oid", oid), c.ResponseWriter(), err, "Failed to upload object", http.StatusText(http.StatusInternalServerError)) - return + service.Serve(logger.Scoped("lfs").Logger, c.ResponseWriter(), c.Request().Request, lfs.Access{ + Repo: repoName, + Href: fmt.Sprintf("%s/%s/info/lfs/objects", externalURL, repoName), + CanRead: perm == forge.PermissionRead || perm == forge.PermissionWrite, + CanWrite: perm == forge.PermissionWrite, + }) } - - w := c.ResponseWriter() - switch { - case result.Redirect != nil: - for k, v := range result.Redirect.Header { - w.Header().Set(k, v) - } - w.Header().Set("Location", result.Redirect.URL) - w.WriteHeader(http.StatusTemporaryRedirect) - case result.Existed: - w.WriteHeader(http.StatusOK) - default: - w.WriteHeader(http.StatusNoContent) - } -} - -func serveDownload(c flamego.Context, logger *logx.Logger, perm forge.Permission, service *lfs.Service) { - logger = logger.Scoped("download") - ctx := c.Request().Context() - - if perm != forge.PermissionRead && perm != forge.PermissionWrite { - writeLFSError(c.ResponseWriter(), http.StatusForbidden, "read permission required for download") - return - } - - oid := c.Param("oid") - result, err := service.Download(ctx, repoNameFromContext(c), oid) - if err != nil { - writeServiceError(ctx, logger.With("oid", oid), c.ResponseWriter(), err, "Failed to download object", http.StatusText(http.StatusInternalServerError)) - return - } - - w := c.ResponseWriter() - if result.Redirect != nil { - w.Header().Set("Location", result.Redirect.URL) - w.WriteHeader(http.StatusTemporaryRedirect) - return - } - - defer func() { _ = result.Content.Close() }() - w.Header().Set("Content-Type", "application/octet-stream") - w.Header().Set("Content-Length", strconv.FormatInt(result.Size, 10)) - w.WriteHeader(http.StatusOK) - if _, err = io.Copy(w, result.Content); err != nil { - logger.ErrorContext(ctx, "Failed to stream object", "oid", oid, "error", err) - } -} - -func serveVerify(c flamego.Context, logger *logx.Logger, perm forge.Permission, service *lfs.Service) { - logger = logger.Scoped("verify") - ctx := c.Request().Context() - - if perm != forge.PermissionWrite { - writeLFSError(c.ResponseWriter(), http.StatusForbidden, "write permission required for verify") - return - } - - body := http.MaxBytesReader(c.ResponseWriter(), c.Request().Body().ReadCloser(), maxJSONRequestBytes) - defer func() { _ = body.Close() }() - - var req lfs.VerifyRequest - if err := json.NewDecoder(body).Decode(&req); err != nil { - writeLFSError(c.ResponseWriter(), http.StatusBadRequest, fmt.Sprintf("invalid JSON body: %v", err)) - return - } - - if err := service.Verify(ctx, logger.Logger, repoNameFromContext(c), req); err != nil { - writeServiceError(ctx, logger.With("oid", req.OID), c.ResponseWriter(), err, "Failed to verify object", http.StatusText(http.StatusInternalServerError)) - return - } - c.ResponseWriter().WriteHeader(http.StatusNoContent) -} - -// writeServiceError responds with the client error carried by err as an -// *lfs.Error. Any other error is internal, so it is logged with logMessage and -// the response only carries internalMessage. -func writeServiceError(ctx context.Context, logger *logx.Logger, w http.ResponseWriter, err error, logMessage, internalMessage string) { - var lfsErr *lfs.Error - if errors.As(err, &lfsErr) { - writeLFSError(w, lfsErr.Code, lfsErr.Message) - return - } - logger.ErrorContext(ctx, logMessage, "error", err) - writeLFSError(w, http.StatusInternalServerError, internalMessage) -} - -// writeLFSError writes an error response in the format required by the LFS -// Batch API spec: a JSON body of {"message": "..."} with the -// application/vnd.git-lfs+json content type. All LFS endpoints (batch, upload, -// download, verify, and the auth middleware that fronts them) must use this -// helper instead of http.Error so clients get a parseable body and the right -// content type on every error. -func writeLFSError(w http.ResponseWriter, code int, msg string) { - w.Header().Set("Content-Type", mediaTypeLFS) - w.WriteHeader(code) - _ = json.NewEncoder(w).Encode(map[string]string{"message": msg}) } diff --git a/cmd/lfsd/main.go b/cmd/lfsd/main.go index b071bd9..dc8cc0d 100644 --- a/cmd/lfsd/main.go +++ b/cmd/lfsd/main.go @@ -70,12 +70,13 @@ func main() { CrossRepoDedup: true, } } + serve := serveLFS(config.Server.ExternalURL, services) f.Group("/{host}/{**}/info/lfs/objects", func() { - f.Post("/batch", serveBatch(config.Server.ExternalURL)) - f.Get("/{oid}", serveDownload) - f.Put("/{oid}", serveUpload) - f.Post("/verify", serveVerify) - }, authorize(config.Forges), mapService(services)) + f.Post("/batch", serve) + f.Get("/{oid}", serve) + f.Put("/{oid}", serve) + f.Post("/verify", serve) + }, authorize(config.Forges)) ctx, stop := signal.NotifyContext(ctx, syscall.SIGINT, syscall.SIGTERM) defer stop() diff --git a/lfs/http.go b/lfs/http.go new file mode 100644 index 0000000..2a7e674 --- /dev/null +++ b/lfs/http.go @@ -0,0 +1,209 @@ +package lfs + +import ( + "encoding/json" + "fmt" + "io" + "log/slog" + "net/http" + "path" + "strconv" + + "github.com/cockroachdb/errors" +) + +const ( + // MediaType is the content type of LFS API requests and responses. + MediaType = "application/vnd.git-lfs+json" + + // maxJSONRequestBytes caps JSON request bodies (batch, verify) to prevent + // authenticated callers from exhausting memory with oversized payloads. + maxJSONRequestBytes = 10 << 20 +) + +// Access is what the caller resolved for a request after authenticating it. +type Access struct { + // Repo identifies the repository to the Store. + Repo string + // Href is the URL that the transfer endpoints of the repository are served + // under, without a trailing slash: objects at Href/{oid} and verifications + // at Href/verify. + Href string + // CanRead allows downloads. + CanRead bool + // CanWrite allows uploads and verifications. + CanWrite bool +} + +// Serve handles an LFS API request to the repository in access. It dispatches +// on the method and the last segment of the URL path, so it serves any URL +// layout whose endpoints end in "batch", "verify", and the OID: +// +// - POST .../batch +// - GET .../{oid} +// - PUT .../{oid} +// - POST .../verify +// +// The request's Authorization header is forwarded in the actions of batch +// responses, so the client authenticates transfers with the same credentials. +func (s *Service) Serve(logger *slog.Logger, w http.ResponseWriter, r *http.Request, access Access) { + name := path.Base(r.URL.Path) + switch { + case r.Method == http.MethodPost && name == "batch": + s.serveBatch(logger, w, r, access) + case r.Method == http.MethodPost && name == "verify": + s.serveVerify(logger, w, r, access) + case r.Method == http.MethodGet: + s.serveDownload(logger, w, r, access, name) + case r.Method == http.MethodPut: + s.serveUpload(logger, w, r, access, name) + default: + WriteError(w, http.StatusMethodNotAllowed, http.StatusText(http.StatusMethodNotAllowed)) + } +} + +func (s *Service) serveBatch(logger *slog.Logger, w http.ResponseWriter, r *http.Request, access Access) { + ctx := r.Context() + + body := http.MaxBytesReader(w, r.Body, maxJSONRequestBytes) + defer func() { _ = body.Close() }() + + var req BatchRequest + if err := json.NewDecoder(body).Decode(&req); err != nil { + WriteError(w, http.StatusBadRequest, fmt.Sprintf("invalid JSON body: %v", err)) + return + } + + switch req.Operation { + case OperationUpload: + if !access.CanWrite { + WriteError(w, http.StatusForbidden, "write permission required for upload") + return + } + case OperationDownload: + if !access.CanRead { + WriteError(w, http.StatusForbidden, "read permission required for download") + return + } + default: + WriteError(w, http.StatusBadRequest, `operation must be "upload" or "download"`) + return + } + + endpoint := Endpoint{Href: access.Href} + if authz := r.Header.Get("Authorization"); authz != "" { + endpoint.Header = map[string]string{"Authorization": authz} + endpoint.Authenticated = true + } + resp, err := s.Batch(ctx, logger, access.Repo, endpoint, req) + if err != nil { + writeServiceError(logger, w, r, err, "failed to process batch") + return + } + + w.Header().Set("Content-Type", MediaType) + if err := json.NewEncoder(w).Encode(resp); err != nil { + logger.ErrorContext(ctx, "Failed to encode batch response", "error", err) + } +} + +func (s *Service) serveUpload(logger *slog.Logger, w http.ResponseWriter, r *http.Request, access Access, oid string) { + if !access.CanWrite { + WriteError(w, http.StatusForbidden, "write permission required for upload") + return + } + + defer func() { _ = r.Body.Close() }() + result, err := s.Upload(r.Context(), logger, access.Repo, oid, r.ContentLength, r.Body) + if err != nil { + writeServiceError(logger.With("oid", oid), w, r, err, http.StatusText(http.StatusInternalServerError)) + return + } + + switch { + case result.Redirect != nil: + writeRedirect(w, result.Redirect) + case result.Existed: + w.WriteHeader(http.StatusOK) + default: + w.WriteHeader(http.StatusNoContent) + } +} + +func (s *Service) serveDownload(logger *slog.Logger, w http.ResponseWriter, r *http.Request, access Access, oid string) { + if !access.CanRead { + WriteError(w, http.StatusForbidden, "read permission required for download") + return + } + + result, err := s.Download(r.Context(), access.Repo, oid) + if err != nil { + writeServiceError(logger.With("oid", oid), w, r, err, http.StatusText(http.StatusInternalServerError)) + return + } + if result.Redirect != nil { + writeRedirect(w, result.Redirect) + return + } + + defer func() { _ = result.Content.Close() }() + w.Header().Set("Content-Type", "application/octet-stream") + w.Header().Set("Content-Length", strconv.FormatInt(result.Size, 10)) + w.WriteHeader(http.StatusOK) + if _, err = io.Copy(w, result.Content); err != nil { + logger.ErrorContext(r.Context(), "Failed to stream object", "oid", oid, "error", err) + } +} + +func (s *Service) serveVerify(logger *slog.Logger, w http.ResponseWriter, r *http.Request, access Access) { + if !access.CanWrite { + WriteError(w, http.StatusForbidden, "write permission required for verify") + return + } + + body := http.MaxBytesReader(w, r.Body, maxJSONRequestBytes) + defer func() { _ = body.Close() }() + + var req VerifyRequest + if err := json.NewDecoder(body).Decode(&req); err != nil { + WriteError(w, http.StatusBadRequest, fmt.Sprintf("invalid JSON body: %v", err)) + return + } + + if err := s.Verify(r.Context(), logger, access.Repo, req); err != nil { + writeServiceError(logger.With("oid", req.OID), w, r, err, http.StatusText(http.StatusInternalServerError)) + return + } + w.WriteHeader(http.StatusNoContent) +} + +func writeRedirect(w http.ResponseWriter, redirect *Redirect) { + for k, v := range redirect.Header { + w.Header().Set(k, v) + } + w.Header().Set("Location", redirect.URL) + w.WriteHeader(http.StatusTemporaryRedirect) +} + +// writeServiceError responds with the client error carried by err as an +// *Error. Any other error is internal, so it is logged and the response only +// carries internalMessage. +func writeServiceError(logger *slog.Logger, w http.ResponseWriter, r *http.Request, err error, internalMessage string) { + var lfsErr *Error + if errors.As(err, &lfsErr) { + WriteError(w, lfsErr.Code, lfsErr.Message) + return + } + logger.ErrorContext(r.Context(), "Failed to serve LFS request", "path", r.URL.Path, "error", err) + WriteError(w, http.StatusInternalServerError, internalMessage) +} + +// WriteError writes an error response in the format required by the LFS API: a +// JSON body of {"message": "..."} with the MediaType content type. Callers use +// it for the errors of their own middleware, e.g., authentication, so clients +// get a parseable body on every error. +func WriteError(w http.ResponseWriter, code int, msg string) { + w.Header().Set("Content-Type", MediaType) + w.WriteHeader(code) + _ = json.NewEncoder(w).Encode(map[string]string{"message": msg}) +} diff --git a/lfs/http_test.go b/lfs/http_test.go new file mode 100644 index 0000000..22cbcea --- /dev/null +++ b/lfs/http_test.go @@ -0,0 +1,132 @@ +package lfs + +import ( + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "strconv" + "strings" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestService_Serve(t *testing.T) { + content := "served content" + oid := sha256Hex(content) + size := len(content) + const href = "https://lfs.example.com/repo/info/lfs/objects" + readWrite := Access{Repo: "repo", Href: href, CanRead: true, CanWrite: true} + + serve := func(s *Service, access Access, method, target, body string) *httptest.ResponseRecorder { + r := httptest.NewRequest(method, href+target, strings.NewReader(body)) + r.SetBasicAuth("user", "token") + w := httptest.NewRecorder() + s.Serve(discardLogger, w, r, access) + return w + } + requireError := func(t *testing.T, w *httptest.ResponseRecorder, wantCode int) { + t.Helper() + require.Equal(t, wantCode, w.Code, w.Body.String()) + assert.Equal(t, MediaType, w.Header().Get("Content-Type")) + var body struct{ Message string } + require.NoError(t, json.Unmarshal(w.Body.Bytes(), &body)) + assert.NotEmpty(t, body.Message) + } + batch := func(operation string) string { + return `{"operation":"` + operation + `","objects":[{"oid":"` + oid + `","size":` + strconv.Itoa(size) + `}]}` + } + verify := `{"oid":"` + oid + `","size":` + strconv.Itoa(size) + `}` + + t.Run("proxied transfer", func(t *testing.T) { + s := &Service{Store: newMemStore(), Storage: newFilesystemBackend(t, "fs", "file://")} + + w := serve(s, readWrite, http.MethodPost, "/batch", batch(OperationUpload)) + require.Equal(t, http.StatusOK, w.Code, w.Body.String()) + assert.Equal(t, MediaType, w.Header().Get("Content-Type")) + var resp BatchResponse + require.NoError(t, json.Unmarshal(w.Body.Bytes(), &resp)) + require.Len(t, resp.Objects, 1) + authz := "Basic dXNlcjp0b2tlbg==" + assert.True(t, resp.Objects[0].Authenticated) + assert.Equal(t, Action{ + Href: href + "/" + oid, + Header: map[string]string{ + "Authorization": authz, + "Content-Type": "application/octet-stream", + "x-amz-checksum-sha256": "97OiZ/pYWtQ/MOiPfTFzQY0nD/6zXoSoWwc00FhqN48=", + }, + }, resp.Objects[0].Actions["upload"]) + assert.Equal(t, Action{Href: href + "/verify", Header: map[string]string{"Authorization": authz}}, resp.Objects[0].Actions["verify"]) + + requireError(t, serve(s, readWrite, http.MethodPut, "/"+oid, strings.Repeat("x", size)), http.StatusUnprocessableEntity) + requireError(t, serve(s, readWrite, http.MethodPost, "/verify", verify), http.StatusNotFound) + + w = serve(s, readWrite, http.MethodPut, "/"+oid, content) + assert.Equal(t, http.StatusNoContent, w.Code, w.Body.String()) + w = serve(s, readWrite, http.MethodPost, "/verify", verify) + assert.Equal(t, http.StatusNoContent, w.Code, w.Body.String()) + + w = serve(s, readWrite, http.MethodGet, "/"+oid, "") + require.Equal(t, http.StatusOK, w.Code, w.Body.String()) + assert.Equal(t, "application/octet-stream", w.Header().Get("Content-Type")) + assert.Equal(t, strconv.Itoa(size), w.Header().Get("Content-Length")) + assert.Equal(t, content, w.Body.String()) + }) + + t.Run("presigned transfer redirects", func(t *testing.T) { + s := &Service{Store: newMemStore(), Storage: newS3PresignBackend(t)} + + w := serve(s, readWrite, http.MethodPut, "/"+oid, "") + require.Equal(t, http.StatusTemporaryRedirect, w.Code, w.Body.String()) + assert.Equal(t, "97OiZ/pYWtQ/MOiPfTFzQY0nD/6zXoSoWwc00FhqN48=", w.Header().Get("x-amz-checksum-sha256")) + + req, err := http.NewRequestWithContext(t.Context(), http.MethodPut, w.Header().Get("Location"), strings.NewReader(content)) + require.NoError(t, err) + req.Header.Set("x-amz-checksum-sha256", w.Header().Get("x-amz-checksum-sha256")) + resp, err := http.DefaultClient.Do(req) + require.NoError(t, err) + require.NoError(t, resp.Body.Close()) + require.Equal(t, http.StatusOK, resp.StatusCode) + + w = serve(s, readWrite, http.MethodPost, "/verify", verify) + require.Equal(t, http.StatusNoContent, w.Code, w.Body.String()) + + w = serve(s, readWrite, http.MethodGet, "/"+oid, "") + require.Equal(t, http.StatusTemporaryRedirect, w.Code, w.Body.String()) + resp, err = http.Get(w.Header().Get("Location")) + require.NoError(t, err) + got, err := io.ReadAll(resp.Body) + require.NoError(t, resp.Body.Close()) + require.NoError(t, err) + assert.Equal(t, content, string(got)) + }) + + t.Run("checks permissions", func(t *testing.T) { + s := &Service{Store: newMemStore(), Storage: newFilesystemBackend(t, "fs", "file://")} + readOnly := Access{Repo: "repo", Href: href, CanRead: true} + requireError(t, serve(s, readOnly, http.MethodPost, "/batch", batch(OperationUpload)), http.StatusForbidden) + requireError(t, serve(s, readOnly, http.MethodPut, "/"+oid, content), http.StatusForbidden) + requireError(t, serve(s, readOnly, http.MethodPost, "/verify", verify), http.StatusForbidden) + assert.Equal(t, http.StatusOK, serve(s, readOnly, http.MethodPost, "/batch", batch(OperationDownload)).Code) + + writeOnly := Access{Repo: "repo", Href: href, CanWrite: true} + requireError(t, serve(s, writeOnly, http.MethodPost, "/batch", batch(OperationDownload)), http.StatusForbidden) + requireError(t, serve(s, writeOnly, http.MethodGet, "/"+oid, ""), http.StatusForbidden) + + _, err := s.Upload(t.Context(), discardLogger, "repo", oid, int64(size), strings.NewReader(content)) + require.NoError(t, err) + assert.Equal(t, http.StatusOK, serve(s, readOnly, http.MethodGet, "/"+oid, "").Code) + }) + + t.Run("rejects invalid requests", func(t *testing.T) { + s := &Service{Store: newMemStore(), Storage: newFilesystemBackend(t, "fs", "file://")} + requireError(t, serve(s, readWrite, http.MethodPost, "/batch", "{"), http.StatusBadRequest) + requireError(t, serve(s, readWrite, http.MethodPost, "/batch", `{"operation":"delete"}`), http.StatusBadRequest) + requireError(t, serve(s, readWrite, http.MethodPost, "/verify", "{"), http.StatusBadRequest) + requireError(t, serve(s, readWrite, http.MethodGet, "/not-an-oid", ""), http.StatusBadRequest) + requireError(t, serve(s, readWrite, http.MethodDelete, "/"+oid, ""), http.StatusMethodNotAllowed) + }) +} diff --git a/lfs/service.go b/lfs/service.go index 6bfd0e5..d01cde8 100644 --- a/lfs/service.go +++ b/lfs/service.go @@ -19,9 +19,9 @@ import ( ) // Service serves LFS operations for repositories. Callers authenticate and -// authorize each request before calling a method, and respond with the Code -// and Message of a returned *Error. Any other returned error is an internal -// error. +// authorize each request, then either hand it to Serve, or call the operation +// methods and respond with the Code and Message of a returned *Error. Any other +// error returned by an operation method is an internal error. type Service struct { Store Store // Storage receives uploads and decides the transfer mode: content is @@ -138,8 +138,11 @@ func (s *Service) Batch(ctx context.Context, logger *slog.Logger, repo string, e out.Error = &ObjectError{Code: http.StatusConflict, Message: "OID exists with different size"} default: rawOID, _ := hex.DecodeString(in.OID) // Validated above. - uploadHeader := make(map[string]string, len(endpoint.Header)+1) + uploadHeader := make(map[string]string, len(endpoint.Header)+2) maps.Copy(uploadHeader, endpoint.Header) + // The Git LFS client otherwise sends a Content-Type detected from + // the file content. + uploadHeader["Content-Type"] = "application/octet-stream" // Presigned uploads require the client to send the checksum that // the storage verifies the content against. uploadHeader["x-amz-checksum-sha256"] = base64.StdEncoding.EncodeToString(rawOID) diff --git a/lfs/service_test.go b/lfs/service_test.go index 830bcad..e3326e7 100644 --- a/lfs/service_test.go +++ b/lfs/service_test.go @@ -290,6 +290,7 @@ func TestService_Batch(t *testing.T) { Href: endpoint.Href + "/" + newOID, Header: map[string]string{ "Authorization": "Basic Zm9vOmJhcg==", + "Content-Type": "application/octet-stream", "x-amz-checksum-sha256": "/jJgjJ71ts9+P5RkgCU/928k9OwGePPQ8H+YRMv/lgE=", }, },