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