modules/lfstransfer: add a backend and runner

Also add handler in runServ()

The protocol lib supports locking but the backend does not,
as neither does Gitea. Support can be added later and the
capability advertised.

Author: ConcurrentCrab

cmd/serv.go

@@ -22,6 +22,7 @@ import (
 	"code.gitea.io/gitea/models/perm"
 	"code.gitea.io/gitea/modules/git"
 	"code.gitea.io/gitea/modules/json"
+	"code.gitea.io/gitea/modules/lfstransfer"
 	"code.gitea.io/gitea/modules/log"
 	"code.gitea.io/gitea/modules/pprof"
 	"code.gitea.io/gitea/modules/private"
@@ -40,6 +41,7 @@ const (
 	verbUploadArchive   = "git-upload-archive"
 	verbReceivePack     = "git-receive-pack"
 	verbLfsAuthenticate = "git-lfs-authenticate"
+	verbLfsTransfer     = "git-lfs-transfer"
 )
 
 // CmdServ represents the available serv sub-command.
@@ -83,9 +85,11 @@ var (
 		verbUploadArchive:   true,
 		verbReceivePack:     true,
 		verbLfsAuthenticate: true,
+		verbLfsTransfer:     true,
 	}
 	allowedCommandsLfs = map[string]bool{
 		verbLfsAuthenticate: true,
+		verbLfsTransfer:     true,
 	}
 	alphaDashDotPattern = regexp.MustCompile(`[^\w-\.]`)
 )
@@ -138,7 +142,7 @@ func getAccessMode(verb string, lfsVerb string) perm.AccessMode {
 		return perm.AccessModeRead
 	case verbReceivePack:
 		return perm.AccessModeWrite
-	case verbLfsAuthenticate:
+	case verbLfsAuthenticate, verbLfsTransfer:
 		switch lfsVerb {
 		case "upload":
 			return perm.AccessModeWrite
@@ -297,6 +301,15 @@ func runServ(c *cli.Context) error {
 		return fail(ctx, extra.UserMsg, "ServCommand failed: %s", extra.Error)
 	}
 
+	// LFS SSH protocol
+	if verb == verbLfsTransfer {
+		token, err := getLFSAuthToken(ctx, lfsVerb, results)
+		if err != nil {
+			return err
+		}
+		return lfstransfer.Main(ctx, repoPath, lfsVerb, token)
+	}
+
 	// LFS token authentication
 	if verb == verbLfsAuthenticate {
 		url := fmt.Sprintf("%s%s/%s.git/info/lfs", setting.AppURL, url.PathEscape(results.OwnerName), url.PathEscape(results.RepoName))

modules/lfstransfer/backend/backend.go

@@ -0,0 +1,298 @@
+// Copyright 2024 The Gitea Authors. All rights reserved.
+// SPDX-License-Identifier: MIT
+
+package backend
+
+import (
+	"bytes"
+	"context"
+	"encoding/base64"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"net/url"
+	"strconv"
+
+	"code.gitea.io/gitea/modules/lfs"
+	"code.gitea.io/gitea/modules/lfstransfer/transfer"
+	"code.gitea.io/gitea/modules/setting"
+)
+
+// Version is the git-lfs-transfer protocol version number.
+const Version = "1"
+
+// Capabilities is a list of Git LFS capabilities supported by this package.
+var Capabilities = []string{
+	"version=" + Version,
+	// "locking", // no support yet in gitea backend
+}
+
+var _ transfer.Backend = &GiteaBackend{}
+
+// GiteaBackend is an adapter between git-lfs-transfer library and Gitea's internal LFS API
+type GiteaBackend struct {
+	ctx    context.Context
+	server *url.URL
+	op     string
+	token  string
+	logger transfer.Logger
+}
+
+func New(ctx context.Context, repo string, op string, token string, logger transfer.Logger) (transfer.Backend, error) {
+	// runServ guarantees repo will be in form [owner]/[name].git
+	server, err := url.Parse(setting.LocalURL)
+	if err != nil {
+		return nil, err
+	}
+	server = server.JoinPath(repo, "info/lfs")
+	return &GiteaBackend{ctx: ctx, server: server, op: op, token: token, logger: logger}, nil
+}
+
+// Batch implements transfer.Backend
+func (g *GiteaBackend) Batch(_ string, pointers []transfer.BatchItem, args transfer.Args) ([]transfer.BatchItem, error) {
+	reqBody := lfs.BatchRequest{Operation: g.op}
+	if transfer, ok := args[argTransfer]; ok {
+		reqBody.Transfers = []string{transfer}
+	}
+	if ref, ok := args[argRefname]; ok {
+		reqBody.Ref = &lfs.Reference{Name: ref}
+	}
+	reqBody.Objects = make([]lfs.Pointer, len(pointers))
+	for i := range pointers {
+		reqBody.Objects[i].Oid = pointers[i].Oid
+		reqBody.Objects[i].Size = pointers[i].Size
+	}
+
+	bodyBytes, err := json.Marshal(reqBody)
+	if err != nil {
+		g.logger.Log("json marshal error", err)
+		return nil, err
+	}
+	url := g.server.JoinPath("objects/batch").String()
+	headers := map[string]string{
+		headerAuthorisation: g.token,
+		headerAccept:        mimeGitLFS,
+		headerContentType:   mimeGitLFS,
+	}
+	req := newInternalRequest(g.ctx, url, http.MethodPost, headers, bodyBytes)
+	resp, err := req.Response()
+	if err != nil {
+		g.logger.Log("http request error", err)
+		return nil, err
+	}
+	if resp.StatusCode != http.StatusOK {
+		g.logger.Log("http statuscode error", resp.StatusCode, statusCodeToErr(resp.StatusCode))
+		return nil, statusCodeToErr(resp.StatusCode)
+	}
+	defer resp.Body.Close()
+	respBytes, err := io.ReadAll(resp.Body)
+	if err != nil {
+		g.logger.Log("http read error", err)
+		return nil, err
+	}
+	var respBody lfs.BatchResponse
+	err = json.Unmarshal(respBytes, &respBody)
+	if err != nil {
+		g.logger.Log("json umarshal error", err)
+		return nil, err
+	}
+
+	// rebuild slice, we can't rely on order in resp being the same as req
+	pointers = pointers[:0]
+	opNum := opMap[g.op]
+	for _, obj := range respBody.Objects {
+		pointer := transfer.Pointer{Oid: obj.Pointer.Oid, Size: obj.Pointer.Size}
+		item := transfer.BatchItem{Pointer: pointer, Args: map[string]string{}}
+		switch opNum {
+		case opDownload:
+			if action, ok := obj.Actions[actionDownload]; ok {
+				item.Present = true
+				idMap := obj.Actions
+				idMapBytes, err := json.Marshal(idMap)
+				if err != nil {
+					g.logger.Log("json marshal error", err)
+					return nil, err
+				}
+				idMapStr := base64.StdEncoding.EncodeToString(idMapBytes)
+				item.Args[argID] = idMapStr
+				if authHeader, ok := action.Header[headerAuthorisation]; ok {
+					authHeaderB64 := base64.StdEncoding.EncodeToString([]byte(authHeader))
+					item.Args[argToken] = authHeaderB64
+				}
+				if action.ExpiresAt != nil {
+					item.Args[argExpiresAt] = action.ExpiresAt.String()
+				}
+			} else {
+				// must be an error, but the SSH protocol can't propagate individual errors
+				g.logger.Log("object not found", obj.Pointer.Oid, obj.Pointer.Size)
+				item.Present = false
+			}
+		case opUpload:
+			if action, ok := obj.Actions[actionUpload]; ok {
+				item.Present = false
+				idMap := obj.Actions
+				idMapBytes, err := json.Marshal(idMap)
+				if err != nil {
+					g.logger.Log("json marshal error", err)
+					return nil, err
+				}
+				idMapStr := base64.StdEncoding.EncodeToString(idMapBytes)
+				item.Args[argID] = idMapStr
+				if authHeader, ok := action.Header[headerAuthorisation]; ok {
+					authHeaderB64 := base64.StdEncoding.EncodeToString([]byte(authHeader))
+					item.Args[argToken] = authHeaderB64
+				}
+				if action.ExpiresAt != nil {
+					item.Args[argExpiresAt] = action.ExpiresAt.String()
+				}
+			} else {
+				item.Present = true
+			}
+		}
+		pointers = append(pointers, item)
+	}
+	return pointers, nil
+}
+
+// Download implements transfer.Backend. The returned reader must be closed by the
+// caller.
+func (g *GiteaBackend) Download(oid string, args transfer.Args) (io.ReadCloser, int64, error) {
+	idMapStr, exists := args[argID]
+	if !exists {
+		return nil, 0, ErrMissingID
+	}
+	idMapBytes, err := base64.StdEncoding.DecodeString(idMapStr)
+	if err != nil {
+		g.logger.Log("base64 decode error", err)
+		return nil, 0, transfer.ErrCorruptData
+	}
+	idMap := map[string]*lfs.Link{}
+	err = json.Unmarshal(idMapBytes, &idMap)
+	if err != nil {
+		g.logger.Log("json unmarshal error", err)
+		return nil, 0, transfer.ErrCorruptData
+	}
+	action, exists := idMap[actionDownload]
+	if !exists {
+		g.logger.Log("argument id incorrect")
+		return nil, 0, transfer.ErrCorruptData
+	}
+	url := action.Href
+	headers := map[string]string{
+		headerAuthorisation: g.token,
+		headerAccept:        mimeOctetStream,
+	}
+	req := newInternalRequest(g.ctx, url, http.MethodGet, headers, nil)
+	resp, err := req.Response()
+	if err != nil {
+		return nil, 0, err
+	}
+	if resp.StatusCode != http.StatusOK {
+		return nil, 0, statusCodeToErr(resp.StatusCode)
+	}
+	defer resp.Body.Close()
+	respBytes, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return nil, 0, err
+	}
+	respSize := int64(len(respBytes))
+	respBuf := io.NopCloser(bytes.NewBuffer(respBytes))
+	return respBuf, respSize, nil
+}
+
+// StartUpload implements transfer.Backend.
+func (g *GiteaBackend) Upload(oid string, size int64, r io.Reader, args transfer.Args) error {
+	idMapStr, exists := args[argID]
+	if !exists {
+		return ErrMissingID
+	}
+	idMapBytes, err := base64.StdEncoding.DecodeString(idMapStr)
+	if err != nil {
+		g.logger.Log("base64 decode error", err)
+		return transfer.ErrCorruptData
+	}
+	idMap := map[string]*lfs.Link{}
+	err = json.Unmarshal(idMapBytes, &idMap)
+	if err != nil {
+		g.logger.Log("json unmarshal error", err)
+		return transfer.ErrCorruptData
+	}
+	action, exists := idMap[actionUpload]
+	if !exists {
+		g.logger.Log("argument id incorrect")
+		return transfer.ErrCorruptData
+	}
+	url := action.Href
+	headers := map[string]string{
+		headerAuthorisation: g.token,
+		headerContentType:   mimeOctetStream,
+		headerContentLength: strconv.FormatInt(size, 10),
+	}
+	reqBytes, err := io.ReadAll(r)
+	if err != nil {
+		return err
+	}
+	req := newInternalRequest(g.ctx, url, http.MethodPut, headers, reqBytes)
+	resp, err := req.Response()
+	if err != nil {
+		return err
+	}
+	if resp.StatusCode != http.StatusOK {
+		return statusCodeToErr(resp.StatusCode)
+	}
+	return nil
+}
+
+// Verify implements transfer.Backend.
+func (g *GiteaBackend) Verify(oid string, size int64, args transfer.Args) (transfer.Status, error) {
+	reqBody := lfs.Pointer{Oid: oid, Size: size}
+
+	bodyBytes, err := json.Marshal(reqBody)
+	if err != nil {
+		return transfer.NewStatus(transfer.StatusInternalServerError), err
+	}
+	idMapStr, exists := args[argID]
+	if !exists {
+		return transfer.NewStatus(transfer.StatusBadRequest, "missing argument: id"), ErrMissingID
+	}
+	idMapBytes, err := base64.StdEncoding.DecodeString(idMapStr)
+	if err != nil {
+		g.logger.Log("base64 decode error", err)
+		return transfer.NewStatus(transfer.StatusBadRequest, "corrupt argument: id"), transfer.ErrCorruptData
+	}
+	idMap := map[string]*lfs.Link{}
+	err = json.Unmarshal(idMapBytes, &idMap)
+	if err != nil {
+		g.logger.Log("json unmarshal error", err)
+		return transfer.NewStatus(transfer.StatusBadRequest, "corrupt argument: id"), transfer.ErrCorruptData
+	}
+	action, exists := idMap[actionVerify]
+	if !exists {
+		// the server sent no verify action
+		return transfer.SuccessStatus(), nil
+	}
+	url := action.Href
+	headers := map[string]string{
+		headerAuthorisation: g.token,
+		headerAccept:        mimeGitLFS,
+		headerContentType:   mimeGitLFS,
+	}
+	req := newInternalRequest(g.ctx, url, http.MethodPost, headers, bodyBytes)
+	resp, err := req.Response()
+	if err != nil {
+		return transfer.NewStatus(transfer.StatusInternalServerError), err
+	}
+	if resp.StatusCode != http.StatusOK {
+		return transfer.NewStatus(uint32(resp.StatusCode), http.StatusText(resp.StatusCode)), statusCodeToErr(resp.StatusCode)
+	}
+	return transfer.SuccessStatus(), nil
+}
+
+// LockBackend implements transfer.Backend.
+func (g *GiteaBackend) LockBackend(_ transfer.Args) transfer.LockBackend {
+	// Gitea doesn't support the locking API
+	// this should never be called as we don't advertise the capability
+	panic(fmt.Errorf("backend doesn't implement locking"))
+	return (transfer.LockBackend)(nil)
+}