SSH Upload/ReceivePack Integration Tests213-test

11 files changed, 457 insertions, 16 deletions

diff --git a/.gitignore b/.gitignore
index 0aba6d488..9e359f4d6 100644
--- a/.gitignore
+++ b/.gitignore
@@ -1,5 +1,6 @@
 /_build
 /gitaly
+/gitaly-*
 /*.deb
 /_support/package/bin
 /_support/bin
diff --git a/client/client.go b/client/client.go
new file mode 100644
index 000000000..2e5093283
--- /dev/null
+++ b/client/client.go
@@ -0,0 +1,34 @@
+package client
+
+import (
+	"net"
+	"time"
+
+	"google.golang.org/grpc"
+)
+
+// Client holds a single gRPC connection
+type Client struct {
+	conn *grpc.ClientConn
+}
+
+// NewClient connects to a gRPC-server and returns the Client
+func NewClient(addr string) (*Client, error) {
+	connOpts := []grpc.DialOption{
+		grpc.WithInsecure(),
+		grpc.WithDialer(func(addr string, _ time.Duration) (net.Conn, error) {
+			return net.Dial("unix", addr)
+		}),
+	}
+	conn, err := grpc.Dial(addr, connOpts...)
+	if err != nil {
+		return nil, err
+	}
+
+	return &Client{conn: conn}, nil
+}
+
+// Close the gRPC-connection
+func (cli *Client) Close() error {
+	return cli.conn.Close()
+}
diff --git a/client/receivepack.go b/client/receivepack.go
new file mode 100644
index 000000000..69f5ce702
--- /dev/null
+++ b/client/receivepack.go
@@ -0,0 +1,101 @@
+package client
+
+import (
+	"io"
+	"io/ioutil"
+	"time"
+
+	"golang.org/x/net/context"
+
+	pb "gitlab.com/gitlab-org/gitaly-proto/go"
+	pbhelper "gitlab.com/gitlab-org/gitaly-proto/go/helper"
+)
+
+// ReceivePack gets git-pack from Gitaly
+func (cli *Client) ReceivePack(stdin io.Reader, stdout, stderr io.Writer, repo, glID string) (int32, error) {
+	ssh := pb.NewSSHClient(cli.conn)
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+	stream, err := ssh.SSHReceivePack(ctx)
+	if err != nil {
+		return 1, err
+	}
+
+	req := &pb.SSHReceivePackRequest{
+		Repository: &pb.Repository{Path: repo},
+		GlId:       glID,
+	}
+
+	if err = stream.Send(req); err != nil {
+		return 1, err
+	}
+
+	inWriter := pbhelper.NewSendWriter(func(p []byte) error {
+		return stream.Send(&pb.SSHReceivePackRequest{Stdin: p})
+	})
+
+	errC := make(chan error, 1)
+
+	go func() {
+		buf := make([]byte, 1024)
+		for {
+			n, errRead := stdin.Read(buf)
+			if errRead == io.EOF {
+				// NOTE: nasty hack to close stdin
+				inWriter.Write(nil)
+			}
+			if errRead != nil {
+				errC <- errRead
+			}
+			if n > 0 {
+				_, errWrite := inWriter.Write(buf[:n])
+				if errWrite != nil {
+					errC <- errWrite
+				}
+			}
+		}
+	}()
+
+	exitStatus := int32(-1)
+	recv := pbhelper.NewReceiveReader(func() ([]byte, error) {
+		resp, errRecv := stream.Recv()
+		if errRecv != nil {
+			return nil, errRecv
+		}
+		if resp.ExitStatus != nil {
+			exitStatus = resp.GetExitStatus().GetValue()
+			return nil, io.EOF
+		}
+
+		if len(resp.Stderr) > 0 {
+			if _, errWrite := stderr.Write(resp.Stderr); errWrite != nil {
+				return nil, errWrite
+			}
+		}
+
+		if len(resp.Stdout) > 0 {
+			if _, errWrite := stdout.Write(resp.Stdout); errWrite != nil {
+				return nil, errWrite
+			}
+		}
+
+		return nil, nil
+	})
+
+	if _, err = io.Copy(ioutil.Discard, recv); err != nil {
+		return 1, err
+	}
+
+	if err = stream.CloseSend(); err != nil {
+		return 1, err
+	}
+
+	if _, err = io.Copy(ioutil.Discard, recv); err != nil {
+		return 1, err
+	}
+
+	if err := <-errC; err != nil && err != io.EOF {
+		return 1, err
+	}
+	return exitStatus, nil
+}
diff --git a/client/uploadpack.go b/client/uploadpack.go
new file mode 100644
index 000000000..eca4f5443
--- /dev/null
+++ b/client/uploadpack.go
@@ -0,0 +1,104 @@
+package client
+
+import (
+	"io"
+	"io/ioutil"
+	"time"
+
+	"golang.org/x/net/context"
+
+	pb "gitlab.com/gitlab-org/gitaly-proto/go"
+	pbhelper "gitlab.com/gitlab-org/gitaly-proto/go/helper"
+)
+
+// UploadPack sends a git-pack payload to Gitaly
+func (cli *Client) UploadPack(stdin io.Reader, stdout, stderr io.Writer, repo string) (int32, error) {
+	ssh := pb.NewSSHClient(cli.conn)
+	// TODO: Set a proper timeout
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+	//ctx := context.Background()
+	stream, err := ssh.SSHUploadPack(ctx)
+	if err != nil {
+		return 1, err
+	}
+
+	req := &pb.SSHUploadPackRequest{
+		Repository: &pb.Repository{Path: repo},
+	}
+
+	if err = stream.Send(req); err != nil {
+		return 1, err
+	}
+
+	inWriter := pbhelper.NewSendWriter(func(p []byte) error {
+		return stream.Send(&pb.SSHUploadPackRequest{Stdin: p})
+	})
+
+	errC := make(chan error, 1)
+
+	go func() {
+		buf := make([]byte, 1024)
+		for {
+			n, errRead := stdin.Read(buf)
+			if errRead == io.EOF {
+				// NOTE: nasty hack to close stdin
+				inWriter.Write(nil)
+				errC <- nil
+			}
+			if errRead != nil {
+				errC <- errRead
+			}
+			if n > 0 {
+				_, errWrite := inWriter.Write(buf[:n])
+				if errWrite != nil {
+					errC <- errWrite
+				}
+			}
+		}
+	}()
+
+	exitStatus := int32(-1)
+	recv := pbhelper.NewReceiveReader(func() ([]byte, error) {
+		resp, errRecv := stream.Recv()
+		if errRecv != nil {
+			return nil, errRecv
+		}
+		if resp.ExitStatus != nil {
+			exitStatus = resp.GetExitStatus().GetValue()
+			return nil, io.EOF
+		}
+
+		if len(resp.Stderr) > 0 {
+			if _, errWrite := stderr.Write(resp.Stderr); errWrite != nil {
+				return nil, errWrite
+			}
+			return nil, io.EOF
+		}
+
+		if len(resp.Stdout) > 0 {
+			if _, errWrite := stdout.Write(resp.Stdout); errWrite != nil {
+				return nil, errWrite
+			}
+		}
+
+		return nil, nil
+	})
+
+	if _, err = io.Copy(ioutil.Discard, recv); err != nil {
+		return 1, err
+	}
+
+	if err = stream.CloseSend(); err != nil {
+		return 1, err
+	}
+
+	if _, err = io.Copy(ioutil.Discard, recv); err != nil {
+		return 1, err
+	}
+
+	if err := <-errC; err != nil && err != io.EOF {
+		return 1, err
+	}
+	return exitStatus, nil
+}
diff --git a/cmd/gitaly-receive-pack/main.go b/cmd/gitaly-receive-pack/main.go
new file mode 100644
index 000000000..0eaaea643
--- /dev/null
+++ b/cmd/gitaly-receive-pack/main.go
@@ -0,0 +1,34 @@
+package main
+
+import (
+	"log"
+	"os"
+	"strings"
+
+	"gitlab.com/gitlab-org/gitaly/client"
+)
+
+func main() {
+	if !(len(os.Args) >= 3 && strings.HasPrefix(os.Args[2], "git-receive-pack")) {
+		log.Fatalf("Not a valid command")
+	}
+
+	addr := os.Getenv("GITALY_SOCKET")
+	if len(addr) == 0 {
+		log.Fatalf("GITALY_SOCKET not set")
+	}
+
+	cli, err := client.NewClient(addr)
+	if err != nil {
+		log.Fatalf("Error: %v", err)
+	}
+	defer cli.Close()
+
+	code, err := cli.ReceivePack(os.Stdin, os.Stdout, os.Stderr, os.Getenv("GL_REPOSITORY"), os.Getenv("GL_ID"))
+	if err != nil {
+		log.Printf("Error: %v", err)
+		os.Exit(int(code))
+	}
+
+	os.Exit(int(code))
+}
diff --git a/cmd/gitaly-upload-pack/main.go b/cmd/gitaly-upload-pack/main.go
new file mode 100644
index 000000000..b2731c7d5
--- /dev/null
+++ b/cmd/gitaly-upload-pack/main.go
@@ -0,0 +1,34 @@
+package main
+
+import (
+	"log"
+	"os"
+	"strings"
+
+	"gitlab.com/gitlab-org/gitaly/client"
+)
+
+func main() {
+	if !(len(os.Args) >= 3 && strings.HasPrefix(os.Args[2], "git-upload-pack")) {
+		log.Fatalf("Not a valid command")
+	}
+
+	addr := os.Getenv("GITALY_SOCKET")
+	if len(addr) == 0 {
+		log.Fatalf("GITALY_SOCKET not set")
+	}
+
+	cli, err := client.NewClient(addr)
+	if err != nil {
+		log.Fatalf("Error: %v", err)
+	}
+	defer cli.Close()
+
+	code, err := cli.UploadPack(os.Stdin, os.Stdout, os.Stderr, os.Getenv("GL_REPOSITORY"))
+	if err != nil {
+		log.Printf("Error: %v", err)
+		os.Exit(int(code))
+	}
+
+	os.Exit(int(code))
+}
diff --git a/internal/service/ssh/receive_pack.go b/internal/service/ssh/receive_pack.go
index a9081516f..6c225d44b 100644
--- a/internal/service/ssh/receive_pack.go
+++ b/internal/service/ssh/receive_pack.go
@@ -2,6 +2,7 @@ package ssh
 
 import (
 	"fmt"
+	"io"
 	"os/exec"
 
 	log "github.com/sirupsen/logrus"
@@ -23,8 +24,12 @@ func (s *server) SSHReceivePack(stream pb.SSH_SSHReceivePackServer) error {
 	}
 
 	stdin := pbhelper.NewReceiveReader(func() ([]byte, error) {
-		request, err := stream.Recv()
-		return request.GetStdin(), err
+		request, errRecv := stream.Recv()
+		if len(request.Stdin) <= 0 {
+			// NOTE: Nasty hack to close stdin
+			return nil, io.EOF
+		}
+		return request.GetStdin(), errRecv
 	})
 	stdout := pbhelper.NewSendWriter(func(p []byte) error {
 		return stream.Send(&pb.SSHReceivePackResponse{Stdout: p})
@@ -42,6 +47,12 @@ func (s *server) SSHReceivePack(stream pb.SSH_SSHReceivePackServer) error {
 
 	repoPath, err := helper.GetRepoPath(req.Repository)
 	if err != nil {
+
+		log.WithFields(log.Fields{
+			"RepoPath":     repoPath,
+			"GlID":         req.GlId,
+			"GlRepository": req.GlRepository,
+		}).Debug("SSHReceivePack GetRepoPath")
 		return err
 	}
 
@@ -69,7 +80,10 @@ func (s *server) SSHReceivePack(stream pb.SSH_SSHReceivePackServer) error {
 		return grpc.Errorf(codes.Unavailable, "SSHReceivePack: cmd wait for %v: %v", cmd.Args, err)
 	}
 
-	return nil
+	return helper.DecorateError(
+		codes.Internal,
+		stream.Send(&pb.SSHReceivePackResponse{ExitStatus: &pb.ExitStatus{Value: int32(0)}}),
+	)
 }
 
 func validateFirstReceivePackRequest(req *pb.SSHReceivePackRequest) error {
diff --git a/internal/service/ssh/receive_pack_test.go b/internal/service/ssh/receive_pack_test.go
index 813e292c8..4a619875e 100644
--- a/internal/service/ssh/receive_pack_test.go
+++ b/internal/service/ssh/receive_pack_test.go
@@ -1,7 +1,13 @@
 package ssh
 
 import (
+	"bytes"
+	"fmt"
+	"os"
+	"os/exec"
+	"path"
 	"testing"
+	"time"
 
 	"gitlab.com/gitlab-org/gitaly/internal/testhelper"
 
@@ -12,9 +18,6 @@ import (
 )
 
 func TestFailedReceivePackRequestDueToValidationError(t *testing.T) {
-	server := runSSHServer(t)
-	defer server.Stop()
-
 	client := newSSHClient(t)
 
 	rpcRequests := []pb.SSHReceivePackRequest{
@@ -41,6 +44,68 @@ func TestFailedReceivePackRequestDueToValidationError(t *testing.T) {
 	}
 }
 
+func TestSuccessReceivePack(t *testing.T) {
+	remoteRepoPath := path.Join(testRepoRoot, "gitlab-test-remote")
+	localRepoPath := path.Join(testRepoRoot, "gitlab-test-local")
+	// Make a bare clone of the test repo to act as a remote one and to leave the original repo intact for other tests
+	testhelper.MustRunCommand(t, nil, "git", "clone", "--bare", testRepoPath, remoteRepoPath)
+	// Make a non-bare clone of the test repo to act as a local one
+	testhelper.MustRunCommand(t, nil, "git", "clone", remoteRepoPath, localRepoPath)
+	// We need git thinking we're pushing over SSH...
+	testhelper.MustRunCommand(t, nil, "git", "-C", localRepoPath, "remote", "set-url", "origin", "git@localhost:test/test.git")
+	defer os.RemoveAll(remoteRepoPath)
+	defer os.RemoveAll(localRepoPath)
+
+	makeCommit(t, localRepoPath)
+
+	cmd := exec.Command("git", "-C", localRepoPath, "push", "origin", "master")
+	cmd.Env = []string{
+		// Running inside the context of the Repo... hence the ugly paths...
+		fmt.Sprintf("GITALY_SOCKET=../../../%s", serverSocketPath),
+		fmt.Sprintf("GL_REPOSITORY=%s", remoteRepoPath),
+		"GL_ID=1",
+		"GIT_SSH_COMMAND='../../../../../../gitaly-receive-pack'",
+	}
+	out, err := cmd.CombinedOutput()
+	if err != nil {
+		t.Fatalf("Error pushing: %v: %q", err, out)
+	}
+	if !cmd.ProcessState.Success() {
+		t.Fatalf("Failed to run `git clone`: %q", out)
+	}
+
+	localHead := bytes.TrimSpace(testhelper.MustRunCommand(t, nil, "git", "-C", localRepoPath, "rev-parse", "master"))
+	remoteHead := bytes.TrimSpace(testhelper.MustRunCommand(t, nil, "git", "-C", remoteRepoPath, "rev-parse", "master"))
+
+	if bytes.Compare(localHead, remoteHead) != 0 {
+		t.Errorf("local and remote head not equal. push failed: %q != %q", localHead, remoteHead)
+	}
+}
+
+// makeCommit creates a new commit and returns oldHead, newHead, success
+func makeCommit(t *testing.T, localRepoPath string) ([]byte, []byte, bool) {
+	commitMsg := fmt.Sprintf("Testing ReceivePack RPC around %d", time.Now().Unix())
+	committerName := "Scrooge McDuck"
+	committerEmail := "scrooge@mcduck.com"
+
+	// The latest commit ID on the remote repo
+	oldHead := bytes.TrimSpace(testhelper.MustRunCommand(t, nil, "git", "-C", localRepoPath, "rev-parse", "master"))
+
+	testhelper.MustRunCommand(t, nil, "git", "-C", localRepoPath,
+		"-c", fmt.Sprintf("user.name=%s", committerName),
+		"-c", fmt.Sprintf("user.email=%s", committerEmail),
+		"commit", "--allow-empty", "-m", commitMsg)
+	if t.Failed() {
+		return nil, nil, false
+	}
+
+	// The commit ID we want to push to the remote repo
+	newHead := bytes.TrimSpace(testhelper.MustRunCommand(t, nil, "git", "-C", localRepoPath, "rev-parse", "master"))
+
+	return oldHead, newHead, t.Failed()
+
+}
+
 func drainPostReceivePackResponse(stream pb.SSH_SSHReceivePackClient) error {
 	var err error
 	for err == nil {
diff --git a/internal/service/ssh/testhelper_test.go b/internal/service/ssh/testhelper_test.go
index 833217e3a..0650c6e74 100644
--- a/internal/service/ssh/testhelper_test.go
+++ b/internal/service/ssh/testhelper_test.go
@@ -10,35 +10,50 @@ import (
 	log "github.com/sirupsen/logrus"
 
 	pb "gitlab.com/gitlab-org/gitaly-proto/go"
+	"gitlab.com/gitlab-org/gitaly/internal/testhelper"
 
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/reflection"
 )
 
 const (
-	scratchDir = "testdata/scratch"
+	scratchDir   = "testdata/scratch"
+	testRepoRoot = "testdata/data"
 )
 
 var (
 	serverSocketPath = path.Join(scratchDir, "gitaly.sock")
+	testRepoPath     = ""
 )
 
 func TestMain(m *testing.M) {
+	testRepoPath = testhelper.GitlabTestRepoPath()
+
+	os.RemoveAll(testRepoRoot)
+	if err := os.MkdirAll(testRepoRoot, 0755); err != nil {
+		log.Fatal(err)
+	}
+	defer os.RemoveAll(testRepoRoot)
+
+	os.RemoveAll(scratchDir)
 	if err := os.MkdirAll(scratchDir, 0755); err != nil {
 		log.Fatal(err)
 	}
 	defer os.RemoveAll(scratchDir)
 
+	server := runSSHServer()
+	defer server.Stop()
+
 	os.Exit(func() int {
 		return m.Run()
 	}())
 }
 
-func runSSHServer(t *testing.T) *grpc.Server {
+func runSSHServer() *grpc.Server {
 	server := grpc.NewServer()
 	listener, err := net.Listen("unix", serverSocketPath)
 	if err != nil {
-		t.Fatal(err)
+		log.Fatal(err)
 	}
 
 	pb.RegisterSSHServer(server, NewServer())
diff --git a/internal/service/ssh/upload_pack_test.go b/internal/service/ssh/upload_pack_test.go
index da223e100..cc38a165f 100644
--- a/internal/service/ssh/upload_pack_test.go
+++ b/internal/service/ssh/upload_pack_test.go
@@ -1,6 +1,10 @@
 package ssh
 
 import (
+	"fmt"
+	"os"
+	"os/exec"
+	"path"
 	"testing"
 
 	"gitlab.com/gitlab-org/gitaly/internal/testhelper"
@@ -12,9 +16,6 @@ import (
 )
 
 func TestFailedUploadPackRequestDueToValidationError(t *testing.T) {
-	server := runSSHServer(t)
-	defer server.Stop()
-
 	client := newSSHClient(t)
 
 	rpcRequests := []pb.SSHUploadPackRequest{
@@ -40,6 +41,30 @@ func TestFailedUploadPackRequestDueToValidationError(t *testing.T) {
 	}
 }
 
+func TestSuccessUploadPack(t *testing.T) {
+	remoteRepoPath := path.Join(testRepoRoot, "gitlab-test-remote")
+	localRepoPath := path.Join(testRepoRoot, "gitlab-test-local")
+	// Make a bare clone of the test repo to act as a remote one and to leave the original repo intact for other tests
+	testhelper.MustRunCommand(t, nil, "git", "clone", "--bare", testRepoPath, remoteRepoPath)
+	defer os.RemoveAll(remoteRepoPath)
+	defer os.RemoveAll(localRepoPath)
+
+	cmd := exec.Command("git", "clone", fmt.Sprintf("git@localhost:%s", remoteRepoPath), localRepoPath)
+	cmd.Env = []string{
+		fmt.Sprintf("GITALY_SOCKET=%s", serverSocketPath),
+		fmt.Sprintf("GL_REPOSITORY=%s", remoteRepoPath),
+		"GIT_SSH_COMMAND='../../../gitaly-upload-pack'",
+	}
+	out, err := cmd.CombinedOutput()
+	if err != nil {
+		t.Fatalf("Error cloning: %v: %q", err, out)
+	}
+	if !cmd.ProcessState.Success() {
+		t.Fatalf("Failed to run `git clone`: %q", out)
+	}
+	testhelper.MustRunCommand(t, nil, "git", "-C", localRepoPath, "status")
+}
+
 func drainPostUploadPackResponse(stream pb.SSH_SSHUploadPackClient) error {
 	var err error
 	for err == nil {
diff --git a/internal/service/ssh/uploadpack.go b/internal/service/ssh/uploadpack.go
index 276d4dd36..e208440d5 100644
--- a/internal/service/ssh/uploadpack.go
+++ b/internal/service/ssh/uploadpack.go
@@ -1,6 +1,7 @@
 package ssh
 
 import (
+	"io"
 	"os/exec"
 
 	log "github.com/sirupsen/logrus"
@@ -20,9 +21,14 @@ func (s *server) SSHUploadPack(stream pb.SSH_SSHUploadPackServer) error {
 		return err
 	}
 
-	stdin := pbhelper.NewReceiveReader(func() ([]byte, error) {
-		request, err := stream.Recv()
-		return request.GetStdin(), err
+	var stdin io.Reader
+	stdin = pbhelper.NewReceiveReader(func() ([]byte, error) {
+		request, errRecv := stream.Recv()
+		if len(request.Stdin) <= 0 {
+			// NOTE: Nasty hack to close stdin
+			return nil, io.EOF
+		}
+		return request.GetStdin(), errRecv
 	})
 	stdout := pbhelper.NewSendWriter(func(p []byte) error {
 		return stream.Send(&pb.SSHUploadPackResponse{Stdout: p})
@@ -49,6 +55,7 @@ func (s *server) SSHUploadPack(stream pb.SSH_SSHUploadPackServer) error {
 
 	if err := cmd.Wait(); err != nil {
 		if status, ok := helper.ExitStatus(err); ok {
+
 			return helper.DecorateError(
 				codes.Internal,
 				stream.Send(&pb.SSHUploadPackResponse{ExitStatus: &pb.ExitStatus{Value: int32(status)}}),
@@ -57,7 +64,14 @@ func (s *server) SSHUploadPack(stream pb.SSH_SSHUploadPackServer) error {
 		return grpc.Errorf(codes.Unavailable, "SSHUploadPack: cmd wait for %v: %v", cmd.Args, err)
 	}
 
-	return nil
+	log.WithFields(log.Fields{
+		"RepoPath": repoPath,
+	}).Debug("SSHUploadPack Done")
+
+	return helper.DecorateError(
+		codes.Internal,
+		stream.Send(&pb.SSHUploadPackResponse{ExitStatus: &pb.ExitStatus{Value: int32(0)}}),
+	)
 }
 
 func validateFirstUploadPackRequest(req *pb.SSHUploadPackRequest) error {