1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3// Package blobs3 implements the blob port with the object, multipart and
4// listing S3 APIs. Checksums are WhenRequired, credentials are explicit, and
5// no bucket creation, policy, ACL, tagging or lifecycle API is ever called.
6package blobs3
7
8import (
9 "bytes"
10 "context"
11 "crypto/sha256"
12 "encoding/hex"
13 "errors"
14 "fmt"
15 "io"
16 "net/http"
17 "sort"
18 "strconv"
19 "strings"
20 "sync"
21 "time"
22
23 "bnema.dev/ermite/internal/domain"
24 "bnema.dev/ermite/internal/ports"
25 "github.com/aws/aws-sdk-go-v2/aws"
26 "github.com/aws/aws-sdk-go-v2/credentials"
27 "github.com/aws/aws-sdk-go-v2/service/s3"
28 "github.com/aws/aws-sdk-go-v2/service/s3/types"
29 "github.com/bnema/zerowrap"
30)
31
32// Store is a provider-independent S3 blob store.
33type Store struct {
34 client *s3.Client
35 cfg Config
36}
37
38var _ ports.BlobStore = (*Store)(nil)
39
40// New builds a client without consulting ambient configuration or probing.
41func New(ctx context.Context, cfg Config) (*Store, error) {
42 if e := ctx.Err(); e != nil {
43 return nil, e
44 }
45 if cfg.Bucket == "" || cfg.Region == "" || cfg.AccessKeyID == "" || cfg.SecretAccessKey == "" {
46 return nil, fmt.Errorf("bucket, region and explicit credentials are required: %w", domain.ErrInvalid)
47 }
48 cfg.Prefix = strings.Trim(cfg.Prefix, "/")
49 if cfg.Prefix != "" {
50 cfg.Prefix += "/"
51 }
52 client := cfg.HTTPClient
53 if client == nil {
54 transport := http.DefaultTransport.(*http.Transport).Clone()
55 transport.IdleConnTimeout = 60 * time.Second
56 transport.ResponseHeaderTimeout = 60 * time.Second
57 client = &http.Client{Transport: transport}
58 }
59 ac := aws.Config{Region: cfg.Region, Credentials: credentials.NewStaticCredentialsProvider(cfg.AccessKeyID, cfg.SecretAccessKey, ""), HTTPClient: client, RequestChecksumCalculation: aws.RequestChecksumCalculationWhenRequired, ResponseChecksumValidation: aws.ResponseChecksumValidationWhenRequired, RetryMaxAttempts: 3}
60 c := s3.NewFromConfig(ac, func(o *s3.Options) {
61 if cfg.Endpoint != "" {
62 o.BaseEndpoint = aws.String(cfg.Endpoint)
63 }
64 o.UsePathStyle = cfg.ForcePathStyle
65 })
66 return &Store{client: c, cfg: cfg}, nil
67}
68func valid(key string) error {
69 if !ports.ValidBlobKey(key) {
70 return domain.ErrInvalid
71 }
72 return nil
73}
74func (s *Store) key(key string) *string { return aws.String(s.cfg.Prefix + key) }
75
76type statusError interface{ HTTPStatusCode() int }
77type codeError interface{ ErrorCode() string }
78
79func status(err error) int {
80 var e statusError
81 if errors.As(err, &e) {
82 return e.HTTPStatusCode()
83 }
84 return 0
85}
86func translate(err error) error {
87 if err == nil {
88 return nil
89 }
90 var c codeError
91 if errors.As(err, &c) {
92 switch c.ErrorCode() {
93 case "InvalidRange":
94 return domain.ErrOutOfRange
95 case "NoSuchKey", "NotFound":
96 return domain.ErrNotFound
97 case "EntityTooSmall":
98 return domain.ErrInvalid
99 }
100 }
101 switch status(err) {
102 case 404:
103 return domain.ErrNotFound
104 case 416:
105 return domain.ErrOutOfRange
106 }
107 return err
108}
109
110type ctxReader struct {
111 ctx context.Context
112 r io.Reader
113}
114
115func (r ctxReader) Read(p []byte) (int, error) {
116 if e := r.ctx.Err(); e != nil {
117 return 0, e
118 }
119 return r.r.Read(p)
120}
121
122// One pooled buffer per in-flight upload bounds staging to BlobPartSize.
123var buffers = sync.Pool{New: func() any { b := make([]byte, ports.BlobPartSize); return &b }}
124
125func exactEnd(ctx context.Context, r io.Reader) error {
126 var b [1]byte
127 n, e := io.ReadFull(ctxReader{ctx, r}, b[:])
128 if n != 0 {
129 return domain.ErrInvalid
130 }
131 if e == io.EOF {
132 return nil
133 }
134 return e
135}
136func readPart(ctx context.Context, r io.Reader, b []byte) error {
137 _, e := io.ReadFull(ctxReader{ctx, r}, b)
138 if e == io.EOF || e == io.ErrUnexpectedEOF {
139 return domain.ErrInvalid
140 }
141 return e
142}
143
144// Put uses signed, seekable bodies, multiparting objects larger than one buffer.
145func (s *Store) Put(ctx context.Context, key string, r io.Reader, size int64) (ports.BlobInfo, error) {
146 logEntry := zerowrap.FromCtx(ctx)
147 logEntry.Debug().Msg("blobs3: Put")
148 if e := valid(key); e != nil {
149 return ports.BlobInfo{}, e
150 }
151 if size < 0 || size > int64(ports.BlobPartSize)*ports.BlobMaxParts {
152 return ports.BlobInfo{}, domain.ErrInvalid
153 }
154 bp := buffers.Get().(*[]byte)
155 defer buffers.Put(bp)
156 h := sha256.New()
157 source := io.TeeReader(r, h)
158 if size <= ports.BlobPartSize {
159 b := (*bp)[:int(size)]
160 if e := readPart(ctx, source, b); e != nil {
161 return ports.BlobInfo{}, e
162 }
163 if e := exactEnd(ctx, r); e != nil {
164 return ports.BlobInfo{}, e
165 }
166 _, e := s.client.PutObject(ctx, &s3.PutObjectInput{Bucket: aws.String(s.cfg.Bucket), Key: s.key(key), Body: bytes.NewReader(b), ContentLength: aws.Int64(size)})
167 if e != nil {
168 return ports.BlobInfo{}, translate(e)
169 }
170 } else {
171 u, e := s.NewMultipart(ctx, key)
172 if e != nil {
173 return ports.BlobInfo{}, e
174 }
175 completed := false
176 defer func() {
177 if !completed {
178 cleanup, cancel := context.WithTimeout(context.WithoutCancel(ctx), 60*time.Second)
179 defer cancel()
180 _ = s.AbortMultipart(cleanup, key, u.ID())
181 }
182 }()
183 remaining := size
184 for n := 1; remaining > 0; n++ {
185 count := min(remaining, int64(ports.BlobPartSize))
186 b := (*bp)[:int(count)]
187 if e = readPart(ctx, source, b); e != nil {
188 return ports.BlobInfo{}, e
189 }
190 if e = u.UploadPart(ctx, n, bytes.NewReader(b), count); e != nil {
191 return ports.BlobInfo{}, e
192 }
193 remaining -= count
194 }
195 if e = exactEnd(ctx, r); e != nil {
196 return ports.BlobInfo{}, e
197 }
198 if _, e = u.Complete(ctx); e != nil {
199 return ports.BlobInfo{}, e
200 }
201 completed = true
202 }
203 log := zerowrap.FromCtx(ctx)
204 log.Debug().Str("key", key).Int64("size", size).Msg("blob stored")
205 return ports.BlobInfo{Key: key, Size: size, Digest: "sha256:" + hex.EncodeToString(h.Sum(nil)), ModTime: time.Now().UTC()}, nil
206}
207
208// Stat retrieves metadata without downloading content.
209func (s *Store) Stat(ctx context.Context, key string) (ports.BlobInfo, error) {
210 if e := valid(key); e != nil {
211 return ports.BlobInfo{}, e
212 }
213 o, e := s.client.HeadObject(ctx, &s3.HeadObjectInput{Bucket: aws.String(s.cfg.Bucket), Key: s.key(key)})
214 if e != nil {
215 return ports.BlobInfo{}, translate(e)
216 }
217 return ports.BlobInfo{Key: key, Size: aws.ToInt64(o.ContentLength), ModTime: aws.ToTime(o.LastModified)}, nil
218}
219
220// Open streams a single range of an object.
221//
222//hexcheck:ignore suspicious-business-logic-in-adapter translates the port offset/length contract into an HTTP Range header; range rules belong to the port
223func (s *Store) Open(ctx context.Context, key string, offset, length int64) (io.ReadCloser, ports.BlobInfo, error) {
224 if e := valid(key); e != nil {
225 return nil, ports.BlobInfo{}, e
226 }
227 if offset < 0 {
228 return nil, ports.BlobInfo{}, domain.ErrInvalid
229 }
230 if length == 0 {
231 info, e := s.Stat(ctx, key)
232 if e != nil {
233 return nil, info, e
234 }
235 if offset >= info.Size {
236 return nil, info, domain.ErrOutOfRange
237 }
238 return io.NopCloser(strings.NewReader("")), info, nil
239 }
240 rg := fmt.Sprintf("bytes=%d-", offset)
241 if length > 0 && length-1 <= int64(^uint64(0)>>1)-offset {
242 rg += strconv.FormatInt(offset+length-1, 10)
243 }
244 o, e := s.client.GetObject(ctx, &s3.GetObjectInput{Bucket: aws.String(s.cfg.Bucket), Key: s.key(key), Range: aws.String(rg)})
245 if e != nil {
246 return nil, ports.BlobInfo{}, translate(e)
247 }
248 size := aws.ToInt64(o.ContentLength)
249 if _, total, ok := strings.Cut(aws.ToString(o.ContentRange), "/"); ok {
250 if n, e := strconv.ParseInt(total, 10, 64); e == nil {
251 size = n
252 }
253 }
254 return o.Body, ports.BlobInfo{Key: key, Size: size, ModTime: aws.ToTime(o.LastModified)}, nil
255}
256
257// Delete removes a key; absence is success.
258func (s *Store) Delete(ctx context.Context, key string) error {
259 logEntry := zerowrap.FromCtx(ctx)
260 logEntry.Debug().Msg("blobs3: Delete")
261 if e := valid(key); e != nil {
262 return e
263 }
264 _, e := s.client.DeleteObject(ctx, &s3.DeleteObjectInput{Bucket: aws.String(s.cfg.Bucket), Key: s.key(key)})
265 if status(e) == 404 {
266 return nil
267 }
268 return translate(e)
269}
270
271// List visits matching objects lexicographically.
272func (s *Store) List(ctx context.Context, prefix string, fn func(ports.BlobInfo) error) error {
273 logEntry := zerowrap.FromCtx(ctx)
274 logEntry.Debug().Msg("blobs3: List")
275 p := s3.NewListObjectsV2Paginator(s.client, &s3.ListObjectsV2Input{Bucket: aws.String(s.cfg.Bucket), Prefix: s.key(prefix)})
276 for p.HasMorePages() {
277 o, e := p.NextPage(ctx)
278 if e != nil {
279 return translate(e)
280 }
281 for _, v := range o.Contents {
282 if e = fn(ports.BlobInfo{Key: strings.TrimPrefix(aws.ToString(v.Key), s.cfg.Prefix), Size: aws.ToInt64(v.Size), ModTime: aws.ToTime(v.LastModified)}); e != nil {
283 return e
284 }
285 }
286 }
287 return nil
288}
289
290// Check probes the existing bucket without modifying it.
291func (s *Store) Check(ctx context.Context) error {
292 logEntry := zerowrap.FromCtx(ctx)
293 logEntry.Debug().Msg("blobs3: Check")
294 _, e := s.client.HeadBucket(ctx, &s3.HeadBucketInput{Bucket: aws.String(s.cfg.Bucket)})
295 if status(e) == 403 {
296 _, e = s.client.ListObjectsV2(ctx, &s3.ListObjectsV2Input{Bucket: aws.String(s.cfg.Bucket), Prefix: aws.String(s.cfg.Prefix), MaxKeys: aws.Int32(1)})
297 }
298 if e == nil {
299 return nil
300 }
301 where := s.cfg.Endpoint
302 if where == "" {
303 where = s.cfg.Region
304 }
305 return fmt.Errorf("bucket %q is not reachable at %s: %w", s.cfg.Bucket, where, e)
306}
307
308type part struct {
309 etag string
310 size int64
311}
312type upload struct {
313 s *Store
314 key, id string
315 parts map[int]part
316 done bool
317}
318
319// NewMultipart initiates an object upload.
320func (s *Store) NewMultipart(ctx context.Context, key string) (ports.MultipartUpload, error) {
321 logEntry := zerowrap.FromCtx(ctx)
322 logEntry.Debug().Msg("blobs3: NewMultipart")
323 if e := valid(key); e != nil {
324 return nil, e
325 }
326 o, e := s.client.CreateMultipartUpload(ctx, &s3.CreateMultipartUploadInput{Bucket: aws.String(s.cfg.Bucket), Key: s.key(key)})
327 if e != nil {
328 return nil, translate(e)
329 }
330 return &upload{s: s, key: key, id: aws.ToString(o.UploadId), parts: map[int]part{}}, nil
331}
332func (u *upload) ID() string { return u.id }
333func (u *upload) UploadPart(ctx context.Context, n int, r io.Reader, size int64) error {
334 logEntry := zerowrap.FromCtx(ctx)
335 logEntry.Debug().Msg("blobs3: UploadPart")
336 if u.done || n < 1 || n > ports.BlobMaxParts || size < 0 || size > ports.BlobPartSize {
337 return domain.ErrInvalid
338 }
339 if e := ctx.Err(); e != nil {
340 return e
341 }
342 var body io.ReadSeeker
343 if seeker, ok := r.(io.ReadSeeker); ok {
344 start, e := seeker.Seek(0, io.SeekCurrent)
345 if e != nil {
346 return e
347 }
348 end, e := seeker.Seek(0, io.SeekEnd)
349 if e != nil {
350 return e
351 }
352 if _, e = seeker.Seek(start, io.SeekStart); e != nil {
353 return e
354 }
355 if end-start != size {
356 return domain.ErrInvalid
357 }
358 body = seeker
359 } else {
360 bp := buffers.Get().(*[]byte)
361 defer buffers.Put(bp)
362 b := (*bp)[:int(size)]
363 if e := readPart(ctx, r, b); e != nil {
364 return e
365 }
366 if e := exactEnd(ctx, r); e != nil {
367 return e
368 }
369 body = bytes.NewReader(b)
370 }
371 o, e := u.s.client.UploadPart(ctx, &s3.UploadPartInput{Bucket: aws.String(u.s.cfg.Bucket), Key: u.s.key(u.key), UploadId: aws.String(u.id), PartNumber: aws.Int32(int32(n)), Body: body, ContentLength: aws.Int64(size)})
372 if e != nil {
373 return translate(e)
374 }
375 u.parts[n] = part{aws.ToString(o.ETag), size}
376 log := zerowrap.FromCtx(ctx)
377 log.Debug().Int("part", n).Msg("blob part uploaded")
378 return nil
379}
380func (u *upload) Complete(ctx context.Context) (ports.BlobInfo, error) {
381 logEntry := zerowrap.FromCtx(ctx)
382 logEntry.Debug().Msg("blobs3: Complete")
383 if u.done {
384 return ports.BlobInfo{}, domain.ErrInvalid
385 }
386 u.done = true
387 var nums []int
388 for n := range u.parts {
389 nums = append(nums, n)
390 }
391 sort.Ints(nums)
392 var parts []types.CompletedPart
393 var size int64
394 invalid := len(nums) == 0
395 for i, n := range nums {
396 p := u.parts[n]
397 if i < len(nums)-1 && p.size < ports.BlobMinPartSize {
398 invalid = true
399 }
400 size += p.size
401 parts = append(parts, types.CompletedPart{PartNumber: aws.Int32(int32(n)), ETag: aws.String(p.etag)})
402 }
403 if invalid {
404 _ = u.s.AbortMultipart(ctx, u.key, u.id)
405 return ports.BlobInfo{}, domain.ErrInvalid
406 }
407 _, e := u.s.client.CompleteMultipartUpload(ctx, &s3.CompleteMultipartUploadInput{Bucket: aws.String(u.s.cfg.Bucket), Key: u.s.key(u.key), UploadId: aws.String(u.id), MultipartUpload: &types.CompletedMultipartUpload{Parts: parts}})
408 if e != nil {
409 cleanup, cancel := context.WithTimeout(context.WithoutCancel(ctx), 60*time.Second)
410 defer cancel()
411 _ = u.s.AbortMultipart(cleanup, u.key, u.id)
412 return ports.BlobInfo{}, translate(e)
413 }
414 return ports.BlobInfo{Key: u.key, Size: size, ModTime: time.Now().UTC()}, nil
415}
416func (u *upload) Abort(ctx context.Context) error {
417 logEntry := zerowrap.FromCtx(ctx)
418 logEntry.Debug().Msg("blobs3: Abort")
419 if u.done {
420 return domain.ErrInvalid
421 }
422 u.done = true
423 return u.s.AbortMultipart(ctx, u.key, u.id)
424}
425
426// AbortMultipart discards a listed upload.
427func (s *Store) AbortMultipart(ctx context.Context, key, id string) error {
428 logEntry := zerowrap.FromCtx(ctx)
429 logEntry.Debug().Msg("blobs3: AbortMultipart")
430 if e := valid(key); e != nil {
431 return e
432 }
433 _, e := s.client.AbortMultipartUpload(ctx, &s3.AbortMultipartUploadInput{Bucket: aws.String(s.cfg.Bucket), Key: s.key(key), UploadId: aws.String(id)})
434 if status(e) == 404 {
435 return nil
436 }
437 return translate(e)
438}
439
440// ListMultipart visits unfinished uploads matching prefix.
441func (s *Store) ListMultipart(ctx context.Context, prefix string, fn func(ports.MultipartInfo) error) error {
442 logEntry := zerowrap.FromCtx(ctx)
443 logEntry.Debug().Msg("blobs3: ListMultipart")
444 input := &s3.ListMultipartUploadsInput{Bucket: aws.String(s.cfg.Bucket), Prefix: s.key(prefix)}
445 for {
446 o, e := s.client.ListMultipartUploads(ctx, input)
447 if e != nil {
448 return translate(e)
449 }
450 for _, v := range o.Uploads {
451 if e = fn(ports.MultipartInfo{Key: strings.TrimPrefix(aws.ToString(v.Key), s.cfg.Prefix), ID: aws.ToString(v.UploadId), Started: aws.ToTime(v.Initiated)}); e != nil {
452 return e
453 }
454 }
455 if !aws.ToBool(o.IsTruncated) {
456 return nil
457 }
458 input.KeyMarker = o.NextKeyMarker
459 input.UploadIdMarker = o.NextUploadIdMarker
460 }
461}