1package registry
2
3import (
4 "context"
5 "crypto/sha256"
6 "encoding/json"
7 "fmt"
8 "io"
9 "net/http"
10 "strconv"
11 "strings"
12
13 "gordon/internal/config"
14 "gordon/internal/events"
15 "gordon/internal/middleware"
16 "gordon/pkg/manifest"
17
18 "github.com/rs/zerolog/log"
19)
20
21type Server struct {
22 config *config.Config
23 mux *http.ServeMux
24 storage Storage
25 eventBus events.EventBus
26}
27
28func NewServer(cfg *config.Config, eventBus events.EventBus) (*Server, error) {
29 // Initialize storage
30 registryPath := cfg.Server.DataDir + "/registry"
31 log.Debug().Str("data_dir", cfg.Server.DataDir).Str("registry_path", registryPath).Msg("Initializing registry storage")
32
33 storage, err := NewFilesystemStorage(registryPath)
34 if err != nil {
35 return nil, fmt.Errorf("failed to initialize storage: %w", err)
36 }
37
38 s := &Server{
39 config: cfg,
40 mux: http.NewServeMux(),
41 storage: storage,
42 eventBus: eventBus,
43 }
44 s.setupRoutes()
45 return s, nil
46}
47
48func (s *Server) setupRoutes() {
49 // Docker Registry API v2 endpoints
50 // Use traditional patterns since repository names can contain slashes
51 s.mux.HandleFunc("/v2/", s.handleRegistryRoutes)
52}
53
54func (s *Server) handleRegistryRoutes(w http.ResponseWriter, r *http.Request) {
55 path := r.URL.Path
56
57 // Route manifest operations: /v2/{name}/manifests/{reference}
58 if strings.Contains(path, "/manifests/") {
59 s.handleManifestRoutes(w, r)
60 return
61 }
62
63 // Route blob operations: /v2/{name}/blobs/{digest}
64 if strings.Contains(path, "/blobs/") && !strings.Contains(path, "/uploads/") {
65 s.handleBlobRoutes(w, r)
66 return
67 }
68
69 // Route blob upload operations: /v2/{name}/blobs/uploads/
70 if strings.Contains(path, "/blobs/uploads/") {
71 s.handleBlobUploadRoutes(w, r)
72 return
73 }
74
75 // Route tag list operations: /v2/{name}/tags/list
76 if strings.Contains(path, "/tags/list") {
77 s.handleTagListRoutes(w, r)
78 return
79 }
80
81 // Base endpoint: /v2/
82 if path == "/v2/" {
83 s.handleBase(w, r)
84 return
85 }
86
87 http.NotFound(w, r)
88}
89
90func (s *Server) handleManifestRoutes(w http.ResponseWriter, r *http.Request) {
91 // Parse path: /v2/{name}/manifests/{reference}
92 parts := strings.Split(strings.TrimPrefix(r.URL.Path, "/v2/"), "/")
93 if len(parts) < 3 || parts[len(parts)-2] != "manifests" {
94 http.NotFound(w, r)
95 return
96 }
97
98 reference := parts[len(parts)-1]
99 name := strings.Join(parts[:len(parts)-2], "/")
100
101 // Set path values for compatibility with existing handlers
102 r.SetPathValue("name", name)
103 r.SetPathValue("reference", reference)
104
105 switch r.Method {
106 case "HEAD", "GET":
107 s.handleGetManifest(w, r)
108 case "PUT":
109 s.handlePutManifest(w, r)
110 default:
111 http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
112 }
113}
114
115func (s *Server) handleBlobRoutes(w http.ResponseWriter, r *http.Request) {
116 // Parse path: /v2/{name}/blobs/{digest}
117 parts := strings.Split(strings.TrimPrefix(r.URL.Path, "/v2/"), "/")
118 if len(parts) < 3 || parts[len(parts)-2] != "blobs" {
119 http.NotFound(w, r)
120 return
121 }
122
123 digest := parts[len(parts)-1]
124 name := strings.Join(parts[:len(parts)-2], "/")
125
126 r.SetPathValue("name", name)
127 r.SetPathValue("digest", digest)
128
129 switch r.Method {
130 case "HEAD", "GET":
131 s.handleGetBlob(w, r)
132 default:
133 http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
134 }
135}
136
137func (s *Server) handleBlobUploadRoutes(w http.ResponseWriter, r *http.Request) {
138 // Parse path: /v2/{name}/blobs/uploads/{uuid?}
139 path := strings.TrimPrefix(r.URL.Path, "/v2/")
140 uploadIndex := strings.Index(path, "/blobs/uploads/")
141 if uploadIndex == -1 {
142 http.NotFound(w, r)
143 return
144 }
145
146 name := path[:uploadIndex]
147 uploadPart := path[uploadIndex+15:] // len("/blobs/uploads/") = 15
148
149 r.SetPathValue("name", name)
150
151 if uploadPart == "" {
152 // POST /v2/{name}/blobs/uploads/
153 switch r.Method {
154 case "POST":
155 s.handleStartBlobUpload(w, r)
156 default:
157 http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
158 }
159 } else {
160 // PATCH/PUT /v2/{name}/blobs/uploads/{uuid}
161 r.SetPathValue("uuid", uploadPart)
162 switch r.Method {
163 case "PATCH", "PUT":
164 s.handleBlobUpload(w, r)
165 default:
166 http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
167 }
168 }
169}
170
171func (s *Server) handleTagListRoutes(w http.ResponseWriter, r *http.Request) {
172 // Parse path: /v2/{name}/tags/list
173 path := strings.TrimPrefix(r.URL.Path, "/v2/")
174 if !strings.HasSuffix(path, "/tags/list") {
175 http.NotFound(w, r)
176 return
177 }
178
179 name := strings.TrimSuffix(path, "/tags/list")
180 r.SetPathValue("name", name)
181
182 switch r.Method {
183 case "GET":
184 s.handleListTags(w, r)
185 default:
186 http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
187 }
188}
189
190func (s *Server) handleBase(w http.ResponseWriter, r *http.Request) {
191 // For Docker clients, we need to respond with a 200 OK to this endpoint
192 // even if authentication is required. The client will then check the
193 // WWW-Authenticate header.
194 // see: https://docs.docker.com/registry/spec/api/#base
195 //
196 // Podman, as of v4.4.1, will not send credentials in the initial /v2/
197 // request. It expects a 401 Unauthorized to then send credentials.
198 //
199 // To support both, we'll check for the "Docker-Client" user agent.
200 // For now, we will just return a 200 OK to all clients.
201 w.Header().Set("Docker-Distribution-API-Version", "registry/2.0")
202 w.WriteHeader(http.StatusOK)
203}
204
205func (s *Server) handleGetManifest(w http.ResponseWriter, r *http.Request) {
206 name := r.PathValue("name")
207 reference := r.PathValue("reference")
208
209 log.Debug().Str("name", name).Str("reference", reference).Msg("GET manifest")
210
211 manifest, contentType, err := s.storage.GetManifest(name, reference)
212 if err != nil {
213 log.Warn().Err(err).Str("name", name).Str("reference", reference).Msg("Manifest not found")
214 http.NotFound(w, r)
215 return
216 }
217
218 log.Debug().Str("name", name).Str("reference", reference).Str("content_type", contentType).Msg("GET manifest - serving")
219
220 // Return the appropriate content type based on what was stored
221 w.Header().Set("Content-Type", contentType)
222 w.Header().Set("Content-Length", strconv.Itoa(len(manifest)))
223 w.WriteHeader(http.StatusOK)
224
225 if r.Method == "GET" {
226 _, _ = w.Write(manifest)
227 }
228}
229
230func (s *Server) handlePutManifest(w http.ResponseWriter, r *http.Request) {
231 name := r.PathValue("name")
232 reference := r.PathValue("reference")
233
234 contentType := r.Header.Get("Content-Type")
235 log.Debug().Str("name", name).Str("reference", reference).Str("content_type", contentType).Msg("PUT manifest")
236
237 if contentType == "" {
238 log.Warn().Str("name", name).Str("reference", reference).Msg("Content-Type header missing for PUT manifest")
239 http.Error(w, "Content-Type header required", http.StatusBadRequest)
240 return
241 }
242
243 // Read manifest data
244 data, err := io.ReadAll(r.Body)
245 if err != nil {
246 log.Error().Err(err).Msg("Failed to read manifest data")
247 http.Error(w, "Bad Request", http.StatusBadRequest)
248 return
249 }
250
251 // Store manifest
252 if err := s.storage.PutManifest(name, reference, contentType, data); err != nil {
253 log.Error().Err(err).Str("name", name).Str("reference", reference).Msg("Failed to store manifest")
254 http.Error(w, "Internal Server Error", http.StatusInternalServerError)
255 return
256 }
257
258 // Calculate digest for Docker-Content-Digest header
259 digest := fmt.Sprintf("sha256:%x", sha256.Sum256(data))
260 w.Header().Set("Docker-Content-Digest", digest)
261 w.Header().Set("Location", fmt.Sprintf("/v2/%s/manifests/%s", name, reference))
262 w.WriteHeader(http.StatusCreated)
263
264 // Parse manifest annotations
265 annotations, err := manifest.ParseManifestAnnotations(data, contentType)
266 if err != nil {
267 log.Warn().Err(err).Str("name", name).Str("reference", reference).Msg("Failed to parse manifest annotations, continuing without them")
268 annotations = map[string]string{}
269 }
270
271 // Log discovered annotations
272 if len(annotations) > 0 {
273 log.Info().
274 Str("name", name).
275 Str("reference", reference).
276 Interface("annotations", annotations).
277 Msg("Manifest annotations discovered")
278 }
279
280 // Emit image pushed event
281 s.eventBus.Publish(events.ImagePushed, events.ImagePushedPayload{
282 Name: name,
283 Reference: reference,
284 Manifest: data,
285 Annotations: annotations,
286 })
287}
288
289func (s *Server) handleGetBlob(w http.ResponseWriter, r *http.Request) {
290 name := r.PathValue("name")
291 digest := r.PathValue("digest")
292
293 log.Debug().Str("name", name).Str("digest", digest).Msg("GET blob")
294
295 path, err := s.storage.GetBlobPath(digest)
296 if err != nil {
297 log.Warn().Err(err).Str("name", name).Str("digest", digest).Msg("Blob not found")
298 http.NotFound(w, r)
299 return
300 }
301
302 // Serve the file directly
303 http.ServeFile(w, r, path)
304}
305
306func (s *Server) handleStartBlobUpload(w http.ResponseWriter, r *http.Request) {
307 name := r.PathValue("name")
308
309 log.Debug().Str("name", name).Msg("Starting blob upload")
310
311 // Generate a unique upload ID
312 uuid, err := s.storage.StartBlobUpload(name)
313 if err != nil {
314 log.Error().Err(err).Msg("Failed to start blob upload")
315 http.Error(w, "Internal Server Error", http.StatusInternalServerError)
316 return
317 }
318
319 // Set headers for the response
320 w.Header().Set("Location", fmt.Sprintf("/v2/%s/blobs/uploads/%s", name, uuid))
321 w.Header().Set("Range", "0-0") // Indicate that no bytes have been received yet
322 w.WriteHeader(http.StatusAccepted)
323}
324
325func (s *Server) handleBlobUpload(w http.ResponseWriter, r *http.Request) {
326 name := r.PathValue("name")
327 uuid := r.PathValue("uuid")
328 digest := r.URL.Query().Get("digest")
329
330 log.Debug().
331 Str("name", name).
332 Str("uuid", uuid).
333 Str("digest", digest).
334 Str("method", r.Method).
335 Msg("Handling blob upload chunk")
336
337 // Read the chunk from the request body
338 chunk, err := io.ReadAll(r.Body)
339 if err != nil {
340 log.Error().Err(err).Msg("Failed to read blob chunk")
341 http.Error(w, "Bad Request", http.StatusBadRequest)
342 return
343 }
344
345 // Append the chunk to the upload
346 length, err := s.storage.AppendBlobChunk(name, uuid, chunk)
347 if err != nil {
348 log.Error().Err(err).Msg("Failed to append blob chunk")
349 http.Error(w, "Internal Server Error", http.StatusInternalServerError)
350 return
351 }
352
353 // If this is the final chunk (PUT request with digest), finalize the upload
354 if r.Method == "PUT" && digest != "" {
355 if err := s.storage.FinishBlobUpload(uuid, digest); err != nil {
356 log.Error().Err(err).Str("digest", digest).Msg("Failed to finalize blob upload")
357 // The upload is invalid, delete it
358 _ = s.storage.CancelBlobUpload(uuid)
359 http.Error(w, "Bad Request: digest mismatch", http.StatusBadRequest)
360 return
361 }
362
363 // Respond with 201 Created
364 w.Header().Set("Location", fmt.Sprintf("/v2/%s/blobs/%s", name, digest))
365 w.Header().Set("Docker-Content-Digest", digest)
366 w.WriteHeader(http.StatusCreated)
367 return
368 }
369
370 // For PATCH requests, respond with 202 Accepted
371 w.Header().Set("Location", fmt.Sprintf("/v2/%s/blobs/uploads/%s", name, uuid))
372 w.Header().Set("Range", fmt.Sprintf("0-%d", length-1))
373 w.WriteHeader(http.StatusAccepted)
374}
375
376func (s *Server) handleListTags(w http.ResponseWriter, r *http.Request) {
377 name := r.PathValue("name")
378
379 log.Debug().Str("name", name).Msg("Listing tags")
380
381 tags, err := s.storage.ListTags(name)
382 if err != nil {
383 log.Warn().Err(err).Str("name", name).Msg("Tag not found")
384 w.WriteHeader(http.StatusNotFound)
385 return
386 }
387
388 // Construct the response JSON properly
389 tagsJSON, err := json.Marshal(tags)
390 if err != nil {
391 log.Error().Err(err).Str("name", name).Msg("Failed to marshal tags to JSON")
392 http.Error(w, "Internal Server Error", http.StatusInternalServerError)
393 return
394 }
395
396 response := fmt.Sprintf(`{"name":"%s","tags":%s}`, name, string(tagsJSON))
397
398 w.Header().Set("Content-Type", "application/json")
399 w.Header().Set("Content-Length", strconv.Itoa(len(response)))
400 w.WriteHeader(http.StatusOK)
401 w.Write([]byte(response))
402}
403
404func (s *Server) Start(ctx context.Context) error {
405 addr := ":" + strconv.Itoa(s.config.Server.RegistryPort)
406
407 // Build middleware chain
408 middlewares := []func(http.Handler) http.Handler{
409 middleware.PanicRecovery,
410 middleware.RequestLogger,
411 }
412
413 // Add registry authentication if enabled
414 if s.config.RegistryAuth.Enabled {
415 middlewares = append(middlewares, middleware.RegistryAuth(s.config.RegistryAuth.Username, s.config.RegistryAuth.Password))
416 }
417
418 // Wrap the mux with middleware
419 handler := middleware.Chain(middlewares...)(s.mux)
420
421 server := &http.Server{
422 Addr: addr,
423 Handler: handler,
424 }
425
426 log.Info().Str("address", addr).Msg("Registry server starting")
427
428 // Start server in a goroutine
429 errChan := make(chan error, 1)
430 go func() {
431 if err := server.ListenAndServe(); err != nil && err != http.ErrServerClosed {
432 errChan <- err
433 }
434 }()
435
436 // Wait for context cancellation or server error
437 select {
438 case err := <-errChan:
439 return err
440 case <-ctx.Done():
441 log.Info().Msg("Registry server shutting down...")
442 return server.Shutdown(context.Background())
443 }
444}
445
446// ServeHTTP implements http.Handler interface
447func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
448 s.mux.ServeHTTP(w, r)
449}
450
451// GetHandler returns the handler with middleware applied (for testing)
452func (s *Server) GetHandler() http.Handler {
453 // Build middleware chain
454 middlewares := []func(http.Handler) http.Handler{
455 middleware.PanicRecovery,
456 middleware.RequestLogger,
457 }
458
459 // Add registry authentication if enabled
460 if s.config.RegistryAuth.Enabled {
461 middlewares = append(middlewares, middleware.RegistryAuth(s.config.RegistryAuth.Username, s.config.RegistryAuth.Password))
462 }
463
464 // Wrap the mux with middleware
465 return middleware.Chain(middlewares...)(s.mux)
466}