libgit2: ensure context timeout cancels transfer

go.mod

@@ -11,7 +11,7 @@ require (
 	github.com/ProtonMail/go-crypto v0.0.0-20210428141323-04723f9f07d7
 	github.com/cyphar/filepath-securejoin v0.2.2
 	github.com/fluxcd/pkg/apis/meta v0.10.0
-	github.com/fluxcd/pkg/gittestserver v0.4.1
+	github.com/fluxcd/pkg/gittestserver v0.4.2
 	github.com/fluxcd/pkg/gitutil v0.1.0
 	github.com/fluxcd/pkg/helmtestserver v0.2.0
 	github.com/fluxcd/pkg/lockedfile v0.1.0

go.sum

@@ -266,8 +266,8 @@ github.com/fatih/color v1.7.0 h1:DkWD4oS2D8LGGgTQ6IvwJJXSL5Vp2ffcQg58nFV38Ys=
 github.com/fatih/color v1.7.0/go.mod h1:Zm6kSWBoL9eyXnKyktHP6abPY2pDugNf5KwzbycvMj4=
 github.com/fluxcd/pkg/apis/meta v0.10.0 h1:N7wVGHC1cyPdT87hrDC7UwCwRwnZdQM46PBSLjG2rlE=
 github.com/fluxcd/pkg/apis/meta v0.10.0/go.mod h1:CW9X9ijMTpNe7BwnokiUOrLl/h13miwVr/3abEQLbKE=
-github.com/fluxcd/pkg/gittestserver v0.4.1 h1:knghRrVEEPnpO0VJYjoz0H2YMc4fnKAVt5hDGsB1IHc=
-github.com/fluxcd/pkg/gittestserver v0.4.1/go.mod h1:hUPx21fe/6oox336Wih/XF1fnmzLmptNMOvATbTZXNY=
+github.com/fluxcd/pkg/gittestserver v0.4.2 h1:XqoiemTnnUNldnOw8N7OTdalu2iZp1FTRhp9uUauDJQ=
+github.com/fluxcd/pkg/gittestserver v0.4.2/go.mod h1:hUPx21fe/6oox336Wih/XF1fnmzLmptNMOvATbTZXNY=
 github.com/fluxcd/pkg/gitutil v0.1.0 h1:VO3kJY/CKOCO4ysDNqfdpTg04icAKBOSb3lbR5uE/IE=
 github.com/fluxcd/pkg/gitutil v0.1.0/go.mod h1:Ybz50Ck5gkcnvF0TagaMwtlRy3X3wXuiri1HVsK5id4=
 github.com/fluxcd/pkg/helmtestserver v0.2.0 h1:cE7YHDmrWI0hr9QpaaeQ0vQ16Z0IiqZKiINDpqdY610=

pkg/git/libgit2/checkout.go

@@ -63,7 +63,7 @@ func (c *CheckoutBranch) Checkout(ctx context.Context, path, url string, opts *g
 	repo, err := git2go.Clone(url, path, &git2go.CloneOptions{
 		FetchOptions: &git2go.FetchOptions{
 			DownloadTags:    git2go.DownloadTagsNone,
-			RemoteCallbacks: RemoteCallbacks(opts),
+			RemoteCallbacks: RemoteCallbacks(ctx, opts),
 		},
 		CheckoutBranch: c.Branch,
 	})
@@ -92,7 +92,7 @@ func (c *CheckoutTag) Checkout(ctx context.Context, path, url string, opts *git.
 	repo, err := git2go.Clone(url, path, &git2go.CloneOptions{
 		FetchOptions: &git2go.FetchOptions{
 			DownloadTags:    git2go.DownloadTagsAll,
-			RemoteCallbacks: RemoteCallbacks(opts),
+			RemoteCallbacks: RemoteCallbacks(ctx, opts),
 		},
 	})
 	if err != nil {
@@ -115,7 +115,7 @@ func (c *CheckoutCommit) Checkout(ctx context.Context, path, url string, opts *g
 	repo, err := git2go.Clone(url, path, &git2go.CloneOptions{
 		FetchOptions: &git2go.FetchOptions{
 			DownloadTags:    git2go.DownloadTagsNone,
-			RemoteCallbacks: RemoteCallbacks(opts),
+			RemoteCallbacks: RemoteCallbacks(ctx, opts),
 		},
 	})
 	if err != nil {
@@ -146,7 +146,7 @@ func (c *CheckoutSemVer) Checkout(ctx context.Context, path, url string, opts *g
 	repo, err := git2go.Clone(url, path, &git2go.CloneOptions{
 		FetchOptions: &git2go.FetchOptions{
 			DownloadTags:    git2go.DownloadTagsAll,
-			RemoteCallbacks: RemoteCallbacks(opts),
+			RemoteCallbacks: RemoteCallbacks(ctx, opts),
 		},
 	})
 	if err != nil {

pkg/git/libgit2/transport.go

@@ -19,6 +19,7 @@ package libgit2
 import (
 	"bufio"
 	"bytes"
+	"context"
 	"crypto/md5"
 	"crypto/sha1"
 	"crypto/sha256"
@@ -43,16 +44,69 @@ var (
 
 // RemoteCallbacks constructs RemoteCallbacks with credentialsCallback and
 // certificateCallback, and the given options if the given opts is not nil.
-func RemoteCallbacks(opts *git.AuthOptions) git2go.RemoteCallbacks {
+func RemoteCallbacks(ctx context.Context, opts *git.AuthOptions) git2go.RemoteCallbacks {
 	if opts != nil {
 		return git2go.RemoteCallbacks{
-			CredentialsCallback:      credentialsCallback(opts),
-			CertificateCheckCallback: certificateCallback(opts),
+			SidebandProgressCallback:     transportMessageCallback(ctx),
+			TransferProgressCallback:     transferProgressCallback(ctx),
+			PushTransferProgressCallback: pushTransferProgressCallback(ctx),
+			CredentialsCallback:          credentialsCallback(opts),
+			CertificateCheckCallback:     certificateCallback(opts),
 		}
 	}
 	return git2go.RemoteCallbacks{}
 }
 
+// transferProgressCallback constructs TransferProgressCallbacks which signals
+// libgit2 it should stop the transfer when the given context is closed (due to
+// e.g. a timeout).
+func transferProgressCallback(ctx context.Context) git2go.TransferProgressCallback {
+	return func(p git2go.TransferProgress) git2go.ErrorCode {
+		// Early return if all the objects have been received.
+		if p.ReceivedObjects == p.TotalObjects {
+			return git2go.ErrorCodeOK
+		}
+		select {
+		case <-ctx.Done():
+			return git2go.ErrorCodeUser
+		default:
+			return git2go.ErrorCodeOK
+		}
+	}
+}
+
+// transportMessageCallback constructs TransportMessageCallback which signals
+// libgit2 it should cancel the network operation when the given context is
+// closed.
+func transportMessageCallback(ctx context.Context) git2go.TransportMessageCallback {
+	return func(_ string) git2go.ErrorCode {
+		select {
+		case <-ctx.Done():
+			return git2go.ErrorCodeUser
+		default:
+			return git2go.ErrorCodeOK
+		}
+	}
+}
+
+// pushTransferProgressCallback constructs PushTransferProgressCallback which
+// signals libgit2 it should stop the push transfer when the given context is
+// closed (due to e.g. a timeout).
+func pushTransferProgressCallback(ctx context.Context) git2go.PushTransferProgressCallback {
+	return func(current, total uint32, _ uint) git2go.ErrorCode {
+		// Early return if current equals total.
+		if current == total {
+			return git2go.ErrorCodeOK
+		}
+		select {
+		case <-ctx.Done():
+			return git2go.ErrorCodeUser
+		default:
+			return git2go.ErrorCodeOK
+		}
+	}
+}
+
 // credentialsCallback constructs CredentialsCallbacks with the given options
 // for git.Transport, and returns the result.
 func credentialsCallback(opts *git.AuthOptions) git2go.CredentialsCallback {

pkg/git/libgit2/transport_test.go

@@ -18,6 +18,7 @@ package libgit2
 
 import (
 	"bytes"
+	"context"
 	"crypto/x509"
 	"encoding/base64"
 	"encoding/pem"
@@ -346,6 +347,155 @@ gitlab.com ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIAfuCHKVTjquxvt6CM6tdG4SLp1Btn/nO
 	}
 }
 
+func Test_transferProgressCallback(t *testing.T) {
+	tests := []struct {
+		name       string
+		progress   git2go.TransferProgress
+		cancelFunc func(context.CancelFunc)
+		wantErr    git2go.ErrorCode
+	}{
+		{
+			name: "ok - in progress",
+			progress: git2go.TransferProgress{
+				TotalObjects:    30,
+				ReceivedObjects: 21,
+			},
+			cancelFunc: func(cf context.CancelFunc) {},
+			wantErr:    git2go.ErrorCodeOK,
+		},
+		{
+			name: "ok - transfer complete",
+			progress: git2go.TransferProgress{
+				TotalObjects:    30,
+				ReceivedObjects: 30,
+			},
+			cancelFunc: func(cf context.CancelFunc) {},
+			wantErr:    git2go.ErrorCodeOK,
+		},
+		{
+			name: "ok - transfer complete, context cancelled",
+			progress: git2go.TransferProgress{
+				TotalObjects:    30,
+				ReceivedObjects: 30,
+			},
+			cancelFunc: func(cf context.CancelFunc) { cf() },
+			wantErr:    git2go.ErrorCodeOK,
+		},
+		{
+			name: "error - context cancelled",
+			progress: git2go.TransferProgress{
+				TotalObjects:    30,
+				ReceivedObjects: 21,
+			},
+			cancelFunc: func(cf context.CancelFunc) { cf() },
+			wantErr:    git2go.ErrorCodeUser,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			g := NewWithT(t)
+
+			ctx, cancel := context.WithCancel(context.TODO())
+			defer cancel()
+
+			tpcb := transferProgressCallback(ctx)
+
+			tt.cancelFunc(cancel)
+
+			g.Expect(tpcb(tt.progress)).To(Equal(tt.wantErr))
+		})
+	}
+}
+
+func Test_transportMessageCallback(t *testing.T) {
+	tests := []struct {
+		name       string
+		cancelFunc func(context.CancelFunc)
+		wantErr    git2go.ErrorCode
+	}{
+		{
+			name:       "ok - transport open",
+			cancelFunc: func(cf context.CancelFunc) {},
+			wantErr:    git2go.ErrorCodeOK,
+		},
+		{
+			name:       "error - transport closed",
+			cancelFunc: func(cf context.CancelFunc) { cf() },
+			wantErr:    git2go.ErrorCodeUser,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			g := NewWithT(t)
+
+			ctx, cancel := context.WithCancel(context.TODO())
+			defer cancel()
+
+			tmcb := transportMessageCallback(ctx)
+
+			tt.cancelFunc(cancel)
+
+			g.Expect(tmcb("")).To(Equal(tt.wantErr))
+		})
+	}
+}
+
+func Test_pushTransferProgressCallback(t *testing.T) {
+	type pushProgress struct {
+		current uint32
+		total   uint32
+		bytes   uint
+	}
+	tests := []struct {
+		name       string
+		progress   pushProgress
+		cancelFunc func(context.CancelFunc)
+		wantErr    git2go.ErrorCode
+	}{
+		{
+			name:       "ok - in progress",
+			progress:   pushProgress{current: 20, total: 25},
+			cancelFunc: func(cf context.CancelFunc) {},
+			wantErr:    git2go.ErrorCodeOK,
+		},
+		{
+			name:       "ok - transfer complete",
+			progress:   pushProgress{current: 25, total: 25},
+			cancelFunc: func(cf context.CancelFunc) {},
+			wantErr:    git2go.ErrorCodeOK,
+		},
+		{
+			name:       "ok - transfer complete, context cancelled",
+			progress:   pushProgress{current: 25, total: 25},
+			cancelFunc: func(cf context.CancelFunc) { cf() },
+			wantErr:    git2go.ErrorCodeOK,
+		},
+		{
+			name:       "error - context cancelled",
+			progress:   pushProgress{current: 20, total: 25},
+			cancelFunc: func(cf context.CancelFunc) { cf() },
+			wantErr:    git2go.ErrorCodeUser,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			g := NewWithT(t)
+
+			ctx, cancel := context.WithCancel(context.TODO())
+			defer cancel()
+
+			ptpcb := pushTransferProgressCallback(ctx)
+
+			tt.cancelFunc(cancel)
+
+			g.Expect(ptpcb(tt.progress.current, tt.progress.total, tt.progress.bytes)).To(Equal(tt.wantErr))
+		})
+	}
+}
+
 func md5Fingerprint(in string) [16]byte {
 	var out [16]byte
 	copy(out[:], in)