cmd/gitbayd/hook.go

  1package main
  2
  3import (
  4	"bufio"
  5	"context"
  6	"fmt"
  7	"io"
  8	"os"
  9	"os/exec"
 10	"strconv"
 11	"strings"
 12
 13	"github.com/spf13/cobra"
 14
 15	"gitbay.org/gitbay/internal/gitutil"
 16	"gitbay.org/gitbay/internal/hookd"
 17	"gitbay.org/gitbay/internal/policy"
 18)
 19
 20// incomingSHAs lists the commits this push introduces, in order, without
 21// duplicates. It runs in the hook process, which inherits git's quarantine
 22// environment — the daemon cannot see these objects yet.
 23func incomingSHAs(updates []policy.RefUpdate) ([]string, error) {
 24	seen := map[string]bool{}
 25	var out []string
 26	for _, u := range updates {
 27		if u.IsDelete {
 28			continue
 29		}
 30		// Everything reachable from the new tip that no existing ref has.
 31		raw, err := exec.Command("git", "rev-list", u.New, "--not", "--all").Output()
 32		if err != nil {
 33			return nil, fmt.Errorf("rev-list %s: %w", u.New, err)
 34		}
 35		for _, sha := range strings.Fields(string(raw)) {
 36			if seen[sha] {
 37				continue
 38			}
 39			seen[sha] = true
 40			out = append(out, sha)
 41		}
 42	}
 43	return out, nil
 44}
 45
 46// streamIncomingCommits reads every incoming commit through one
 47// `cat-file --batch` and hands each to emit as it arrives.
 48//
 49// This used to fork a cat-file per commit and build the whole payload in
 50// memory before sending it: a 50k-commit first push to a protected branch
 51// forked 50k processes and held 50k raw commits at once (#100). One
 52// subprocess now serves the whole push, and nothing is accumulated.
 53func streamIncomingCommits(updates []policy.RefUpdate, emit func(hookd.RawCommit) error) error {
 54	shas, err := incomingSHAs(updates)
 55	if err != nil {
 56		return err
 57	}
 58	if len(shas) == 0 {
 59		return nil
 60	}
 61	// Cancelling kills git on an early return. Without it, bailing out
 62	// part-way through a large push leaves git blocked writing into a
 63	// pipe nobody is reading and Wait blocked on git.
 64	ctx, cancel := context.WithCancel(context.Background())
 65	cmd := exec.CommandContext(ctx, "git", "cat-file", "--batch")
 66	stdin, err := cmd.StdinPipe()
 67	if err != nil {
 68		cancel()
 69		return err
 70	}
 71	stdout, err := cmd.StdoutPipe()
 72	if err != nil {
 73		cancel()
 74		return err
 75	}
 76	if err := cmd.Start(); err != nil {
 77		cancel()
 78		return fmt.Errorf("cat-file --batch: %w", err)
 79	}
 80	defer cmd.Wait() // second: reaps the process cancel just signalled
 81	defer cancel()
 82	// Feeding stdin from another goroutine: the pipe buffer is smaller
 83	// than 50k object names, so writing them all before reading would
 84	// block against a git that is blocked writing its own output.
 85	writeErr := make(chan error, 1)
 86	go func() {
 87		defer stdin.Close()
 88		w := bufio.NewWriter(stdin)
 89		for _, sha := range shas {
 90			if _, err := fmt.Fprintln(w, sha); err != nil {
 91				writeErr <- err
 92				return
 93			}
 94		}
 95		writeErr <- w.Flush()
 96	}()
 97
 98	r := bufio.NewReader(stdout)
 99	for range shas {
100		// Each record is "<oid> <type> <size>\n", then size bytes, then
101		// a newline.
102		header, err := r.ReadString('\n')
103		if err != nil {
104			return fmt.Errorf("cat-file --batch: %w", err)
105		}
106		fields := strings.Fields(header)
107		if len(fields) != 3 {
108			return fmt.Errorf("cat-file --batch: unexpected %q", strings.TrimSpace(header))
109		}
110		size, err := strconv.Atoi(fields[2])
111		if err != nil {
112			return fmt.Errorf("cat-file --batch: bad size in %q", strings.TrimSpace(header))
113		}
114		raw := make([]byte, size)
115		if _, err := io.ReadFull(r, raw); err != nil {
116			return fmt.Errorf("cat-file %s: %w", fields[0], err)
117		}
118		if _, err := r.Discard(1); err != nil {
119			return fmt.Errorf("cat-file %s: %w", fields[0], err)
120		}
121		if err := emit(hookd.RawCommit{SHA: fields[0], Raw: raw}); err != nil {
122			return err
123		}
124	}
125	if err := <-writeErr; err != nil {
126		return fmt.Errorf("cat-file --batch: %w", err)
127	}
128	return nil
129}
130
131// hookCmd runs inside a git hook. It computes git facts here — the hook
132// process inherits git's quarantine environment, so incoming objects are
133// visible — and asks the daemon for a policy decision over the unix socket.
134func hookCmd() *cobra.Command {
135	return &cobra.Command{
136		Use:    "hook <pre-receive|post-receive>",
137		Hidden: true,
138		Args:   cobra.ExactArgs(1),
139		RunE: func(cmd *cobra.Command, args []string) error {
140			sock := os.Getenv(hookd.EnvSocket)
141			repoID, err1 := strconv.ParseInt(os.Getenv(hookd.EnvRepoID), 10, 64)
142			userID, err2 := strconv.ParseInt(os.Getenv(hookd.EnvUserID), 10, 64)
143			if sock == "" || err1 != nil || err2 != nil {
144				return fmt.Errorf("missing GITBAY_* environment; this command only runs as a git hook")
145			}
146
147			var updates []policy.RefUpdate
148			scanner := bufio.NewScanner(os.Stdin)
149			for scanner.Scan() {
150				fields := strings.Fields(scanner.Text())
151				if len(fields) != 3 {
152					continue
153				}
154				u := policy.RefUpdate{Old: fields[0], New: fields[1], Ref: fields[2]}
155				u.IsDelete = gitutil.ZeroSHA(u.New)
156				if !u.IsDelete && !gitutil.ZeroSHA(u.Old) {
157					anc, err := gitutil.IsAncestor(".", u.Old, u.New)
158					if err != nil {
159						return fmt.Errorf("checking ancestry for %s: %w", u.Ref, err)
160					}
161					u.IsForce = !anc
162				}
163				updates = append(updates, u)
164			}
165			if err := scanner.Err(); err != nil {
166				return err
167			}
168
169			resp, err := hookd.Ask(sock, hookd.Request{
170				Hook:    args[0],
171				RepoID:  repoID,
172				UserID:  userID,
173				Updates: updates,
174			}, func(emit func(hookd.RawCommit) error) error {
175				return streamIncomingCommits(updates, emit)
176			})
177			if err != nil {
178				return fmt.Errorf("gitbay daemon unreachable: %w", err)
179			}
180			if !resp.Allow {
181				fmt.Fprintln(os.Stderr, resp.Message)
182				os.Exit(1)
183			}
184			return nil
185		},
186	}
187}