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