1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package githttp
4
5import (
6 "bufio"
7 "context"
8 "crypto/rand"
9 "encoding/hex"
10 "encoding/json"
11 "errors"
12 "fmt"
13 "io"
14 "net"
15 "os"
16 "sync"
17 "time"
18
19 "github.com/bnema/zerowrap"
20
21 "bnema.dev/ermite/internal/domain"
22)
23
24// PushHooks is what the hook socket asks of the use cases. app.Pushes fits.
25type PushHooks interface {
26 // Check decides a push before Git applies it; an error matching
27 // domain.ErrRejected refuses it, any other error refuses it as a failure.
28 Check(ctx context.Context, p domain.Principal, name domain.RepoName, updates []domain.RefUpdate) error
29 // Record handles an accepted push.
30 Record(ctx context.Context, p domain.Principal, name domain.RepoName, updates []domain.RefUpdate) error
31}
32
33// readTimeout bounds how long a connection may take to send its request line.
34const readTimeout = 30 * time.Second
35
36// recordTimeout bounds the post-receive recording of a push, which outlives
37// the shutdown of the server.
38const recordTimeout = 30 * time.Second
39
40type hookEntry struct {
41 repo domain.RepoName
42 principal domain.Principal
43}
44
45// HookServer registers the per-request hook tokens and serves the Unix
46// socket the hooks call back on. It implements HookTokens.
47type HookServer struct {
48 path string
49 hooks PushHooks
50
51 mu sync.RWMutex
52 tokens map[string]hookEntry
53
54 connMu sync.Mutex
55 ln net.Listener
56 conns map[net.Conn]struct{}
57 wg sync.WaitGroup
58}
59
60var _ HookTokens = (*HookServer)(nil)
61
62// NewHookServer builds a server for the socket at path. Call Serve to listen.
63func NewHookServer(path string, hooks PushHooks) *HookServer {
64 return &HookServer{path: path, hooks: hooks, tokens: make(map[string]hookEntry), conns: make(map[net.Conn]struct{})}
65}
66
67// Socket returns the path of the socket.
68func (s *HookServer) Socket() string { return s.path }
69
70// Register binds a fresh random token to the repository and principal of a
71// push. The returned release revokes it.
72func (s *HookServer) Register(ctx context.Context, repo domain.RepoName, p domain.Principal) (string, func()) {
73 var raw [32]byte
74 _, _ = rand.Read(raw[:]) // never fails
75 token := hex.EncodeToString(raw[:])
76 s.mu.Lock()
77 s.tokens[token] = hookEntry{repo: repo, principal: p}
78 s.mu.Unlock()
79 log := zerowrap.FromCtx(ctx)
80 log.Debug().Str("repo", repo.String()).Str("token_id", p.TokenID()).Msg("hook token registered")
81 return token, func() {
82 s.mu.Lock()
83 delete(s.tokens, token)
84 s.mu.Unlock()
85 }
86}
87
88func (s *HookServer) lookup(token string) (hookEntry, bool) {
89 s.mu.RLock()
90 defer s.mu.RUnlock()
91 e, ok := s.tokens[token]
92 return e, ok
93}
94
95// Listen creates the socket with mode 0600 (replacing a stale file) and
96// returns once it accepts connections. Serve then handles them.
97func (s *HookServer) Listen(ctx context.Context) error {
98 // sun_path holds 108 bytes on Linux, 104 on the BSDs and macOS.
99 if len(s.path) >= 100 {
100 return fmt.Errorf("hook socket path is too long for a Unix socket (%d bytes, limit about 100): %s; use a shorter %s", len(s.path), s.path, "ERMITE_DATA_DIR")
101 }
102 if err := os.Remove(s.path); err != nil && !errors.Is(err, os.ErrNotExist) {
103 return fmt.Errorf("remove stale hook socket: %w", err)
104 }
105 ln, err := net.Listen("unix", s.path)
106 if err != nil {
107 return fmt.Errorf("listen on hook socket %s: %w", s.path, err)
108 }
109 if err := os.Chmod(s.path, 0o600); err != nil {
110 _ = ln.Close()
111 return fmt.Errorf("chmod hook socket: %w", err)
112 }
113 s.connMu.Lock()
114 s.ln = ln
115 s.connMu.Unlock()
116 log := zerowrap.FromCtx(ctx)
117 log.Debug().Str("socket", s.path).Msg("hook socket created")
118 return nil
119}
120
121// Serve accepts connections until ctx ends or Close is called, then waits for
122// the requests in progress. The logger of ctx is used throughout. It returns
123// nil after a clean stop.
124func (s *HookServer) Serve(ctx context.Context) error {
125 s.connMu.Lock()
126 ln := s.ln
127 s.connMu.Unlock()
128 if ln == nil {
129 return errors.New("hook server is not listening")
130 }
131 log := zerowrap.FromCtx(ctx)
132 log.Info().Str("socket", s.path).Msg("hook socket listening")
133
134 stopped := make(chan struct{})
135 go func() {
136 select {
137 case <-ctx.Done():
138 case <-stopped:
139 }
140 _ = ln.Close()
141 }()
142 defer close(stopped)
143
144 for {
145 conn, err := ln.Accept()
146 if err != nil {
147 s.wg.Wait()
148 _ = os.Remove(s.path)
149 if ctx.Err() != nil || errors.Is(err, net.ErrClosed) {
150 return nil
151 }
152 return fmt.Errorf("accept on hook socket: %w", err)
153 }
154 s.connMu.Lock()
155 s.conns[conn] = struct{}{}
156 s.connMu.Unlock()
157 s.wg.Add(1)
158 go func() {
159 defer s.wg.Done()
160 defer func() {
161 _ = conn.Close()
162 s.connMu.Lock()
163 delete(s.conns, conn)
164 s.connMu.Unlock()
165 }()
166 s.handle(ctx, conn)
167 }()
168 }
169}
170
171// Close stops listening and drops the connections in progress.
172func (s *HookServer) Close(ctx context.Context) {
173 log := zerowrap.FromCtx(ctx)
174 log.Debug().Str("socket", s.path).Msg("closing hook socket")
175 s.connMu.Lock()
176 defer s.connMu.Unlock()
177 if s.ln != nil {
178 _ = s.ln.Close()
179 }
180 for c := range s.conns {
181 _ = c.Close()
182 }
183}
184
185// handle serves one connection: one request line, one reply line.
186func (s *HookServer) handle(ctx context.Context, conn net.Conn) {
187 log := zerowrap.FromCtx(ctx)
188 _ = conn.SetReadDeadline(time.Now().Add(readTimeout))
189 line, err := readLine(conn)
190 if err != nil {
191 log.Warn().Err(err).Msg("read hook request")
192 s.reply(ctx, conn, HookReply{Message: "ermite: malformed hook request"})
193 return
194 }
195 _ = conn.SetReadDeadline(time.Time{})
196
197 var req HookRequest
198 if err := json.Unmarshal(line, &req); err != nil {
199 log.Warn().Err(err).Msg("decode hook request")
200 s.reply(ctx, conn, HookReply{Message: "ermite: malformed hook request"})
201 return
202 }
203 updates, err := req.refUpdates()
204 if err != nil {
205 log.Warn().Err(err).Str("hook", req.Hook).Msg("invalid hook request")
206 s.reply(ctx, conn, HookReply{Message: "ermite: invalid hook request"})
207 return
208 }
209 entry, ok := s.lookup(req.Token)
210 if !ok {
211 log.Warn().Str("hook", req.Hook).Msg("hook request with an unknown or expired token")
212 s.reply(ctx, conn, HookReply{Message: "ermite: unknown or expired hook token"})
213 return
214 }
215
216 rlog := log.WithField("repo", entry.repo.String())
217 switch req.Hook {
218 case HookPreReceive:
219 err := s.hooks.Check(ctx, entry.principal, entry.repo, updates)
220 switch {
221 case err == nil:
222 s.reply(ctx, conn, HookReply{OK: true})
223 case errors.Is(err, domain.ErrRejected):
224 rlog.Warn().Err(err).Int("refs", len(updates)).Msg("push rejected by pre-receive")
225 s.reply(ctx, conn, HookReply{Message: "ermite: push rejected: " + err.Error()})
226 default:
227 rlog.Error().Err(err).Msg("pre-receive check failed")
228 s.reply(ctx, conn, HookReply{Message: "ermite: push refused: internal error"})
229 }
230 case HookPostReceive:
231 // The push is already in the repository: recording it must not be
232 // dropped because the server is shutting down.
233 rctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), recordTimeout)
234 defer cancel()
235 if err := s.hooks.Record(rctx, entry.principal, entry.repo, updates); err != nil {
236 rlog.Error().Err(err).Msg("post-receive recording failed")
237 s.reply(ctx, conn, HookReply{OK: true, Message: "ermite: the push succeeded but recording it failed"})
238 return
239 }
240 s.reply(ctx, conn, HookReply{OK: true})
241 }
242}
243
244func (s *HookServer) reply(ctx context.Context, conn net.Conn, r HookReply) {
245 b, err := encodeLine(r)
246 if err != nil {
247 return
248 }
249 _ = conn.SetWriteDeadline(time.Now().Add(readTimeout))
250 if _, err := conn.Write(b); err != nil {
251 log := zerowrap.FromCtx(ctx)
252 log.Warn().Err(err).Msg("write hook reply")
253 }
254}
255
256// readLine reads one line of at most maxHookMessage bytes.
257func readLine(r io.Reader) ([]byte, error) {
258 br := bufio.NewReaderSize(io.LimitReader(r, maxHookMessage+1), 64<<10)
259 line, err := br.ReadBytes('\n')
260 if err != nil && (!errors.Is(err, io.EOF) || len(line) == 0) {
261 return nil, err
262 }
263 if len(line) > maxHookMessage {
264 return nil, errors.New("hook message too long")
265 }
266 return line, nil
267}