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 "fmt"
10 "io"
11 "strings"
12 "sync"
13 "time"
14
15 "github.com/bnema/zerowrap"
16 "go.opentelemetry.io/otel"
17 "go.opentelemetry.io/otel/attribute"
18 "go.opentelemetry.io/otel/metric"
19 "go.opentelemetry.io/otel/trace"
20
21 "github.com/bnema/gordon/internal/adapters/out/telemetry"
22 "github.com/bnema/gordon/internal/boundaries/out"
23 "github.com/bnema/gordon/internal/domain"
24 "github.com/bnema/gordon/internal/usecase/registrystate"
25 "github.com/bnema/gordon/pkg/validation"
26)
27
28var registryTracer = otel.Tracer("gordon.registry")
29
30// Service implements the RegistryService interface.
31type Service struct {
32 blobStorage out.BlobStorage
33 manifestStorage out.ManifestStorage
34 eventBus out.EventPublisher
35 metrics *telemetry.Metrics
36 mutationMu *sync.RWMutex
37 registryState *registrystate.State
38}
39
40// SetMetrics sets the telemetry metrics for the registry service.
41func (s *Service) SetMetrics(m *telemetry.Metrics) {
42 s.metrics = m
43}
44
45// NewService creates a new registry service.
46func NewService(
47 blobStorage out.BlobStorage,
48 manifestStorage out.ManifestStorage,
49 eventBus out.EventPublisher,
50 states ...*registrystate.State,
51) *Service {
52 registryState := registrystate.New()
53 if len(states) > 0 && states[0] != nil {
54 registryState = states[0]
55 }
56 return &Service{
57 blobStorage: blobStorage,
58 manifestStorage: manifestStorage,
59 eventBus: eventBus,
60 mutationMu: ®istryState.MutationMu,
61 registryState: registryState,
62 }
63}
64
65// GetManifest retrieves a manifest by name and reference.
66func (s *Service) GetManifest(ctx context.Context, name, reference string) (*domain.Manifest, error) {
67 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
68 zerowrap.FieldLayer: "usecase",
69 zerowrap.FieldUseCase: "GetManifest",
70 "name": name,
71 "reference": reference,
72 })
73 log := zerowrap.FromCtx(ctx)
74
75 data, contentType, err := s.manifestStorage.GetManifest(name, reference)
76 if err != nil {
77 return nil, log.WrapErr(err, "failed to get manifest")
78 }
79
80 return &domain.Manifest{
81 Name: name,
82 Reference: reference,
83 ContentType: contentType,
84 Data: data,
85 }, nil
86}
87
88// PutManifest stores a manifest and returns the calculated digest.
89func (s *Service) PutManifest(ctx context.Context, manifest *domain.Manifest) (string, error) {
90 s.mutationMu.RLock()
91 defer s.mutationMu.RUnlock()
92
93 ctx, span := registryTracer.Start(ctx, "registry.put_manifest",
94 trace.WithAttributes(
95 attribute.String("name", manifest.Name),
96 attribute.String("reference", manifest.Reference),
97 attribute.Int("manifest_size", len(manifest.Data)),
98 ))
99 defer span.End()
100
101 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
102 zerowrap.FieldLayer: "usecase",
103 zerowrap.FieldUseCase: "PutManifest",
104 "name": manifest.Name,
105 "reference": manifest.Reference,
106 })
107 log := zerowrap.FromCtx(ctx)
108
109 // Calculate and verify the digest before mutating storage. A digest-addressed
110 // manifest must match its URL reference or clients could later retrieve
111 // attacker-controlled bytes under a trusted content address.
112 digest := fmt.Sprintf("sha256:%x", sha256.Sum256(manifest.Data))
113 if validation.IsDigest(manifest.Reference) {
114 matches, err := manifestDigestMatches(manifest.Reference, manifest.Data)
115 if err != nil {
116 return "", fmt.Errorf("validate manifest digest: %w", err)
117 }
118 if !matches {
119 return "", fmt.Errorf("%w: manifest content does not match %s", domain.ErrDigestMismatch, manifest.Reference)
120 }
121 digest = manifest.Reference
122 }
123
124 if err := s.validateManifestBlobs(manifest.Name, manifest.Data); err != nil {
125 return "", log.WrapErr(err, "manifest references unowned content")
126 }
127
128 if err := s.manifestStorage.PutManifest(manifest.Name, manifest.Reference, manifest.ContentType, manifest.Data); err != nil {
129 return "", log.WrapErr(err, "failed to store manifest")
130 }
131 s.registryState.MarkPublished(manifestReferencedDigests(manifest.Data))
132
133 // Record push metrics
134 if s.metrics != nil {
135 attrs := metric.WithAttributes(
136 attribute.String("name", manifest.Name),
137 attribute.String("reference", manifest.Reference),
138 )
139 s.metrics.ImagePushTotal.Add(ctx, 1, attrs)
140 s.metrics.ImagePushSize.Add(ctx, int64(len(manifest.Data)), attrs)
141 }
142
143 // Publish image pushed event only for tag references (not digests).
144 // A docker push sends manifests by both digest and tag; firing only on
145 // tag prevents duplicate deploy triggers for the same push.
146 if s.eventBus != nil && !validation.IsDigest(manifest.Reference) {
147 if err := s.eventBus.Publish(domain.EventImagePushed, domain.ImagePushedPayload{
148 Name: manifest.Name,
149 Reference: manifest.Reference,
150 Manifest: manifest.Data,
151 Annotations: manifest.Annotations,
152 }); err != nil {
153 log.Warn().Err(err).Msg("failed to publish image pushed event")
154 }
155 }
156
157 log.Info().Str("digest", digest).Msg("manifest stored")
158 return digest, nil
159}
160
161// DeleteManifest removes a manifest.
162func (s *Service) DeleteManifest(ctx context.Context, name, reference string) error {
163 s.mutationMu.RLock()
164 defer s.mutationMu.RUnlock()
165
166 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
167 zerowrap.FieldLayer: "usecase",
168 zerowrap.FieldUseCase: "DeleteManifest",
169 "name": name,
170 "reference": reference,
171 })
172 log := zerowrap.FromCtx(ctx)
173
174 if err := s.manifestStorage.DeleteManifest(name, reference); err != nil {
175 return log.WrapErr(err, "failed to delete manifest")
176 }
177
178 log.Info().Msg("manifest deleted")
179 return nil
180}
181
182// GetBlob retrieves a blob by digest.
183func (s *Service) GetBlob(ctx context.Context, digest string) (io.ReadCloser, error) {
184 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
185 zerowrap.FieldLayer: "usecase",
186 zerowrap.FieldUseCase: "GetBlob",
187 "digest": digest,
188 })
189 log := zerowrap.FromCtx(ctx)
190
191 reader, err := s.blobStorage.GetBlob(digest)
192 if err != nil {
193 return nil, log.WrapErr(err, "failed to get blob")
194 }
195
196 return reader, nil
197}
198
199// GetBlobPath returns the filesystem path to a blob only when the
200// repository completed an upload of that digest. Ownership is never
201// inferred from manifest references: a repository writer controls its own
202// manifests, so a manifest naming a foreign digest must not confer access.
203func (s *Service) GetBlobPath(ctx context.Context, name, digest string) (string, error) {
204 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
205 zerowrap.FieldLayer: "usecase",
206 zerowrap.FieldUseCase: "GetBlobPath",
207 "name": name,
208 "digest": digest,
209 })
210 log := zerowrap.FromCtx(ctx)
211
212 owned, err := s.blobStorage.BlobOwnedByRepository(name, digest)
213 if err != nil {
214 return "", log.WrapErr(err, "failed to verify blob ownership")
215 }
216 if !owned {
217 return "", domain.ErrBlobNotFound
218 }
219
220 path, err := s.blobStorage.GetBlobPath(digest)
221 if err != nil {
222 return "", log.WrapErr(err, "failed to get blob path")
223 }
224
225 return path, nil
226}
227
228type manifestDescriptor struct {
229 Digest string `json:"digest"`
230}
231
232type manifestReferences struct {
233 Config manifestDescriptor `json:"config"`
234 Layers []manifestDescriptor `json:"layers"`
235 Manifests []manifestDescriptor `json:"manifests"`
236 Blobs []manifestDescriptor `json:"blobs"`
237 Subject *manifestDescriptor `json:"subject"`
238}
239
240const maxManifestTraversal = 10000
241
242// validateManifestBlobs requires every config/layer blob a manifest
243// references to be owned by the repository, and every child manifest to
244// already exist in it. A manifest can therefore never manufacture access to
245// content the repository did not receive through an authenticated push.
246func (s *Service) validateManifestBlobs(name string, data []byte) error {
247 var refs manifestReferences
248 if err := json.Unmarshal(data, &refs); err != nil {
249 return fmt.Errorf("%w: manifest is not a JSON descriptor document", domain.ErrManifestBlobUnknown)
250 }
251
252 blobs := make([]manifestDescriptor, 0, 1+len(refs.Layers)+len(refs.Blobs))
253 if refs.Config.Digest != "" {
254 blobs = append(blobs, refs.Config)
255 }
256 blobs = append(blobs, refs.Layers...)
257 blobs = append(blobs, refs.Blobs...)
258 for _, descriptor := range blobs {
259 if descriptor.Digest == "" {
260 continue
261 }
262 owned, err := s.blobStorage.BlobOwnedByRepository(name, descriptor.Digest)
263 if err != nil {
264 return fmt.Errorf("verify blob ownership: %w", err)
265 }
266 if !owned {
267 return fmt.Errorf("%w: %s", domain.ErrManifestBlobUnknown, descriptor.Digest)
268 }
269 }
270
271 children := refs.Manifests
272 if refs.Subject != nil && refs.Subject.Digest != "" {
273 children = append(append([]manifestDescriptor(nil), children...), *refs.Subject)
274 }
275 if len(children) > maxManifestTraversal {
276 return fmt.Errorf("%w: manifest references too many children", domain.ErrManifestBlobUnknown)
277 }
278 for _, descriptor := range children {
279 if descriptor.Digest == "" {
280 continue
281 }
282 if _, _, err := s.manifestStorage.GetManifest(name, descriptor.Digest); err != nil {
283 return fmt.Errorf("%w: %s", domain.ErrManifestBlobUnknown, descriptor.Digest)
284 }
285 }
286 return nil
287}
288
289func manifestReferencedDigests(data []byte) []string {
290 var refs manifestReferences
291 if json.Unmarshal(data, &refs) != nil {
292 return nil
293 }
294 digests := make([]string, 0, 1+len(refs.Layers)+len(refs.Manifests)+len(refs.Blobs))
295 if refs.Config.Digest != "" {
296 digests = append(digests, refs.Config.Digest)
297 }
298 for _, descriptors := range [][]manifestDescriptor{refs.Layers, refs.Manifests, refs.Blobs} {
299 for _, descriptor := range descriptors {
300 if descriptor.Digest != "" {
301 digests = append(digests, descriptor.Digest)
302 }
303 }
304 }
305 if refs.Subject != nil && refs.Subject.Digest != "" {
306 digests = append(digests, refs.Subject.Digest)
307 }
308 return digests
309}
310
311func manifestDigestMatches(reference string, data []byte) (bool, error) {
312 algorithm, _, ok := strings.Cut(reference, ":")
313 if !ok {
314 return false, domain.ErrInvalidDigest
315 }
316 switch algorithm {
317 case "sha256":
318 return reference == fmt.Sprintf("sha256:%x", sha256.Sum256(data)), nil
319 case "sha512":
320 return reference == fmt.Sprintf("sha512:%x", sha512.Sum512(data)), nil
321 default:
322 return false, domain.ErrInvalidDigest
323 }
324}
325
326// PutBlob stores a blob with the given digest.
327func (s *Service) PutBlob(ctx context.Context, digest string, data io.Reader, size int64) error {
328 s.mutationMu.RLock()
329 defer s.mutationMu.RUnlock()
330
331 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
332 zerowrap.FieldLayer: "usecase",
333 zerowrap.FieldUseCase: "PutBlob",
334 "digest": digest,
335 zerowrap.FieldSize: size,
336 })
337 log := zerowrap.FromCtx(ctx)
338
339 if err := s.blobStorage.PutBlob(digest, data, size); err != nil {
340 return log.WrapErr(err, "failed to store blob")
341 }
342 s.registryState.AddPending(digest, time.Now().UTC())
343
344 log.Info().Msg("blob stored")
345 return nil
346}
347
348// BlobExists checks if a blob exists.
349func (s *Service) BlobExists(_ context.Context, digest string) bool {
350 return s.blobStorage.BlobExists(digest)
351}
352
353// StartUpload starts a new blob upload.
354func (s *Service) StartUpload(ctx context.Context, name string) (string, error) {
355 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
356 zerowrap.FieldLayer: "usecase",
357 zerowrap.FieldUseCase: "StartUpload",
358 "name": name,
359 })
360 log := zerowrap.FromCtx(ctx)
361
362 uuid, err := s.blobStorage.StartBlobUpload(name)
363 if err != nil {
364 return "", log.WrapErr(err, "failed to start blob upload")
365 }
366
367 log.Info().Str("uuid", uuid).Msg("blob upload started")
368 return uuid, nil
369}
370
371// AppendBlobChunk appends data to an in-progress blob upload.
372func (s *Service) AppendBlobChunk(ctx context.Context, name, uuid string, data io.Reader, contentLength, maxBlobSize int64) (int64, error) {
373 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
374 zerowrap.FieldLayer: "usecase",
375 zerowrap.FieldUseCase: "AppendBlobChunk",
376 "name": name,
377 "uuid": uuid,
378 })
379 log := zerowrap.FromCtx(ctx)
380
381 length, err := s.blobStorage.AppendBlobChunk(name, uuid, data, contentLength, maxBlobSize)
382 if err != nil {
383 return 0, log.WrapErr(err, "failed to append blob chunk")
384 }
385
386 return length, nil
387}
388
389// FinishUpload completes a blob upload for the named repository and records
390// the repository/blob association.
391func (s *Service) FinishUpload(ctx context.Context, name, uuid, digest string) error {
392 // Keep the transition from upload to blob storage atomic with respect to
393 // registry garbage collection. PruneRegistry holds the exclusive lock, so
394 // it cannot observe a finalized blob before it is marked pending.
395 s.mutationMu.RLock()
396 defer s.mutationMu.RUnlock()
397
398 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
399 zerowrap.FieldLayer: "usecase",
400 zerowrap.FieldUseCase: "FinishUpload",
401 "name": name,
402 "uuid": uuid,
403 "digest": digest,
404 })
405 log := zerowrap.FromCtx(ctx)
406
407 if err := s.blobStorage.FinishBlobUpload(name, uuid, digest); err != nil {
408 return log.WrapErr(err, "failed to finish blob upload")
409 }
410 s.registryState.AddPending(digest, time.Now().UTC())
411
412 log.Info().Msg("blob upload finished")
413 return nil
414}
415
416// CancelUpload cancels an in-progress upload.
417func (s *Service) CancelUpload(ctx context.Context, name, uuid string) error {
418 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
419 zerowrap.FieldLayer: "usecase",
420 zerowrap.FieldUseCase: "CancelUpload",
421 "name": name,
422 "uuid": uuid,
423 })
424 log := zerowrap.FromCtx(ctx)
425
426 if err := s.blobStorage.CancelBlobUpload(name, uuid); err != nil {
427 return log.WrapErr(err, "failed to cancel blob upload")
428 }
429
430 log.Info().Msg("blob upload cancelled")
431 return nil
432}
433
434// ListTags returns all tags for a repository.
435func (s *Service) ListTags(ctx context.Context, name string) ([]string, error) {
436 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
437 zerowrap.FieldLayer: "usecase",
438 zerowrap.FieldUseCase: "ListTags",
439 "name": name,
440 })
441 log := zerowrap.FromCtx(ctx)
442
443 tags, err := s.manifestStorage.ListTags(name)
444 if err != nil {
445 return nil, log.WrapErr(err, "failed to list tags")
446 }
447
448 filtered := make([]string, 0, len(tags))
449 for _, tag := range tags {
450 if validation.IsDigest(tag) {
451 continue
452 }
453 filtered = append(filtered, tag)
454 }
455
456 return filtered, nil
457}
458
459// ListRepositories returns all repository names.
460func (s *Service) ListRepositories(ctx context.Context) ([]string, error) {
461 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
462 zerowrap.FieldLayer: "usecase",
463 zerowrap.FieldUseCase: "ListRepositories",
464 })
465 log := zerowrap.FromCtx(ctx)
466
467 repos, err := s.manifestStorage.ListRepositories()
468 if err != nil {
469 return nil, log.WrapErr(err, "failed to list repositories")
470 }
471
472 return repos, nil
473}