diff --git a/.env.example b/.env.example index 9ac12dd..e514954 100644 --- a/.env.example +++ b/.env.example @@ -50,6 +50,17 @@ BACKEND_PUBLIC_URL=http://localhost/api/v1 BACKEND_INTERNAL_URL=http://backend:8080 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 # values win over these env vars; deleting the in-app value reverts here. RESEND_API_KEY= diff --git a/CHANGELOG.md b/CHANGELOG.md index cf2cc01..46184bc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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 --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. - **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 diff --git a/DEPLOYMENT_GUIDE.md b/DEPLOYMENT_GUIDE.md index 5dc1e93..6132c43 100644 --- a/DEPLOYMENT_GUIDE.md +++ b/DEPLOYMENT_GUIDE.md @@ -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_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 | -| `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..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 | | `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 | diff --git a/ROADMAP.md b/ROADMAP.md index f402f2d..1f1a698 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -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. Deno or Bun isolates in a sidecar; secrets via env, payload via stdin. The job runner already supplies the scheduling half. -- **Phase 10 — Storage backends** — S3-compatible object storage behind - the existing bucket abstraction (deferred from Phase 6), plus - presigned-URL uploads for large files. -- **HA mode** — leader-elected scheduler + distributed locks so two - backend replicas don't double-fire jobs; Postgres advisory locks are - enough at this scale. -- **Email surface** — transactional templates + send log in the - dashboard, backed by the auth service's existing SMTP/Resend wiring. +- **Phase 10 — Storage backends** — ✅ S3-compatible object storage shipped: + `BACKEND_STORAGE_DRIVER=s3` + `S3_*` env vars, stdlib SigV4, verified + against a real S3 server. Presigned-URL uploads for large files remain a + candidate. +- **HA mode** — ✅ shipped: PostgreSQL advisory-lock leadership for the + scheduler and alert evaluator; failover verified with two replicas. +- **Email surface** — ✅ shipped: templated sends + `core.email_log` + + dashboard card; more templates land as new flows need them. --- diff --git a/apps/backend/internal/app/app.go b/apps/backend/internal/app/app.go index f118103..9680c83 100644 --- a/apps/backend/internal/app/app.go +++ b/apps/backend/internal/app/app.go @@ -81,9 +81,30 @@ func Bootstrap(ctx context.Context) (*App, error) { } } - store, err := storage.NewLocalStore(cfg.StorageRoot) - if err != nil { - return nil, err + 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 { + return nil, err + } + store = localStore } dbxClient := dbx.NewClient(logger, cfg.DBXDataDir) repo := repositories.NewCoreRepository(dbPool) diff --git a/apps/backend/internal/config/config.go b/apps/backend/internal/config/config.go index 972f26c..f699364 100644 --- a/apps/backend/internal/config/config.go +++ b/apps/backend/internal/config/config.go @@ -35,6 +35,16 @@ type Config struct { SMTPSecure bool 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. @@ -63,6 +73,13 @@ func Load() (Config, error) { SMTPUser: os.Getenv("SMTP_USER"), SMTPPassword: os.Getenv("SMTP_PASSWORD"), 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")) @@ -95,6 +112,12 @@ func Load() (Config, error) { } 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 if cfg.DatabaseURL == "" { missing = append(missing, "DATABASE_URL") @@ -102,8 +125,23 @@ func Load() (Config, error) { if cfg.JWTSecret == "" { missing = append(missing, "JWT_SECRET") } - if cfg.StorageRoot == "" { - missing = append(missing, "BACKEND_STORAGE_ROOT") + switch cfg.StorageDriver { + case "local": + if cfg.StorageRoot == "" { + 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 == "" { missing = append(missing, "AUTH_INTERNAL_BASE_URL") diff --git a/apps/backend/internal/services/platform_service.go b/apps/backend/internal/services/platform_service.go index 4996b9b..e693c58 100644 --- a/apps/backend/internal/services/platform_service.go +++ b/apps/backend/internal/services/platform_service.go @@ -29,7 +29,7 @@ import ( type PlatformService struct { repo *repositories.CoreRepository - store *storage.LocalStore + store storage.Store mailer *Mailer publicURL string dbx *dbx.Client @@ -224,7 +224,7 @@ type InvitationSummary struct { 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} if enc != nil { s.dispatcher = NewWebhookDispatcher(repo.Queries(), enc, logger) diff --git a/apps/backend/internal/storage/local.go b/apps/backend/internal/storage/local.go index 249cfe1..7a4a9d9 100644 --- a/apps/backend/internal/storage/local.go +++ b/apps/backend/internal/storage/local.go @@ -65,7 +65,7 @@ func (s *LocalStore) Put(ctx context.Context, bucketID, objectKey string, reader }, 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) if err != nil { return nil, "", err diff --git a/apps/backend/internal/storage/s3.go b/apps/backend/internal/storage/s3.go new file mode 100644 index 0000000..7081b17 --- /dev/null +++ b/apps/backend/internal/storage/s3.go @@ -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)) +} diff --git a/apps/backend/internal/storage/s3_test.go b/apps/backend/internal/storage/s3_test.go new file mode 100644 index 0000000..f47df02 --- /dev/null +++ b/apps/backend/internal/storage/s3_test.go @@ -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(``) + for k := range f.objects { + if strings.HasPrefix(k, prefix) { + fmt.Fprintf(&out, "%s", k) + } + } + out.WriteString(``) + 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) + } +} diff --git a/apps/backend/internal/storage/store.go b/apps/backend/internal/storage/store.go new file mode 100644 index 0000000..e70e504 --- /dev/null +++ b/apps/backend/internal/storage/store.go @@ -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 +}