Merge remote-tracking branch 'dev/master'

This includes changes for the security release, merged on master though
not yet on the master branch of gitaly @ gitlab.com

9 files changed, 326 insertions, 18 deletions

diff --git a/changelogs/unreleased/security-limit-rpc-negotiation-phase.yml b/changelogs/unreleased/security-limit-rpc-negotiation-phase.yml
new file mode 100644
index 000000000..0f27721e8
--- /dev/null
+++ b/changelogs/unreleased/security-limit-rpc-negotiation-phase.yml
@@ -0,0 +1,5 @@
+---
+title: Limit the negotiation phase for certain Gitaly RPCs
+merge_request:
+author:
+type: security
diff --git a/internal/git/pktline/pktline.go b/internal/git/pktline/pktline.go
index 75eb490f9..d012c8c7d 100644
--- a/internal/git/pktline/pktline.go
+++ b/internal/git/pktline/pktline.go
@@ -16,10 +16,6 @@ const (
 	pktDelim   = "0001"
 )
 
-var (
-	flush = []byte("0000")
-)
-
 // NewScanner returns a bufio.Scanner that splits on Git pktline boundaries
 func NewScanner(r io.Reader) *bufio.Scanner {
 	scanner := bufio.NewScanner(r)
@@ -37,7 +33,7 @@ func Data(pkt []byte) []byte {
 
 // IsFlush detects the special flush packet '0000'
 func IsFlush(pkt []byte) bool {
-	return bytes.Equal(pkt, flush)
+	return bytes.Equal(pkt, PktFlush())
 }
 
 // WriteString writes a string with pkt-line framing
@@ -53,7 +49,7 @@ func WriteString(w io.Writer, str string) (int, error) {
 
 // WriteFlush writes a pkt flush packet.
 func WriteFlush(w io.Writer) error {
-	_, err := w.Write(flush)
+	_, err := w.Write(PktFlush())
 	return err
 }
 
@@ -63,6 +59,16 @@ func WriteDelim(w io.Writer) error {
 	return err
 }
 
+// PktDone returns the bytes for a "done" packet.
+func PktDone() []byte {
+	return []byte("0009done\n")
+}
+
+// PktFlush returns the bytes for a "flush" packet.
+func PktFlush() []byte {
+	return []byte("0000")
+}
+
 func pktLineSplitter(data []byte, atEOF bool) (advance int, token []byte, err error) {
 	if len(data) < 4 {
 		if atEOF && len(data) > 0 {
diff --git a/internal/git/pktline/read_monitor.go b/internal/git/pktline/read_monitor.go
new file mode 100644
index 000000000..4f74fda05
--- /dev/null
+++ b/internal/git/pktline/read_monitor.go
@@ -0,0 +1,82 @@
+package pktline
+
+import (
+	"bytes"
+	"context"
+	"io"
+	"io/ioutil"
+	"os"
+	"time"
+)
+
+// ReadMonitor monitors an io.Reader, waiting for a specified packet. If the
+// packet doesn't come within a timeout, a cancel function is called. This can
+// be used to place a timeout on the *negotiation* phase of some git commands,
+// aborting them if it is exceeded.
+//
+// This timeout prevents a class of "use-after-check" security issue when the
+// access check for a git command is run before the command itself. The user
+// has control of stdin for the git command, and if they can delay input for
+// an arbitrarily long time, they can gain access days or weeks after the
+// access check has completed.
+//
+// This approach is better than placing a timeout on the overall git operation
+// because there is a conflict between mitigating the use-after-check with a
+// short timeout, and allowing long-lived git operations to complete. The
+// negotiation phase is a small proportion of the time taken for a large git
+// fetch, for instance, so tighter limits can be placed on it, leading to a
+// better mitigation.
+type ReadMonitor struct {
+	pr         *os.File
+	pw         *os.File
+	underlying io.Reader
+}
+
+// NewReadMonitor wraps the provided reader with an os.Pipe(), returning the
+// read end for onward use.
+//
+// Call Monitor(pkt, timeout, cancelFn) to start streaming from the reader to
+// to the pipe. The stream will be monitored for a pktline-formatted packet
+// matching pkt. If it isn't seen within the timeout, cancelFn will be called.
+//
+// Resources will be freed when the context is done, but you should close the
+// returned *os.File earlier if possible.
+func NewReadMonitor(ctx context.Context, r io.Reader) (*os.File, *ReadMonitor, error) {
+	pr, pw, err := os.Pipe()
+	if err != nil {
+		return nil, nil, err
+	}
+
+	// Ensure all resources are closed once the context is done
+	go func() {
+		<-ctx.Done()
+
+		pr.Close()
+		pw.Close()
+	}()
+
+	return pr, &ReadMonitor{
+		pr:         pr,
+		pw:         pw,
+		underlying: r,
+	}, nil
+}
+
+// Monitor should be called at most once. It scans the stream, looking for the
+// specified packet, and will call cancelFn if it isn't seen within the timeout
+func (m *ReadMonitor) Monitor(pkt []byte, timeout time.Duration, cancelFn func()) {
+	timer := time.AfterFunc(timeout, cancelFn)
+	teeReader := io.TeeReader(m.underlying, m.pw)
+
+	scanner := NewScanner(teeReader)
+	for scanner.Scan() {
+		if bytes.Equal(scanner.Bytes(), pkt) {
+			timer.Stop()
+			break
+		}
+	}
+
+	// Complete the read loop, then signal completion on pr by closing pw
+	_, _ = io.Copy(ioutil.Discard, teeReader)
+	_ = m.pw.Close()
+}
diff --git a/internal/git/pktline/read_monitor_test.go b/internal/git/pktline/read_monitor_test.go
new file mode 100644
index 000000000..63f26f1db
--- /dev/null
+++ b/internal/git/pktline/read_monitor_test.go
@@ -0,0 +1,88 @@
+package pktline
+
+import (
+	"bytes"
+	"context"
+	"io"
+	"io/ioutil"
+	"os"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"gitlab.com/gitlab-org/gitaly/internal/testhelper"
+)
+
+func TestReadMonitorTimeout(t *testing.T) {
+	waitPipeR, waitPipeW := io.Pipe()
+	defer waitPipeW.Close()
+
+	ctx, cancel := testhelper.Context()
+	defer cancel()
+
+	in := io.MultiReader(
+		strings.NewReader("000ftest string"),
+		waitPipeR, // this pipe reader lets us block the multi reader
+	)
+
+	r, monitor, err := NewReadMonitor(ctx, in)
+	require.NoError(t, err)
+
+	startTime := time.Now()
+	go monitor.Monitor(PktDone(), 10*time.Millisecond, cancel)
+
+	// We should be done quickly
+	<-ctx.Done()
+
+	elapsed := time.Since(startTime)
+	require.Error(t, ctx.Err())
+	require.Equal(t, ctx.Err(), context.Canceled)
+	require.True(t, elapsed < time.Second, "Expected context to be cancelled quickly, but it was not")
+
+	// Verify that pipe is closed
+	_, err = ioutil.ReadAll(r)
+	require.Error(t, err)
+	require.IsType(t, &os.PathError{}, err)
+}
+
+func TestReadMonitorSuccess(t *testing.T) {
+	waitPipeR, waitPipeW := io.Pipe()
+
+	ctx, cancel := testhelper.Context()
+	defer cancel()
+
+	preTimeoutPayload := "000ftest string"
+	postTimeoutPayload := "0017post-timeout string"
+
+	in := io.MultiReader(
+		strings.NewReader(preTimeoutPayload),
+		bytes.NewReader(PktFlush()),
+		waitPipeR, // this pipe reader lets us block the multi reader
+		strings.NewReader(postTimeoutPayload),
+	)
+
+	r, monitor, err := NewReadMonitor(ctx, in)
+	require.NoError(t, err)
+
+	go monitor.Monitor(PktFlush(), 10*time.Millisecond, cancel)
+
+	// Verify the data is passed through correctly
+	scanner := NewScanner(r)
+	require.True(t, scanner.Scan())
+	require.Equal(t, preTimeoutPayload, scanner.Text())
+	require.True(t, scanner.Scan())
+	require.Equal(t, PktFlush(), scanner.Bytes())
+
+	// If the timeout *has* been stopped, a wait this long won't break it
+	time.Sleep(100 * time.Millisecond)
+
+	// Close the pipe to skip to next reader
+	require.NoError(t, waitPipeW.Close())
+
+	// Verify that more data can be sent through the pipe
+	require.True(t, scanner.Scan())
+	require.Equal(t, postTimeoutPayload, scanner.Text())
+
+	require.NoError(t, ctx.Err())
+}
diff --git a/internal/service/ssh/monitor_stdin_command.go b/internal/service/ssh/monitor_stdin_command.go
new file mode 100644
index 000000000..e5de7741e
--- /dev/null
+++ b/internal/service/ssh/monitor_stdin_command.go
@@ -0,0 +1,26 @@
+package ssh
+
+import (
+	"context"
+	"fmt"
+	"io"
+
+	"gitlab.com/gitlab-org/gitaly/internal/command"
+	"gitlab.com/gitlab-org/gitaly/internal/git"
+	"gitlab.com/gitlab-org/gitaly/internal/git/pktline"
+)
+
+func monitorStdinCommand(ctx context.Context, stdin io.Reader, stdout, stderr io.Writer, env []string, globals []git.Option, sc git.SubCmd) (*command.Command, *pktline.ReadMonitor, error) {
+	stdinPipe, monitor, err := pktline.NewReadMonitor(ctx, stdin)
+	if err != nil {
+		return nil, nil, fmt.Errorf("create monitor: %v", err)
+	}
+
+	cmd, err := git.SafeBareCmd(ctx, stdinPipe, stdout, stderr, env, globals, sc)
+	stdinPipe.Close() // this now belongs to cmd
+	if err != nil {
+		return nil, nil, fmt.Errorf("start cmd: %v", err)
+	}
+
+	return cmd, monitor, err
+}
diff --git a/internal/service/ssh/upload_archive.go b/internal/service/ssh/upload_archive.go
index 1d25b0b21..ae8cdcc3c 100644
--- a/internal/service/ssh/upload_archive.go
+++ b/internal/service/ssh/upload_archive.go
@@ -1,15 +1,22 @@
 package ssh
 
 import (
+	"context"
 	"fmt"
+	"time"
 
 	"gitlab.com/gitlab-org/gitaly/internal/command"
 	"gitlab.com/gitlab-org/gitaly/internal/git"
+	"gitlab.com/gitlab-org/gitaly/internal/git/pktline"
 	"gitlab.com/gitlab-org/gitaly/internal/helper"
 	"gitlab.com/gitlab-org/gitaly/proto/go/gitalypb"
 	"gitlab.com/gitlab-org/gitaly/streamio"
 )
 
+var (
+	uploadArchiveRequestTimeout = time.Minute
+)
+
 func (s *server) SSHUploadArchive(stream gitalypb.SSHService_SSHUploadArchiveServer) error {
 	req, err := stream.Recv() // First request contains Repository only
 	if err != nil {
@@ -27,10 +34,14 @@ func (s *server) SSHUploadArchive(stream gitalypb.SSHService_SSHUploadArchiveSer
 }
 
 func sshUploadArchive(stream gitalypb.SSHService_SSHUploadArchiveServer, req *gitalypb.SSHUploadArchiveRequest) error {
+	ctx, cancelCtx := context.WithCancel(stream.Context())
+	defer cancelCtx()
+
 	repoPath, err := helper.GetRepoPath(req.Repository)
 	if err != nil {
 		return err
 	}
+
 	stdin := streamio.NewReader(func() ([]byte, error) {
 		request, err := stream.Recv()
 		return request.GetStdin(), err
@@ -42,15 +53,21 @@ func sshUploadArchive(stream gitalypb.SSHService_SSHUploadArchiveServer, req *gi
 		return stream.Send(&gitalypb.SSHUploadArchiveResponse{Stderr: p})
 	})
 
-	cmd, err := git.SafeBareCmd(stream.Context(), stdin, stdout, stderr, nil, nil, git.SubCmd{
+	cmd, monitor, err := monitorStdinCommand(ctx, stdin, stdout, stderr, nil, nil, git.SubCmd{
 		Name: "upload-archive",
 		Args: []string{repoPath},
 	})
-
 	if err != nil {
-		return fmt.Errorf("start cmd: %v", err)
+		return err
 	}
 
+	// upload-archive expects a list of options terminated by a flush packet:
+	// https://github.com/git/git/blob/v2.22.0/builtin/upload-archive.c#L38
+	//
+	// Place a timeout on receiving the flush packet to mitigate use-after-check
+	// attacks
+	go monitor.Monitor(pktline.PktFlush(), uploadArchiveRequestTimeout, cancelCtx)
+
 	if err := cmd.Wait(); err != nil {
 		if status, ok := command.ExitStatus(err); ok {
 			return stream.Send(&gitalypb.SSHUploadArchiveResponse{
diff --git a/internal/service/ssh/upload_archive_test.go b/internal/service/ssh/upload_archive_test.go
index f9576b1cd..f58013a5c 100644
--- a/internal/service/ssh/upload_archive_test.go
+++ b/internal/service/ssh/upload_archive_test.go
@@ -3,9 +3,11 @@ package ssh
 import (
 	"context"
 	"fmt"
+	"io"
 	"os"
 	"os/exec"
 	"testing"
+	"time"
 
 	"github.com/golang/protobuf/jsonpb"
 	"github.com/stretchr/testify/require"
@@ -14,6 +16,34 @@ import (
 	"google.golang.org/grpc/codes"
 )
 
+var (
+	originalUploadArchiveRequestTimeout = uploadArchiveRequestTimeout
+)
+
+func TestFailedUploadArchiveRequestDueToTimeout(t *testing.T) {
+	uploadArchiveRequestTimeout = time.Millisecond
+	defer func() { uploadArchiveRequestTimeout = originalUploadArchiveRequestTimeout }()
+
+	server, serverSocketPath := runSSHServer(t)
+	defer server.Stop()
+
+	client, conn := newSSHClient(t, serverSocketPath)
+	defer conn.Close()
+
+	ctx, cancel := testhelper.Context()
+	defer cancel()
+
+	stream, err := client.SSHUploadArchive(ctx)
+	require.NoError(t, err)
+
+	// The first request is not limited by timeout, but also not under attacker control
+	require.NoError(t, stream.Send(&gitalypb.SSHUploadArchiveRequest{Repository: testRepo}))
+
+	// The RPC should time out after a short period of sending no data
+	err = testUploadArchiveFailedResponse(t, stream)
+	require.Equal(t, io.EOF, err)
+}
+
 func TestFailedUploadArchiveRequestDueToValidationError(t *testing.T) {
 	server, serverSocketPath := runSSHServer(t)
 	defer server.Stop()
@@ -57,7 +87,7 @@ func TestFailedUploadArchiveRequestDueToValidationError(t *testing.T) {
 			}
 			stream.CloseSend()
 
-			err = drainUploadArchiveResponse(stream)
+			err = testUploadArchiveFailedResponse(t, stream)
 			testhelper.RequireGrpcError(t, err, test.Code)
 		})
 	}
@@ -98,10 +128,14 @@ func testArchive(t *testing.T, serverSocketPath string, testRepo *gitalypb.Repos
 	return nil
 }
 
-func drainUploadArchiveResponse(stream gitalypb.SSHService_SSHUploadArchiveClient) error {
+func testUploadArchiveFailedResponse(t *testing.T, stream gitalypb.SSHService_SSHUploadArchiveClient) error {
 	var err error
+	var res *gitalypb.SSHUploadArchiveResponse
+
 	for err == nil {
-		_, err = stream.Recv()
+		res, err = stream.Recv()
+		require.Nil(t, res.GetStdout())
 	}
+
 	return err
 }
diff --git a/internal/service/ssh/upload_pack.go b/internal/service/ssh/upload_pack.go
index 180dd0221..094afab2c 100644
--- a/internal/service/ssh/upload_pack.go
+++ b/internal/service/ssh/upload_pack.go
@@ -1,15 +1,22 @@
 package ssh
 
 import (
+	"context"
 	"fmt"
+	"time"
 
 	"gitlab.com/gitlab-org/gitaly/internal/command"
 	"gitlab.com/gitlab-org/gitaly/internal/git"
+	"gitlab.com/gitlab-org/gitaly/internal/git/pktline"
 	"gitlab.com/gitlab-org/gitaly/internal/helper"
 	"gitlab.com/gitlab-org/gitaly/proto/go/gitalypb"
 	"gitlab.com/gitlab-org/gitaly/streamio"
 )
 
+var (
+	uploadPackRequestTimeout = 10 * time.Minute
+)
+
 func (s *server) SSHUploadPack(stream gitalypb.SSHService_SSHUploadPackServer) error {
 	req, err := stream.Recv() // First request contains Repository only
 	if err != nil {
@@ -28,7 +35,8 @@ func (s *server) SSHUploadPack(stream gitalypb.SSHService_SSHUploadPackServer) e
 }
 
 func sshUploadPack(stream gitalypb.SSHService_SSHUploadPackServer, req *gitalypb.SSHUploadPackRequest) error {
-	ctx := stream.Context()
+	ctx, cancelCtx := context.WithCancel(stream.Context())
+	defer cancelCtx()
 
 	stdin := streamio.NewReader(func() ([]byte, error) {
 		request, err := stream.Recv()
@@ -55,14 +63,22 @@ func sshUploadPack(stream gitalypb.SSHService_SSHUploadPackServer, req *gitalypb
 		globalOpts = append(globalOpts, git.ValueFlag{"-c", o})
 	}
 
-	cmd, err := git.SafeBareCmd(ctx, stdin, stdout, stderr, env, globalOpts, git.SubCmd{
+	cmd, monitor, err := monitorStdinCommand(ctx, stdin, stdout, stderr, env, globalOpts, git.SubCmd{
 		Name: "upload-pack",
 		Args: []string{repoPath},
 	})
 	if err != nil {
-		return fmt.Errorf("start cmd: %v", err)
+		return err
 	}
 
+	// upload-pack negotiation is terminated by either a flush, or the "done"
+	// packet: https://github.com/git/git/blob/v2.20.0/Documentation/technical/pack-protocol.txt#L335
+	//
+	// "flush" tells the server it can terminate, while "done" tells it to start
+	// generating a packfile. Add a timeout to the second case to mitigate
+	// use-after-check attacks.
+	go monitor.Monitor(pktline.PktDone(), uploadPackRequestTimeout, cancelCtx)
+
 	if err := cmd.Wait(); err != nil {
 		if status, ok := command.ExitStatus(err); ok {
 			return stream.Send(&gitalypb.SSHUploadPackResponse{
diff --git a/internal/service/ssh/upload_pack_test.go b/internal/service/ssh/upload_pack_test.go
index ea08292db..d72b35b6f 100644
--- a/internal/service/ssh/upload_pack_test.go
+++ b/internal/service/ssh/upload_pack_test.go
@@ -4,11 +4,13 @@ import (
 	"bytes"
 	"context"
 	"fmt"
+	"io"
 	"os"
 	"os/exec"
 	"path"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/golang/protobuf/jsonpb"
 	"github.com/stretchr/testify/require"
@@ -18,6 +20,34 @@ import (
 	"google.golang.org/grpc/codes"
 )
 
+var (
+	originalUploadPackRequestTimeout = uploadPackRequestTimeout
+)
+
+func TestFailedUploadPackRequestDueToTimeout(t *testing.T) {
+	uploadPackRequestTimeout = time.Millisecond
+	defer func() { uploadPackRequestTimeout = originalUploadPackRequestTimeout }()
+
+	server, serverSocketPath := runSSHServer(t)
+	defer server.Stop()
+
+	client, conn := newSSHClient(t, serverSocketPath)
+	defer conn.Close()
+
+	ctx, cancel := testhelper.Context()
+	defer cancel()
+
+	stream, err := client.SSHUploadPack(ctx)
+	require.NoError(t, err)
+
+	// The first request is not limited by timeout, but also not under attacker control
+	require.NoError(t, stream.Send(&gitalypb.SSHUploadPackRequest{Repository: testRepo}))
+
+	// The RPC should time out after a short period of sending no data
+	err = testPostUploadPackFailedResponse(t, stream)
+	require.Equal(t, io.EOF, err)
+}
+
 func TestFailedUploadPackRequestDueToValidationError(t *testing.T) {
 	server, serverSocketPath := runSSHServer(t)
 	defer server.Stop()
@@ -61,7 +91,7 @@ func TestFailedUploadPackRequestDueToValidationError(t *testing.T) {
 			}
 			stream.CloseSend()
 
-			err = drainPostUploadPackResponse(stream)
+			err = testPostUploadPackFailedResponse(t, stream)
 			testhelper.RequireGrpcError(t, err, test.Code)
 		})
 	}
@@ -211,10 +241,14 @@ func testClone(t *testing.T, serverSocketPath, storageName, relativePath, localR
 	return string(localHead), string(remoteHead), string(localTags), string(remoteTags), nil
 }
 
-func drainPostUploadPackResponse(stream gitalypb.SSHService_SSHUploadPackClient) error {
+func testPostUploadPackFailedResponse(t *testing.T, stream gitalypb.SSHService_SSHUploadPackClient) error {
 	var err error
+	var res *gitalypb.SSHUploadPackResponse
+
 	for err == nil {
-		_, err = stream.Recv()
+		res, err = stream.Recv()
+		require.Nil(t, res.GetStdout())
 	}
+
 	return err
 }