1// Package registry implements the container registry use case.
2package registry
3
4import (
5 "context"
6 "crypto/sha256"
7 "crypto/sha512"
8 "encoding/json"
9 "errors"
10 "fmt"
11 "io"
12 "strings"
13 "sync"
14 "time"
15
16 "github.com/bnema/zerowrap"
17 "go.opentelemetry.io/otel"
18 "go.opentelemetry.io/otel/attribute"
19 "go.opentelemetry.io/otel/metric"
20 "go.opentelemetry.io/otel/trace"
21
22 "github.com/bnema/gordon/internal/adapters/out/telemetry"
23 "github.com/bnema/gordon/internal/boundaries/out"
24 "github.com/bnema/gordon/internal/domain"
25 "github.com/bnema/gordon/internal/usecase/registrystate"
26 "github.com/bnema/gordon/pkg/validation"
27)
28
29var registryTracer = otel.Tracer("gordon.registry")
30
31// Service implements the RegistryService interface.
32type Service struct {
33 blobStorage out.BlobStorage
34 manifestStorage out.ManifestStorage
35 eventBus out.EventPublisher
36 metrics *telemetry.Metrics
37 suppressedImages sync.Map // imageName -> *time.Timer
38 mutationMu *sync.RWMutex
39 registryState *registrystate.State
40}
41
42// SetMetrics sets the telemetry metrics for the registry service.
43func (s *Service) SetMetrics(m *telemetry.Metrics) {
44 s.metrics = m
45}
46
47// NewService creates a new registry service.
48func NewService(
49 blobStorage out.BlobStorage,
50 manifestStorage out.ManifestStorage,
51 eventBus out.EventPublisher,
52 states ...*registrystate.State,
53) *Service {
54 registryState := registrystate.New()
55 if len(states) > 0 && states[0] != nil {
56 registryState = states[0]
57 }
58 return &Service{
59 blobStorage: blobStorage,
60 manifestStorage: manifestStorage,
61 eventBus: eventBus,
62 mutationMu: ®istryState.MutationMu,
63 registryState: registryState,
64 }
65}
66
67// SuppressDeployEvent marks an image name to skip image.pushed events.
68// The suppression auto-expires after 2 minutes to prevent leaks.
69func (s *Service) SuppressDeployEvent(imageName string) {
70 imageName = ExtractImageName(strings.TrimSpace(imageName))
71 if imageName == "" {
72 return
73 }
74
75 var timer *time.Timer
76 timer = time.AfterFunc(2*time.Minute, func() {
77 // Only delete if this timer is still the current one, preventing an
78 // old timer's callback from removing a newer suppression entry.
79 if v, ok := s.suppressedImages.Load(imageName); ok && v == timer {
80 s.suppressedImages.Delete(imageName)
81 }
82 })
83 if existing, loaded := s.suppressedImages.LoadOrStore(imageName, timer); loaded {
84 existing.(*time.Timer).Stop()
85 s.suppressedImages.Store(imageName, timer)
86 }
87}
88
89// ClearDeployEventSuppression removes event suppression for an image.
90func (s *Service) ClearDeployEventSuppression(imageName string) {
91 imageName = ExtractImageName(strings.TrimSpace(imageName))
92 if imageName == "" {
93 return
94 }
95
96 if v, loaded := s.suppressedImages.LoadAndDelete(imageName); loaded {
97 v.(*time.Timer).Stop()
98 }
99}
100
101// ExtractImageName returns just the repository path of a container image
102// reference, stripping any registry host prefix, tag, and digest.
103// Examples:
104//
105// "reg.example.com/team/my-app:latest" -> "team/my-app"
106// "reg.example.com/my-app@sha256:abc" -> "my-app"
107// "my-app:v1.2" -> "my-app"
108func ExtractImageName(imageRef string) string {
109 name := imageRef
110 // Strip digest
111 if idx := strings.Index(name, "@"); idx != -1 {
112 name = name[:idx]
113 }
114 // Strip tag
115 // Find the last colon, but only strip it if it comes after any slash
116 // (to avoid treating a port number in the host as a tag).
117 if idx := strings.LastIndex(name, ":"); idx != -1 {
118 slashIdx := strings.LastIndex(name, "/")
119 if idx > slashIdx {
120 name = name[:idx]
121 }
122 }
123 // Strip registry host: if the first segment contains a dot or colon it is
124 // a registry hostname; remove it.
125 parts := strings.SplitN(name, "/", 2)
126 if len(parts) == 2 {
127 host := parts[0]
128 if strings.ContainsAny(host, ".:") || host == "localhost" {
129 name = parts[1]
130 }
131 }
132 return name
133}
134
135// IsDeployEventSuppressed checks if deploy events are suppressed for an image.
136func (s *Service) IsDeployEventSuppressed(imageName string) bool {
137 imageName = ExtractImageName(strings.TrimSpace(imageName))
138 if imageName == "" {
139 return false
140 }
141
142 _, exists := s.suppressedImages.Load(imageName)
143 return exists
144}
145
146// GetManifest retrieves a manifest by name and reference.
147func (s *Service) GetManifest(ctx context.Context, name, reference string) (*domain.Manifest, error) {
148 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
149 zerowrap.FieldLayer: "usecase",
150 zerowrap.FieldUseCase: "GetManifest",
151 "name": name,
152 "reference": reference,
153 })
154 log := zerowrap.FromCtx(ctx)
155
156 data, contentType, err := s.manifestStorage.GetManifest(name, reference)
157 if err != nil {
158 return nil, log.WrapErr(err, "failed to get manifest")
159 }
160
161 return &domain.Manifest{
162 Name: name,
163 Reference: reference,
164 ContentType: contentType,
165 Data: data,
166 }, nil
167}
168
169// PutManifest stores a manifest and returns the calculated digest.
170func (s *Service) PutManifest(ctx context.Context, manifest *domain.Manifest) (string, error) {
171 s.mutationMu.RLock()
172 defer s.mutationMu.RUnlock()
173
174 ctx, span := registryTracer.Start(ctx, "registry.put_manifest",
175 trace.WithAttributes(
176 attribute.String("name", manifest.Name),
177 attribute.String("reference", manifest.Reference),
178 attribute.Int("manifest_size", len(manifest.Data)),
179 ))
180 defer span.End()
181
182 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
183 zerowrap.FieldLayer: "usecase",
184 zerowrap.FieldUseCase: "PutManifest",
185 "name": manifest.Name,
186 "reference": manifest.Reference,
187 })
188 log := zerowrap.FromCtx(ctx)
189
190 // Calculate and verify the digest before mutating storage. A digest-addressed
191 // manifest must match its URL reference or clients could later retrieve
192 // attacker-controlled bytes under a trusted content address.
193 digest := fmt.Sprintf("sha256:%x", sha256.Sum256(manifest.Data))
194 if validation.IsDigest(manifest.Reference) {
195 matches, err := manifestDigestMatches(manifest.Reference, manifest.Data)
196 if err != nil {
197 return "", fmt.Errorf("validate manifest digest: %w", err)
198 }
199 if !matches {
200 return "", fmt.Errorf("%w: manifest content does not match %s", domain.ErrDigestMismatch, manifest.Reference)
201 }
202 digest = manifest.Reference
203 }
204
205 if err := s.manifestStorage.PutManifest(manifest.Name, manifest.Reference, manifest.ContentType, manifest.Data); err != nil {
206 return "", log.WrapErr(err, "failed to store manifest")
207 }
208 s.registryState.MarkPublished(manifestReferencedDigests(manifest.Data))
209
210 // Record push metrics
211 if s.metrics != nil {
212 attrs := metric.WithAttributes(
213 attribute.String("name", manifest.Name),
214 attribute.String("reference", manifest.Reference),
215 )
216 s.metrics.ImagePushTotal.Add(ctx, 1, attrs)
217 s.metrics.ImagePushSize.Add(ctx, int64(len(manifest.Data)), attrs)
218 }
219
220 // Publish image pushed event only for tag references (not digests).
221 // A docker push sends manifests by both digest and tag; firing only on
222 // tag prevents duplicate deploy triggers for the same push.
223 if s.eventBus != nil && !validation.IsDigest(manifest.Reference) {
224 if s.IsDeployEventSuppressed(manifest.Name) {
225 log.Info().Str("image", manifest.Name).Msg("skipping image.pushed event: CLI deploy intent active")
226 } else {
227 if err := s.eventBus.Publish(domain.EventImagePushed, domain.ImagePushedPayload{
228 Name: manifest.Name,
229 Reference: manifest.Reference,
230 Manifest: manifest.Data,
231 Annotations: manifest.Annotations,
232 }); err != nil {
233 log.Warn().Err(err).Msg("failed to publish image pushed event")
234 }
235 }
236 }
237
238 log.Info().Str("digest", digest).Msg("manifest stored")
239 return digest, nil
240}
241
242// DeleteManifest removes a manifest.
243func (s *Service) DeleteManifest(ctx context.Context, name, reference string) error {
244 s.mutationMu.RLock()
245 defer s.mutationMu.RUnlock()
246
247 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
248 zerowrap.FieldLayer: "usecase",
249 zerowrap.FieldUseCase: "DeleteManifest",
250 "name": name,
251 "reference": reference,
252 })
253 log := zerowrap.FromCtx(ctx)
254
255 if err := s.manifestStorage.DeleteManifest(name, reference); err != nil {
256 return log.WrapErr(err, "failed to delete manifest")
257 }
258
259 log.Info().Msg("manifest deleted")
260 return nil
261}
262
263// GetBlob retrieves a blob by digest.
264func (s *Service) GetBlob(ctx context.Context, digest string) (io.ReadCloser, error) {
265 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
266 zerowrap.FieldLayer: "usecase",
267 zerowrap.FieldUseCase: "GetBlob",
268 "digest": digest,
269 })
270 log := zerowrap.FromCtx(ctx)
271
272 reader, err := s.blobStorage.GetBlob(digest)
273 if err != nil {
274 return nil, log.WrapErr(err, "failed to get blob")
275 }
276
277 return reader, nil
278}
279
280// GetBlobPath returns the filesystem path to a blob only when a manifest in
281// the requested repository references it.
282func (s *Service) GetBlobPath(ctx context.Context, name, digest string) (string, error) {
283 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
284 zerowrap.FieldLayer: "usecase",
285 zerowrap.FieldUseCase: "GetBlobPath",
286 "name": name,
287 "digest": digest,
288 })
289 log := zerowrap.FromCtx(ctx)
290
291 referenced, err := s.repositoryReferencesDigest(name, digest)
292 if err != nil {
293 return "", log.WrapErr(err, "failed to verify blob ownership")
294 }
295 if !referenced {
296 return "", domain.ErrBlobNotFound
297 }
298
299 path, err := s.blobStorage.GetBlobPath(digest)
300 if err != nil {
301 return "", log.WrapErr(err, "failed to get blob path")
302 }
303
304 return path, nil
305}
306
307type manifestDescriptor struct {
308 Digest string `json:"digest"`
309}
310
311type manifestReferences struct {
312 Config manifestDescriptor `json:"config"`
313 Layers []manifestDescriptor `json:"layers"`
314 Manifests []manifestDescriptor `json:"manifests"`
315 Blobs []manifestDescriptor `json:"blobs"`
316 Subject *manifestDescriptor `json:"subject"`
317}
318
319const maxManifestTraversal = 10000
320
321func (s *Service) repositoryReferencesDigest(name, target string) (bool, error) {
322 tags, err := s.manifestStorage.ListTags(name)
323 if err != nil {
324 if errors.Is(err, domain.ErrManifestNotFound) {
325 return false, nil
326 }
327 return false, fmt.Errorf("list tags for repository %s: %w", name, err)
328 }
329
330 if len(tags) > maxManifestTraversal {
331 return false, fmt.Errorf("%w: repository %s exceeds manifest traversal limit", domain.ErrBlobNotFound, name)
332 }
333 queue := append([]string(nil), tags...)
334 seen := make(map[string]struct{}, len(queue))
335 for len(queue) > 0 {
336 reference := queue[0]
337 queue = queue[1:]
338 if _, ok := seen[reference]; ok {
339 continue
340 }
341 seen[reference] = struct{}{}
342 if len(seen) > maxManifestTraversal {
343 return false, fmt.Errorf("%w: repository %s exceeds manifest traversal limit", domain.ErrBlobNotFound, name)
344 }
345
346 data, _, err := s.manifestStorage.GetManifest(name, reference)
347 if err != nil {
348 return false, fmt.Errorf("get manifest %s for repository %s: %w", reference, name, err)
349 }
350 var refs manifestReferences
351 if err := json.Unmarshal(data, &refs); err != nil {
352 return false, fmt.Errorf("decode manifest %s: %w", reference, err)
353 }
354
355 if manifestReferencesTarget(refs, target) {
356 return true, nil
357 }
358 nestedManifests := refs.Manifests
359 if refs.Subject != nil && refs.Subject.Digest != "" {
360 nestedManifests = append(append([]manifestDescriptor(nil), refs.Manifests...), *refs.Subject)
361 }
362 if len(seen)+len(queue)+len(nestedManifests) > maxManifestTraversal {
363 return false, fmt.Errorf("%w: repository %s exceeds manifest traversal limit", domain.ErrBlobNotFound, name)
364 }
365 queue = appendManifestDigests(queue, nestedManifests)
366 }
367 return false, nil
368}
369
370func manifestReferencesTarget(refs manifestReferences, target string) bool {
371 return refs.Config.Digest == target ||
372 descriptorListContains(refs.Layers, target) ||
373 descriptorListContains(refs.Manifests, target) ||
374 descriptorListContains(refs.Blobs, target) ||
375 (refs.Subject != nil && refs.Subject.Digest == target)
376}
377
378func appendManifestDigests(queue []string, descriptors []manifestDescriptor) []string {
379 for _, descriptor := range descriptors {
380 if descriptor.Digest != "" {
381 queue = append(queue, descriptor.Digest)
382 }
383 }
384 return queue
385}
386
387func manifestReferencedDigests(data []byte) []string {
388 var refs manifestReferences
389 if json.Unmarshal(data, &refs) != nil {
390 return nil
391 }
392 digests := make([]string, 0, 1+len(refs.Layers)+len(refs.Manifests)+len(refs.Blobs))
393 if refs.Config.Digest != "" {
394 digests = append(digests, refs.Config.Digest)
395 }
396 for _, descriptors := range [][]manifestDescriptor{refs.Layers, refs.Manifests, refs.Blobs} {
397 for _, descriptor := range descriptors {
398 if descriptor.Digest != "" {
399 digests = append(digests, descriptor.Digest)
400 }
401 }
402 }
403 if refs.Subject != nil && refs.Subject.Digest != "" {
404 digests = append(digests, refs.Subject.Digest)
405 }
406 return digests
407}
408
409func descriptorListContains(descriptors []manifestDescriptor, digest string) bool {
410 for _, descriptor := range descriptors {
411 if descriptor.Digest == digest {
412 return true
413 }
414 }
415 return false
416}
417
418func manifestDigestMatches(reference string, data []byte) (bool, error) {
419 algorithm, _, ok := strings.Cut(reference, ":")
420 if !ok {
421 return false, domain.ErrInvalidDigest
422 }
423 switch algorithm {
424 case "sha256":
425 return reference == fmt.Sprintf("sha256:%x", sha256.Sum256(data)), nil
426 case "sha512":
427 return reference == fmt.Sprintf("sha512:%x", sha512.Sum512(data)), nil
428 default:
429 return false, domain.ErrInvalidDigest
430 }
431}
432
433// PutBlob stores a blob with the given digest.
434func (s *Service) PutBlob(ctx context.Context, digest string, data io.Reader, size int64) error {
435 s.mutationMu.RLock()
436 defer s.mutationMu.RUnlock()
437
438 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
439 zerowrap.FieldLayer: "usecase",
440 zerowrap.FieldUseCase: "PutBlob",
441 "digest": digest,
442 zerowrap.FieldSize: size,
443 })
444 log := zerowrap.FromCtx(ctx)
445
446 if err := s.blobStorage.PutBlob(digest, data, size); err != nil {
447 return log.WrapErr(err, "failed to store blob")
448 }
449 s.registryState.AddPending(digest, time.Now().UTC())
450
451 log.Info().Msg("blob stored")
452 return nil
453}
454
455// BlobExists checks if a blob exists.
456func (s *Service) BlobExists(_ context.Context, digest string) bool {
457 return s.blobStorage.BlobExists(digest)
458}
459
460// StartUpload starts a new blob upload.
461func (s *Service) StartUpload(ctx context.Context, name string) (string, error) {
462 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
463 zerowrap.FieldLayer: "usecase",
464 zerowrap.FieldUseCase: "StartUpload",
465 "name": name,
466 })
467 log := zerowrap.FromCtx(ctx)
468
469 uuid, err := s.blobStorage.StartBlobUpload(name)
470 if err != nil {
471 return "", log.WrapErr(err, "failed to start blob upload")
472 }
473
474 log.Info().Str("uuid", uuid).Msg("blob upload started")
475 return uuid, nil
476}
477
478// AppendBlobChunk appends data to an in-progress blob upload.
479func (s *Service) AppendBlobChunk(ctx context.Context, name, uuid string, data io.Reader, contentLength, maxBlobSize int64) (int64, error) {
480 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
481 zerowrap.FieldLayer: "usecase",
482 zerowrap.FieldUseCase: "AppendBlobChunk",
483 "name": name,
484 "uuid": uuid,
485 })
486 log := zerowrap.FromCtx(ctx)
487
488 length, err := s.blobStorage.AppendBlobChunk(name, uuid, data, contentLength, maxBlobSize)
489 if err != nil {
490 return 0, log.WrapErr(err, "failed to append blob chunk")
491 }
492
493 return length, nil
494}
495
496// FinishUpload completes a blob upload.
497func (s *Service) FinishUpload(ctx context.Context, uuid, digest string) error {
498 // Keep the transition from upload to blob storage atomic with respect to
499 // registry garbage collection. PruneRegistry holds the exclusive lock, so
500 // it cannot observe a finalized blob before it is marked pending.
501 s.mutationMu.RLock()
502 defer s.mutationMu.RUnlock()
503
504 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
505 zerowrap.FieldLayer: "usecase",
506 zerowrap.FieldUseCase: "FinishUpload",
507 "uuid": uuid,
508 "digest": digest,
509 })
510 log := zerowrap.FromCtx(ctx)
511
512 if err := s.blobStorage.FinishBlobUpload(uuid, digest); err != nil {
513 return log.WrapErr(err, "failed to finish blob upload")
514 }
515 s.registryState.AddPending(digest, time.Now().UTC())
516
517 log.Info().Msg("blob upload finished")
518 return nil
519}
520
521// CancelUpload cancels an in-progress upload.
522func (s *Service) CancelUpload(ctx context.Context, uuid string) error {
523 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
524 zerowrap.FieldLayer: "usecase",
525 zerowrap.FieldUseCase: "CancelUpload",
526 "uuid": uuid,
527 })
528 log := zerowrap.FromCtx(ctx)
529
530 if err := s.blobStorage.CancelBlobUpload(uuid); err != nil {
531 return log.WrapErr(err, "failed to cancel blob upload")
532 }
533
534 log.Info().Msg("blob upload cancelled")
535 return nil
536}
537
538// ListTags returns all tags for a repository.
539func (s *Service) ListTags(ctx context.Context, name string) ([]string, error) {
540 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
541 zerowrap.FieldLayer: "usecase",
542 zerowrap.FieldUseCase: "ListTags",
543 "name": name,
544 })
545 log := zerowrap.FromCtx(ctx)
546
547 tags, err := s.manifestStorage.ListTags(name)
548 if err != nil {
549 return nil, log.WrapErr(err, "failed to list tags")
550 }
551
552 filtered := make([]string, 0, len(tags))
553 for _, tag := range tags {
554 if validation.IsDigest(tag) {
555 continue
556 }
557 filtered = append(filtered, tag)
558 }
559
560 return filtered, nil
561}
562
563// ListRepositories returns all repository names.
564func (s *Service) ListRepositories(ctx context.Context) ([]string, error) {
565 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
566 zerowrap.FieldLayer: "usecase",
567 zerowrap.FieldUseCase: "ListRepositories",
568 })
569 log := zerowrap.FromCtx(ctx)
570
571 repos, err := s.manifestStorage.ListRepositories()
572 if err != nil {
573 return nil, log.WrapErr(err, "failed to list repositories")
574 }
575
576 return repos, nil
577}