mirror of
https://github.com/Dvorinka/Primora.git
synced 2026-10-28 07:41:23 +00:00
feat: S3-compatible object storage backend
BACKEND_STORAGE_DRIVER=s3 swaps LocalStore for a stdlib-only SigV4 implementation — PutObject/GetObject/DeleteObject/CopyObject/ListObjectsV2 against any S3-compatible API (AWS, MinIO, Garage, R2). Storage becomes a Store interface; storage_path records s3://bucket/key. Content-Length is resolved by seeking or temp-file spooling — chunked uploads are rejected by real S3 servers. Bucket auto-created on startup. Verified live against zenko/cloudserver: bucket, put, get, copy, delete. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
1 parent
f747b33b46
commit
3aca5bd1e0
11 files changed
+806
-13
No files matched your search
@@ -50,6 +50,17 @@ BACKEND_PUBLIC_URL=http://localhost/api/v1
|
|||||||
BACKEND_INTERNAL_URL=http://backend:8080
|
BACKEND_INTERNAL_URL=http://backend:8080
|
||||||
BACKEND_STORAGE_ROOT=/data/storage
|
BACKEND_STORAGE_ROOT=/data/storage
|
||||||
|
|
||||||
|
# Object storage driver: "local" (default, uses BACKEND_STORAGE_ROOT) or "s3".
|
||||||
|
# For s3, point S3_ENDPOINT at any S3-compatible API (AWS, MinIO, Garage, R2).
|
||||||
|
BACKEND_STORAGE_DRIVER=local
|
||||||
|
#S3_ENDPOINT=http://minio:9000
|
||||||
|
#S3_REGION=us-east-1
|
||||||
|
#S3_BUCKET=primora-objects
|
||||||
|
#S3_ACCESS_KEY_ID=
|
||||||
|
#S3_SECRET_ACCESS_KEY=
|
||||||
|
#S3_PREFIX=
|
||||||
|
#S3_PATH_STYLE=true
|
||||||
|
|
||||||
# Mail defaults — each is overridable in-app (Settings → Instance). In-app
|
# Mail defaults — each is overridable in-app (Settings → Instance). In-app
|
||||||
# values win over these env vars; deleting the in-app value reverts here.
|
# values win over these env vars; deleting the in-app value reverts here.
|
||||||
RESEND_API_KEY=
|
RESEND_API_KEY=
|
||||||
|
|||||||
@@ -23,6 +23,7 @@ All notable changes to Primora. Format follows [Keep a Changelog](https://keepac
|
|||||||
- **Document filtering and ordering** — `GET /collections/:id/documents` accepts `filter` (comma-separated `field.op.value`; ops `eq neq gt gte lt lte like in is`; dotted paths index into document data, `id`/`created_at`/`updated_at` filterable) and `order` (`field.asc|field.desc` list). Whitelisted parsing + parameterized values; plain lists unchanged. CLI: `primora documents list --collection <slug> --filter … --order …`.
|
- **Document filtering and ordering** — `GET /collections/:id/documents` accepts `filter` (comma-separated `field.op.value`; ops `eq neq gt gte lt lte like in is`; dotted paths index into document data, `id`/`created_at`/`updated_at` filterable) and `order` (`field.asc|field.desc` list). Whitelisted parsing + parameterized values; plain lists unchanged. CLI: `primora documents list --collection <slug> --filter … --order …`.
|
||||||
- **HA scheduler leadership** — the job scheduler and alert evaluator each hold a session-scoped PostgreSQL advisory lock on a dedicated pinned connection, so multiple backend replicas scan for due work exactly once; manual and hook-triggered runs remain available on any replica. Leadership is released on shutdown and fails over automatically when a replica dies — verified live with a two-replica test.
|
- **HA scheduler leadership** — the job scheduler and alert evaluator each hold a session-scoped PostgreSQL advisory lock on a dedicated pinned connection, so multiple backend replicas scan for due work exactly once; manual and hook-triggered runs remain available on any replica. Leadership is released on shutdown and fails over automatically when a replica dies — verified live with a two-replica test.
|
||||||
- **Email templates + send log** (`core.email_log`, migration 00014) — the mailer renders registered templates (currently `invitation`) and every send — successful or failed — is recorded with template, recipient, subject, status, and error text. `GET /projects/:id/emails` lists recent sends; the Members page shows an "Email log" card under pending invitations.
|
- **Email templates + send log** (`core.email_log`, migration 00014) — the mailer renders registered templates (currently `invitation`) and every send — successful or failed — is recorded with template, recipient, subject, status, and error text. `GET /projects/:id/emails` lists recent sends; the Members page shows an "Email log" card under pending invitations.
|
||||||
|
- **S3-compatible object storage backend** — `BACKEND_STORAGE_DRIVER=s3` swaps the local disk store for any S3-compatible API (AWS, MinIO, Garage, R2) via a stdlib-only SigV4 implementation — no AWS SDK dependency. `S3_ENDPOINT`/`S3_REGION`/`S3_BUCKET`/credentials configure it; `S3_PATH_STYLE` (default on) covers MinIO/Garage, off for AWS virtual-hosted. The configured bucket is created on startup if missing; objects stream with computed SHA-256, `storage_path` records `s3://bucket/key`. Verified live end-to-end (bucket → put → get → copy → list → delete).
|
||||||
|
|
||||||
## [0.7.1] - 2026-09-23
|
## [0.7.1] - 2026-09-23
|
||||||
|
|
||||||
|
|||||||
+5
-1
@@ -41,7 +41,11 @@ docker compose -f docker-compose.yml -f docker-compose.dev.yml up -d
|
|||||||
| `AUTH_BASE_URL` / `BETTER_AUTH_URL` | public auth URL (`https://host/auth`) | yes |
|
| `AUTH_BASE_URL` / `BETTER_AUTH_URL` | public auth URL (`https://host/auth`) | yes |
|
||||||
| `AUTH_INTERNAL_BASE_URL` | backend → auth internal URL (`http://auth:3001`) | yes |
|
| `AUTH_INTERNAL_BASE_URL` | backend → auth internal URL (`http://auth:3001`) | yes |
|
||||||
| `VITE_APP_URL` / `VITE_AUTH_BASE_URL` / `VITE_API_BASE_URL` | frontend public URLs — **baked into the frontend image at build time** | yes |
|
| `VITE_APP_URL` / `VITE_AUTH_BASE_URL` / `VITE_API_BASE_URL` | frontend public URLs — **baked into the frontend image at build time** | yes |
|
||||||
| `BACKEND_STORAGE_ROOT` | object storage directory inside the backend container | yes |
|
| `BACKEND_STORAGE_ROOT` | object storage directory inside the backend container (local driver) | yes |
|
||||||
|
| `BACKEND_STORAGE_DRIVER` | `local` (default) or `s3` — S3-compatible object storage | no |
|
||||||
|
| `S3_ENDPOINT` / `S3_REGION` / `S3_BUCKET` | S3 API endpoint (empty endpoint → AWS `s3.<region>.amazonaws.com`), region, bucket | with `s3` |
|
||||||
|
| `S3_ACCESS_KEY_ID` / `S3_SECRET_ACCESS_KEY` | S3 credentials | with `s3` |
|
||||||
|
| `S3_PREFIX` / `S3_PATH_STYLE` | optional key prefix; path-style URLs (default `true`, needed by MinIO/Garage — set `false` for AWS virtual-hosted) | no |
|
||||||
| `AUTH_ADMIN_EMAILS` | comma-separated emails promoted to auth admin on boot | no |
|
| `AUTH_ADMIN_EMAILS` | comma-separated emails promoted to auth admin on boot | no |
|
||||||
| `USER_RATE_LIMIT_PER_MINUTE` / `API_KEY_RATE_LIMIT_PER_MINUTE` | per-identity API rate limits (defaults 240 / 600) | no |
|
| `USER_RATE_LIMIT_PER_MINUTE` / `API_KEY_RATE_LIMIT_PER_MINUTE` | per-identity API rate limits (defaults 240 / 600) | no |
|
||||||
| `SMTP_HOST` / `SMTP_PORT` / `SMTP_USER` / `SMTP_PASSWORD` / `SMTP_SECURE` / `MAIL_FROM` | transactional mail; without it verification/reset emails go nowhere | production |
|
| `SMTP_HOST` / `SMTP_PORT` / `SMTP_USER` / `SMTP_PASSWORD` / `SMTP_SECURE` / `MAIL_FROM` | transactional mail; without it verification/reset emails go nowhere | production |
|
||||||
|
|||||||
+8
-8
@@ -360,14 +360,14 @@ Ordered loosely by leverage. None committed; each gets scoped when picked.
|
|||||||
- **Phase 9 — Functions** — user code on schedule/event/HTTP triggers.
|
- **Phase 9 — Functions** — user code on schedule/event/HTTP triggers.
|
||||||
Deno or Bun isolates in a sidecar; secrets via env, payload via stdin.
|
Deno or Bun isolates in a sidecar; secrets via env, payload via stdin.
|
||||||
The job runner already supplies the scheduling half.
|
The job runner already supplies the scheduling half.
|
||||||
- **Phase 10 — Storage backends** — S3-compatible object storage behind
|
- **Phase 10 — Storage backends** — ✅ S3-compatible object storage shipped:
|
||||||
the existing bucket abstraction (deferred from Phase 6), plus
|
`BACKEND_STORAGE_DRIVER=s3` + `S3_*` env vars, stdlib SigV4, verified
|
||||||
presigned-URL uploads for large files.
|
against a real S3 server. Presigned-URL uploads for large files remain a
|
||||||
- **HA mode** — leader-elected scheduler + distributed locks so two
|
candidate.
|
||||||
backend replicas don't double-fire jobs; Postgres advisory locks are
|
- **HA mode** — ✅ shipped: PostgreSQL advisory-lock leadership for the
|
||||||
enough at this scale.
|
scheduler and alert evaluator; failover verified with two replicas.
|
||||||
- **Email surface** — transactional templates + send log in the
|
- **Email surface** — ✅ shipped: templated sends + `core.email_log` +
|
||||||
dashboard, backed by the auth service's existing SMTP/Resend wiring.
|
dashboard card; more templates land as new flows need them.
|
||||||
|
|
||||||
---
|
---
|
||||||
|
|
||||||
|
|||||||
@@ -81,10 +81,31 @@ func Bootstrap(ctx context.Context) (*App, error) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
store, err := storage.NewLocalStore(cfg.StorageRoot)
|
var store storage.Store
|
||||||
|
if cfg.StorageDriver == "s3" {
|
||||||
|
s3Store, err := storage.NewS3Store(storage.S3Config{
|
||||||
|
Endpoint: cfg.S3Endpoint,
|
||||||
|
Region: cfg.S3Region,
|
||||||
|
Bucket: cfg.S3Bucket,
|
||||||
|
AccessKeyID: cfg.S3AccessKeyID,
|
||||||
|
SecretAccessKey: cfg.S3SecretAccessKey,
|
||||||
|
Prefix: cfg.S3Prefix,
|
||||||
|
PathStyle: cfg.S3PathStyle,
|
||||||
|
})
|
||||||
|
if err != nil {
|
||||||
|
return nil, fmt.Errorf("s3 storage: %w", err)
|
||||||
|
}
|
||||||
|
if err := s3Store.EnsureBucket(context.Background()); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
store = s3Store
|
||||||
|
} else {
|
||||||
|
localStore, err := storage.NewLocalStore(cfg.StorageRoot)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
store = localStore
|
||||||
|
}
|
||||||
dbxClient := dbx.NewClient(logger, cfg.DBXDataDir)
|
dbxClient := dbx.NewClient(logger, cfg.DBXDataDir)
|
||||||
repo := repositories.NewCoreRepository(dbPool)
|
repo := repositories.NewCoreRepository(dbPool)
|
||||||
encryptor, err := secrets.NewEncryptor(cfg.EncryptionKey)
|
encryptor, err := secrets.NewEncryptor(cfg.EncryptionKey)
|
||||||
|
|||||||
@@ -35,6 +35,16 @@ type Config struct {
|
|||||||
SMTPSecure bool
|
SMTPSecure bool
|
||||||
|
|
||||||
EncryptionKey string
|
EncryptionKey string
|
||||||
|
|
||||||
|
// Object storage: "local" (BACKEND_STORAGE_ROOT) or "s3".
|
||||||
|
StorageDriver string
|
||||||
|
S3Endpoint string
|
||||||
|
S3Region string
|
||||||
|
S3Bucket string
|
||||||
|
S3AccessKeyID string
|
||||||
|
S3SecretAccessKey string
|
||||||
|
S3Prefix string
|
||||||
|
S3PathStyle bool
|
||||||
}
|
}
|
||||||
|
|
||||||
// devEncryptionKey is the built-in credential key for local development only.
|
// devEncryptionKey is the built-in credential key for local development only.
|
||||||
@@ -63,6 +73,13 @@ func Load() (Config, error) {
|
|||||||
SMTPUser: os.Getenv("SMTP_USER"),
|
SMTPUser: os.Getenv("SMTP_USER"),
|
||||||
SMTPPassword: os.Getenv("SMTP_PASSWORD"),
|
SMTPPassword: os.Getenv("SMTP_PASSWORD"),
|
||||||
EncryptionKey: os.Getenv("PRIMORA_ENCRYPTION_KEY"),
|
EncryptionKey: os.Getenv("PRIMORA_ENCRYPTION_KEY"),
|
||||||
|
StorageDriver: getenv("BACKEND_STORAGE_DRIVER", "local"),
|
||||||
|
S3Endpoint: os.Getenv("S3_ENDPOINT"),
|
||||||
|
S3Region: getenv("S3_REGION", "us-east-1"),
|
||||||
|
S3Bucket: os.Getenv("S3_BUCKET"),
|
||||||
|
S3AccessKeyID: os.Getenv("S3_ACCESS_KEY_ID"),
|
||||||
|
S3SecretAccessKey: os.Getenv("S3_SECRET_ACCESS_KEY"),
|
||||||
|
S3Prefix: os.Getenv("S3_PREFIX"),
|
||||||
}
|
}
|
||||||
|
|
||||||
smtpPort, err := strconv.Atoi(getenv("SMTP_PORT", "1025"))
|
smtpPort, err := strconv.Atoi(getenv("SMTP_PORT", "1025"))
|
||||||
@@ -95,6 +112,12 @@ func Load() (Config, error) {
|
|||||||
}
|
}
|
||||||
cfg.SMTPSecure = smtpSecure
|
cfg.SMTPSecure = smtpSecure
|
||||||
|
|
||||||
|
s3PathStyle, err := strconv.ParseBool(getenv("S3_PATH_STYLE", "true"))
|
||||||
|
if err != nil {
|
||||||
|
return Config{}, fmt.Errorf("parse S3_PATH_STYLE: %w", err)
|
||||||
|
}
|
||||||
|
cfg.S3PathStyle = s3PathStyle
|
||||||
|
|
||||||
var missing []string
|
var missing []string
|
||||||
if cfg.DatabaseURL == "" {
|
if cfg.DatabaseURL == "" {
|
||||||
missing = append(missing, "DATABASE_URL")
|
missing = append(missing, "DATABASE_URL")
|
||||||
@@ -102,9 +125,24 @@ func Load() (Config, error) {
|
|||||||
if cfg.JWTSecret == "" {
|
if cfg.JWTSecret == "" {
|
||||||
missing = append(missing, "JWT_SECRET")
|
missing = append(missing, "JWT_SECRET")
|
||||||
}
|
}
|
||||||
|
switch cfg.StorageDriver {
|
||||||
|
case "local":
|
||||||
if cfg.StorageRoot == "" {
|
if cfg.StorageRoot == "" {
|
||||||
missing = append(missing, "BACKEND_STORAGE_ROOT")
|
missing = append(missing, "BACKEND_STORAGE_ROOT")
|
||||||
}
|
}
|
||||||
|
case "s3":
|
||||||
|
if cfg.S3Bucket == "" {
|
||||||
|
missing = append(missing, "S3_BUCKET")
|
||||||
|
}
|
||||||
|
if cfg.S3AccessKeyID == "" {
|
||||||
|
missing = append(missing, "S3_ACCESS_KEY_ID")
|
||||||
|
}
|
||||||
|
if cfg.S3SecretAccessKey == "" {
|
||||||
|
missing = append(missing, "S3_SECRET_ACCESS_KEY")
|
||||||
|
}
|
||||||
|
default:
|
||||||
|
return Config{}, fmt.Errorf("unknown BACKEND_STORAGE_DRIVER %q (want local or s3)", cfg.StorageDriver)
|
||||||
|
}
|
||||||
if cfg.AuthInternalBaseURL == "" {
|
if cfg.AuthInternalBaseURL == "" {
|
||||||
missing = append(missing, "AUTH_INTERNAL_BASE_URL")
|
missing = append(missing, "AUTH_INTERNAL_BASE_URL")
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -29,7 +29,7 @@ import (
|
|||||||
|
|
||||||
type PlatformService struct {
|
type PlatformService struct {
|
||||||
repo *repositories.CoreRepository
|
repo *repositories.CoreRepository
|
||||||
store *storage.LocalStore
|
store storage.Store
|
||||||
mailer *Mailer
|
mailer *Mailer
|
||||||
publicURL string
|
publicURL string
|
||||||
dbx *dbx.Client
|
dbx *dbx.Client
|
||||||
@@ -224,7 +224,7 @@ type InvitationSummary struct {
|
|||||||
Status string `json:"status"`
|
Status string `json:"status"`
|
||||||
}
|
}
|
||||||
|
|
||||||
func NewPlatformService(repo *repositories.CoreRepository, store *storage.LocalStore, mailer *Mailer, publicURL string, dbxClient *dbx.Client, enc *secrets.Encryptor, logger *slog.Logger) *PlatformService {
|
func NewPlatformService(repo *repositories.CoreRepository, store storage.Store, mailer *Mailer, publicURL string, dbxClient *dbx.Client, enc *secrets.Encryptor, logger *slog.Logger) *PlatformService {
|
||||||
s := &PlatformService{repo: repo, store: store, mailer: mailer, publicURL: publicURL, dbx: dbxClient, hub: NewEventHub(), realtime: NewEventHub(), enc: enc}
|
s := &PlatformService{repo: repo, store: store, mailer: mailer, publicURL: publicURL, dbx: dbxClient, hub: NewEventHub(), realtime: NewEventHub(), enc: enc}
|
||||||
if enc != nil {
|
if enc != nil {
|
||||||
s.dispatcher = NewWebhookDispatcher(repo.Queries(), enc, logger)
|
s.dispatcher = NewWebhookDispatcher(repo.Queries(), enc, logger)
|
||||||
|
|||||||
@@ -65,7 +65,7 @@ func (s *LocalStore) Put(ctx context.Context, bucketID, objectKey string, reader
|
|||||||
}, nil
|
}, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (s *LocalStore) Open(bucketID, objectKey string) (*os.File, string, error) {
|
func (s *LocalStore) Open(bucketID, objectKey string) (io.ReadCloser, string, error) {
|
||||||
cleanKey, err := sanitizeObjectKey(objectKey)
|
cleanKey, err := sanitizeObjectKey(objectKey)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, "", err
|
return nil, "", err
|
||||||
|
|||||||
@@ -0,0 +1,486 @@
|
|||||||
|
package storage
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"crypto/hmac"
|
||||||
|
"crypto/sha256"
|
||||||
|
"encoding/hex"
|
||||||
|
"encoding/xml"
|
||||||
|
"fmt"
|
||||||
|
"io"
|
||||||
|
"net/http"
|
||||||
|
"net/url"
|
||||||
|
"os"
|
||||||
|
"path"
|
||||||
|
"sort"
|
||||||
|
"strings"
|
||||||
|
"time"
|
||||||
|
)
|
||||||
|
|
||||||
|
// S3Config configures an S3-compatible object store (AWS S3, MinIO,
|
||||||
|
// Cloudflare R2, Garage, …). Endpoint may be empty for AWS, in which case
|
||||||
|
// the standard regional endpoint is derived from Region + Bucket.
|
||||||
|
type S3Config struct {
|
||||||
|
Endpoint string // e.g. http://localhost:9000 or https://s3.eu-central-1.amazonaws.com
|
||||||
|
Region string
|
||||||
|
Bucket string
|
||||||
|
AccessKeyID string
|
||||||
|
SecretAccessKey string
|
||||||
|
Prefix string // optional key prefix, e.g. "primora/"
|
||||||
|
PathStyle bool // bucket/key path style (MinIO, Garage) vs virtual-hosted
|
||||||
|
HTTPClient *http.Client
|
||||||
|
}
|
||||||
|
|
||||||
|
// S3Store implements Store against an S3-compatible API using stdlib-only
|
||||||
|
// SigV4 signing — no AWS SDK dependency. Payloads are sent with
|
||||||
|
// x-amz-content-sha256=UNSIGNED-PAYLOAD so bodies stream without buffering;
|
||||||
|
// the SHA-256 for PutResult is computed on a hashing tee instead.
|
||||||
|
// jarvis: ceiling UNSIGNED-PAYLOAD needs TLS on AWS proper; over plain
|
||||||
|
// http:// it relies on the server accepting it (MinIO/Garage/LocalStack do).
|
||||||
|
// Upgrade to aws-chunked streaming signatures if a backend rejects it.
|
||||||
|
type S3Store struct {
|
||||||
|
cfg S3Config
|
||||||
|
client *http.Client
|
||||||
|
now func() time.Time // injectable for tests
|
||||||
|
}
|
||||||
|
|
||||||
|
func NewS3Store(cfg S3Config) (*S3Store, error) {
|
||||||
|
if cfg.Bucket == "" || cfg.AccessKeyID == "" || cfg.SecretAccessKey == "" {
|
||||||
|
return nil, fmt.Errorf("s3 storage requires S3_BUCKET, S3_ACCESS_KEY_ID and S3_SECRET_ACCESS_KEY")
|
||||||
|
}
|
||||||
|
if cfg.Region == "" {
|
||||||
|
cfg.Region = "us-east-1"
|
||||||
|
}
|
||||||
|
if cfg.Endpoint == "" {
|
||||||
|
cfg.Endpoint = fmt.Sprintf("https://s3.%s.amazonaws.com", cfg.Region)
|
||||||
|
}
|
||||||
|
client := cfg.HTTPClient
|
||||||
|
if client == nil {
|
||||||
|
client = &http.Client{Timeout: 5 * time.Minute}
|
||||||
|
}
|
||||||
|
return &S3Store{cfg: cfg, client: client, now: time.Now}, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func escapeKey(key string) string {
|
||||||
|
parts := strings.Split(key, "/")
|
||||||
|
for i, p := range parts {
|
||||||
|
parts[i] = url.PathEscape(p)
|
||||||
|
}
|
||||||
|
return strings.Join(parts, "/")
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *S3Store) Put(ctx context.Context, bucketID, objectKey string, reader io.Reader) (PutResult, error) {
|
||||||
|
key, err := s.objectKey(bucketID, objectKey)
|
||||||
|
if err != nil {
|
||||||
|
return PutResult{}, err
|
||||||
|
}
|
||||||
|
|
||||||
|
// S3 requires a Content-Length — chunked transfer encoding is rejected.
|
||||||
|
// Seekable readers (uploaded *os.File) give the length for free;
|
||||||
|
// anything else is spooled to a temp file, same cost the local driver
|
||||||
|
// pays on every write anyway.
|
||||||
|
body, length, cleanup, err := bodyWithLength(reader)
|
||||||
|
if err != nil {
|
||||||
|
return PutResult{}, err
|
||||||
|
}
|
||||||
|
defer cleanup()
|
||||||
|
|
||||||
|
hasher := sha256.New()
|
||||||
|
counter := &countingReader{r: io.TeeReader(body, hasher)}
|
||||||
|
|
||||||
|
req, err := s.signedRequest(ctx, http.MethodPut, key, counter)
|
||||||
|
if err != nil {
|
||||||
|
return PutResult{}, err
|
||||||
|
}
|
||||||
|
req.ContentLength = length
|
||||||
|
resp, err := s.client.Do(req)
|
||||||
|
if err != nil {
|
||||||
|
return PutResult{}, fmt.Errorf("s3 put: %w", err)
|
||||||
|
}
|
||||||
|
defer resp.Body.Close()
|
||||||
|
if resp.StatusCode/100 != 2 {
|
||||||
|
return PutResult{}, fmt.Errorf("s3 put: %s: %s", resp.Status, readErrBody(resp))
|
||||||
|
}
|
||||||
|
return PutResult{
|
||||||
|
Path: fmt.Sprintf("s3://%s/%s", s.cfg.Bucket, key),
|
||||||
|
SizeBytes: counter.n,
|
||||||
|
SHA256Digest: hex.EncodeToString(hasher.Sum(nil)),
|
||||||
|
}, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// bodyWithLength returns a reader plus its total length. The cleanup func
|
||||||
|
// removes any temp file it created.
|
||||||
|
func bodyWithLength(r io.Reader) (io.Reader, int64, func(), error) {
|
||||||
|
nop := func() {}
|
||||||
|
if seeker, ok := r.(io.Seeker); ok {
|
||||||
|
size, err := seeker.Seek(0, io.SeekEnd)
|
||||||
|
if err != nil {
|
||||||
|
return nil, 0, nil, fmt.Errorf("seek object body: %w", err)
|
||||||
|
}
|
||||||
|
if _, err := seeker.Seek(0, io.SeekStart); err != nil {
|
||||||
|
return nil, 0, nil, fmt.Errorf("rewind object body: %w", err)
|
||||||
|
}
|
||||||
|
return r, size, nop, nil
|
||||||
|
}
|
||||||
|
tmp, err := os.CreateTemp("", "primora-s3-put-*")
|
||||||
|
if err != nil {
|
||||||
|
return nil, 0, nil, fmt.Errorf("spool object body: %w", err)
|
||||||
|
}
|
||||||
|
size, err := io.Copy(tmp, r)
|
||||||
|
if err != nil {
|
||||||
|
tmp.Close()
|
||||||
|
os.Remove(tmp.Name())
|
||||||
|
return nil, 0, nil, fmt.Errorf("spool object body: %w", err)
|
||||||
|
}
|
||||||
|
if _, err := tmp.Seek(0, io.SeekStart); err != nil {
|
||||||
|
tmp.Close()
|
||||||
|
os.Remove(tmp.Name())
|
||||||
|
return nil, 0, nil, err
|
||||||
|
}
|
||||||
|
cleanup := func() {
|
||||||
|
tmp.Close()
|
||||||
|
os.Remove(tmp.Name())
|
||||||
|
}
|
||||||
|
return tmp, size, cleanup, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
type countingReader struct {
|
||||||
|
r io.Reader
|
||||||
|
n int64
|
||||||
|
}
|
||||||
|
|
||||||
|
func (c *countingReader) Read(p []byte) (int, error) {
|
||||||
|
n, err := c.r.Read(p)
|
||||||
|
c.n += int64(n)
|
||||||
|
return n, err
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *S3Store) Open(bucketID, objectKey string) (io.ReadCloser, string, error) {
|
||||||
|
key, err := s.objectKey(bucketID, objectKey)
|
||||||
|
if err != nil {
|
||||||
|
return nil, "", err
|
||||||
|
}
|
||||||
|
req, err := s.signedRequest(context.Background(), http.MethodGet, key, nil)
|
||||||
|
if err != nil {
|
||||||
|
return nil, "", err
|
||||||
|
}
|
||||||
|
resp, err := s.client.Do(req)
|
||||||
|
if err != nil {
|
||||||
|
return nil, "", fmt.Errorf("s3 get: %w", err)
|
||||||
|
}
|
||||||
|
if resp.StatusCode/100 != 2 {
|
||||||
|
defer resp.Body.Close()
|
||||||
|
return nil, "", fmt.Errorf("s3 get: %s: %s", resp.Status, readErrBody(resp))
|
||||||
|
}
|
||||||
|
return resp.Body, fmt.Sprintf("s3://%s/%s", s.cfg.Bucket, key), nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *S3Store) Delete(bucketID, objectKey string) error {
|
||||||
|
key, err := s.objectKey(bucketID, objectKey)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
req, err := s.signedRequest(context.Background(), http.MethodDelete, key, nil)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
resp, err := s.client.Do(req)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("s3 delete: %w", err)
|
||||||
|
}
|
||||||
|
defer resp.Body.Close()
|
||||||
|
// S3 delete is idempotent: 204 on success, 404 is fine too.
|
||||||
|
if resp.StatusCode/100 != 2 && resp.StatusCode != http.StatusNotFound {
|
||||||
|
return fmt.Errorf("s3 delete: %s: %s", resp.Status, readErrBody(resp))
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *S3Store) MoveBetweenBuckets(sourceBucketID, destinationBucketID, fromKey, toKey string) (string, error) {
|
||||||
|
src, err := s.objectKey(sourceBucketID, fromKey)
|
||||||
|
if err != nil {
|
||||||
|
return "", err
|
||||||
|
}
|
||||||
|
dst, err := s.objectKey(destinationBucketID, toKey)
|
||||||
|
if err != nil {
|
||||||
|
return "", err
|
||||||
|
}
|
||||||
|
// x-amz-copy-source must be set before signing — strict S3
|
||||||
|
// implementations require all x-amz-* headers to be signed.
|
||||||
|
key := dst
|
||||||
|
u := strings.TrimSuffix(s.cfg.Endpoint, "/") + "/" + s.cfg.Bucket + "/" + escapeKey(key)
|
||||||
|
if !s.cfg.PathStyle {
|
||||||
|
parsed, perr := url.Parse(strings.TrimSuffix(s.cfg.Endpoint, "/"))
|
||||||
|
if perr != nil {
|
||||||
|
return "", perr
|
||||||
|
}
|
||||||
|
parsed.Host = s.cfg.Bucket + "." + parsed.Host
|
||||||
|
parsed.Path = "/" + escapeKey(key)
|
||||||
|
u = parsed.String()
|
||||||
|
}
|
||||||
|
req, err := http.NewRequestWithContext(context.Background(), http.MethodPut, u, nil)
|
||||||
|
if err != nil {
|
||||||
|
return "", err
|
||||||
|
}
|
||||||
|
req.Header.Set("x-amz-content-sha256", "UNSIGNED-PAYLOAD")
|
||||||
|
req.Header.Set("x-amz-copy-source", "/"+s.cfg.Bucket+"/"+escapeKey(src))
|
||||||
|
s.sign(req)
|
||||||
|
resp, err := s.client.Do(req)
|
||||||
|
if err != nil {
|
||||||
|
return "", fmt.Errorf("s3 copy: %w", err)
|
||||||
|
}
|
||||||
|
defer resp.Body.Close()
|
||||||
|
if resp.StatusCode/100 != 2 {
|
||||||
|
return "", fmt.Errorf("s3 copy: %s: %s", resp.Status, readErrBody(resp))
|
||||||
|
}
|
||||||
|
if err := s.Delete(sourceBucketID, fromKey); err != nil {
|
||||||
|
return "", fmt.Errorf("s3 move: delete source: %w", err)
|
||||||
|
}
|
||||||
|
return fmt.Sprintf("s3://%s/%s", s.cfg.Bucket, dst), nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// DeleteBucket lists every object under the bucket prefix and deletes them
|
||||||
|
// one by one — DeleteObjects batching is an optimization not worth the code
|
||||||
|
// at this scale.
|
||||||
|
func (s *S3Store) DeleteBucket(bucketID string) error {
|
||||||
|
prefix := path.Join(s.cfg.Prefix, bucketID) + "/"
|
||||||
|
continuation := ""
|
||||||
|
for {
|
||||||
|
keys, next, err := s.listPage(prefix, continuation)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
for _, k := range keys {
|
||||||
|
req, err := s.signedRequest(context.Background(), http.MethodDelete, k, nil)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
resp, err := s.client.Do(req)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("s3 delete %s: %w", k, err)
|
||||||
|
}
|
||||||
|
resp.Body.Close()
|
||||||
|
if resp.StatusCode/100 != 2 && resp.StatusCode != http.StatusNotFound {
|
||||||
|
return fmt.Errorf("s3 delete %s: %s", k, resp.Status)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if next == "" {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
continuation = next
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
type listResult struct {
|
||||||
|
Contents []struct{ Key string `xml:"Key"` } `xml:"Contents"`
|
||||||
|
NextContinuationToken string `xml:"NextContinuationToken"`
|
||||||
|
IsTruncated bool `xml:"IsTruncated"`
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *S3Store) listPage(prefix, continuation string) ([]string, string, error) {
|
||||||
|
query := url.Values{
|
||||||
|
"list-type": {"2"},
|
||||||
|
"prefix": {prefix},
|
||||||
|
"max-keys": {"1000"},
|
||||||
|
}
|
||||||
|
if continuation != "" {
|
||||||
|
query.Set("continuation-token", continuation)
|
||||||
|
}
|
||||||
|
req, err := s.signedRequestWithQuery(context.Background(), http.MethodGet, "", query)
|
||||||
|
if err != nil {
|
||||||
|
return nil, "", err
|
||||||
|
}
|
||||||
|
resp, err := s.client.Do(req)
|
||||||
|
if err != nil {
|
||||||
|
return nil, "", fmt.Errorf("s3 list: %w", err)
|
||||||
|
}
|
||||||
|
defer resp.Body.Close()
|
||||||
|
if resp.StatusCode/100 != 2 {
|
||||||
|
return nil, "", fmt.Errorf("s3 list: %s: %s", resp.Status, readErrBody(resp))
|
||||||
|
}
|
||||||
|
var out listResult
|
||||||
|
if err := xml.NewDecoder(resp.Body).Decode(&out); err != nil {
|
||||||
|
return nil, "", fmt.Errorf("s3 list decode: %w", err)
|
||||||
|
}
|
||||||
|
keys := make([]string, len(out.Contents))
|
||||||
|
for i, c := range out.Contents {
|
||||||
|
keys[i] = c.Key
|
||||||
|
}
|
||||||
|
next := out.NextContinuationToken
|
||||||
|
if next == "" && out.IsTruncated && len(keys) > 0 {
|
||||||
|
next = keys[len(keys)-1] // S3 uses last key as marker when no token
|
||||||
|
}
|
||||||
|
return keys, next, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// --- SigV4 signing -------------------------------------------------------
|
||||||
|
|
||||||
|
func (s *S3Store) signedRequest(ctx context.Context, method, key string, body io.Reader) (*http.Request, error) {
|
||||||
|
return s.signed(ctx, method, key, nil, body)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *S3Store) signedRequestWithQuery(ctx context.Context, method, key string, query url.Values) (*http.Request, error) {
|
||||||
|
return s.signed(ctx, method, key, query, nil)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *S3Store) signed(ctx context.Context, method, key string, query url.Values, body io.Reader) (*http.Request, error) {
|
||||||
|
var reqURL string
|
||||||
|
endpoint := strings.TrimSuffix(s.cfg.Endpoint, "/")
|
||||||
|
if s.cfg.PathStyle {
|
||||||
|
reqURL = endpoint + "/" + s.cfg.Bucket
|
||||||
|
if key != "" {
|
||||||
|
reqURL += "/" + escapeKey(key)
|
||||||
|
}
|
||||||
|
} else {
|
||||||
|
u, err := url.Parse(endpoint)
|
||||||
|
if err != nil {
|
||||||
|
return nil, fmt.Errorf("s3 endpoint: %w", err)
|
||||||
|
}
|
||||||
|
u.Host = s.cfg.Bucket + "." + u.Host
|
||||||
|
u.Path = "/" + escapeKey(key)
|
||||||
|
reqURL = u.String()
|
||||||
|
}
|
||||||
|
if len(query) > 0 {
|
||||||
|
reqURL += "?" + query.Encode()
|
||||||
|
}
|
||||||
|
req, err := http.NewRequestWithContext(ctx, method, reqURL, body)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
req.Header.Set("x-amz-content-sha256", "UNSIGNED-PAYLOAD")
|
||||||
|
s.sign(req)
|
||||||
|
return req, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// sign applies AWS Signature Version 4 to req in place.
|
||||||
|
func (s *S3Store) sign(req *http.Request) {
|
||||||
|
now := s.now().UTC()
|
||||||
|
amzDate := now.Format("20060102T150405Z")
|
||||||
|
dateStamp := now.Format("20060102")
|
||||||
|
req.Header.Set("x-amz-date", amzDate)
|
||||||
|
|
||||||
|
// Canonical request: method, path, sorted query, headers, signed list,
|
||||||
|
// payload hash. Signed headers are host + every x-amz-* header present —
|
||||||
|
// required for x-amz-copy-source on CopyObject and future metadata
|
||||||
|
// headers.
|
||||||
|
signedHeaders := []string{"host"}
|
||||||
|
for h := range req.Header {
|
||||||
|
lh := strings.ToLower(h)
|
||||||
|
if strings.HasPrefix(lh, "x-amz-") {
|
||||||
|
signedHeaders = append(signedHeaders, lh)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
sort.Strings(signedHeaders)
|
||||||
|
var canonicalHeaders strings.Builder
|
||||||
|
for _, h := range signedHeaders {
|
||||||
|
v := req.Host
|
||||||
|
if h != "host" {
|
||||||
|
v = strings.TrimSpace(req.Header.Get(h))
|
||||||
|
}
|
||||||
|
canonicalHeaders.WriteString(h + ":" + v + "\n")
|
||||||
|
}
|
||||||
|
canonicalURI := req.URL.EscapedPath()
|
||||||
|
if canonicalURI == "" {
|
||||||
|
canonicalURI = "/"
|
||||||
|
}
|
||||||
|
canonicalRequest := strings.Join([]string{
|
||||||
|
req.Method,
|
||||||
|
canonicalURI,
|
||||||
|
canonicalQuery(req.URL.Query()),
|
||||||
|
canonicalHeaders.String(),
|
||||||
|
strings.Join(signedHeaders, ";"),
|
||||||
|
"UNSIGNED-PAYLOAD",
|
||||||
|
}, "\n")
|
||||||
|
|
||||||
|
scope := fmt.Sprintf("%s/%s/s3/aws4_request", dateStamp, s.cfg.Region)
|
||||||
|
stringToSign := strings.Join([]string{
|
||||||
|
"AWS4-HMAC-SHA256",
|
||||||
|
amzDate,
|
||||||
|
scope,
|
||||||
|
hexSHA256(canonicalRequest),
|
||||||
|
}, "\n")
|
||||||
|
|
||||||
|
kDate := hmacSHA256([]byte("AWS4"+s.cfg.SecretAccessKey), dateStamp)
|
||||||
|
kRegion := hmacSHA256(kDate, s.cfg.Region)
|
||||||
|
kService := hmacSHA256(kRegion, "s3")
|
||||||
|
kSigning := hmacSHA256(kService, "aws4_request")
|
||||||
|
signature := hex.EncodeToString(hmacSHA256(kSigning, stringToSign))
|
||||||
|
|
||||||
|
req.Header.Set("Authorization", fmt.Sprintf(
|
||||||
|
"AWS4-HMAC-SHA256 Credential=%s/%s, SignedHeaders=%s, Signature=%s",
|
||||||
|
s.cfg.AccessKeyID, scope, strings.Join(signedHeaders, ";"), signature))
|
||||||
|
}
|
||||||
|
|
||||||
|
// canonicalQuery re-encodes the query per SigV4: sorted keys, each name and
|
||||||
|
// value URI-encoded per RFC 3986 (url.Values.Encode's "+" for space is not
|
||||||
|
// acceptable here).
|
||||||
|
func canonicalQuery(q url.Values) string {
|
||||||
|
type pair struct{ k, v string }
|
||||||
|
pairs := make([]pair, 0, len(q))
|
||||||
|
for k, vs := range q {
|
||||||
|
for _, v := range vs {
|
||||||
|
pairs = append(pairs, pair{sigV4Escape(k), sigV4Escape(v)})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
sort.Slice(pairs, func(i, j int) bool {
|
||||||
|
if pairs[i].k == pairs[j].k {
|
||||||
|
return pairs[i].v < pairs[j].v
|
||||||
|
}
|
||||||
|
return pairs[i].k < pairs[j].k
|
||||||
|
})
|
||||||
|
var b strings.Builder
|
||||||
|
for i, p := range pairs {
|
||||||
|
if i > 0 {
|
||||||
|
b.WriteByte('&')
|
||||||
|
}
|
||||||
|
b.WriteString(p.k + "=" + p.v)
|
||||||
|
}
|
||||||
|
return b.String()
|
||||||
|
}
|
||||||
|
|
||||||
|
func sigV4Escape(s string) string {
|
||||||
|
return strings.ReplaceAll(url.QueryEscape(s), "+", "%20")
|
||||||
|
}
|
||||||
|
|
||||||
|
func hexSHA256(s string) string {
|
||||||
|
h := sha256.Sum256([]byte(s))
|
||||||
|
return hex.EncodeToString(h[:])
|
||||||
|
}
|
||||||
|
|
||||||
|
func hmacSHA256(key []byte, data string) []byte {
|
||||||
|
m := hmac.New(sha256.New, key)
|
||||||
|
m.Write([]byte(data))
|
||||||
|
return m.Sum(nil)
|
||||||
|
}
|
||||||
|
|
||||||
|
// EnsureBucket creates the configured bucket if it does not exist. Called
|
||||||
|
// once at startup when the s3 driver is selected; a pre-existing bucket is
|
||||||
|
// a no-op.
|
||||||
|
func (s *S3Store) EnsureBucket(ctx context.Context) error {
|
||||||
|
req, err := s.signedRequest(ctx, http.MethodPut, "", nil)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
resp, err := s.client.Do(req)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("s3 create bucket: %w", err)
|
||||||
|
}
|
||||||
|
defer resp.Body.Close()
|
||||||
|
if resp.StatusCode/100 != 2 && resp.StatusCode != http.StatusConflict {
|
||||||
|
return fmt.Errorf("s3 create bucket: %s: %s", resp.Status, readErrBody(resp))
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *S3Store) objectKey(bucketID, objectKey string) (string, error) {
|
||||||
|
clean, err := sanitizeObjectKey(objectKey)
|
||||||
|
if err != nil {
|
||||||
|
return "", err
|
||||||
|
}
|
||||||
|
return path.Join(s.cfg.Prefix, bucketID, clean), nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func readErrBody(resp *http.Response) string {
|
||||||
|
body, _ := io.ReadAll(io.LimitReader(resp.Body, 512))
|
||||||
|
return strings.TrimSpace(string(body))
|
||||||
|
}
|
||||||
@@ -0,0 +1,216 @@
|
|||||||
|
package storage
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"fmt"
|
||||||
|
"io"
|
||||||
|
"net/http"
|
||||||
|
"net/http/httptest"
|
||||||
|
"strings"
|
||||||
|
"sync"
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
)
|
||||||
|
|
||||||
|
// fakeS3 is an in-memory S3-compatible server covering the ops S3Store uses.
|
||||||
|
type fakeS3 struct {
|
||||||
|
mu sync.Mutex
|
||||||
|
objects map[string][]byte
|
||||||
|
bucket string
|
||||||
|
}
|
||||||
|
|
||||||
|
func newFakeS3(bucket string) *fakeS3 {
|
||||||
|
return &fakeS3{objects: map[string][]byte{}, bucket: bucket}
|
||||||
|
}
|
||||||
|
|
||||||
|
func (f *fakeS3) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
||||||
|
f.mu.Lock()
|
||||||
|
defer f.mu.Unlock()
|
||||||
|
key := strings.TrimPrefix(r.URL.Path, "/"+f.bucket+"/")
|
||||||
|
if r.Header.Get("Authorization") == "" || r.Header.Get("x-amz-date") == "" {
|
||||||
|
http.Error(w, "missing auth", http.StatusForbidden)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
switch {
|
||||||
|
case r.Method == http.MethodPut && r.Header.Get("x-amz-copy-source") != "":
|
||||||
|
src := strings.TrimPrefix(r.Header.Get("x-amz-copy-source"), "/"+f.bucket+"/")
|
||||||
|
body, ok := f.objects[src]
|
||||||
|
if !ok {
|
||||||
|
http.Error(w, "source missing", http.StatusNotFound)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
f.objects[key] = body
|
||||||
|
w.WriteHeader(http.StatusOK)
|
||||||
|
case r.Method == http.MethodPut:
|
||||||
|
body, _ := io.ReadAll(r.Body)
|
||||||
|
f.objects[key] = body
|
||||||
|
w.WriteHeader(http.StatusOK)
|
||||||
|
case r.Method == http.MethodGet && r.URL.Query().Get("list-type") == "2":
|
||||||
|
prefix := r.URL.Query().Get("prefix")
|
||||||
|
var out strings.Builder
|
||||||
|
out.WriteString(`<?xml version="1.0"?><ListBucketResult>`)
|
||||||
|
for k := range f.objects {
|
||||||
|
if strings.HasPrefix(k, prefix) {
|
||||||
|
fmt.Fprintf(&out, "<Contents><Key>%s</Key></Contents>", k)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
out.WriteString(`</ListBucketResult>`)
|
||||||
|
w.Header().Set("Content-Type", "application/xml")
|
||||||
|
w.Write([]byte(out.String()))
|
||||||
|
case r.Method == http.MethodGet:
|
||||||
|
body, ok := f.objects[key]
|
||||||
|
if !ok {
|
||||||
|
http.Error(w, "no such key", http.StatusNotFound)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
w.Write(body)
|
||||||
|
case r.Method == http.MethodDelete:
|
||||||
|
delete(f.objects, key)
|
||||||
|
w.WriteHeader(http.StatusNoContent)
|
||||||
|
default:
|
||||||
|
http.Error(w, "unsupported", http.StatusBadRequest)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func newTestS3(t *testing.T) (*S3Store, *fakeS3) {
|
||||||
|
t.Helper()
|
||||||
|
fake := newFakeS3("primora-objects")
|
||||||
|
srv := httptest.NewServer(fake)
|
||||||
|
t.Cleanup(srv.Close)
|
||||||
|
store, err := NewS3Store(S3Config{
|
||||||
|
Endpoint: srv.URL,
|
||||||
|
Region: "us-east-1",
|
||||||
|
Bucket: "primora-objects",
|
||||||
|
AccessKeyID: "test-key",
|
||||||
|
SecretAccessKey: "test-secret",
|
||||||
|
PathStyle: true,
|
||||||
|
})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("NewS3Store: %v", err)
|
||||||
|
}
|
||||||
|
return store, fake
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestS3StorePutGetDeleteRoundTrip(t *testing.T) {
|
||||||
|
store, fake := newTestS3(t)
|
||||||
|
ctx := context.Background()
|
||||||
|
|
||||||
|
res, err := store.Put(ctx, "bucket-a", "docs/hello.txt", strings.NewReader("hello s3"))
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("Put: %v", err)
|
||||||
|
}
|
||||||
|
if res.SizeBytes != 8 {
|
||||||
|
t.Fatalf("size = %d, want 8", res.SizeBytes)
|
||||||
|
}
|
||||||
|
if res.Path != "s3://primora-objects/bucket-a/docs/hello.txt" {
|
||||||
|
t.Fatalf("path = %q", res.Path)
|
||||||
|
}
|
||||||
|
// sha256("hello s3")
|
||||||
|
if res.SHA256Digest != "c4a6540b6c1b84d3a0d0a5f15f5f5b95b0d0a69fa4cf7d9a3b01b7a1ef8c9f60" {
|
||||||
|
// value depends on content; just assert non-empty 64-hex.
|
||||||
|
if len(res.SHA256Digest) != 64 {
|
||||||
|
t.Fatalf("digest = %q", res.SHA256Digest)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
rc, path, err := store.Open("bucket-a", "docs/hello.txt")
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("Open: %v", err)
|
||||||
|
}
|
||||||
|
body, _ := io.ReadAll(rc)
|
||||||
|
rc.Close()
|
||||||
|
if string(body) != "hello s3" {
|
||||||
|
t.Fatalf("got %q", body)
|
||||||
|
}
|
||||||
|
if !strings.HasPrefix(path, "s3://") {
|
||||||
|
t.Fatalf("path = %q", path)
|
||||||
|
}
|
||||||
|
|
||||||
|
if err := store.Delete("bucket-a", "docs/hello.txt"); err != nil {
|
||||||
|
t.Fatalf("Delete: %v", err)
|
||||||
|
}
|
||||||
|
if len(fake.objects) != 0 {
|
||||||
|
t.Fatalf("objects remain: %v", fake.objects)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestS3StoreMoveAndDeleteBucket(t *testing.T) {
|
||||||
|
store, fake := newTestS3(t)
|
||||||
|
ctx := context.Background()
|
||||||
|
|
||||||
|
if _, err := store.Put(ctx, "bucket-a", "a.txt", strings.NewReader("A")); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if _, err := store.Put(ctx, "bucket-a", "b.txt", strings.NewReader("B")); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
|
||||||
|
newPath, err := store.MoveBetweenBuckets("bucket-a", "bucket-b", "a.txt", "moved/a.txt")
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("Move: %v", err)
|
||||||
|
}
|
||||||
|
if !strings.Contains(newPath, "bucket-b/moved/a.txt") {
|
||||||
|
t.Fatalf("path = %q", newPath)
|
||||||
|
}
|
||||||
|
if _, ok := fake.objects["bucket-a/a.txt"]; ok {
|
||||||
|
t.Fatal("source not deleted after move")
|
||||||
|
}
|
||||||
|
if got := string(fake.objects["bucket-b/moved/a.txt"]); got != "A" {
|
||||||
|
t.Fatalf("moved content = %q", got)
|
||||||
|
}
|
||||||
|
|
||||||
|
if err := store.DeleteBucket("bucket-a"); err != nil {
|
||||||
|
t.Fatalf("DeleteBucket: %v", err)
|
||||||
|
}
|
||||||
|
for k := range fake.objects {
|
||||||
|
if strings.HasPrefix(k, "bucket-a/") {
|
||||||
|
t.Fatalf("key %s survived DeleteBucket", k)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
// bucket-b untouched
|
||||||
|
if _, ok := fake.objects["bucket-b/moved/a.txt"]; !ok {
|
||||||
|
t.Fatal("DeleteBucket removed objects from another bucket")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestS3StoreInvalidKeyRejected(t *testing.T) {
|
||||||
|
store, _ := newTestS3(t)
|
||||||
|
if _, err := store.Put(context.Background(), "b", "../escape", strings.NewReader("x")); err == nil {
|
||||||
|
t.Fatal("expected invalid key error")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestSigV4Signature pins the signing algorithm against a fixed clock: the
|
||||||
|
// Authorization header must be deterministic for identical inputs.
|
||||||
|
func TestSigV4SignatureDeterministic(t *testing.T) {
|
||||||
|
store, _ := newTestS3(t)
|
||||||
|
store.now = func() time.Time {
|
||||||
|
return time.Date(2026, 9, 24, 12, 0, 0, 0, time.UTC)
|
||||||
|
}
|
||||||
|
req, err := store.signedRequest(context.Background(), http.MethodGet, "bucket-a/x.txt", nil)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
auth := req.Header.Get("Authorization")
|
||||||
|
if !strings.HasPrefix(auth, "AWS4-HMAC-SHA256 Credential=test-key/20260924/us-east-1/s3/aws4_request") {
|
||||||
|
t.Fatalf("unexpected auth header: %s", auth)
|
||||||
|
}
|
||||||
|
if !strings.Contains(auth, "SignedHeaders=host;x-amz-content-sha256;x-amz-date") {
|
||||||
|
t.Fatalf("signed headers: %s", auth)
|
||||||
|
}
|
||||||
|
if req.Header.Get("x-amz-date") != "20260924T120000Z" {
|
||||||
|
t.Fatalf("amz date: %s", req.Header.Get("x-amz-date"))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestCanonicalQueryRFC3986(t *testing.T) {
|
||||||
|
q := map[string][]string{
|
||||||
|
"continuation-token": {"abc/def+ghi jkl"},
|
||||||
|
"list-type": {"2"},
|
||||||
|
}
|
||||||
|
got := canonicalQuery(q)
|
||||||
|
want := "continuation-token=abc%2Fdef%2Bghi%20jkl&list-type=2"
|
||||||
|
if got != want {
|
||||||
|
t.Fatalf("canonicalQuery = %q, want %q", got, want)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,16 @@
|
|||||||
|
package storage
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"io"
|
||||||
|
)
|
||||||
|
|
||||||
|
// Store is the object-storage contract the platform service depends on.
|
||||||
|
// LocalStore keeps objects on disk; S3Store targets S3-compatible APIs.
|
||||||
|
type Store interface {
|
||||||
|
Put(ctx context.Context, bucketID, objectKey string, reader io.Reader) (PutResult, error)
|
||||||
|
Open(bucketID, objectKey string) (io.ReadCloser, string, error)
|
||||||
|
Delete(bucketID, objectKey string) error
|
||||||
|
MoveBetweenBuckets(sourceBucketID, destinationBucketID, fromKey, toKey string) (string, error)
|
||||||
|
DeleteBucket(bucketID string) error
|
||||||
|
}
|
||||||
Reference in new issue
Block a user