Files
Gitea/services/lfs/server.go
T
hsdfat 7ca64566f5 fix(lfs): failed upload deletes a concurrent upload's meta object (#38693)
UploadHandler creates the LFS meta object only as the last step of
uploadOrVerify, after the content is already in the store, so a request
that errors has never created a row of its own. The removal on the error
path was a real compensating action when it was added in #14726, where
the meta object was created before contentStore.Put, but #16865 moved
creation after the Put and left the removal behind.

Since then it can only ever delete a row created by a different request:
a stalled git-lfs PUT that fails after its own retry has already
succeeded wipes the winner's meta object, leaving the content in the
store unreachable and eligible for orphan cleanup.

Drop the removal and add a regression test asserting that a failing
upload keeps a pre-existing meta object.

Fixes: #38424
Assisted-by: Claude Code:claude-opus-5
Co-authored-by: wxiaoguang <wxiaoguang@gmail.com>
2026-07-31 11:05:41 +00:00

642 lines
21 KiB
Go

// Copyright 2021 The Gitea Authors. All rights reserved.
// SPDX-License-Identifier: MIT
package lfs
import (
stdCtx "context"
"crypto/sha256"
"encoding/base64"
"encoding/hex"
"errors"
"fmt"
"io"
"io/fs"
"net/http"
"net/url"
"regexp"
"strconv"
"strings"
"time"
auth_model "gitea.dev/models/auth"
git_model "gitea.dev/models/git"
perm_model "gitea.dev/models/perm"
access_model "gitea.dev/models/perm/access"
repo_model "gitea.dev/models/repo"
"gitea.dev/models/unit"
user_model "gitea.dev/models/user"
"gitea.dev/modules/auth/httpauth"
"gitea.dev/modules/httplib"
"gitea.dev/modules/json"
lfs_module "gitea.dev/modules/lfs"
"gitea.dev/modules/log"
"gitea.dev/modules/setting"
"gitea.dev/modules/storage"
"gitea.dev/modules/util"
"gitea.dev/services/context"
"github.com/golang-jwt/jwt/v5"
)
// requestContext contain variables from the HTTP request.
type requestContext struct {
User string
Repo string
Authorization string
RepoGitURL string
}
// Claims is a JWT Token Claims
type Claims struct {
RepoID int64
Op string
UserID int64
jwt.RegisteredClaims
}
type AuthTokenOptions struct {
Op string
UserID int64
RepoID int64
}
func GetLFSAuthTokenWithBearer(opts AuthTokenOptions) (string, error) {
now := time.Now()
claims := Claims{
RegisteredClaims: jwt.RegisteredClaims{
ExpiresAt: jwt.NewNumericDate(now.Add(setting.LFS.HTTPAuthExpiry)),
NotBefore: jwt.NewNumericDate(now),
},
RepoID: opts.RepoID,
Op: opts.Op,
UserID: opts.UserID,
}
token := jwt.NewWithClaims(jwt.SigningMethodHS256, claims)
// Sign and get the complete encoded token as a string using the secret
tokenString, err := token.SignedString(setting.LFS.JWTSecretBytes)
if err != nil {
return "", fmt.Errorf("failed to sign LFS JWT token: %w", err)
}
return "Bearer " + tokenString, nil
}
// DownloadLink builds a URL to download the object.
func (rc *requestContext) DownloadLink(p lfs_module.Pointer) string {
return rc.RepoGitURL + "/info/lfs/objects/" + url.PathEscape(p.Oid)
}
// UploadLink builds a URL to upload the object.
func (rc *requestContext) UploadLink(p lfs_module.Pointer) string {
return rc.RepoGitURL + "/info/lfs/objects/" + url.PathEscape(p.Oid) + "/" + strconv.FormatInt(p.Size, 10)
}
// VerifyLink builds a URL for verifying the object.
func (rc *requestContext) VerifyLink(p lfs_module.Pointer) string {
return rc.RepoGitURL + "/info/lfs/verify"
}
// CheckAcceptMediaType checks if the client accepts the LFS media type.
func CheckAcceptMediaType(ctx *context.Context) {
mediaParts := strings.Split(ctx.Req.Header.Get("Accept"), ";")
if mediaParts[0] != lfs_module.MediaType {
log.Trace("Calling a LFS method without accepting the correct media type: %s", lfs_module.MediaType)
writeStatus(ctx, http.StatusUnsupportedMediaType)
return
}
}
var rangeHeaderRegexp = regexp.MustCompile(`bytes=(\d+)-(\d*).*`)
// DownloadHandler gets the content from the content store
func DownloadHandler(ctx *context.Context) {
rc := getRequestContext(ctx)
p := lfs_module.Pointer{Oid: ctx.PathParam("oid")}
meta := getAuthenticatedMeta(ctx, rc, p, false)
if meta == nil {
return
}
// Support resume download using Range header
var fromByte, toByte int64
toByte = meta.Size - 1
statusCode := http.StatusOK
if rangeHdr := ctx.Req.Header.Get("Range"); rangeHdr != "" {
match := rangeHeaderRegexp.FindStringSubmatch(rangeHdr)
if len(match) > 1 {
statusCode = http.StatusPartialContent
fromByte, _ = strconv.ParseInt(match[1], 10, 32)
if fromByte >= meta.Size {
writeStatus(ctx, http.StatusRequestedRangeNotSatisfiable)
return
}
if match[2] != "" {
_toByte, _ := strconv.ParseInt(match[2], 10, 32)
if _toByte >= fromByte && _toByte < toByte {
toByte = _toByte
}
}
ctx.Resp.Header().Set("Content-Range", fmt.Sprintf("bytes %d-%d/%d", fromByte, toByte, meta.Size))
ctx.Resp.Header().Set("Access-Control-Expose-Headers", "Content-Range")
}
}
contentStore := lfs_module.NewContentStore()
content, err := contentStore.Get(meta.Pointer)
if err != nil {
writeStatus(ctx, http.StatusNotFound)
return
}
defer content.Close()
if fromByte > 0 {
_, err = content.Seek(fromByte, io.SeekStart)
if err != nil {
log.Error("Whilst trying to read LFS OID[%s]: Unable to seek to %d Error: %v", meta.Oid, fromByte, err)
writeStatus(ctx, http.StatusInternalServerError)
return
}
}
contentLength := toByte + 1 - fromByte
contentLengthStr := strconv.FormatInt(contentLength, 10)
ctx.Resp.Header().Set("Content-Length", contentLengthStr)
ctx.Resp.Header().Set("X-Gitea-LFS-Content-Length", contentLengthStr) // we need this header to make sure it won't be affected by reverse proxy or compression
ctx.Resp.Header().Set("Content-Type", "application/octet-stream")
filename := ctx.PathParam("filename")
if len(filename) > 0 {
decodedFilename, err := base64.RawURLEncoding.DecodeString(filename)
if err == nil {
ctx.Resp.Header().Set("Content-Disposition", httplib.EncodeContentDispositionAttachment(string(decodedFilename)))
ctx.Resp.Header().Set("Access-Control-Expose-Headers", "Content-Disposition")
}
}
ctx.Resp.WriteHeader(statusCode)
if written, err := io.CopyN(ctx.Resp, content, contentLength); err != nil {
log.Error("Error whilst copying LFS OID[%s] to the response after %d bytes. Error: %v", meta.Oid, written, err)
}
}
// BatchHandler provides the batch api
func BatchHandler(ctx *context.Context) {
var br lfs_module.BatchRequest
if err := decodeJSON(ctx.Req, &br); err != nil {
log.Trace("Unable to decode BATCH request vars: Error: %v", err)
writeStatus(ctx, http.StatusBadRequest)
return
}
var isUpload bool
switch br.Operation {
case "upload":
isUpload = true
case "download":
isUpload = false
default:
log.Trace("Attempt to BATCH with invalid operation: %s", br.Operation)
writeStatus(ctx, http.StatusBadRequest)
return
}
rc := getRequestContext(ctx)
repository := getAuthenticatedRepository(ctx, rc, isUpload)
if repository == nil {
return
}
if setting.LFS.MaxBatchSize != 0 && len(br.Objects) > setting.LFS.MaxBatchSize {
writeStatus(ctx, http.StatusRequestEntityTooLarge)
return
}
contentStore := lfs_module.NewContentStore()
var responseObjects []*lfs_module.ObjectResponse
for _, p := range br.Objects {
if !p.IsValid() {
responseObjects = append(responseObjects, buildObjectResponse(rc, p, false, false, &lfs_module.ObjectError{
Code: http.StatusUnprocessableEntity,
Message: "Oid or size are invalid",
}))
continue
}
exists, err := contentStore.Exists(p)
if err != nil {
log.Error("Unable to check if LFS object with ID '%s' exists for %s/%s. Error: %v", p.Oid, rc.User, rc.Repo, err)
writeStatus(ctx, http.StatusInternalServerError)
return
}
meta, err := git_model.GetLFSMetaObjectByOid(ctx, repository.ID, p.Oid)
if err != nil && err != git_model.ErrLFSObjectNotExist {
log.Error("Unable to get LFS MetaObject [%s] for %s/%s. Error: %v", p.Oid, rc.User, rc.Repo, err)
writeStatus(ctx, http.StatusInternalServerError)
return
}
if meta != nil && p.Size != meta.Size {
responseObjects = append(responseObjects, buildObjectResponse(rc, p, false, false, &lfs_module.ObjectError{
Code: http.StatusUnprocessableEntity,
Message: fmt.Sprintf("Object %s is not %d bytes", p.Oid, p.Size),
}))
continue
}
var responseObject *lfs_module.ObjectResponse
if isUpload {
if exists && meta == nil {
// The object exists in the content store but is not linked to this
// repo. Do not auto-link it based on cross-repo access: the token
// only authorizes this repo, and for deploy keys ctx.Doer is the
// repo owner, so trusting it here would let a single-repo key pull
// objects from any repo the owner can see. Require proof of
// possession by making the client upload the (hash-verified) bytes,
// and treat it as a new upload for size-limit enforcement below.
exists = false
}
var err *lfs_module.ObjectError
if !exists && setting.LFS.MaxFileSize > 0 && p.Size > setting.LFS.MaxFileSize {
err = &lfs_module.ObjectError{
Code: http.StatusUnprocessableEntity,
Message: fmt.Sprintf("Size must be less than or equal to %d", setting.LFS.MaxFileSize),
}
}
responseObject = buildObjectResponse(rc, p, false, !exists, err)
} else {
var err *lfs_module.ObjectError
if !exists || meta == nil {
err = &lfs_module.ObjectError{
Code: http.StatusNotFound,
Message: http.StatusText(http.StatusNotFound),
}
}
responseObject = buildObjectResponse(rc, p, true, false, err)
}
responseObjects = append(responseObjects, responseObject)
}
respobj := &lfs_module.BatchResponse{Objects: responseObjects}
ctx.Resp.Header().Set("Content-Type", lfs_module.MediaType)
enc := json.NewEncoder(ctx.Resp)
if err := enc.Encode(respobj); err != nil {
log.Error("Failed to encode representation as json. Error: %v", err)
}
}
// UploadHandler receives data from the client and puts it into the content store
func UploadHandler(ctx *context.Context) {
rc := getRequestContext(ctx)
p := lfs_module.Pointer{Oid: ctx.PathParam("oid")}
var err error
if p.Size, err = strconv.ParseInt(ctx.PathParam("size"), 10, 64); err != nil {
writeStatusMessage(ctx, http.StatusUnprocessableEntity, err.Error())
}
if !p.IsValid() {
log.Trace("Attempt to access invalid LFS OID[%s] in %s/%s", p.Oid, rc.User, rc.Repo)
writeStatus(ctx, http.StatusUnprocessableEntity)
return
}
repository := getAuthenticatedRepository(ctx, rc, true)
if repository == nil {
return
}
uploadOrVerify := func() error {
contentStore := lfs_module.NewContentStore()
stat, err := contentStore.Stat(p)
if stat != nil {
// The bytes already exist in the content store. Only skip proof of
// possession when the object is already linked to *this* repo; never
// trust cross-repo access (ctx.Doer is the repo owner for deploy keys),
// which would let a caller link an object it cannot produce.
meta, err := git_model.GetLFSMetaObjectByOid(ctx, repository.ID, p.Oid)
if err != nil && !errors.Is(err, util.ErrNotExist) {
log.Error("Unable to get LFS MetaObject [%s]. Error: %v", p.Oid, err)
return err
}
if meta == nil {
// The file exists but is not linked to this repo, or the file is being uploaded.
// The upload gets verified by hashing and size comparison to prove access to it.
// Keep in mind: here the file might be incomplete due to concurrent uploading, so the verification might fail.
// ATTENTION: it's impossible to handle corrupted file on server-side at the moment,
// we don't know whether a file is really corrupted, or it is being uploaded.
hash := sha256.New()
written, err := io.Copy(hash, ctx.Req.Body)
if err != nil {
log.Error("Error creating hash. Error: %v", err)
return err
}
if written != p.Size {
return lfs_module.ErrSizeMismatch
}
if hex.EncodeToString(hash.Sum(nil)) != p.Oid {
return lfs_module.ErrHashMismatch
}
}
} else if errors.Is(err, fs.ErrNotExist) {
// not exist, store it into the store
if err := contentStore.Put(p, ctx.Req.Body); err != nil {
log.Error("Error putting LFS MetaObject [%s] into content store. Error: %v", p.Oid, err)
return err
}
} else {
log.Error("Unable to check if LFS OID[%s] stat. Error: %v", p.Oid, err)
return err
}
_, err = git_model.NewLFSMetaObject(ctx, repository.ID, p)
return err
}
defer ctx.Req.Body.Close()
if err := uploadOrVerify(); err != nil {
if errors.Is(err, lfs_module.ErrSizeMismatch) || errors.Is(err, lfs_module.ErrHashMismatch) {
log.Error("Upload does not match LFS MetaObject [%s]. Error: %v", p.Oid, err)
writeStatusMessage(ctx, http.StatusUnprocessableEntity, err.Error())
} else {
log.Error("Error whilst uploadOrVerify LFS OID[%s]: %v", p.Oid, err)
writeStatus(ctx, http.StatusInternalServerError)
}
// Do not remove the LFS MetaObject here: this request only creates it after the content is verified and stored,
// an invalid request should not remove the existing correct record.
// If two requests are loading (the file is incomplete):
// * one will keep writing the file content
// * one will fail the verification because it reads an incomplete file, the failure should be just ignore
// In the end, the first one will complete the upload and insert a LFS MetaObject record.
return
}
writeStatus(ctx, http.StatusOK)
}
// VerifyHandler verify oid and its size from the content store
func VerifyHandler(ctx *context.Context) {
var p lfs_module.Pointer
if err := decodeJSON(ctx.Req, &p); err != nil {
writeStatus(ctx, http.StatusUnprocessableEntity)
return
}
rc := getRequestContext(ctx)
meta := getAuthenticatedMeta(ctx, rc, p, true)
if meta == nil {
return
}
contentStore := lfs_module.NewContentStore()
ok, err := contentStore.Verify(meta.Pointer)
status := http.StatusOK
if err != nil {
log.Error("Error whilst verifying LFS OID[%s]: %v", p.Oid, err)
status = http.StatusInternalServerError
} else if !ok {
status = http.StatusNotFound
}
writeStatus(ctx, status)
}
func decodeJSON(req *http.Request, v any) error {
defer req.Body.Close()
dec := json.NewDecoder(req.Body)
return dec.Decode(v)
}
func getRequestContext(ctx *context.Context) *requestContext {
ownerName := ctx.PathParam("username")
repoName := strings.TrimSuffix(ctx.PathParam("reponame"), ".git")
return &requestContext{
User: ownerName,
Repo: repoName,
Authorization: ctx.Req.Header.Get("Authorization"),
RepoGitURL: httplib.GuessCurrentAppURL(ctx) + url.PathEscape(ownerName) + "/" + url.PathEscape(repoName+".git"),
}
}
func getAuthenticatedMeta(ctx *context.Context, rc *requestContext, p lfs_module.Pointer, requireWrite bool) *git_model.LFSMetaObject {
if !p.IsValid() {
log.Info("Attempt to access invalid LFS OID[%s] in %s/%s", p.Oid, rc.User, rc.Repo)
writeStatusMessage(ctx, http.StatusUnprocessableEntity, "Oid or size are invalid")
return nil
}
repository := getAuthenticatedRepository(ctx, rc, requireWrite)
if repository == nil {
return nil
}
meta, err := git_model.GetLFSMetaObjectByOid(ctx, repository.ID, p.Oid)
if err != nil {
log.Error("Unable to get LFS OID[%s] Error: %v", p.Oid, err)
writeStatus(ctx, http.StatusNotFound)
return nil
}
return meta
}
func getAuthenticatedRepository(ctx *context.Context, rc *requestContext, requireWrite bool) *repo_model.Repository {
repository, err := repo_model.GetRepositoryByOwnerAndName(ctx, rc.User, rc.Repo)
if err != nil {
log.Error("Unable to get repository: %s/%s Error: %v", rc.User, rc.Repo, err)
writeStatus(ctx, http.StatusNotFound)
return nil
}
if !authenticate(ctx, repository, rc.Authorization, false, requireWrite) {
requireAuth(ctx)
return nil
}
if requireWrite {
context.CheckRepoScopedToken(ctx, repository, auth_model.Write)
} else {
context.CheckRepoScopedToken(ctx, repository, auth_model.Read)
}
if ctx.Written() {
return nil
}
return repository
}
func buildObjectResponse(rc *requestContext, pointer lfs_module.Pointer, download, upload bool, err *lfs_module.ObjectError) *lfs_module.ObjectResponse {
rep := &lfs_module.ObjectResponse{Pointer: pointer}
if err != nil {
rep.Error = err
} else {
rep.Actions = make(map[string]*lfs_module.Link)
if download {
var link *lfs_module.Link
if setting.LFS.Storage.ServeDirect() {
// If we have a signed url (S3, object storage), redirect to this directly.
// DO NOT USE the http POST method coming from the lfs batch endpoint
u, err := storage.LFS.ServeDirectURL(pointer.RelativePath(), pointer.Oid, http.MethodGet, nil)
if u != nil && err == nil {
link = lfs_module.NewLink(u.String()) // Presigned url does not need the Authorization header
}
}
if link == nil {
link = lfs_module.NewLink(rc.DownloadLink(pointer)).WithHeader("Authorization", rc.Authorization)
}
rep.Actions["download"] = link
}
if upload {
// Set Transfer-Encoding header to enable chunked uploads. Required by git-lfs client to do chunked transfer.
// See: https://github.com/git-lfs/git-lfs/blob/main/tq/basic_upload.go#L58-59
rep.Actions["upload"] = lfs_module.NewLink(rc.UploadLink(pointer)).
WithHeader("Authorization", rc.Authorization).
WithHeader("Transfer-Encoding", "chunked")
// "Accept" header is the workaround for git-lfs < 2.8.0 (before 2019).
// This workaround could be removed in the future: https://github.com/git-lfs/git-lfs/issues/3662
rep.Actions["verify"] = lfs_module.NewLink(rc.VerifyLink(pointer)).
WithHeader("Authorization", rc.Authorization).
WithHeader("Accept", lfs_module.AcceptHeader)
}
}
return rep
}
func writeStatus(ctx *context.Context, status int) {
writeStatusMessage(ctx, status, http.StatusText(status))
}
func writeStatusMessage(ctx *context.Context, status int, message string) {
ctx.Resp.Header().Set("Content-Type", lfs_module.MediaType)
ctx.Resp.WriteHeader(status)
er := lfs_module.ErrorResponse{Message: message}
enc := json.NewEncoder(ctx.Resp)
if err := enc.Encode(er); err != nil {
log.Error("Failed to encode error response as json. Error: %v", err)
}
}
// authenticate uses the authorization string to determine whether
// to proceed. This server assumes an HTTP Basic auth format.
func authenticate(ctx *context.Context, repository *repo_model.Repository, authorization string, requireSigned, requireWrite bool) bool {
accessMode := perm_model.AccessModeRead
if requireWrite {
accessMode = perm_model.AccessModeWrite
}
if taskID, ok := user_model.GetActionsUserTaskID(ctx.Doer); ok {
perm, err := access_model.GetActionsUserRepoPermission(ctx, repository, ctx.Doer, taskID)
if err != nil {
log.Error("Unable to GetActionsUserRepoPermission for task[%d] Error: %v", taskID, err)
return false
}
return perm.CanAccess(accessMode, unit.TypeCode)
}
// it works for both anonymous request and signed-in user, then perm.CanAccess will do the permission check
perm, err := access_model.GetDoerRepoPermission(ctx, repository, ctx.Doer)
if err != nil {
log.Error("Unable to GetDoerRepoPermission for user %-v in repo %-v Error: %v", ctx.Doer, repository, err)
return false
}
canAccess := perm.CanAccess(accessMode, unit.TypeCode)
// if it doesn't require sign-in and anonymous user has access, return true
// if the user is already signed in (for example: by session auth method), and the doer can access, return true
if canAccess && (!requireSigned || ctx.IsSigned) {
return true
}
// now, either sign-in is required or the ctx.Doer cannot access, check the LFS token
// however, "ctx.Doer exists but cannot access then check LFS token" should not really happen:
// * why a request can be sent with both valid user session and valid LFS token then use LFS token to access?
user, err := parseToken(ctx, authorization, repository, accessMode)
if err != nil {
// Most of these are Warn level - the true internal server errors are logged in parseToken already
log.Warn("Authentication failure for provided token with Error: %v", err)
return false
}
ctx.Doer = user
return true
}
func handleLFSToken(ctx stdCtx.Context, tokenSHA string, target *repo_model.Repository, mode perm_model.AccessMode) (*user_model.User, error) {
token, err := jwt.ParseWithClaims(tokenSHA, &Claims{}, func(t *jwt.Token) (any, error) {
if _, ok := t.Method.(*jwt.SigningMethodHMAC); !ok {
return nil, fmt.Errorf("unexpected signing method: %v", t.Header["alg"])
}
return setting.LFS.JWTSecretBytes, nil
})
if err != nil {
return nil, errors.New("invalid token")
}
claims, claimsOk := token.Claims.(*Claims)
if !token.Valid || !claimsOk {
return nil, errors.New("invalid token claim")
}
if claims.RepoID != target.ID {
return nil, errors.New("invalid token claim")
}
if mode == perm_model.AccessModeWrite && claims.Op != "upload" {
return nil, errors.New("invalid token claim")
}
u, err := user_model.GetUserByID(ctx, claims.UserID)
if err != nil {
log.Error("Unable to GetUserById[%d]: Error: %v", claims.UserID, err)
return nil, err
}
if !u.IsActive || u.ProhibitLogin {
return nil, util.NewPermissionDeniedErrorf("not allowed to access any repository")
}
perm, err := access_model.GetDoerRepoPermission(ctx, target, u)
if err != nil {
log.Error("Unable to GetDoerRepoPermission for user[%d] repo[%d]: %v", claims.UserID, target.ID, err)
return nil, err
}
if !perm.CanAccess(mode, unit.TypeCode) {
return nil, util.NewPermissionDeniedErrorf("no permission to access the repository")
}
return u, nil
}
func parseToken(ctx stdCtx.Context, authorization string, target *repo_model.Repository, mode perm_model.AccessMode) (*user_model.User, error) {
if authorization == "" {
return nil, errors.New("no token")
}
parsed, ok := httpauth.ParseAuthorizationHeader(authorization)
if !ok || parsed.BearerToken == nil {
return nil, errors.New("token not found")
}
return handleLFSToken(ctx, parsed.BearerToken.Token, target, mode)
}
func requireAuth(ctx *context.Context) {
ctx.Resp.Header().Set("WWW-Authenticate", `Basic realm="gitea-lfs"`)
writeStatus(ctx, http.StatusUnauthorized)
}