Limit the negotiation phase for certain Gitaly RPCs

In most cases, Gitaly trusts that the caller of the RPC has validated
that the user is permitted to perform the action represented by the RPC
and doesn't repeat any access control checks. Where an RPC reads data
from a client-controlled stream before acting, the time between the
check and the operation can be artificially extended. This can lead to
security issues where

Solve this by placing a limit on the *negotiation phase* of two RPCs
that are known to be vulnerable:

* ssh.SSHUploadPack
* ssh.SSHUploadArchive

These RPCs are known not to be vulnerable, for one reason or another:

* ssh.SSHReceivePack
* smarthttp.ReceivePack

The smarthttp.UploadPack RPC is vulnerable, but the vulnerability is
being handled in Workhorse.

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
 }