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