diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 8cff2362..0727ad8a 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -26,7 +26,6 @@ env: CLOUDFRONT_KEY_ID: ${{ secrets.CLOUDFRONT_KEY_ID }} CLOUDFRONT_PRIVATE_KEY: ${{ secrets.CLOUDFRONT_PRIVATE_KEY }} CLOUDFRONT_DOMAIN: ${{ secrets.CLOUDFRONT_DOMAIN }} - CLOUDFRONT_PRIVATE_KEY_PATH: /tmp/dummy_key.pem ONSHAPE_API: ${{ secrets.ONSHAPE_API }} ONSHAPE_ACCESS_KEY: ${{ secrets.ONSHAPE_ACCESS_KEY }} ONSHAPE_SECRET_KEY: ${{ secrets.ONSHAPE_SECRET_KEY }} diff --git a/store/app/routers/artifacts.py b/store/app/routers/artifacts.py index 56991ea3..bdbf1593 100644 --- a/store/app/routers/artifacts.py +++ b/store/app/routers/artifacts.py @@ -22,7 +22,6 @@ can_write_listing, check_content_type, get_artifact_type, - get_artifact_url, get_artifact_urls, ) from store.app.security.user import ( @@ -65,30 +64,19 @@ async def artifact_url( # Initialize CloudFront signer signer = CloudFrontUrlSigner( key_id=settings.cloudfront.key_id, - private_key_path=settings.cloudfront.private_key_path, + private_key=settings.cloudfront.private_key, ) - # Generate base URL based on environment - if settings.environment == "local": - base_url = get_artifact_url( - artifact_type=artifact.artifact_type, - artifact_id=artifact.id, - listing_id=listing_id, - name=s3_filename, - size=size, - ) - else: - # For production, use CloudFront domain - base_url = f"https://{settings.cloudfront.domain}/{artifact.artifact_type}/{listing_id}/{s3_filename}" - if size and artifact.artifact_type == "image": - base_url = f"{base_url}_{size}" + # Always use CloudFront domain and sign the URL + base_url = f"https://{settings.cloudfront.domain}/{artifact.artifact_type}/{listing_id}/{s3_filename}" + if size and artifact.artifact_type == "image": + base_url = f"{base_url}_{size}" - # Create and sign URL for production environment - if settings.environment != "local": - policy = signer.create_custom_policy(url=base_url, expire_days=180) - base_url = signer.generate_presigned_url(base_url, policy=policy) + # Create and sign URL + policy = signer.create_custom_policy(url=base_url, expire_days=180) + signed_url = signer.generate_presigned_url(base_url, policy=policy) - return RedirectResponse(url=base_url) + return RedirectResponse(url=signed_url) class ArtifactUrls(BaseModel): @@ -107,18 +95,24 @@ def get_artifact_url_response(artifact: Artifact) -> ArtifactUrls: signer = CloudFrontUrlSigner( key_id=settings.cloudfront.key_id, - private_key_path=settings.cloudfront.private_key_path, + private_key=settings.cloudfront.private_key, ) expire_days = 180 expiration_time = int((datetime.utcnow() + timedelta(days=expire_days)).timestamp()) - # Explicitly iterate over the literal types sizes: list[Literal["small", "large"]] = ["small", "large"] for size in sizes: try: - url = artifact_urls[size] - cf_url = f"https://{settings.cloudfront.domain}/{url}" + cf_url = ( + f"https://{settings.cloudfront.domain}/{artifact.artifact_type}/{artifact.listing_id}/{artifact.id}" + ) + if size == "small": + cf_url += "_small_256x256" + elif size == "large": + cf_url += "_large_1536x1536" + cf_url += f"_{artifact.name}" + policy = signer.create_custom_policy(url=cf_url, expire_days=expire_days) artifact_urls[size] = signer.generate_presigned_url(cf_url, policy=policy) except KeyError: diff --git a/store/app/utils/cloudfront_signer.py b/store/app/utils/cloudfront_signer.py index cee8c05d..aba700d6 100644 --- a/store/app/utils/cloudfront_signer.py +++ b/store/app/utils/cloudfront_signer.py @@ -16,14 +16,14 @@ class CloudFrontUrlSigner: """A class to generate signed URLs for AWS CloudFront using RSA keys.""" - def __init__(self, key_id: str, private_key_path: str) -> None: - """Initialize the CloudFrontUrlSigner with a key ID and the path to the private key file. + def __init__(self, key_id: str, private_key: str) -> None: + """Initialize the CloudFrontUrlSigner with a key ID and private key content. :param key_id: The CloudFront key ID associated with the public key in your CloudFront key group. - :param private_key_path: The path to the private key PEM file. + :param private_key: The private key content in PEM format. """ self.key_id = key_id - self.private_key_path = private_key_path + self.private_key = private_key self.cf_signer = CloudFrontSigner(key_id, self._rsa_signer) def _rsa_signer(self, message: bytes) -> bytes: @@ -38,16 +38,15 @@ def _rsa_signer(self, message: bytes) -> bytes: Raises: ValueError: If the loaded key is not an RSA private key. """ - with open(self.private_key_path, "rb") as key_file: - private_key = serialization.load_pem_private_key( - key_file.read(), - password=None, - ) + private_key = serialization.load_pem_private_key( + self.private_key.encode("utf-8"), + password=None, + ) - if not isinstance(private_key, RSAPrivateKey): - raise ValueError("The provided key is not an RSA private key") + if not isinstance(private_key, RSAPrivateKey): + raise ValueError("The provided key is not an RSA private key") - return private_key.sign(message, padding.PKCS1v15(), hashes.SHA1()) + return private_key.sign(message, padding.PKCS1v15(), hashes.SHA1()) def generate_presigned_url(self, url: str, policy: Optional[str] = None) -> str: """Generate a presigned URL for CloudFront using an optional custom policy. diff --git a/store/settings/environment.py b/store/settings/environment.py index 1e534251..90990845 100644 --- a/store/settings/environment.py +++ b/store/settings/environment.py @@ -75,7 +75,7 @@ class StripeSettings: class CloudFrontSettings: domain: str = field(default=II("oc.env:CLOUDFRONT_DOMAIN")) key_id: str = field(default=II("oc.env:CLOUDFRONT_KEY_ID")) - private_key_path: str = field(default=II("oc.env:CLOUDFRONT_PRIVATE_KEY_PATH")) + private_key: str = field(default=II("oc.env:CLOUDFRONT_PRIVATE_KEY")) @dataclass