Add CloudFront signed URL support (#531)

Author: shayancoin

backend/README.md

@@ -30,6 +30,10 @@ The backend expects the following environment variables (or entries in `.env.dev
 - `SHOPIFY_STORE` – Shopify store subdomain used to build Admin and Storefront API URLs.
 - `SHOPIFY_TOKEN` – Private access token used for Shopify Admin REST and Storefront GraphQL calls.
 - `SHOPIFY_WEBHOOK_SECRET` – Secret shared with Shopify to verify webhook signatures.
+- `S3_USE_CLOUDFRONT_SIGNER` – Enable to serve assets via CloudFront using signed URLs.
+- `CLOUDFRONT_KEY_PAIR_ID` – AWS CloudFront key pair identifier used for signing URLs.
+- `CLOUDFRONT_PRIVATE_KEY` – PEM-encoded private key that pairs with the CloudFront key pair ID.
+- `CLOUDFRONT_SIGNED_URL_TTL_SECONDS` – Optional override for the CloudFront signed URL expiration (defaults to one hour).
 
 ## Project Structure
 

backend/api/config.py

@@ -6,7 +6,7 @@
 from functools import lru_cache
 from typing import Any
 
-from pydantic import Field, field_validator
+from pydantic import Field, field_validator, model_validator
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
 
@@ -123,6 +123,24 @@ class Settings(BaseSettings):
         default="cnc-exports",
         description="Object key prefix applied to uploaded CNC artifacts.",
     )
+    s3_use_cloudfront_signer: bool = Field(
+        default=False,
+        description="When true, generate signed URLs for the public S3 base URL via CloudFront.",
+    )
+    cloudfront_key_pair_id: str | None = Field(
+        default=None,
+        description="CloudFront key pair identifier used to sign URLs.",
+    )
+    cloudfront_private_key: str | None = Field(
+        default=None,
+        description="PEM-encoded private key associated with the CloudFront key pair.",
+        repr=False,
+    )
+    cloudfront_signed_url_ttl_seconds: int = Field(
+        default=3600,
+        description="Default expiration, in seconds, for signed CloudFront URLs.",
+        ge=1,
+    )
 
     model_config = SettingsConfigDict(
         env_file=".env.development",
@@ -176,6 +194,7 @@ def _normalize_api_write_token(cls, value: str | None) -> str | None:
         "supabase_anon_key",
         "supabase_service_role_key",
         "supabase_jwt_audience",
+        "cloudfront_key_pair_id",
     )
     @classmethod
     def _trim_optional_str(cls, value: str | None) -> str | None:
@@ -184,6 +203,33 @@ def _trim_optional_str(cls, value: str | None) -> str | None:
         value = value.strip()
         return value or None
 
+    @field_validator("cloudfront_private_key")
+    @classmethod
+    def _normalize_cloudfront_private_key(cls, value: str | None) -> str | None:
+        if value is None:
+            return None
+        value = value.strip()
+        if not value:
+            return None
+        return value.replace("\\n", "\n")
+
+    @model_validator(mode="after")
+    def _validate_cloudfront_config(self) -> "Settings":
+        if self.s3_use_cloudfront_signer:
+            if not self.s3_public_base_url:
+                raise ValueError(
+                    "S3_USE_CLOUDFRONT_SIGNER requires S3_PUBLIC_BASE_URL to be set."
+                )
+            if not self.cloudfront_key_pair_id:
+                raise ValueError(
+                    "CLOUDFRONT_KEY_PAIR_ID must be provided when CloudFront signing is enabled."
+                )
+            if not self.cloudfront_private_key:
+                raise ValueError(
+                    "CLOUDFRONT_PRIVATE_KEY must be provided when CloudFront signing is enabled."
+                )
+        return self
+
 
 @lru_cache(maxsize=1)
 def _settings_cache(secret: str | None, shopify_secret: str | None) -> Settings:

backend/api/storage_dependencies.py

@@ -5,7 +5,11 @@
 from fastapi import Depends
 
 from api.config import Settings, get_settings
-from backend.services.storage import StorageClient, create_storage
+from backend.services.storage import (
+    CloudFrontSigningConfig,
+    StorageClient,
+    create_storage,
+)
 
 
 def get_storage_client(settings: Settings = Depends(get_settings)) -> StorageClient:
@@ -14,6 +18,14 @@ def get_storage_client(settings: Settings = Depends(get_settings)) -> StorageCli
     return create_storage(
         bucket=settings.s3_bucket_name,
         base_url=settings.s3_public_base_url,
+        cloudfront_signing=
+        CloudFrontSigningConfig(
+            key_pair_id=settings.cloudfront_key_pair_id,
+            private_key=settings.cloudfront_private_key,
+            url_ttl_seconds=settings.cloudfront_signed_url_ttl_seconds,
+        )
+        if settings.s3_use_cloudfront_signer
+        else None,
     )
 
 

backend/services/storage.py

@@ -3,13 +3,19 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
+from datetime import datetime, timedelta, timezone
 from typing import BinaryIO, Dict, Protocol
 
 try:  # pragma: no cover - optional dependency
     import boto3
 except Exception:  # pragma: no cover - dependency may be absent in tests
     boto3 = None  # type: ignore[assignment]
 
+try:  # pragma: no cover - optional dependency
+    from botocore.signers import CloudFrontSigner
+except Exception:  # pragma: no cover - dependency may be absent in tests
+    CloudFrontSigner = None  # type: ignore[assignment]
+
 
 @dataclass(frozen=True)
 class StoredObject:
@@ -19,6 +25,19 @@ class StoredObject:
     content_type: str | None = None
 
 
+@dataclass(frozen=True)
+class CloudFrontSigningConfig:
+    """Configuration describing how to sign CloudFront URLs."""
+
+    key_pair_id: str
+    private_key: str
+    url_ttl_seconds: int = 3600
+
+    def __post_init__(self) -> None:  # pragma: no cover - trivial validation
+        if self.url_ttl_seconds <= 0:
+            raise ValueError("CloudFront signed URL TTL must be positive")
+
+
 class StorageClient(Protocol):
     """Minimal protocol for uploading file-like objects."""
 
@@ -71,12 +90,16 @@ def __init__(
         *,
         client: "boto3.client" | None = None,
         base_url: str | None = None,
+        cloudfront_signer: "CloudFrontSigner" | None = None,
+        signed_url_ttl_seconds: int | None = None,
     ) -> None:
         if boto3 is None:  # pragma: no cover - only triggered when boto3 missing
             raise RuntimeError("boto3 is required for S3 uploads but is not installed")
         self._bucket = bucket
         self._client = client or boto3.client("s3")
         self._base_url = base_url.rstrip("/") if base_url else None
+        self._cloudfront_signer = cloudfront_signer
+        self._signed_url_ttl_seconds = signed_url_ttl_seconds
 
     def upload_fileobj(
         self,
@@ -94,7 +117,14 @@ def upload_fileobj(
 
     def _build_url(self, key: str) -> str:
         if self._base_url:
-            return f"{self._base_url}/{key}"
+            url = f"{self._base_url}/{key}"
+            if self._cloudfront_signer:
+                expires_in = self._signed_url_ttl_seconds or 3600
+                expiration = datetime.now(timezone.utc) + timedelta(seconds=expires_in)
+                return self._cloudfront_signer.generate_presigned_url(
+                    url, date_less_than=expiration
+                )
+            return url
         region = getattr(self._client.meta, "region_name", "us-east-1")
         return f"https://{self._bucket}.s3.{region}.amazonaws.com/{key}"
 
@@ -103,12 +133,26 @@ def create_storage(
     *,
     bucket: str | None,
     base_url: str | None = None,
+    cloudfront_signing: CloudFrontSigningConfig | None = None,
 ) -> StorageClient:
     """Instantiate the preferred storage backend."""
 
     if bucket:
         try:
-            return S3Storage(bucket=bucket, base_url=base_url)
+            cloudfront_signer = None
+            signed_url_ttl_seconds = None
+            if cloudfront_signing and base_url:
+                cloudfront_signer = _create_cloudfront_signer(
+                    key_pair_id=cloudfront_signing.key_pair_id,
+                    private_key_pem=cloudfront_signing.private_key,
+                )
+                signed_url_ttl_seconds = cloudfront_signing.url_ttl_seconds
+            return S3Storage(
+                bucket=bucket,
+                base_url=base_url,
+                cloudfront_signer=cloudfront_signer,
+                signed_url_ttl_seconds=signed_url_ttl_seconds,
+            )
         except RuntimeError:
             # Fall back to in-memory storage when boto3 is unavailable.
             return InMemoryStorage(bucket=bucket, base_url=base_url)
@@ -120,5 +164,32 @@ def create_storage(
     "StorageClient",
     "InMemoryStorage",
     "S3Storage",
+    "CloudFrontSigningConfig",
     "create_storage",
 ]
+
+
+def _create_cloudfront_signer(
+    *, key_pair_id: str, private_key_pem: str
+) -> "CloudFrontSigner" | None:
+    """Instantiate a CloudFront signer from the provided credentials."""
+
+    if CloudFrontSigner is None:  # pragma: no cover - dependency may be absent
+        raise RuntimeError("botocore is required for CloudFront signing but is not installed")
+
+    try:  # pragma: no cover - dependency may be absent in tests
+        from cryptography.hazmat.primitives import hashes, serialization
+        from cryptography.hazmat.primitives.asymmetric import padding
+    except Exception as exc:  # pragma: no cover - dependency may be absent
+        raise RuntimeError(
+            "cryptography is required for CloudFront signing but is not installed"
+        ) from exc
+
+    private_key = serialization.load_pem_private_key(
+        private_key_pem.encode("utf-8"), password=None
+    )
+
+    def _rsa_signer(message: bytes) -> bytes:
+        return private_key.sign(message, padding.PKCS1v15(), hashes.SHA1())
+
+    return CloudFrontSigner(key_pair_id, _rsa_signer)

backend/tests/test_config.py

@@ -19,3 +19,23 @@ def test_get_settings_cache_clear(monkeypatch):
     refreshed_settings = get_settings()
     assert refreshed_settings is not settings_initial
     assert refreshed_settings.api_write_token == "updated"
+
+
+def test_cloudfront_private_key_normalization(monkeypatch):
+    monkeypatch.setenv("HYGRAPH_WEBHOOK_SECRET", "secret")
+    monkeypatch.setenv("API_WRITE_TOKEN", "token")
+    monkeypatch.setenv("S3_USE_CLOUDFRONT_SIGNER", "true")
+    monkeypatch.setenv("S3_PUBLIC_BASE_URL", "https://d111111abcdef8.cloudfront.net")
+    monkeypatch.setenv("CLOUDFRONT_KEY_PAIR_ID", "K123")
+    monkeypatch.setenv(
+        "CLOUDFRONT_PRIVATE_KEY",
+        "-----BEGIN PRIVATE KEY-----\\nline1\\nline2\\n-----END PRIVATE KEY-----",
+    )
+
+    get_settings.cache_clear()
+    settings = get_settings()
+
+    assert "\n" in settings.cloudfront_private_key
+    assert settings.s3_use_cloudfront_signer is True
+
+    get_settings.cache_clear()

backend/tests/test_storage.py

@@ -0,0 +1,93 @@
+import datetime
+from types import SimpleNamespace
+
+import pytest
+
+from backend.services import storage
+
+
+class DummyS3Client:
+    def __init__(self, region: str = "us-west-2") -> None:
+        self.meta = SimpleNamespace(region_name=region)
+
+    def upload_fileobj(self, *_args, **_kwargs) -> None:  # pragma: no cover - unused
+        return None
+
+
+class DummySigner:
+    def __init__(self) -> None:
+        self.calls: list[tuple[str, datetime.datetime]] = []
+
+    def generate_presigned_url(self, url: str, date_less_than: datetime.datetime) -> str:
+        self.calls.append((url, date_less_than))
+        return f"signed:{url}"
+
+
+@pytest.fixture(autouse=True)
+def fake_boto3(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(
+        storage,
+        "boto3",
+        SimpleNamespace(client=lambda _service: DummyS3Client()),
+    )
+
+
+def test_s3_storage_cloudfront_signing() -> None:
+    signer = DummySigner()
+    client = DummyS3Client()
+    s3_storage = storage.S3Storage(
+        bucket="example-bucket",
+        client=client,
+        base_url="https://d111111abcdef8.cloudfront.net",
+        cloudfront_signer=signer,
+        signed_url_ttl_seconds=600,
+    )
+
+    url = s3_storage._build_url("folder/file.txt")
+
+    assert url == "signed:https://d111111abcdef8.cloudfront.net/folder/file.txt"
+    assert len(signer.calls) == 1
+    called_url, expires_at = signer.calls[0]
+    assert called_url.endswith("folder/file.txt")
+    assert isinstance(expires_at, datetime.datetime)
+    assert expires_at.tzinfo is not None
+    remaining = (expires_at - datetime.datetime.now(tz=datetime.timezone.utc)).total_seconds()
+    assert 0 < remaining <= 600
+
+
+def test_s3_storage_base_url_without_signer() -> None:
+    client = DummyS3Client(region="us-east-1")
+    s3_storage = storage.S3Storage(
+        bucket="example-bucket",
+        client=client,
+        base_url="https://assets.example.com",
+    )
+
+    url = s3_storage._build_url("a/b.txt")
+
+    assert url == "https://assets.example.com/a/b.txt"
+
+
+def test_create_storage_with_cloudfront_config(monkeypatch: pytest.MonkeyPatch) -> None:
+    dummy_signer = DummySigner()
+
+    def fake_create_signer(*_args, **_kwargs):
+        return dummy_signer
+
+    monkeypatch.setattr(storage, "_create_cloudfront_signer", fake_create_signer)
+
+    config = storage.CloudFrontSigningConfig(
+        key_pair_id="K12345",
+        private_key="dummy",
+        url_ttl_seconds=120,
+    )
+
+    client = storage.create_storage(
+        bucket="example-bucket",
+        base_url="https://d111111abcdef8.cloudfront.net",
+        cloudfront_signing=config,
+    )
+
+    assert isinstance(client, storage.S3Storage)
+    assert client._cloudfront_signer is dummy_signer
+    assert client._signed_url_ttl_seconds == 120