diff --git a/.github/workflows/helm-install-test.yml b/.github/workflows/helm-install-test.yml index ee5b717..aff42c3 100644 --- a/.github/workflows/helm-install-test.yml +++ b/.github/workflows/helm-install-test.yml @@ -147,14 +147,15 @@ jobs: --set image.pullPolicy=Always \ --set "imagePullSecrets[0].name=ghcr-login" \ --set s3.host="http://minio:9000" \ - --set secrets.encryptKey=test-encryption-key-for-ci \ - --set secrets.awsAccessKeyId=minioadmin \ - --set secrets.awsSecretAccessKey=minioadmin \ + --set secrets.credentials[0].accessKey=minioadmin \ + --set secrets.credentials[0].secretKey=minioadmin \ + --set secrets.credentials[0].kek=test-encryption-key-for-ci \ --set redis-ha.enabled=false \ --set externalRedis.url="redis://redis:6379/0" \ --set externalRedis.existingSecret=redis-secret \ --set externalRedis.passwordKey=redis-password \ --set admin.enabled=true \ + --set admin.secret=test-admin-secret-for-ci \ --set replicaCount=3 \ --set resources.limits.cpu=100m \ --set resources.requests.cpu=50m \ diff --git a/README.md b/README.md index 9131063..5d13d25 100644 --- a/README.md +++ b/README.md @@ -45,22 +45,22 @@ S3's server-side encryption is great, but your cloud provider holds the keys. S3 ## Install +Each AWS credential is configured with its **own encryption key** (KEK). The proxy verifies the client's signature with the credential's secret key, then encrypts/decrypts that credential's objects with its KEK. + **Option A** — inline secrets (quick start): ```bash helm install s3proxy oci://ghcr.io/serversidehannes/s3proxy-python/charts/s3proxy-python \ - --set secrets.encryptKey="your-32-byte-key" \ - --set secrets.awsAccessKeyId="AKIA..." \ - --set secrets.awsSecretAccessKey="wJalr..." + --set secrets.credentials[0].accessKey="AKIA..." \ + --set secrets.credentials[0].secretKey="wJalr..." \ + --set secrets.credentials[0].kek="this-credentials-encryption-secret" ``` **Option B** — existing K8s secret (recommended for production): ```bash kubectl create secret generic s3proxy-secrets \ - --from-literal=S3PROXY_ENCRYPT_KEY="your-32-byte-key" \ - --from-literal=AWS_ACCESS_KEY_ID="AKIA..." \ - --from-literal=AWS_SECRET_ACCESS_KEY="wJalr..." + --from-literal=S3PROXY_CREDENTIALS='[{"access_key":"AKIA...","secret_key":"wJalr...","kek":"this-credentials-encryption-secret"}]' helm install s3proxy oci://ghcr.io/serversidehannes/s3proxy-python/charts/s3proxy-python \ --set secrets.existingSecrets.enabled=true \ @@ -106,6 +106,18 @@ Master Key → KEK (derived via SHA-256) └→ encrypts data (AES-256-GCM) ``` +**Per-credential keys** — Each AWS credential has its own KEK. The proxy verifies the client's signature with the credential's secret key, then wraps that credential's DEKs with the credential's KEK. So a leaked KEK only exposes the data written by that one credential. The access key that wrapped each object is recorded in the object's metadata (`isec-kid`), so **decryption always uses the key that actually encrypted the object** — reconfiguring credentials never orphans existing data, as long as that access key's KEK is still present. + +```bash +# Each credential: access_key + secret_key + its own kek (SHA-256'd into the KEK) +S3PROXY_CREDENTIALS='[ + {"access_key":"AKIA-ACME","secret_key":"...","kek":"acme-kek-secret"}, + {"access_key":"AKIA-GLOBEX","secret_key":"...","kek":"globex-kek-secret"} +]' +``` + +A request signed by an access key with no configured KEK is rejected. Via Helm: set `secrets.credentials` (see [chart/values.yaml](chart/values.yaml)). + --- ## Configuration @@ -115,8 +127,9 @@ Master Key → KEK (derived via SHA-256) | `replicaCount` | `3` | Pod replicas | | `s3.host` | `s3.amazonaws.com` | S3 endpoint (AWS, MinIO, R2, etc.) | | `s3.region` | `us-east-1` | AWS region | -| `secrets.encryptKey` | — | Encryption key | +| `secrets.credentials` | `[]` | AWS credentials, each `{accessKey, secretKey, kek}` | | `secrets.existingSecrets.enabled` | `false` | Use existing K8s secret | +| `admin.secret` | `change-me` | Secret signing admin session cookies (when admin UI on) | | `redis-ha.enabled` | `true` | Deploy embedded Redis HA | | `gateway.enabled` | `false` | Create gateway service | | `ingress.enabled` | `false` | Enable ingress | @@ -134,8 +147,8 @@ Yes. S3Proxy detects unencrypted objects and returns them as-is. Migrate by copy
-What if I lose my encryption key? -Data is unrecoverable. Back up your key. +What if I lose an encryption key? +Data written by that credential is unrecoverable. Each object records the access key that encrypted it, so keep every credential's kek as long as objects written by that credential exist. Back up your keys.
@@ -158,8 +171,8 @@ Yes. The proxy verifies the presigned signature, then makes its own authenticate ## Roadmap - [ ] Key rotation (re-encrypt objects with a new master key) -- [ ] Multiple AWS credential pairs (per-client auth) -- [ ] Per-bucket / per-prefix encryption keys +- [x] Multiple AWS credential pairs (per-client auth) +- [x] Per-credential encryption keys - [ ] S3 Select passthrough - [ ] Ceph S3 compatibility > 80% - [ ] Batch re-encryption CLI tool diff --git a/chart/README.md b/chart/README.md index 2a2ac14..1c78c4f 100644 --- a/chart/README.md +++ b/chart/README.md @@ -4,9 +4,9 @@ ```bash helm install s3proxy oci://ghcr.io/serversidehannes/s3proxy-python/charts/s3proxy-python \ - --set secrets.encryptKey="your-key" \ - --set secrets.awsAccessKeyId="AKIA..." \ - --set secrets.awsSecretAccessKey="wJalr..." + --set secrets.credentials[0].accessKey="AKIA..." \ + --set secrets.credentials[0].secretKey="wJalr..." \ + --set secrets.credentials[0].kek="this-credentials-encryption-secret" ``` ## Values @@ -23,21 +23,19 @@ helm install s3proxy oci://ghcr.io/serversidehannes/s3proxy-python/charts/s3prox | `server.noTls` | `true` | Disable TLS (in-cluster only) | | `performance.memoryLimitMb` | `64` | Memory budget for streaming | | `logLevel` | `DEBUG` | Log level | -| `secrets.encryptKey` | `""` | AES-256 encryption key | -| `secrets.awsAccessKeyId` | `""` | AWS access key | -| `secrets.awsSecretAccessKey` | `""` | AWS secret key | +| `secrets.credentials` | `[]` | AWS credentials, each `{accessKey, secretKey, kek}` — the credential's KEK encrypts its objects | | `secrets.existingSecrets.enabled` | `false` | Use pre-created K8s secret | | `secrets.existingSecrets.name` | `""` | Existing secret name | -| `secrets.existingSecrets.keys.encryptKey` | `S3PROXY_ENCRYPT_KEY` | Key name in existing secret | -| `secrets.existingSecrets.keys.awsAccessKeyId` | `AWS_ACCESS_KEY_ID` | Key name in existing secret | -| `secrets.existingSecrets.keys.awsSecretAccessKey` | `AWS_SECRET_ACCESS_KEY` | Key name in existing secret | +| `secrets.existingSecrets.keys.credentials` | `S3PROXY_CREDENTIALS` | Credentials key name in existing secret | | `admin.enabled` | `false` | Enable the admin dashboard | | `admin.path` | `/admin` | URL path prefix for the dashboard | | `admin.username` | `admin` | Dashboard username (stored in the Secret; override in production) | | `admin.password` | `admin` | Dashboard password (stored in the Secret; override in production) | +| `admin.secret` | `change-me` | Secret signing dashboard session cookies (override in production) | | `admin.existingSecret.name` | `""` | Pre-created secret holding admin credentials | | `admin.existingSecret.usernameKey` | `S3PROXY_ADMIN_USERNAME` | Username key in the existing secret | | `admin.existingSecret.passwordKey` | `S3PROXY_ADMIN_PASSWORD` | Password key in the existing secret | +| `admin.existingSecret.secretKey` | `S3PROXY_ADMIN_SECRET` | Session-secret key in the existing secret | | `admin.ingress.enabled` | `false` | Dedicated Ingress for the dashboard (keep off unless intentionally exposing it) | | `admin.ingress.className` | `nginx` | Ingress class for the admin Ingress | | `admin.ingress.host` | `""` | Hostname for the dashboard (required when enabled) | diff --git a/chart/templates/deployment.yaml b/chart/templates/deployment.yaml index c229a8f..7a381ad 100644 --- a/chart/templates/deployment.yaml +++ b/chart/templates/deployment.yaml @@ -40,21 +40,11 @@ spec: env: {{- /* App secrets from existing secret */ -}} {{- if .Values.secrets.existingSecrets.enabled }} - - name: S3PROXY_ENCRYPT_KEY + - name: S3PROXY_CREDENTIALS valueFrom: secretKeyRef: name: {{ .Values.secrets.existingSecrets.name }} - key: {{ .Values.secrets.existingSecrets.keys.encryptKey }} - - name: AWS_ACCESS_KEY_ID - valueFrom: - secretKeyRef: - name: {{ .Values.secrets.existingSecrets.name }} - key: {{ .Values.secrets.existingSecrets.keys.awsAccessKeyId }} - - name: AWS_SECRET_ACCESS_KEY - valueFrom: - secretKeyRef: - name: {{ .Values.secrets.existingSecrets.name }} - key: {{ .Values.secrets.existingSecrets.keys.awsSecretAccessKey }} + key: {{ .Values.secrets.existingSecrets.keys.credentials }} {{- end }} {{- /* Admin dashboard credentials (when not pulled via the chart-secret envFrom) */ -}} {{- if $adminNeedsEnv }} @@ -69,6 +59,11 @@ spec: secretKeyRef: name: {{ $adminSecret }} key: {{ .Values.admin.existingSecret.passwordKey }} + - name: S3PROXY_ADMIN_SECRET + valueFrom: + secretKeyRef: + name: {{ $adminSecret }} + key: {{ .Values.admin.existingSecret.secretKey }} {{- end }} {{- /* Redis password from redis-ha secret */ -}} {{- if and (index .Values "redis-ha" "enabled") (index .Values "redis-ha" "auth") }} diff --git a/chart/templates/secret.yaml b/chart/templates/secret.yaml index 6ee7822..bf9f362 100644 --- a/chart/templates/secret.yaml +++ b/chart/templates/secret.yaml @@ -2,8 +2,8 @@ {{- $createAdmin := and .Values.admin.enabled (not .Values.admin.existingSecret.name) -}} {{- if or $createApp $createAdmin }} # Creating static secrets (provide via helm --set or secure values file, not hardcoded) -# Secret keys created: S3PROXY_ENCRYPT_KEY, AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, -# and (when the admin dashboard is enabled) S3PROXY_ADMIN_USERNAME / S3PROXY_ADMIN_PASSWORD +# Secret keys created: S3PROXY_CREDENTIALS, and (when the admin dashboard is enabled) +# S3PROXY_ADMIN_USERNAME / S3PROXY_ADMIN_PASSWORD / S3PROXY_ADMIN_SECRET apiVersion: v1 kind: Secret metadata: @@ -13,12 +13,15 @@ metadata: type: Opaque stringData: {{- if $createApp }} - S3PROXY_ENCRYPT_KEY: {{ .Values.secrets.encryptKey | quote }} - AWS_ACCESS_KEY_ID: {{ .Values.secrets.awsAccessKeyId | quote }} - AWS_SECRET_ACCESS_KEY: {{ .Values.secrets.awsSecretAccessKey | quote }} + {{- $creds := list }} + {{- range .Values.secrets.credentials }} + {{- $creds = append $creds (dict "access_key" .accessKey "secret_key" .secretKey "kek" .kek) }} + {{- end }} + S3PROXY_CREDENTIALS: {{ toJson $creds | quote }} {{- end }} {{- if $createAdmin }} S3PROXY_ADMIN_USERNAME: {{ .Values.admin.username | quote }} S3PROXY_ADMIN_PASSWORD: {{ .Values.admin.password | quote }} + S3PROXY_ADMIN_SECRET: {{ .Values.admin.secret | quote }} {{- end }} {{- end }} diff --git a/chart/values.yaml b/chart/values.yaml index 167ddbf..7f7be6e 100644 --- a/chart/values.yaml +++ b/chart/values.yaml @@ -87,13 +87,16 @@ secrets: enabled: false name: "" keys: - encryptKey: "S3PROXY_ENCRYPT_KEY" - awsAccessKeyId: "AWS_ACCESS_KEY_ID" - awsSecretAccessKey: "AWS_SECRET_ACCESS_KEY" + credentials: "S3PROXY_CREDENTIALS" - encryptKey: "" - awsAccessKeyId: "" - awsSecretAccessKey: "" + # AWS credentials, each with its own encryption key (KEK). The proxy verifies + # a client's signature with secretKey, then encrypts/decrypts that credential's + # objects with the KEK derived from kek. Rendered into the S3PROXY_CREDENTIALS + # secret as a JSON array. At least one entry is required. + credentials: [] + # - accessKey: "AKIAEXAMPLE" + # secretKey: "wJalr...EXAMPLEKEY" + # kek: "a-long-random-kek-secret" logLevel: "DEBUG" @@ -105,10 +108,14 @@ admin: path: "/admin" username: "admin" password: "admin" + # Stable secret used to sign admin session cookies. Required when enabled. + # Override in production; keep it stable so sessions survive pod restarts. + secret: "change-me" existingSecret: name: "" usernameKey: "S3PROXY_ADMIN_USERNAME" passwordKey: "S3PROXY_ADMIN_PASSWORD" + secretKey: "S3PROXY_ADMIN_SECRET" # Optional dedicated Ingress for the dashboard, separate from the S3 data-plane # ingress above. OFF by default — the admin UI should not be internet-facing diff --git a/e2e/docker-compose.yml b/e2e/docker-compose.yml index d34906d..deb95f1 100644 --- a/e2e/docker-compose.yml +++ b/e2e/docker-compose.yml @@ -221,7 +221,7 @@ services: echo "ECK operator apply failed, retry $$i..." sleep 5 done - kubectl wait --namespace elastic-system --for=condition=ready pod --selector=control-plane=elastic-operator --timeout=300s + kubectl rollout status --namespace elastic-system statefulset/elastic-operator --timeout=300s echo "✓ ECK operator ready" ) & ECK_PID=$$! @@ -413,9 +413,10 @@ services: --set image.repository=s3proxy \ --set image.pullPolicy=IfNotPresent \ --set s3.host="http://minio.minio.svc.cluster.local:9000" \ - --set secrets.encryptKey="$$ENCRYPT_KEY" \ - --set secrets.awsAccessKeyId="minioadmin" \ - --set secrets.awsSecretAccessKey="minioadmin" \ + --set secrets.credentials[0].accessKey="minioadmin" \ + --set secrets.credentials[0].secretKey="minioadmin" \ + --set secrets.credentials[0].kek="$$ENCRYPT_KEY" \ + --set admin.secret="$$ENCRYPT_KEY" \ --set logLevel="DEBUG" \ --set performance.memoryLimitMb=64 \ --set gateway.enabled=true \ diff --git a/s3proxy/admin/auth.py b/s3proxy/admin/auth.py index 0b2bb55..eb54f48 100644 --- a/s3proxy/admin/auth.py +++ b/s3proxy/admin/auth.py @@ -33,8 +33,8 @@ def __init__(self, settings: Settings, credentials_store: dict[str, str]): ) self.username = settings.admin_username self.password = settings.admin_password - # Derive a session-signing secret from the KEK so cookies survive pod restarts. - self.session_secret = hashlib.sha256(b"s3proxy-admin-session|" + settings.kek).digest() + # Stable session-signing secret (survives pod restarts, shared across replicas). + self.session_secret = settings.admin_session_secret def valid(self, username: str, password: str) -> bool: return secrets.compare_digest(username.encode(), self.username.encode()) and ( diff --git a/s3proxy/admin/collectors.py b/s3proxy/admin/collectors.py index d497a5a..513b9eb 100644 --- a/s3proxy/admin/collectors.py +++ b/s3proxy/admin/collectors.py @@ -338,15 +338,20 @@ def _derive_buckets(entries: list[RequestEntry]) -> list[dict]: def _derive_keys(settings: Settings) -> list[dict]: - fp = hashlib.sha256(settings.kek).hexdigest()[:8] - return [ - { - "id": f"key-{fp}", - "type": "Local (KEK)", - "status": "Active", - "created": "—", - } - ] + # One KEK per configured AWS login (access key). The kek secret itself is + # never exposed - only a short fingerprint for identification. + keys = [] + for entry in settings.credentials: + fp = hashlib.sha256(entry.kek.encode()).hexdigest()[:8] + keys.append( + { + "id": entry.access_key, + "type": f"Local (KEK · {fp})", + "status": "Active", + "created": "—", + } + ) + return keys # --------------------------------------------------------------------------- diff --git a/s3proxy/app.py b/s3proxy/app.py index 7ee19bf..67b12cb 100644 --- a/s3proxy/app.py +++ b/s3proxy/app.py @@ -41,16 +41,6 @@ logger: BoundLogger = structlog.get_logger(__name__).bind(pod=pod_name) -def load_credentials() -> dict[str, str]: - """Load AWS credentials from environment variables.""" - credentials_store: dict[str, str] = {} - access_key = os.environ.get("AWS_ACCESS_KEY_ID", "") - secret_key = os.environ.get("AWS_SECRET_ACCESS_KEY", "") - if access_key and secret_key: - credentials_store[access_key] = secret_key - return credentials_store - - def create_lifespan(settings: Settings, credentials_store: dict[str, str]) -> AsyncIterator[None]: """Create lifespan context manager for FastAPI app. @@ -103,7 +93,7 @@ def create_app(settings: Settings | None = None) -> FastAPI: Configured FastAPI application instance. """ settings = settings or Settings() - credentials_store = load_credentials() + credentials_store = settings.credentials_store lifespan = create_lifespan(settings, credentials_store) app = FastAPI(title="S3Proxy", lifespan=lifespan, docs_url=None, redoc_url=None) diff --git a/s3proxy/config.py b/s3proxy/config.py index a679567..98a8e8f 100644 --- a/s3proxy/config.py +++ b/s3proxy/config.py @@ -2,9 +2,23 @@ import hashlib -from pydantic import Field, PrivateAttr +from pydantic import BaseModel, Field, PrivateAttr from pydantic_settings import BaseSettings, SettingsConfigDict +from .keyring import KeyRing, derive_kek + + +class CredentialEntry(BaseModel): + """An AWS credential with its own encryption key, via S3PROXY_CREDENTIALS. + + The proxy verifies the client's SigV4 signature with `secret_key`, then + encrypts/decrypts that credential's objects with the KEK derived from `kek`. + """ + + access_key: str + secret_key: str + kek: str = Field(..., description="Per-credential KEK secret - SHA256 hashed into a KEK") + class Settings(BaseSettings): """S3Proxy configuration settings.""" @@ -16,8 +30,19 @@ class Settings(BaseSettings): region: str = Field(default="us-east-1", description="AWS region") # Encryption settings - encrypt_key: str = Field(..., description="Key Encryption Key (KEK) - will be SHA256 hashed") dektag_name: str = Field(default="isec", description="Metadata tag name for encrypted DEK") + kidtag_name: str = Field( + default="isec-kid", description="Metadata tag name for the key id that wrapped the DEK" + ) + + # Per-access-key encryption. Each entry is an AWS credential with its own + # KEK. The access key that wrote an object is stored as its kid; objects are + # decrypted with the KEK of that access key. Replaces the single + # AWS_ACCESS_KEY_ID / AWS_SECRET_ACCESS_KEY credential. + credentials: list[CredentialEntry] = Field( + default_factory=list, + description="AWS credentials with per-credential KEKs (JSON list)", + ) # Server settings ip: str = Field(default="0.0.0.0", description="Bind address") @@ -54,17 +79,42 @@ class Settings(BaseSettings): admin_path: str = Field(default="/admin", description="URL path prefix for the admin UI") admin_username: str = Field(default="", description="Admin dashboard username") admin_password: str = Field(default="", description="Admin dashboard password") + admin_secret: str = Field( + default="", + description="Stable secret for signing admin session cookies (required when admin_ui)", + ) - # Cached KEK derived from encrypt_key (computed once in model_post_init) - _kek: bytes = PrivateAttr() + # Cached KeyRing + credentials store (computed once in model_post_init). + _keyring: KeyRing = PrivateAttr() + _credentials_store: dict[str, str] = PrivateAttr() def model_post_init(self, __context: object) -> None: - self._kek = hashlib.sha256(self.encrypt_key.encode()).digest() + self._credentials_store = {} + keys: dict[str, bytes] = {} + for entry in self.credentials: + if entry.access_key in self._credentials_store: + raise ValueError(f"Duplicate access key: {entry.access_key!r}") + self._credentials_store[entry.access_key] = entry.secret_key + keys[entry.access_key] = derive_kek(entry.kek) + self._keyring = KeyRing(keys=keys) + + if self.admin_ui and not self.admin_secret: + raise ValueError("S3PROXY_ADMIN_SECRET is required when the admin dashboard is enabled") + + @property + def admin_session_secret(self) -> bytes: + """Stable 32-byte secret for signing admin session cookies.""" + return hashlib.sha256(b"s3proxy-admin-session|" + self.admin_secret.encode()).digest() + + @property + def keyring(self) -> KeyRing: + """Get the KeyRing resolving per-access-key encryption keys.""" + return self._keyring @property - def kek(self) -> bytes: - """Get the 32-byte Key Encryption Key (SHA256 of encrypt_key).""" - return self._kek + def credentials_store(self) -> dict[str, str]: + """Get the access_key -> secret_key map for signature verification.""" + return self._credentials_store @property def s3_endpoint(self) -> str: diff --git a/s3proxy/handlers/base.py b/s3proxy/handlers/base.py index 8a838ac..07d7ce6 100644 --- a/s3proxy/handlers/base.py +++ b/s3proxy/handlers/base.py @@ -70,6 +70,7 @@ def __init__( self.settings = settings self.credentials_store = credentials_store self.multipart_manager = multipart_manager + self.keyring = settings.keyring def _client(self, creds: S3Credentials) -> S3Client: return S3Client(self.settings, creds) @@ -216,13 +217,13 @@ def _check_conditional_headers( return None async def _download_encrypted_single( - self, client: S3Client, bucket: str, key: str, wrapped_dek_b64: str + self, client: S3Client, bucket: str, key: str, wrapped_dek_b64: str, kid: str = "" ) -> bytes: resp = await client.get_object(bucket, key) async with resp["Body"] as body: ciphertext = await body.read() wrapped_dek = base64.b64decode(wrapped_dek_b64) - return crypto.decrypt_object(ciphertext, wrapped_dek, self.settings.kek) + return crypto.decrypt_object(ciphertext, wrapped_dek, self.keyring.key_by_id(kid)) async def _download_encrypted_multipart( self, @@ -234,7 +235,7 @@ async def _download_encrypted_multipart( range_end: int | None = None, ) -> bytes: """Download and decrypt multipart encrypted object, optionally with range.""" - dek = crypto.unwrap_key(meta.wrapped_dek, self.settings.kek) + dek = crypto.unwrap_key(meta.wrapped_dek, self.keyring.key_by_id(meta.kid)) sorted_parts = sorted(meta.parts, key=lambda p: p.part_number) plaintext_chunks = [] diff --git a/s3proxy/handlers/multipart/copy.py b/s3proxy/handlers/multipart/copy.py index c657127..26b12ab 100644 --- a/s3proxy/handlers/multipart/copy.py +++ b/s3proxy/handlers/multipart/copy.py @@ -35,10 +35,12 @@ async def handle_upload_part_copy(self, request: Request, creds: S3Credentials) # Get upload state state = await self.multipart_manager.get_upload(bucket, key, upload_id) if not state: - dek = await load_upload_state(client, bucket, key, upload_id, self.settings.kek) - if not dek: + state_data = await load_upload_state(client, bucket, key, upload_id) + if not state_data: raise S3Error.no_such_upload(upload_id) - state = await self.multipart_manager.create_upload(bucket, key, upload_id, dek) + wrapped_dek, kid = state_data + dek = crypto.unwrap_key(wrapped_dek, self.keyring.key_by_id(kid)) + state = await self.multipart_manager.create_upload(bucket, key, upload_id, dek, kid) # Get source data plaintext = await self._get_copy_source_data( @@ -93,9 +95,11 @@ async def _get_copy_source_data( client, src_bucket, src_key, src_multipart_meta, range_start, range_end ) else: - # Single-part encrypted - use shared helper + # Single-part encrypted - use shared helper. Decrypt with the kid + # the source object was encrypted under, not the current rules. + src_kid = src_metadata.get(self.settings.kidtag_name, "") full_plaintext = await self._download_encrypted_single( - client, src_bucket, src_key, src_wrapped_dek + client, src_bucket, src_key, src_wrapped_dek, src_kid ) if copy_source_range: range_start, range_end = self._parse_copy_source_range( diff --git a/s3proxy/handlers/multipart/lifecycle.py b/s3proxy/handlers/multipart/lifecycle.py index 7b8505d..f385f4c 100644 --- a/s3proxy/handlers/multipart/lifecycle.py +++ b/s3proxy/handlers/multipart/lifecycle.py @@ -47,9 +47,7 @@ async def _recover_upload_state( context=context, ) - state = await reconstruct_upload_state_from_s3( - client, bucket, key, upload_id, self.settings.kek - ) + state = await reconstruct_upload_state_from_s3(client, bucket, key, upload_id, self.keyring) if not state: raise S3Error.no_such_upload(upload_id) @@ -75,12 +73,14 @@ async def handle_create_multipart_upload( cache_control = request.headers.get("cache-control") expires = request.headers.get("expires") + kid, kek = self.keyring.key_for(client.credentials.access_key) dek = crypto.generate_dek() - wrapped_dek = crypto.wrap_key(dek, self.settings.kek) + wrapped_dek = crypto.wrap_key(dek, kek) # Build metadata (include user's x-amz-meta-*) upload_metadata = { self.settings.dektag_name: base64.b64encode(wrapped_dek).decode(), + self.settings.kidtag_name: kid, } for hdr, val in request.headers.items(): if hdr.lower().startswith("x-amz-meta-"): @@ -98,12 +98,12 @@ async def handle_create_multipart_upload( upload_id = resp["UploadId"] # Store state in Redis/memory first, then persist to S3 as backup - await self.multipart_manager.create_upload(bucket, key, upload_id, dek) + await self.multipart_manager.create_upload(bucket, key, upload_id, dek, kid) # Persist DEK to S3 as backup - retry once on failure for attempt in range(2): try: - await persist_upload_state(client, bucket, key, upload_id, wrapped_dek) + await persist_upload_state(client, bucket, key, upload_id, wrapped_dek, kid) break except Exception as e: if attempt == 0: @@ -177,17 +177,25 @@ async def handle_complete_multipart_upload( # Order matters: if metadata save fails, state is preserved # so the upload can be retried. Deleting state first would # lose the DEK, making the object permanently undecryptable. - wrapped_dek = crypto.wrap_key(state.dek, self.settings.kek) + # Prefer the kid recorded when the upload was created; if the state + # predates it (e.g. older recovered state), fall back to the + # completing credential's key. + if state.kid: + kid, kek = state.kid, self.keyring.key_by_id(state.kid) + else: + kid, kek = self.keyring.key_for(creds.access_key) + wrapped_dek = crypto.wrap_key(state.dek, kek) await save_multipart_metadata( client, bucket, key, MultipartMetadata( - version=1, + version=2, part_count=len(completed_parts), total_plaintext_size=total_plaintext, parts=completed_parts, wrapped_dek=wrapped_dek, + kid=kid, ), ) await delete_upload_state(client, bucket, key, upload_id) @@ -223,8 +231,8 @@ def internal_to_client_part(internal_part_number: int) -> int: """Convert internal part number to client part number.""" return ((internal_part_number - 1) // MAX_INTERNAL_PARTS_PER_CLIENT) + 1 - dek = await load_upload_state(client, bucket, key, upload_id, self.settings.kek) - if not dek: + state_data = await load_upload_state(client, bucket, key, upload_id) + if not state_data: # Check if upload exists in S3 before returning NoSuchUpload try: await client.list_parts(bucket, key, upload_id, max_parts=1) @@ -241,7 +249,9 @@ def internal_to_client_part(internal_part_number: int) -> int: pass raise S3Error.no_such_upload(upload_id) - state = await self.multipart_manager.create_upload(bucket, key, upload_id, dek) + wrapped_dek, kid = state_data + dek = crypto.unwrap_key(wrapped_dek, self.keyring.key_by_id(kid)) + state = await self.multipart_manager.create_upload(bucket, key, upload_id, dek, kid) try: parts_resp = await client.list_parts(bucket, key, upload_id) diff --git a/s3proxy/handlers/objects/get.py b/s3proxy/handlers/objects/get.py index afe71f3..b8b5030 100644 --- a/s3proxy/handlers/objects/get.py +++ b/s3proxy/handlers/objects/get.py @@ -77,6 +77,7 @@ async def handle_get_object(self, request: Request, creds: S3Credentials) -> Res # Add user metadata (x-amz-meta-*), excluding internal keys internal_keys = { self.settings.dektag_name.lower(), + self.settings.kidtag_name.lower(), "client-etag", "plaintext-size", } @@ -106,9 +107,10 @@ async def _get_single( client, bucket, key, range_header, head_resp, last_modified ) - # Encrypted single-object - decrypt in memory + # Encrypted single-object - decrypt in memory using the kid that wrapped it + kid = metadata.get(self.settings.kidtag_name, "") return await self._decrypt_single_object( - client, bucket, key, range_header, head_resp, last_modified, wrapped_dek_b64 + client, bucket, key, range_header, head_resp, last_modified, wrapped_dek_b64, kid ) async def _stream_unencrypted( @@ -145,6 +147,7 @@ async def _decrypt_single_object( head_resp: dict, last_modified: str | None, wrapped_dek_b64: str, + kid: str = "", ) -> Response: logger.info("GET_ENCRYPTED_SINGLE", bucket=bucket, key=key) resp = await client.get_object(bucket, key) @@ -161,7 +164,7 @@ async def _decrypt_single_object( wrapped_dek = base64.b64decode(wrapped_dek_b64) async with resp["Body"] as body: ciphertext = await body.read() - plaintext = crypto.decrypt_object(ciphertext, wrapped_dek, self.settings.kek) + plaintext = crypto.decrypt_object(ciphertext, wrapped_dek, self.keyring.key_by_id(kid)) del ciphertext content_type = head_resp.get("ContentType", "application/octet-stream") @@ -204,7 +207,7 @@ async def _get_multipart( last_modified: str | None, creds: S3Credentials, ) -> Response: - dek = crypto.unwrap_key(meta.wrapped_dek, self.settings.kek) + dek = crypto.unwrap_key(meta.wrapped_dek, self.keyring.key_by_id(meta.kid)) total = meta.total_plaintext_size start, end = self._parse_range(range_header, total) if range_header else (0, total - 1) parts = calculate_part_range(meta.parts, start, end) diff --git a/s3proxy/handlers/objects/misc.py b/s3proxy/handlers/objects/misc.py index c008b08..15a4950 100644 --- a/s3proxy/handlers/objects/misc.py +++ b/s3proxy/handlers/objects/misc.py @@ -98,7 +98,12 @@ def _build_head_extra_headers(self, resp: dict, last_modified: str | None) -> di extra["x-amz-tagging-count"] = str(resp["TagCount"]) # Include user metadata (x-amz-meta-*) excluding internal s3proxy keys metadata = resp.get("Metadata", {}) - internal_keys = {self.settings.dektag_name.lower(), "client-etag", "plaintext-size"} + internal_keys = { + self.settings.dektag_name.lower(), + self.settings.kidtag_name.lower(), + "client-etag", + "plaintext-size", + } for key, value in metadata.items(): if key.lower() not in internal_keys: extra[f"x-amz-meta-{key}"] = value @@ -283,21 +288,26 @@ async def _copy_encrypted( ) if src_multipart_meta: + # Multipart source carries its kid in the stored metadata. plaintext = await self._download_encrypted_multipart( client, src_bucket, src_key, src_multipart_meta ) else: + # Single-object source kid is on the head response we already fetched. + src_kid = head_resp.get("Metadata", {}).get(self.settings.kidtag_name, "") plaintext = await self._download_encrypted_single( - client, src_bucket, src_key, src_wrapped_dek + client, src_bucket, src_key, src_wrapped_dek, src_kid ) - # Re-encrypt - encrypted = crypto.encrypt_object(plaintext, self.settings.kek) + # Re-encrypt under the calling credential's key. + dest_kid, dest_kek = self.keyring.key_for(client.credentials.access_key) + encrypted = crypto.encrypt_object(plaintext, dest_kek) etag = hashlib.md5(plaintext, usedforsecurity=False).hexdigest() # Build destination metadata dest_metadata = { self.settings.dektag_name: base64.b64encode(encrypted.wrapped_dek).decode(), + self.settings.kidtag_name: dest_kid, "client-etag": etag, "plaintext-size": str(len(plaintext)), } @@ -308,7 +318,12 @@ async def _copy_encrypted( else: # Copy user metadata from source (excluding our internal keys) src_metadata = head_resp.get("Metadata", {}) - internal_keys = {self.settings.dektag_name.lower(), "client-etag", "plaintext-size"} + internal_keys = { + self.settings.dektag_name.lower(), + self.settings.kidtag_name.lower(), + "client-etag", + "plaintext-size", + } for meta_key, meta_value in src_metadata.items(): if meta_key.lower() not in internal_keys: dest_metadata[meta_key] = meta_value diff --git a/s3proxy/handlers/objects/put.py b/s3proxy/handlers/objects/put.py index f826c7e..6bfa2f6 100644 --- a/s3proxy/handlers/objects/put.py +++ b/s3proxy/handlers/objects/put.py @@ -130,11 +130,13 @@ async def _put_buffered( if needs_chunked_decode: body = decode_aws_chunked(body) - encrypted = crypto.encrypt_object(body, self.settings.kek) + kid, kek = self.keyring.key_for(client.credentials.access_key) + encrypted = crypto.encrypt_object(body, kek) logger.debug( "PUT_ENCRYPTED", bucket=bucket, key=key, + kid=kid, plaintext_mb=round(len(body) / 1024 / 1024, 2), ciphertext_mb=round(len(encrypted.ciphertext) / 1024 / 1024, 2), ) @@ -146,6 +148,7 @@ async def _put_buffered( encrypted.ciphertext, metadata={ self.settings.dektag_name: base64.b64encode(encrypted.wrapped_dek).decode(), + self.settings.kidtag_name: kid, "client-etag": etag, "plaintext-size": str(len(body)), }, @@ -169,8 +172,9 @@ async def _put_streaming( expires: str | None = None, tagging: str | None = None, ) -> Response: + kid, kek = self.keyring.key_for(client.credentials.access_key) dek = crypto.generate_dek() - wrapped_dek = crypto.wrap_key(dek, self.settings.kek) + wrapped_dek = crypto.wrap_key(dek, kek) resp = await client.create_multipart_upload( bucket, @@ -274,11 +278,12 @@ async def upload_part(data: bytes) -> None: bucket, key, MultipartMetadata( - version=1, + version=2, part_count=len(parts_meta), total_plaintext_size=total_plaintext_size, parts=parts_meta, wrapped_dek=wrapped_dek, + kid=kid, ), ) diff --git a/s3proxy/keyring.py b/s3proxy/keyring.py new file mode 100644 index 0000000..e596617 --- /dev/null +++ b/s3proxy/keyring.py @@ -0,0 +1,51 @@ +"""Per-access-key encryption key resolution. + +Each AWS access key has its own KEK (derived from a per-credential secret). The +access key that wrapped an object's DEK is stored alongside the object as its +``kid``, so decryption is always driven by the stored kid - never re-derived +from the request. A verified access key with no configured KEK is rejected. +""" + +from __future__ import annotations + +import hashlib + + +def derive_kek(secret: str) -> bytes: + """Derive a 32-byte KEK from a per-credential secret (SHA256).""" + return hashlib.sha256(secret.encode()).digest() + + +class KeyRing: + """Immutable resolver from access key -> KEK, and kid -> KEK. + + The kid is the access key that wrapped an object. Reads are lock-free; the + ring is built once at startup and never mutated. + """ + + __slots__ = ("_keys",) + + def __init__(self, keys: dict[str, bytes]): + # keys: access_key -> KEK. + self._keys = dict(keys) + + def key_for(self, access_key: str) -> tuple[str, bytes]: + """Resolve the kid + KEK to encrypt an object written by this access key. + + Raises KeyError if the access key has no configured KEK. + """ + try: + return access_key, self._keys[access_key] + except KeyError as e: + raise KeyError(f"No encryption key configured for access key {access_key!r}") from e + + def key_by_id(self, kid: str) -> bytes: + """Resolve the KEK that wrapped an object's DEK, by its stored kid. + + Raises KeyError for an empty or unknown kid - we cannot guess the + wrapping key. + """ + try: + return self._keys[kid] + except KeyError as e: + raise KeyError(f"No key for kid {kid!r}") from e diff --git a/s3proxy/main.py b/s3proxy/main.py index dbfcc04..99c2740 100644 --- a/s3proxy/main.py +++ b/s3proxy/main.py @@ -45,8 +45,8 @@ def main(): os.environ.setdefault("S3PROXY_REGION", args.region) os.environ.setdefault("S3PROXY_LOG_LEVEL", args.log_level) - if not os.environ.get("S3PROXY_ENCRYPT_KEY"): - sys.exit("Error: S3PROXY_ENCRYPT_KEY environment variable required") + if not os.environ.get("S3PROXY_CREDENTIALS"): + sys.exit("Error: S3PROXY_CREDENTIALS environment variable required") settings = Settings() application = create_app(settings) diff --git a/s3proxy/state/manager.py b/s3proxy/state/manager.py index 7c6159a..20d6b4e 100644 --- a/s3proxy/state/manager.py +++ b/s3proxy/state/manager.py @@ -50,6 +50,7 @@ async def create_upload( key: str, upload_id: str, dek: bytes, + kid: str = "", ) -> MultipartUploadState: """Create new upload state.""" state = MultipartUploadState( @@ -57,6 +58,7 @@ async def create_upload( bucket=bucket, key=key, upload_id=upload_id, + kid=kid, ) sk = self._storage_key(bucket, key, upload_id) diff --git a/s3proxy/state/metadata.py b/s3proxy/state/metadata.py index 7f2d09e..d3452d3 100644 --- a/s3proxy/state/metadata.py +++ b/s3proxy/state/metadata.py @@ -38,6 +38,7 @@ def encode_multipart_metadata(meta: MultipartMetadata) -> str: "pc": meta.part_count, "ts": meta.total_plaintext_size, "dek": base64.b64encode(meta.wrapped_dek).decode(), + "kid": meta.kid, "parts": [ { "pn": p.part_number, @@ -90,6 +91,7 @@ def decode_multipart_metadata(encoded: str) -> MultipartMetadata: part_count=data.get("pc", 0), total_plaintext_size=data.get("ts", 0), wrapped_dek=base64.b64decode(data.get("dek", "")), + kid=data.get("kid", ""), parts=[ PartMetadata( part_number=p["pn"], @@ -118,10 +120,11 @@ async def persist_upload_state( key: str, upload_id: str, wrapped_dek: bytes, + kid: str = "", ) -> None: """Persist DEK to S3 during upload (fallback for Redis failures).""" state_key = _internal_upload_key(key, upload_id) - data = {"dek": base64.b64encode(wrapped_dek).decode()} + data = {"dek": base64.b64encode(wrapped_dek).decode(), "kid": kid} logger.info( "PERSIST_UPLOAD_STATE", @@ -151,14 +154,12 @@ async def load_upload_state( bucket: str, key: str, upload_id: str, - kek: bytes, -) -> bytes | None: - """Load DEK from S3 for resumed upload. +) -> tuple[bytes, str] | None: + """Load the wrapped DEK + kid from S3 for a resumed upload. - Returns the unwrapped DEK, or None if not found. + Returns (wrapped_dek, kid), or None if not found. Callers unwrap with the + key resolved from kid - this layer stays free of keyring knowledge. """ - from .. import crypto - state_key = _internal_upload_key(key, upload_id) logger.info( @@ -181,7 +182,7 @@ async def load_upload_state( key=key, upload_id=upload_id[:20] + "...", ) - return crypto.unwrap_key(wrapped_dek, kek) + return wrapped_dek, data.get("kid", "") except Exception as e: logger.warning( diff --git a/s3proxy/state/models.py b/s3proxy/state/models.py index e97ec8f..bcb0338 100644 --- a/s3proxy/state/models.py +++ b/s3proxy/state/models.py @@ -51,6 +51,7 @@ class MultipartUploadState: created_at: datetime = field(default_factory=lambda: datetime.now(UTC)) total_plaintext_size: int = 0 next_internal_part_number: int = 1 # Next S3 part number to use + kid: str = "" # Key id that wraps this upload's DEK ("" = default key) @dataclass(slots=True) @@ -66,6 +67,7 @@ class MultipartMetadata: total_plaintext_size: int = 0 parts: list[PartMetadata] = field(default_factory=list) wrapped_dek: bytes = b"" + kid: str = "" # Key id that wrapped the DEK ("" = legacy/default key) class StateMissingError(Exception): diff --git a/s3proxy/state/recovery.py b/s3proxy/state/recovery.py index d13b373..759163d 100644 --- a/s3proxy/state/recovery.py +++ b/s3proxy/state/recovery.py @@ -7,6 +7,7 @@ from structlog.stdlib import BoundLogger from .. import crypto +from ..keyring import KeyRing from .manager import MAX_INTERNAL_PARTS_PER_CLIENT from .metadata import load_upload_state from .models import InternalPartMetadata, MultipartUploadState, PartMetadata @@ -30,12 +31,12 @@ async def reconstruct_upload_state_from_s3( bucket: str, key: str, upload_id: str, - kek: bytes, + keyring: KeyRing, ) -> MultipartUploadState | None: """Reconstruct upload state from S3 when Redis state is lost. This is a fallback recovery mechanism that: - 1. Loads the DEK from S3 metadata + 1. Loads the wrapped DEK + kid from S3 metadata, unwraps via the keyring 2. Lists all uploaded parts from S3 3. Reconstructs part metadata from the S3 response @@ -49,9 +50,9 @@ async def reconstruct_upload_state_from_s3( upload_id=upload_id[:20] + "..." if len(upload_id) > 20 else upload_id, ) - # Step 1: Load DEK from S3 metadata - dek = await load_upload_state(s3_client, bucket, key, upload_id, kek) - if not dek: + # Step 1: Load wrapped DEK + kid from S3 metadata, unwrap via keyring + state_data = await load_upload_state(s3_client, bucket, key, upload_id) + if not state_data: logger.warning( "RECONSTRUCT_FAILED_NO_DEK", bucket=bucket, @@ -59,6 +60,8 @@ async def reconstruct_upload_state_from_s3( upload_id=upload_id, ) return None + wrapped_dek, kid = state_data + dek = crypto.unwrap_key(wrapped_dek, keyring.key_by_id(kid)) # Step 2: List all uploaded parts from S3 try: @@ -158,6 +161,7 @@ async def reconstruct_upload_state_from_s3( total_plaintext_size=total_plaintext_size, next_internal_part_number=max_internal_part_number + 1, created_at=datetime.now(UTC), + kid=kid, ) logger.info( diff --git a/s3proxy/state/serialization.py b/s3proxy/state/serialization.py index 1a9ac56..5612626 100644 --- a/s3proxy/state/serialization.py +++ b/s3proxy/state/serialization.py @@ -36,6 +36,7 @@ def serialize_upload_state(state: MultipartUploadState) -> bytes: "created_at": state.created_at.isoformat(), "total_plaintext_size": state.total_plaintext_size, "next_internal_part_number": state.next_internal_part_number, + "kid": state.kid, "parts": { str(pn): { "part_number": p.part_number, @@ -132,6 +133,7 @@ def deserialize_upload_state(data: bytes) -> MultipartUploadState | None: created_at=datetime.fromisoformat(obj["created_at"]), total_plaintext_size=obj.get("total_plaintext_size", 0), next_internal_part_number=obj.get("next_internal_part_number", 1), + kid=obj.get("kid", ""), ) except (KeyError, TypeError, ValueError) as e: logger.error( diff --git a/tests/conftest.py b/tests/conftest.py index f6d94fa..add6f32 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -10,8 +10,7 @@ import fakeredis.aioredis import pytest -# Set required environment variables before importing s3proxy modules -os.environ.setdefault("S3PROXY_ENCRYPT_KEY", "test-encryption-key-for-pytest") +# Set environment variables before importing s3proxy modules os.environ.setdefault("S3PROXY_HOST", "http://localhost:9000") from s3proxy.config import Settings @@ -42,20 +41,26 @@ async def mock_redis(): @pytest.fixture def settings(): - """Create test settings with encryption key.""" + """Create test settings with encryption key and one credential.""" return Settings( host="http://localhost:9000", - encrypt_key="test-encryption-key-32bytes!!!!", region="us-east-1", no_tls=True, port=4433, + credentials=[ + { + "access_key": "AKIAIOSFODNN7EXAMPLE", + "secret_key": "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY", + "kek": "test-credential-kek", + } + ], ) @pytest.fixture def kek(settings): - """Get the Key Encryption Key derived from settings.""" - return settings.kek + """Get the KEK for the standard test credential.""" + return settings.keyring.key_for("AKIAIOSFODNN7EXAMPLE")[1] # ============================================================================ @@ -122,11 +127,17 @@ async def __aexit__(self, exc_type, exc_val, exc_tb): class MockS3Client: """Mock S3 client for testing without real S3 backend.""" - def __init__(self): + def __init__(self, credentials: S3Credentials | None = None): self.objects: dict[str, dict[str, Any]] = {} # bucket/key -> {body, metadata, ...} self.buckets: dict[str, dict] = {} self.multipart_uploads: dict[str, dict] = {} # upload_id -> {bucket, key, parts} self.call_history: list[tuple[str, dict]] = [] + # Handlers read client.credentials.access_key to resolve the per-login KEK. + self.credentials = credentials or S3Credentials( + access_key="AKIAIOSFODNN7EXAMPLE", + secret_key="wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY", + region="us-east-1", + ) async def __aenter__(self): """Async context manager entry - returns self.""" diff --git a/tests/docker-compose.yml b/tests/docker-compose.yml index 0627302..58a1b47 100644 --- a/tests/docker-compose.yml +++ b/tests/docker-compose.yml @@ -26,12 +26,10 @@ services: ports: - "4433:4433" environment: - S3PROXY_ENCRYPT_KEY: "test-encryption-key-32-bytes!!" + S3PROXY_CREDENTIALS: '[{"access_key":"minioadmin","secret_key":"minioadmin","kek":"test-encryption-key-32-bytes!!"}]' S3PROXY_HOST: "http://minio:9000" S3PROXY_REGION: "us-east-1" S3PROXY_MEMORY_LIMIT_MB: "48" S3PROXY_LOG_LEVEL: "WARNING" - AWS_ACCESS_KEY_ID: minioadmin - AWS_SECRET_ACCESS_KEY: minioadmin depends_on: - minio diff --git a/tests/ha/test_ha_redis_e2e.py b/tests/ha/test_ha_redis_e2e.py index a3ab06a..4b71569 100644 --- a/tests/ha/test_ha_redis_e2e.py +++ b/tests/ha/test_ha_redis_e2e.py @@ -84,15 +84,16 @@ def s3proxy_pods(redis_server): env = os.environ.copy() env.update( { - "S3PROXY_ENCRYPT_KEY": "test-encryption-key-32-bytes!!", + "S3PROXY_CREDENTIALS": ( + '[{"access_key":"minioadmin","secret_key":"minioadmin",' + '"kek":"test-encryption-key-32-bytes!!"}]' + ), "S3PROXY_HOST": "http://localhost:9000", "S3PROXY_REGION": "us-east-1", "S3PROXY_PORT": str(port), "S3PROXY_NO_TLS": "true", "S3PROXY_REDIS_URL": redis_server, "S3PROXY_LOG_LEVEL": "WARNING", - "AWS_ACCESS_KEY_ID": "minioadmin", - "AWS_SECRET_ACCESS_KEY": "minioadmin", } ) diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py index 9c2a1c2..8de0ca4 100644 --- a/tests/integration/conftest.py +++ b/tests/integration/conftest.py @@ -44,7 +44,10 @@ def run_s3proxy( env = os.environ.copy() env.update( { - "S3PROXY_ENCRYPT_KEY": "test-encryption-key-32-bytes!!", + "S3PROXY_CREDENTIALS": ( + '[{"access_key":"minioadmin","secret_key":"minioadmin",' + '"kek":"test-encryption-key-32-bytes!!"}]' + ), "S3PROXY_HOST": "http://localhost:9000", "S3PROXY_REGION": "us-east-1", "S3PROXY_PORT": str(port), diff --git a/tests/integration/test_elasticsearch_range_scenario.py b/tests/integration/test_elasticsearch_range_scenario.py index bca58f1..720f91b 100644 --- a/tests/integration/test_elasticsearch_range_scenario.py +++ b/tests/integration/test_elasticsearch_range_scenario.py @@ -28,7 +28,7 @@ class TestElasticsearchRangeScenario: """Test the actual scenario from Elasticsearch backup logs.""" @pytest.mark.asyncio - async def test_elasticsearch_backup_range_error(self, handler, settings): + async def test_elasticsearch_backup_range_error(self, handler, settings, kek): """ Test scenario from logs: range bytes=53687203-70464446 fails. @@ -94,7 +94,8 @@ async def test_elasticsearch_backup_range_error(self, handler, settings): part_count=1, total_plaintext_size=64 * 1024 * 1024, parts=[part_meta], - wrapped_dek=crypto.wrap_key(dek, settings.kek), + wrapped_dek=crypto.wrap_key(dek, kek), + kid="AKIAIOSFODNN7EXAMPLE", ) with patch.object(handler, "_client", return_value=mock_client): @@ -132,7 +133,7 @@ async def test_elasticsearch_backup_range_error(self, handler, settings): ) @pytest.mark.asyncio - async def test_partial_object_with_3_of_4_parts(self, handler, settings): + async def test_partial_object_with_3_of_4_parts(self, handler, settings, kek): """ Test when metadata claims 4 internal parts but only 3 were uploaded. @@ -207,7 +208,8 @@ def get_object_side_effect(bucket, key, range_header=None): part_count=1, total_plaintext_size=64 * 1024 * 1024, parts=[part_meta], - wrapped_dek=crypto.wrap_key(dek, settings.kek), + wrapped_dek=crypto.wrap_key(dek, kek), + kid="AKIAIOSFODNN7EXAMPLE", ) with patch.object(handler, "_client", return_value=mock_client): @@ -244,7 +246,7 @@ def get_object_side_effect(bucket, key, range_header=None): ) @pytest.mark.asyncio - async def test_successful_3_part_fetch(self, handler, settings): + async def test_successful_3_part_fetch(self, handler, settings, kek): """Test that fetching 3 complete parts works correctly.""" mock_client = AsyncMock() # Make mock_client an async context manager @@ -324,7 +326,8 @@ def get_object_side_effect(bucket, key, range_header=None): part_count=1, total_plaintext_size=3 * size_per_part, parts=[part_meta], - wrapped_dek=crypto.wrap_key(dek, settings.kek), + wrapped_dek=crypto.wrap_key(dek, kek), + kid="AKIAIOSFODNN7EXAMPLE", ) with patch.object(handler, "_client", return_value=mock_client): diff --git a/tests/integration/test_entity_too_small_errors.py b/tests/integration/test_entity_too_small_errors.py index 37ff598..30a522f 100644 --- a/tests/integration/test_entity_too_small_errors.py +++ b/tests/integration/test_entity_too_small_errors.py @@ -26,7 +26,9 @@ async def test_complete_with_missing_part_rejected(self, handler, settings): # Create upload state with parts 1, 2, 3, 5 (missing 4) dek = crypto.generate_dek() - await handler.multipart_manager.create_upload(bucket, key, upload_id, dek) + await handler.multipart_manager.create_upload( + bucket, key, upload_id, dek, kid="AKIAIOSFODNN7EXAMPLE" + ) # Add parts 1, 2, 3, 5 to state for part_num in [1, 2, 3, 5]: @@ -89,7 +91,9 @@ async def test_entity_too_small_with_small_parts(self, handler, settings): # Create upload with 5 parts of 1KB each (total 5KB) dek = crypto.generate_dek() - await handler.multipart_manager.create_upload(bucket, key, upload_id, dek) + await handler.multipart_manager.create_upload( + bucket, key, upload_id, dek, kid="AKIAIOSFODNN7EXAMPLE" + ) # Add 5 parts of 1KB each for part_num in range(1, 6): diff --git a/tests/integration/test_entity_too_small_fix.py b/tests/integration/test_entity_too_small_fix.py index ff3f1b3..8d1cdcf 100644 --- a/tests/integration/test_entity_too_small_fix.py +++ b/tests/integration/test_entity_too_small_fix.py @@ -30,7 +30,7 @@ async def test_elasticsearch_typical_50mb_part_no_split(self, manager, settings) # Create upload dek = crypto.generate_dek() - state = await manager.create_upload(bucket, key, upload_id, dek) + state = await manager.create_upload(bucket, key, upload_id, dek, kid="AKIAIOSFODNN7EXAMPLE") # Simulate Elasticsearch uploading a 50MB part (typical size) # With PART_SIZE=64MB, this should NOT be split @@ -79,7 +79,7 @@ async def test_elasticsearch_multiple_50mb_parts_no_entity_too_small(self, manag # Create upload dek = crypto.generate_dek() - await manager.create_upload(bucket, key, upload_id, dek) + await manager.create_upload(bucket, key, upload_id, dek, kid="AKIAIOSFODNN7EXAMPLE") # Upload 6 parts of ~50MB each (total ~300MB, similar to shard 3: 305MB) internal_part_num = 1 @@ -141,7 +141,7 @@ async def test_large_100mb_part_splits_correctly(self, manager, settings): # Create upload dek = crypto.generate_dek() - await manager.create_upload(bucket, key, upload_id, dek) + await manager.create_upload(bucket, key, upload_id, dek, kid="AKIAIOSFODNN7EXAMPLE") # 100MB part splits into 64MB + 36MB part = PartMetadata( @@ -197,7 +197,7 @@ async def test_edge_case_130mb_part_with_small_remainder(self, manager, settings # Create upload dek = crypto.generate_dek() - await manager.create_upload(bucket, key, upload_id, dek) + await manager.create_upload(bucket, key, upload_id, dek, kid="AKIAIOSFODNN7EXAMPLE") # 130MB part splits into 64MB + 64MB + 2MB part = PartMetadata( @@ -261,7 +261,7 @@ async def test_production_scenario_shard_3(self, manager, settings): # Create upload dek = crypto.generate_dek() - await manager.create_upload(bucket, key, upload_id, dek) + await manager.create_upload(bucket, key, upload_id, dek, kid="AKIAIOSFODNN7EXAMPLE") # Production showed parts 1, 2, 3, 4, 5 with total 305MB # Average ~61MB per part diff --git a/tests/integration/test_handlers.py b/tests/integration/test_handlers.py index 7b9277f..0018931 100644 --- a/tests/integration/test_handlers.py +++ b/tests/integration/test_handlers.py @@ -16,10 +16,16 @@ def settings(): """Create test settings.""" return Settings( host="http://localhost:9000", - encrypt_key="test-encryption-key", region="us-east-1", no_tls=True, port=4433, + credentials=[ + { + "access_key": "AKIAIOSFODNN7EXAMPLE", + "secret_key": "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY", + "kek": "test-credential-kek", + } + ], ) @@ -89,11 +95,10 @@ class TestSettings: def test_default_settings(self): """Test default settings values.""" - with patch.dict(os.environ, {"S3PROXY_ENCRYPT_KEY": "test-key"}): - settings = Settings() - assert settings.region == "us-east-1" - assert settings.no_tls is False - assert settings.port == 4433 + settings = Settings() + assert settings.region == "us-east-1" + assert settings.no_tls is False + assert settings.port == 4433 class TestRangeParsing: @@ -281,7 +286,7 @@ class TestMultipartDownloadWithInternalParts: """Test downloading multipart objects with internal parts (streaming uploads).""" @pytest.mark.asyncio - async def test_download_multipart_with_internal_parts(self, settings, mock_s3): + async def test_download_multipart_with_internal_parts(self, settings, mock_s3, kek): """Test downloading an object that was uploaded with internal parts.""" from unittest.mock import MagicMock @@ -305,7 +310,7 @@ async def test_download_multipart_with_internal_parts(self, settings, mock_s3): # Part 1: 50MB plaintext split into 4 internal parts (16MB + 16MB + 16MB + 2MB) test_data = b"x" * (50 * 1024 * 1024) # 50MB dek = crypto.generate_dek() - wrapped_dek = crypto.wrap_key(dek, settings.kek) + wrapped_dek = crypto.wrap_key(dek, kek) # Split into internal parts part_size = 16 * 1024 * 1024 @@ -353,6 +358,7 @@ async def test_download_multipart_with_internal_parts(self, settings, mock_s3): total_plaintext_size=len(test_data), parts=[part_meta], wrapped_dek=wrapped_dek, + kid="AKIAIOSFODNN7EXAMPLE", ) # Upload concatenated ciphertext as the S3 object @@ -386,7 +392,7 @@ async def test_download_multipart_with_internal_parts(self, settings, mock_s3): assert response.status_code == 200 @pytest.mark.asyncio - async def test_download_multipart_with_range_request(self, settings, mock_s3): + async def test_download_multipart_with_range_request(self, settings, mock_s3, kek): """Test range download from object with internal parts.""" from unittest.mock import MagicMock @@ -409,7 +415,7 @@ async def test_download_multipart_with_range_request(self, settings, mock_s3): # Create test data with recognizable pattern test_data = b"".join([bytes([i % 256]) * 1024 for i in range(1024)]) # 1MB with pattern dek = crypto.generate_dek() - wrapped_dek = crypto.wrap_key(dek, settings.kek) + wrapped_dek = crypto.wrap_key(dek, kek) # Split into 2 internal parts part_size = len(test_data) // 2 @@ -455,6 +461,7 @@ async def test_download_multipart_with_range_request(self, settings, mock_s3): total_plaintext_size=len(test_data), parts=[part_meta], wrapped_dek=wrapped_dek, + kid="AKIAIOSFODNN7EXAMPLE", ) # Upload concatenated ciphertext diff --git a/tests/integration/test_integration.py b/tests/integration/test_integration.py index 867c4a1..d420279 100644 --- a/tests/integration/test_integration.py +++ b/tests/integration/test_integration.py @@ -17,13 +17,15 @@ class TestObjectEncryptionFlow: """Test full object encryption/decryption workflow.""" @pytest.mark.asyncio - async def test_put_then_get_object(self, mock_s3, settings, credentials, multipart_manager): + async def test_put_then_get_object( + self, mock_s3, settings, credentials, multipart_manager, kek + ): """Test uploading and then downloading an object preserves data.""" S3ProxyHandler(settings, {}, multipart_manager) plaintext = b"Hello, this is secret data!" # Encrypt and store - encrypted = crypto.encrypt_object(plaintext, settings.kek) + encrypted = crypto.encrypt_object(plaintext, kek) metadata = { settings.dektag_name: base64.b64encode(encrypted.wrapped_dek).decode(), "plaintext-size": str(len(plaintext)), @@ -37,15 +39,15 @@ async def test_put_then_get_object(self, mock_s3, settings, credentials, multipa stored_metadata = resp["Metadata"] wrapped_dek = base64.b64decode(stored_metadata[settings.dektag_name]) - decrypted = crypto.decrypt_object(ciphertext, wrapped_dek, settings.kek) + decrypted = crypto.decrypt_object(ciphertext, wrapped_dek, kek) assert decrypted == plaintext @pytest.mark.asyncio - async def test_put_then_head_object(self, mock_s3, settings): + async def test_put_then_head_object(self, mock_s3, settings, kek): """Test HEAD returns correct plaintext size.""" plaintext = b"Test data for head request" - encrypted = crypto.encrypt_object(plaintext, settings.kek) + encrypted = crypto.encrypt_object(plaintext, kek) metadata = { settings.dektag_name: base64.b64encode(encrypted.wrapped_dek).decode(), "plaintext-size": str(len(plaintext)), @@ -68,10 +70,10 @@ async def test_delete_object(self, mock_s3): assert "NoSuchKey" in str(exc_info.value) @pytest.mark.asyncio - async def test_range_request(self, mock_s3, settings): + async def test_range_request(self, mock_s3, settings, kek): """Test partial object download with range request.""" plaintext = b"0123456789ABCDEF" # 16 bytes - encrypted = crypto.encrypt_object(plaintext, settings.kek) + encrypted = crypto.encrypt_object(plaintext, kek) metadata = { settings.dektag_name: base64.b64encode(encrypted.wrapped_dek).decode(), "plaintext-size": str(len(plaintext)), @@ -90,7 +92,7 @@ class TestMultipartEncryptionFlow: """Test multipart upload encryption workflow.""" @pytest.mark.asyncio - async def test_multipart_upload_flow(self, mock_s3, settings, multipart_manager): + async def test_multipart_upload_flow(self, mock_s3, settings, multipart_manager, kek): """Test complete multipart upload flow.""" bucket = "test-bucket" key = "large-file.bin" @@ -104,7 +106,7 @@ async def test_multipart_upload_flow(self, mock_s3, settings, multipart_manager) # Generate encryption key dek = crypto.generate_dek() - crypto.wrap_key(dek, settings.kek) + crypto.wrap_key(dek, kek) # Upload parts part1_plaintext = b"A" * 5242880 # 5MB @@ -388,13 +390,13 @@ class TestEncryptedCopyObject: """Test copy object with encrypted source.""" @pytest.mark.asyncio - async def test_copy_encrypted_object(self, mock_s3, settings): + async def test_copy_encrypted_object(self, mock_s3, settings, kek): """Test copying an encrypted object re-encrypts it.""" await mock_s3.create_bucket("test-bucket") # Store encrypted object plaintext = b"Secret data to copy" - encrypted = crypto.encrypt_object(plaintext, settings.kek) + encrypted = crypto.encrypt_object(plaintext, kek) metadata = { settings.dektag_name: base64.b64encode(encrypted.wrapped_dek).decode(), "plaintext-size": str(len(plaintext)), diff --git a/tests/integration/test_large_file_streaming.py b/tests/integration/test_large_file_streaming.py index 7ba241c..c94c340 100644 --- a/tests/integration/test_large_file_streaming.py +++ b/tests/integration/test_large_file_streaming.py @@ -181,7 +181,7 @@ async def mock_body(): assert len(put_object_calls) >= 1, "Should use single put_object for small files" @pytest.mark.asyncio - async def test_large_file_encryption_and_decryption(self, mock_s3, settings): + async def test_large_file_encryption_and_decryption(self, mock_s3, settings, kek): """Test full workflow: upload large file, download, decrypt, verify.""" bucket = "test-bucket" key = "large-encrypted.bin" @@ -193,7 +193,7 @@ async def test_large_file_encryption_and_decryption(self, mock_s3, settings): # Generate DEK and wrap it dek = crypto.generate_dek() - wrapped_dek = crypto.wrap_key(dek, settings.kek) + wrapped_dek = crypto.wrap_key(dek, kek) # Simulate multipart upload by encrypting in parts upload_id = "test-upload-123" @@ -231,7 +231,7 @@ async def test_large_file_encryption_and_decryption(self, mock_s3, settings): # Unwrap DEK wrapped_dek_bytes = base64.b64decode(metadata[settings.dektag_name]) - decrypted_dek = crypto.unwrap_key(wrapped_dek_bytes, settings.kek) + decrypted_dek = crypto.unwrap_key(wrapped_dek_bytes, kek) # Decrypt all parts # We need to track position in ciphertext since parts may have different sizes diff --git a/tests/integration/test_memory_usage.py b/tests/integration/test_memory_usage.py index 22995ed..d9e24fc 100644 --- a/tests/integration/test_memory_usage.py +++ b/tests/integration/test_memory_usage.py @@ -51,7 +51,10 @@ def s3proxy_with_memory_limit(self): env = os.environ.copy() env.update( { - "S3PROXY_ENCRYPT_KEY": "test-encryption-key-32-bytes!!", + "S3PROXY_CREDENTIALS": ( + '[{"access_key":"minioadmin","secret_key":"minioadmin",' + '"kek":"test-encryption-key-32-bytes!!"}]' + ), "S3PROXY_HOST": "http://localhost:9000", "S3PROXY_REGION": "us-east-1", "S3PROXY_PORT": str(port), @@ -102,7 +105,10 @@ def s3proxy_with_short_backpressure(self): env = os.environ.copy() env.update( { - "S3PROXY_ENCRYPT_KEY": "test-encryption-key-32-bytes!!", + "S3PROXY_CREDENTIALS": ( + '[{"access_key":"minioadmin","secret_key":"minioadmin",' + '"kek":"test-encryption-key-32-bytes!!"}]' + ), "S3PROXY_HOST": "http://localhost:9000", "S3PROXY_REGION": "us-east-1", "S3PROXY_PORT": str(port), diff --git a/tests/integration/test_multipart_range_validation.py b/tests/integration/test_multipart_range_validation.py index 11cbdda..c91e45b 100644 --- a/tests/integration/test_multipart_range_validation.py +++ b/tests/integration/test_multipart_range_validation.py @@ -31,7 +31,9 @@ class TestMultipartRangeValidation: """Test range validation for multipart downloads with internal parts.""" @pytest.mark.asyncio - async def test_invalid_range_detected_before_fetch(self, handler, settings, mock_s3_client): + async def test_invalid_range_detected_before_fetch( + self, handler, settings, mock_s3_client, kek + ): """Test that invalid ranges are detected before making S3 requests.""" # Create metadata with internal parts that exceed actual object size internal_parts = [ @@ -69,7 +71,8 @@ async def test_invalid_range_detected_before_fetch(self, handler, settings, mock part_count=1, total_plaintext_size=48 * 1024 * 1024, parts=[part_meta], - wrapped_dek=crypto.wrap_key(crypto.generate_dek(), settings.kek), + wrapped_dek=crypto.wrap_key(crypto.generate_dek(), kek), + kid="AKIAIOSFODNN7EXAMPLE", ) # Mock head_object to return a size smaller than what metadata expects @@ -134,7 +137,7 @@ async def test_invalid_range_detected_before_fetch(self, handler, settings, mock ) @pytest.mark.asyncio - async def test_handles_s3_invalid_range_error(self, handler, settings): + async def test_handles_s3_invalid_range_error(self, handler, settings, kek): """Test that S3 InvalidRange errors are caught and wrapped properly.""" # Create a mock S3 client that raises InvalidRange mock_client = AsyncMock() @@ -180,7 +183,8 @@ async def test_handles_s3_invalid_range_error(self, handler, settings): part_count=1, total_plaintext_size=1000, parts=[part_meta], - wrapped_dek=crypto.wrap_key(crypto.generate_dek(), settings.kek), + wrapped_dek=crypto.wrap_key(crypto.generate_dek(), kek), + kid="AKIAIOSFODNN7EXAMPLE", ) with patch.object(handler, "_client", return_value=mock_client): @@ -211,7 +215,7 @@ async def test_handles_s3_invalid_range_error(self, handler, settings): ) @pytest.mark.asyncio - async def test_valid_range_succeeds(self, handler, settings): + async def test_valid_range_succeeds(self, handler, settings, kek): """Test that valid ranges work correctly.""" mock_client = AsyncMock() # Make mock_client an async context manager @@ -262,7 +266,8 @@ async def test_valid_range_succeeds(self, handler, settings): part_count=1, total_plaintext_size=len(plaintext), parts=[part_meta], - wrapped_dek=crypto.wrap_key(dek, settings.kek), + wrapped_dek=crypto.wrap_key(dek, kek), + kid="AKIAIOSFODNN7EXAMPLE", ) with patch.object(handler, "_client", return_value=mock_client): @@ -287,7 +292,7 @@ async def test_valid_range_succeeds(self, handler, settings): assert response.status_code == 200 @pytest.mark.asyncio - async def test_multiple_internal_parts_validation(self, handler, settings): + async def test_multiple_internal_parts_validation(self, handler, settings, kek): """Test validation with multiple internal parts.""" mock_client = AsyncMock() # Make mock_client an async context manager @@ -333,7 +338,8 @@ async def test_multiple_internal_parts_validation(self, handler, settings): part_count=1, total_plaintext_size=48 * 1024 * 1024, parts=[part_meta], - wrapped_dek=crypto.wrap_key(crypto.generate_dek(), settings.kek), + wrapped_dek=crypto.wrap_key(crypto.generate_dek(), kek), + kid="AKIAIOSFODNN7EXAMPLE", ) with patch.object(handler, "_client", return_value=mock_client): diff --git a/tests/integration/test_part_ordering.py b/tests/integration/test_part_ordering.py index 4d3521d..e518e06 100644 --- a/tests/integration/test_part_ordering.py +++ b/tests/integration/test_part_ordering.py @@ -34,7 +34,7 @@ async def test_out_of_order_client_parts_sorted_internally(self, manager, settin from s3proxy import crypto dek = crypto.generate_dek() - await manager.create_upload(bucket, key, upload_id, dek) + await manager.create_upload(bucket, key, upload_id, dek, kid="AKIAIOSFODNN7EXAMPLE") # Simulate parts uploaded out of order with direct client→internal mapping # (NEW: no splitting = use client part number as internal part number) @@ -158,7 +158,7 @@ async def test_sequential_parts_remain_sorted(self, manager, settings): from s3proxy import crypto dek = crypto.generate_dek() - await manager.create_upload(bucket, key, upload_id, dek) + await manager.create_upload(bucket, key, upload_id, dek, kid="AKIAIOSFODNN7EXAMPLE") # Upload parts in order with sequential internal parts for i in range(1, 4): diff --git a/tests/integration/test_partial_complete_fix.py b/tests/integration/test_partial_complete_fix.py index cfcdf39..a67d206 100644 --- a/tests/integration/test_partial_complete_fix.py +++ b/tests/integration/test_partial_complete_fix.py @@ -25,7 +25,9 @@ async def test_complete_with_subset_of_parts(self, handler, settings): dek = crypto.generate_dek() # Simulate upload state with 4 parts uploaded - await handler.multipart_manager.create_upload("bucket", "key", "upload-123", dek) + await handler.multipart_manager.create_upload( + "bucket", "key", "upload-123", dek, kid="AKIAIOSFODNN7EXAMPLE" + ) # Create 4 parts in state (all uploaded) for i in range(1, 5): @@ -129,7 +131,9 @@ async def test_complete_logs_size_mismatch(self, handler, settings): mock_client.__aexit__ = AsyncMock(return_value=None) dek = crypto.generate_dek() - await handler.multipart_manager.create_upload("bucket", "key", "upload-123", dek) + await handler.multipart_manager.create_upload( + "bucket", "key", "upload-123", dek, kid="AKIAIOSFODNN7EXAMPLE" + ) # Add 2 parts for i in range(1, 3): @@ -195,7 +199,9 @@ async def test_complete_with_no_parts_fails(self, handler, settings): mock_client.__aexit__ = AsyncMock(return_value=None) dek = crypto.generate_dek() - await handler.multipart_manager.create_upload("bucket", "key", "upload-123", dek) + await handler.multipart_manager.create_upload( + "bucket", "key", "upload-123", dek, kid="AKIAIOSFODNN7EXAMPLE" + ) # Add some parts to state part = PartMetadata( diff --git a/tests/integration/test_per_key_encryption.py b/tests/integration/test_per_key_encryption.py new file mode 100644 index 0000000..8bd6c9a --- /dev/null +++ b/tests/integration/test_per_key_encryption.py @@ -0,0 +1,195 @@ +"""End-to-end tests for per-access-key encryption keys. + +Drives the real handler against the in-memory MockS3Client to verify: +- objects are encrypted under the calling login's key, +- the access key is stored on the object (kid) and used for decryption, +- a different login decrypts correctly as long as its KEK is configured, +- the access key never leaks back to clients as user metadata. +""" + +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from s3proxy.config import Settings +from s3proxy.handlers import S3ProxyHandler +from s3proxy.s3client import S3Credentials +from s3proxy.state import MultipartStateManager + +ACME = "AKIA-ACME" +GLOBEX = "AKIA-GLOBEX" + + +@pytest.fixture +def settings(): + return Settings( + host="http://localhost:9000", + no_tls=True, + credentials=[ + {"access_key": ACME, "secret_key": "acme-sec", "kek": "acme-kek"}, + {"access_key": GLOBEX, "secret_key": "globex-sec", "kek": "globex-kek"}, + ], + ) + + +def _creds(access_key: str) -> S3Credentials: + return S3Credentials(access_key=access_key, secret_key="x", region="us-east-1") + + +def _handler(settings, mock_s3, access_key): + handler = S3ProxyHandler(settings, settings.credentials_store, MultipartStateManager()) + mock_s3.credentials = _creds(access_key) + handler._client = MagicMock(return_value=mock_s3) + return handler + + +def _put_request(path: str, body: bytes): + req = MagicMock() + req.url.path = path + req.headers = {"content-length": str(len(body)), "content-type": "text/plain"} + req.body = AsyncMock(return_value=body) + return req + + +def _get_request(path: str): + req = MagicMock() + req.url.path = path + req.headers = {} + return req + + +async def _read(response) -> bytes: + if hasattr(response, "body_iterator"): + return b"".join([c async for c in response.body_iterator]) + return response.body + + +class TestPerCredentialRoundtrip: + @pytest.mark.asyncio + async def test_object_encrypted_under_calling_login(self, settings, mock_s3): + handler = _handler(settings, mock_s3, ACME) + await mock_s3.create_bucket("data") + + body = b"acme payload" + await handler.handle_put_object(_put_request("/data/report.txt", body), _creds(ACME)) + + obj = mock_s3.objects[mock_s3._key("data", "report.txt")] + assert obj["Metadata"][settings.kidtag_name] == ACME + + resp = await handler.handle_get_object(_get_request("/data/report.txt"), _creds(ACME)) + assert await _read(resp) == body + + @pytest.mark.asyncio + async def test_different_logins_get_different_keys(self, settings, mock_s3): + # ACME writes one object, GLOBEX writes another - same bucket, different keys. + await mock_s3.create_bucket("shared") + + h_acme = _handler(settings, mock_s3, ACME) + await h_acme.handle_put_object(_put_request("/shared/a.txt", b"A"), _creds(ACME)) + + h_globex = _handler(settings, mock_s3, GLOBEX) + await h_globex.handle_put_object(_put_request("/shared/g.txt", b"G"), _creds(GLOBEX)) + + assert ( + mock_s3.objects[mock_s3._key("shared", "a.txt")]["Metadata"][settings.kidtag_name] + == ACME + ) + assert ( + mock_s3.objects[mock_s3._key("shared", "g.txt")]["Metadata"][settings.kidtag_name] + == GLOBEX + ) + + @pytest.mark.asyncio + async def test_decrypt_uses_stored_kid_not_caller(self, settings, mock_s3): + """ACME writes; a different login (GLOBEX) reads. Decryption uses the + stored ACME kid, so it works as long as ACME's KEK is configured.""" + await mock_s3.create_bucket("shared") + body = b"written by acme" + + h_acme = _handler(settings, mock_s3, ACME) + await h_acme.handle_put_object(_put_request("/shared/f.txt", body), _creds(ACME)) + + # GLOBEX reads the same object (its handler shares the same keyring). + h_globex = _handler(settings, mock_s3, GLOBEX) + resp = await h_globex.handle_get_object(_get_request("/shared/f.txt"), _creds(GLOBEX)) + assert await _read(resp) == body + + @pytest.mark.asyncio + async def test_unknown_access_key_rejected_on_write(self, settings, mock_s3): + handler = _handler(settings, mock_s3, "AKIA-NOT-CONFIGURED") + await mock_s3.create_bucket("data") + with pytest.raises(KeyError): + await handler.handle_put_object( + _put_request("/data/x.txt", b"data"), _creds("AKIA-NOT-CONFIGURED") + ) + + +class TestCopyPerCredential: + @pytest.mark.asyncio + async def test_copy_object_decrypts_source_kid_reencrypts_dest(self, settings, mock_s3): + """CopyObject must decrypt the source via its stored kid and re-encrypt + under the calling credential. Regression: the single-object source path + previously dropped the kid and crashed on decrypt.""" + await mock_s3.create_bucket("data") + body = b"copy me across credentials" + + # ACME writes the source. + h_acme = _handler(settings, mock_s3, ACME) + await h_acme.handle_put_object(_put_request("/data/src.txt", body), _creds(ACME)) + + # GLOBEX copies it -> re-encrypted under GLOBEX's kid. + h_globex = _handler(settings, mock_s3, GLOBEX) + copy_req = MagicMock() + copy_req.url.path = "/data/dst.txt" + copy_req.headers = {"x-amz-copy-source": "/data/src.txt"} + await h_globex.handle_copy_object(copy_req, _creds(GLOBEX)) + + assert ( + mock_s3.objects[mock_s3._key("data", "dst.txt")]["Metadata"][settings.kidtag_name] + == GLOBEX + ) + + resp = await h_globex.handle_get_object(_get_request("/data/dst.txt"), _creds(GLOBEX)) + assert await _read(resp) == body + + +class TestMultipartPerCredential: + @pytest.mark.asyncio + async def test_streaming_upload_roundtrip(self, settings, mock_s3): + handler = _handler(settings, mock_s3, ACME) + await mock_s3.create_bucket("data") + + body = b"m" * (12 * 1024 * 1024) # > MAX_BUFFER_SIZE -> multiple parts + + req = MagicMock() + req.headers = {} + + async def stream(): + for i in range(0, len(body), 1024 * 1024): + yield body[i : i + 1024 * 1024] + + req.stream = stream + + await handler._put_streaming(req, mock_s3, "data", "big.bin", "application/octet-stream") + + from s3proxy.state import load_multipart_metadata + + meta = await load_multipart_metadata(mock_s3, "data", "big.bin") + assert meta is not None + assert meta.kid == ACME + + resp = await handler.handle_get_object(_get_request("/data/big.bin"), _creds(ACME)) + assert await _read(resp) == body + + +class TestKidDoesNotLeak: + @pytest.mark.asyncio + async def test_get_response_has_no_kid_header(self, settings, mock_s3): + handler = _handler(settings, mock_s3, ACME) + await mock_s3.create_bucket("data") + + await handler.handle_put_object(_put_request("/data/leak.txt", b"data"), _creds(ACME)) + resp = await handler.handle_get_object(_get_request("/data/leak.txt"), _creds(ACME)) + + leaked = [h for h in resp.headers if "isec" in h.lower()] + assert not leaked, f"internal tags leaked to client: {leaked}" diff --git a/tests/integration/test_sequential_part_numbering.py b/tests/integration/test_sequential_part_numbering.py index 731c37c..08106de 100644 --- a/tests/integration/test_sequential_part_numbering.py +++ b/tests/integration/test_sequential_part_numbering.py @@ -32,7 +32,7 @@ async def test_two_part_upload_sequential_numbering(self, manager, settings): # Create upload dek = crypto.generate_dek() - await manager.create_upload(bucket, key, upload_id, dek) + await manager.create_upload(bucket, key, upload_id, dek, kid="AKIAIOSFODNN7EXAMPLE") # Simulate Client Part 2 uploaded first (smaller, 4.24MB) # With exact allocation (no +5 buffer), should get internal part 1 @@ -110,8 +110,12 @@ async def test_concurrent_uploads_independent_numbering(self, manager, settings) dek = crypto.generate_dek() # Create two independent uploads - await manager.create_upload("bucket", "file-a.tar", "upload-a", dek) - await manager.create_upload("bucket", "file-b.tar", "upload-b", dek) + await manager.create_upload( + "bucket", "file-a.tar", "upload-a", dek, kid="AKIAIOSFODNN7EXAMPLE" + ) + await manager.create_upload( + "bucket", "file-b.tar", "upload-b", dek, kid="AKIAIOSFODNN7EXAMPLE" + ) # Upload A allocates parts start_a1 = await manager.allocate_internal_parts("bucket", "file-a.tar", "upload-a", 1) @@ -145,7 +149,7 @@ async def test_eight_part_upload_all_sequential(self, manager, settings): upload_id = "test-eight-part" dek = crypto.generate_dek() - await manager.create_upload(bucket, key, upload_id, dek) + await manager.create_upload(bucket, key, upload_id, dek, kid="AKIAIOSFODNN7EXAMPLE") internal_numbers = [] for part_num in range(1, 9): @@ -196,7 +200,7 @@ async def test_old_behavior_with_buffer_would_fail(self, manager, settings): upload_id = "test-buffer-demo" dek = crypto.generate_dek() - await manager.create_upload(bucket, key, upload_id, dek) + await manager.create_upload(bucket, key, upload_id, dek, kid="AKIAIOSFODNN7EXAMPLE") # Simulate OLD behavior: allocate with +5 buffer # Client Part 1 (estimated 1 part) → allocates 1+5=6 parts [1-6] diff --git a/tests/unit/test_admin.py b/tests/unit/test_admin.py index 2f2e832..6e8e0ef 100644 --- a/tests/unit/test_admin.py +++ b/tests/unit/test_admin.py @@ -31,10 +31,11 @@ def _clean_state(): def admin_settings(): return Settings( host="http://localhost:9000", - encrypt_key="test-kek-32bytes!!!!!!!!!!!!!!!!", admin_ui=True, admin_username="admin", admin_password="secret", + admin_secret="test-admin-secret", + credentials=[{"access_key": "AKIA-TEST", "secret_key": "s", "kek": "k"}], ) @@ -169,10 +170,10 @@ def test_status_api_401_without_auth(admin_settings) -> None: def test_auth_uses_explicit_credentials_not_aws() -> None: settings = Settings( host="http://localhost:9000", - encrypt_key="test-kek", admin_ui=True, admin_username="admin", admin_password="admin", + admin_secret="test-admin-secret", ) admin = AdminCredentials(settings, {"AKIAEXAMPLE": "secret-key"}) assert admin.valid("admin", "admin") @@ -183,10 +184,10 @@ def test_auth_uses_explicit_credentials_not_aws() -> None: def test_auth_raises_when_credentials_blank() -> None: settings = Settings( host="http://localhost:9000", - encrypt_key="test-kek", admin_ui=True, admin_username="", admin_password="", + admin_secret="test-admin-secret", ) with pytest.raises(RuntimeError): create_auth_dependency(settings, {}) diff --git a/tests/unit/test_keyring.py b/tests/unit/test_keyring.py new file mode 100644 index 0000000..0cbdee1 --- /dev/null +++ b/tests/unit/test_keyring.py @@ -0,0 +1,97 @@ +"""Tests for per-access-key key resolution (KeyRing) and config wiring.""" + +import pytest +from pydantic import ValidationError + +from s3proxy.config import Settings +from s3proxy.keyring import KeyRing, derive_kek + + +def _ring(): + return KeyRing(keys={"AKIA-A": derive_kek("a-kek"), "AKIA-B": derive_kek("b-kek")}) + + +class TestKeyRingResolution: + def test_key_for_known_access_key(self): + ring = _ring() + kid, kek = ring.key_for("AKIA-A") + assert kid == "AKIA-A" + assert kek == derive_kek("a-kek") + + def test_key_for_unknown_access_key_raises(self): + ring = _ring() + with pytest.raises(KeyError): + ring.key_for("AKIA-UNKNOWN") + + def test_key_by_id_roundtrip(self): + ring = _ring() + # kid stored on an object == the access key that wrote it + assert ring.key_by_id("AKIA-B") == derive_kek("b-kek") + + def test_key_by_id_empty_raises(self): + ring = _ring() + with pytest.raises(KeyError): + ring.key_by_id("") + + def test_key_by_id_unknown_raises(self): + ring = _ring() + with pytest.raises(KeyError): + ring.key_by_id("AKIA-GHOST") + + +class TestSettingsKeyRing: + def test_no_credentials(self): + s = Settings() + assert s.credentials_store == {} + + def test_credentials_build_store_and_keyring(self): + s = Settings( + credentials=[ + {"access_key": "AKIA-A", "secret_key": "a-sec", "kek": "a-kek"}, + {"access_key": "AKIA-B", "secret_key": "b-sec", "kek": "b-kek"}, + ], + ) + assert s.credentials_store == {"AKIA-A": "a-sec", "AKIA-B": "b-sec"} + assert s.keyring.key_for("AKIA-A")[0] == "AKIA-A" + assert s.keyring.key_for("AKIA-A")[1] == derive_kek("a-kek") + # decrypt resolves by the stored kid (= access key) + assert s.keyring.key_by_id("AKIA-B") == derive_kek("b-kek") + + def test_credentials_from_env_json(self, monkeypatch): + monkeypatch.setenv( + "S3PROXY_CREDENTIALS", + '[{"access_key":"AKIA-A","secret_key":"s","kek":"e"}]', + ) + s = Settings() + assert s.credentials_store == {"AKIA-A": "s"} + assert s.keyring.key_for("AKIA-A")[1] == derive_kek("e") + + def test_duplicate_access_key_raises(self): + with pytest.raises(ValidationError): + Settings( + credentials=[ + {"access_key": "dup", "secret_key": "1", "kek": "a"}, + {"access_key": "dup", "secret_key": "2", "kek": "b"}, + ], + ) + + def test_unknown_access_key_rejected_at_encrypt(self): + s = Settings(credentials=[{"access_key": "AKIA-A", "secret_key": "s", "kek": "e"}]) + with pytest.raises(KeyError): + s.keyring.key_for("AKIA-NOT-CONFIGURED") + + +class TestAdminSecret: + def test_admin_ui_requires_secret(self): + with pytest.raises(ValidationError): + Settings(admin_ui=True, admin_username="a", admin_password="b") + + def test_admin_session_secret_is_stable(self): + a = Settings( + admin_ui=True, admin_username="a", admin_password="b", admin_secret="sek" + ).admin_session_secret + b = Settings( + admin_ui=True, admin_username="a", admin_password="b", admin_secret="sek" + ).admin_session_secret + assert a == b + assert len(a) == 32 diff --git a/tests/unit/test_metrics.py b/tests/unit/test_metrics.py index 14d90ed..9e5c10d 100644 --- a/tests/unit/test_metrics.py +++ b/tests/unit/test_metrics.py @@ -268,7 +268,6 @@ def client(self): settings = Settings( host="http://localhost:9000", - encrypt_key="test-encryption-key-32bytes!!!!", region="us-east-1", no_tls=True, port=4433, diff --git a/tests/unit/test_multipart.py b/tests/unit/test_multipart.py index 43be43d..3e0becd 100644 --- a/tests/unit/test_multipart.py +++ b/tests/unit/test_multipart.py @@ -7,6 +7,7 @@ import pytest +from s3proxy.keyring import KeyRing from s3proxy.state import ( InternalPartMetadata, MultipartMetadata, @@ -24,6 +25,13 @@ serialize_upload_state, ) +RECOVERY_KID = "AKIA-TEST" + + +def _single_key_ring(kek: bytes) -> KeyRing: + """A KeyRing holding `kek` under the kid the recovery state carries.""" + return KeyRing(keys={RECOVERY_KID: kek}) + class TestMultipartStateManager: """Test multipart state management.""" @@ -666,7 +674,10 @@ async def test_reconstruct_recovers_state_from_s3(self): # Mock the internal upload state object containing wrapped DEK # load_upload_state calls get_object on the internal key - wrapped_dek_data = {"dek": base64.b64encode(b"wrapped-dek-data").decode()} + wrapped_dek_data = { + "dek": base64.b64encode(b"wrapped-dek-data").decode(), + "kid": RECOVERY_KID, + } async def mock_read(): return json.dumps(wrapped_dek_data).encode() @@ -693,7 +704,7 @@ async def mock_read(): # Mock the crypto.unwrap_key to return the original DEK with patch("s3proxy.crypto.unwrap_key", return_value=original_dek): state = await reconstruct_upload_state_from_s3( - mock_client, "bucket", "key", "upload-123", kek + mock_client, "bucket", "key", "upload-123", _single_key_ring(kek) ) # Verify state was reconstructed correctly @@ -731,7 +742,7 @@ async def test_reconstruct_returns_none_when_dek_not_found(self): mock_client.get_object = AsyncMock(side_effect=Exception("NoSuchKey")) state = await reconstruct_upload_state_from_s3( - mock_client, "bucket", "key", "upload-123", kek + mock_client, "bucket", "key", "upload-123", _single_key_ring(kek) ) assert state is None @@ -744,7 +755,10 @@ async def test_reconstruct_returns_none_when_list_parts_fails(self): kek = b"k" * 32 # Mock successful DEK retrieval - wrapped_dek_data = {"dek": base64.b64encode(b"wrapped-dek-data").decode()} + wrapped_dek_data = { + "dek": base64.b64encode(b"wrapped-dek-data").decode(), + "kid": RECOVERY_KID, + } async def mock_read(): return json.dumps(wrapped_dek_data).encode() @@ -758,7 +772,7 @@ async def mock_read(): with patch("s3proxy.crypto.unwrap_key", return_value=original_dek): state = await reconstruct_upload_state_from_s3( - mock_client, "bucket", "key", "upload-123", kek + mock_client, "bucket", "key", "upload-123", _single_key_ring(kek) ) assert state is None @@ -771,7 +785,10 @@ async def test_reconstruct_handles_empty_parts_list(self): kek = b"k" * 32 # Mock successful DEK retrieval - wrapped_dek_data = {"dek": base64.b64encode(b"wrapped-dek-data").decode()} + wrapped_dek_data = { + "dek": base64.b64encode(b"wrapped-dek-data").decode(), + "kid": RECOVERY_KID, + } async def mock_read(): return json.dumps(wrapped_dek_data).encode() @@ -785,7 +802,7 @@ async def mock_read(): with patch("s3proxy.crypto.unwrap_key", return_value=original_dek): state = await reconstruct_upload_state_from_s3( - mock_client, "bucket", "key", "upload-123", kek + mock_client, "bucket", "key", "upload-123", _single_key_ring(kek) ) assert state is not None diff --git a/tests/unit/test_state_recovery_fix.py b/tests/unit/test_state_recovery_fix.py index 6cab43f..56705c1 100644 --- a/tests/unit/test_state_recovery_fix.py +++ b/tests/unit/test_state_recovery_fix.py @@ -12,6 +12,7 @@ import pytest from s3proxy import crypto +from s3proxy.keyring import KeyRing from s3proxy.state import ( MultipartUploadState, PartMetadata, @@ -22,6 +23,14 @@ ENCRYPTION_OVERHEAD = crypto.NONCE_SIZE + crypto.TAG_SIZE # 12 + 16 = 28 +RECOVERY_KID = "AKIA-TEST" + + +def _single_key_ring(kek: bytes) -> KeyRing: + """A KeyRing holding `kek` under the kid the recovery state carries.""" + return KeyRing(keys={RECOVERY_KID: kek}) + + class TestStateReconstruction: """Test state reconstruction from S3.""" @@ -42,7 +51,9 @@ async def test_reconstruct_state_with_multiple_parts(self): from s3proxy.state import json_dumps - state_data = json_dumps({"dek": base64.b64encode(wrapped_dek).decode()}) + state_data = json_dumps( + {"dek": base64.b64encode(wrapped_dek).decode(), "kid": RECOVERY_KID} + ) async def mock_get_object(bucket_name, key_name): return {"Body": AsyncMock(read=AsyncMock(return_value=state_data))} @@ -77,7 +88,9 @@ async def mock_get_object(bucket_name, key_name): ) # Reconstruct state - state = await reconstruct_upload_state_from_s3(mock_s3_client, bucket, key, upload_id, kek) + state = await reconstruct_upload_state_from_s3( + mock_s3_client, bucket, key, upload_id, _single_key_ring(kek) + ) # Verify reconstruction assert state is not None @@ -119,7 +132,9 @@ async def test_reconstruct_state_with_no_parts(self): from s3proxy.state import json_dumps - state_data = json_dumps({"dek": base64.b64encode(wrapped_dek).decode()}) + state_data = json_dumps( + {"dek": base64.b64encode(wrapped_dek).decode(), "kid": RECOVERY_KID} + ) async def mock_get_object(bucket_name, key_name): return {"Body": AsyncMock(read=AsyncMock(return_value=state_data))} @@ -130,7 +145,9 @@ async def mock_get_object(bucket_name, key_name): mock_s3_client.list_parts = AsyncMock(return_value={"Parts": []}) # Reconstruct state - state = await reconstruct_upload_state_from_s3(mock_s3_client, bucket, key, upload_id, kek) + state = await reconstruct_upload_state_from_s3( + mock_s3_client, bucket, key, upload_id, _single_key_ring(kek) + ) # Verify empty state is created assert state is not None @@ -154,7 +171,9 @@ async def mock_get_object(bucket_name, key_name): mock_s3_client.get_object = mock_get_object # Reconstruct state should return None - state = await reconstruct_upload_state_from_s3(mock_s3_client, bucket, key, upload_id, kek) + state = await reconstruct_upload_state_from_s3( + mock_s3_client, bucket, key, upload_id, _single_key_ring(kek) + ) assert state is None @@ -174,7 +193,9 @@ async def test_reconstruct_fails_when_list_parts_fails(self): from s3proxy.state import json_dumps - state_data = json_dumps({"dek": base64.b64encode(wrapped_dek).decode()}) + state_data = json_dumps( + {"dek": base64.b64encode(wrapped_dek).decode(), "kid": RECOVERY_KID} + ) async def mock_get_object(bucket_name, key_name): return {"Body": AsyncMock(read=AsyncMock(return_value=state_data))} @@ -185,7 +206,9 @@ async def mock_get_object(bucket_name, key_name): mock_s3_client.list_parts = AsyncMock(side_effect=Exception("List failed")) # Reconstruct state should return None - state = await reconstruct_upload_state_from_s3(mock_s3_client, bucket, key, upload_id, kek) + state = await reconstruct_upload_state_from_s3( + mock_s3_client, bucket, key, upload_id, _single_key_ring(kek) + ) assert state is None @@ -205,7 +228,9 @@ async def test_reconstruct_with_out_of_order_parts(self): from s3proxy.state import json_dumps - state_data = json_dumps({"dek": base64.b64encode(wrapped_dek).decode()}) + state_data = json_dumps( + {"dek": base64.b64encode(wrapped_dek).decode(), "kid": RECOVERY_KID} + ) async def mock_get_object(bucket_name, key_name): return {"Body": AsyncMock(read=AsyncMock(return_value=state_data))} @@ -240,7 +265,9 @@ async def mock_get_object(bucket_name, key_name): ) # Reconstruct state - state = await reconstruct_upload_state_from_s3(mock_s3_client, bucket, key, upload_id, kek) + state = await reconstruct_upload_state_from_s3( + mock_s3_client, bucket, key, upload_id, _single_key_ring(kek) + ) # Verify all client parts are present assert state is not None diff --git a/tests/unit/test_workflows.py b/tests/unit/test_workflows.py index d4bcdcc..79e0594 100644 --- a/tests/unit/test_workflows.py +++ b/tests/unit/test_workflows.py @@ -25,7 +25,7 @@ class TestPgBackRestWorkflow: """ @pytest.mark.asyncio - async def test_full_backup_workflow(self, mock_s3, settings): + async def test_full_backup_workflow(self, mock_s3, settings, kek): """Test complete backup workflow.""" bucket = "pgbackrest-repo" @@ -36,7 +36,7 @@ async def test_full_backup_workflow(self, mock_s3, settings): manifest_content = ( b'{"backup_label": "20240115-120000F", "start_time": "2024-01-15 12:00:00"}' ) - manifest_encrypted = crypto.encrypt_object(manifest_content, settings.kek) + manifest_encrypted = crypto.encrypt_object(manifest_content, kek) await mock_s3.put_object( bucket, "backup/20240115-120000F/backup.manifest", @@ -49,7 +49,7 @@ async def test_full_backup_workflow(self, mock_s3, settings): # 3. Upload WAL segment (medium file ~16MB simulated with smaller data) wal_content = b"WAL segment data " * 1000 - wal_encrypted = crypto.encrypt_object(wal_content, settings.kek) + wal_encrypted = crypto.encrypt_object(wal_content, kek) await mock_s3.put_object( bucket, "archive/000000010000000000000001", @@ -70,11 +70,11 @@ async def test_full_backup_workflow(self, mock_s3, settings): get_resp = await mock_s3.get_object(bucket, "backup/20240115-120000F/backup.manifest") ciphertext = await get_resp["Body"].read() wrapped_dek = base64.b64decode(get_resp["Metadata"][settings.dektag_name]) - decrypted = crypto.decrypt_object(ciphertext, wrapped_dek, settings.kek) + decrypted = crypto.decrypt_object(ciphertext, wrapped_dek, kek) assert decrypted == manifest_content @pytest.mark.asyncio - async def test_retention_cleanup(self, mock_s3, settings): + async def test_retention_cleanup(self, mock_s3, settings, kek): """Test batch delete for retention policy cleanup.""" bucket = "pgbackrest-repo" await mock_s3.create_bucket(bucket) @@ -82,7 +82,7 @@ async def test_retention_cleanup(self, mock_s3, settings): # Create old backup files for i in range(5): content = f"old backup {i}".encode() - encrypted = crypto.encrypt_object(content, settings.kek) + encrypted = crypto.encrypt_object(content, kek) await mock_s3.put_object( bucket, f"backup/old-backup-{i}/data.bin", @@ -115,14 +115,14 @@ class TestWALGWorkflow: """ @pytest.mark.asyncio - async def test_delta_backup_with_copy(self, mock_s3, settings): + async def test_delta_backup_with_copy(self, mock_s3, settings, kek): """Test delta backup using copy object.""" bucket = "walg-repo" await mock_s3.create_bucket(bucket) # Upload base backup base_content = b"base backup data " * 100 - base_encrypted = crypto.encrypt_object(base_content, settings.kek) + base_encrypted = crypto.encrypt_object(base_content, kek) await mock_s3.put_object( bucket, "basebackups/base_000000010000000000000001/data.tar", @@ -146,7 +146,7 @@ async def test_delta_backup_with_copy(self, mock_s3, settings): assert len(keys) == 2 @pytest.mark.asyncio - async def test_wal_archiving(self, mock_s3, settings): + async def test_wal_archiving(self, mock_s3, settings, kek): """Test WAL file archiving.""" bucket = "walg-repo" await mock_s3.create_bucket(bucket) @@ -154,7 +154,7 @@ async def test_wal_archiving(self, mock_s3, settings): # Archive multiple WAL files for segment_num in range(5): wal_content = f"WAL segment {segment_num:08d}".encode() * 100 - wal_encrypted = crypto.encrypt_object(wal_content, settings.kek) + wal_encrypted = crypto.encrypt_object(wal_content, kek) await mock_s3.put_object( bucket, f"wal_005/00000001000000000000000{segment_num}", @@ -183,7 +183,7 @@ class TestScyllaManagerWorkflow: """ @pytest.mark.asyncio - async def test_sstable_multipart_upload(self, mock_s3, settings): + async def test_sstable_multipart_upload(self, mock_s3, settings, kek): """Test multipart upload for large SSTables.""" bucket = "scylla-backup" key = "keyspace/table/mc-1-big-Data.db" @@ -195,7 +195,7 @@ async def test_sstable_multipart_upload(self, mock_s3, settings): # Generate encryption key for this upload dek = crypto.generate_dek() - crypto.wrap_key(dek, settings.kek) + crypto.wrap_key(dek, kek) # Upload parts (simulated SSTable chunks) part_etags = [] @@ -265,7 +265,7 @@ class TestClickHouseBackupWorkflow: """ @pytest.mark.asyncio - async def test_backup_metadata(self, mock_s3, settings): + async def test_backup_metadata(self, mock_s3, settings, kek): """Test backing up ClickHouse metadata.""" bucket = "clickhouse-backup" await mock_s3.create_bucket(bucket) @@ -277,7 +277,7 @@ async def test_backup_metadata(self, mock_s3, settings): "size": 1024000, } metadata_bytes = str(metadata).encode() - encrypted = crypto.encrypt_object(metadata_bytes, settings.kek) + encrypted = crypto.encrypt_object(metadata_bytes, kek) await mock_s3.put_object( bucket, @@ -306,7 +306,7 @@ class TestElasticsearchSnapshotWorkflow: """ @pytest.mark.asyncio - async def test_snapshot_to_s3(self, mock_s3, settings): + async def test_snapshot_to_s3(self, mock_s3, settings, kek): """Test creating an ES snapshot to S3.""" bucket = "es-snapshots" await mock_s3.create_bucket(bucket) @@ -317,7 +317,7 @@ async def test_snapshot_to_s3(self, mock_s3, settings): # Upload snapshot metadata snapshot_metadata = b'{"snapshot_id": "snapshot_1", "indices": ["logs-2024.01"]}' - encrypted = crypto.encrypt_object(snapshot_metadata, settings.kek) + encrypted = crypto.encrypt_object(snapshot_metadata, kek) await mock_s3.put_object( bucket, @@ -330,7 +330,7 @@ async def test_snapshot_to_s3(self, mock_s3, settings): # Upload index data index_data = b"lucene index data " * 1000 - index_encrypted = crypto.encrypt_object(index_data, settings.kek) + index_encrypted = crypto.encrypt_object(index_data, kek) await mock_s3.put_object( bucket, @@ -350,7 +350,7 @@ class TestBarmanCloudWorkflow: """Simulate Barman Cloud (CloudNativePG) backup workflow.""" @pytest.mark.asyncio - async def test_base_backup(self, mock_s3, settings): + async def test_base_backup(self, mock_s3, settings, kek): """Test Barman Cloud base backup.""" bucket = "barman-backup" await mock_s3.create_bucket(bucket) @@ -360,7 +360,7 @@ async def test_base_backup(self, mock_s3, settings): # Upload backup label backup_label = b"START WAL LOCATION: 0/1000000 (file 000000010000000000000001)" - encrypted = crypto.encrypt_object(backup_label, settings.kek) + encrypted = crypto.encrypt_object(backup_label, kek) await mock_s3.put_object( bucket, @@ -373,7 +373,7 @@ async def test_base_backup(self, mock_s3, settings): # Upload data directory tarball data_tar = b"postgres data directory contents " * 1000 - data_encrypted = crypto.encrypt_object(data_tar, settings.kek) + data_encrypted = crypto.encrypt_object(data_tar, kek) await mock_s3.put_object( bucket, @@ -395,14 +395,14 @@ class TestEncryptionKeyRotation: """Test scenarios involving key rotation.""" @pytest.mark.asyncio - async def test_read_with_original_key(self, mock_s3, settings): + async def test_read_with_original_key(self, mock_s3, settings, kek): """Test reading data encrypted with original key still works.""" bucket = "key-rotation-test" await mock_s3.create_bucket(bucket) # Encrypt with current key plaintext = b"data encrypted before key rotation" - encrypted = crypto.encrypt_object(plaintext, settings.kek) + encrypted = crypto.encrypt_object(plaintext, kek) await mock_s3.put_object( bucket, @@ -418,7 +418,7 @@ async def test_read_with_original_key(self, mock_s3, settings): ciphertext = await get_resp["Body"].read() wrapped_dek = base64.b64decode(get_resp["Metadata"][settings.dektag_name]) - decrypted = crypto.decrypt_object(ciphertext, wrapped_dek, settings.kek) + decrypted = crypto.decrypt_object(ciphertext, wrapped_dek, kek) assert decrypted == plaintext