Source code for multistorageclient.signers.cloudfront

  1# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
  2# SPDX-License-Identifier: Apache-2.0
  3#
  4# Licensed under the Apache License, Version 2.0 (the "License");
  5# you may not use this file except in compliance with the License.
  6# You may obtain a copy of the License at
  7#
  8# http://www.apache.org/licenses/LICENSE-2.0
  9#
 10# Unless required by applicable law or agreed to in writing, software
 11# distributed under the License is distributed on an "AS IS" BASIS,
 12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 13# See the License for the specific language governing permissions and
 14# limitations under the License.
 15
 16from __future__ import annotations
 17
 18import base64
 19import json
 20from datetime import datetime, timedelta, timezone
 21from typing import Any
 22
 23try:
 24    from cryptography.hazmat.primitives.asymmetric.padding import PKCS1v15
 25    from cryptography.hazmat.primitives.hashes import SHA1
 26    from cryptography.hazmat.primitives.serialization import load_pem_private_key
 27except ImportError:
 28    PKCS1v15 = None  # type: ignore[assignment, misc]
 29    SHA1 = None  # type: ignore[assignment, misc]
 30    load_pem_private_key = None  # type: ignore[assignment]
 31
 32from .base import URLSigner
 33
 34DEFAULT_CLOUDFRONT_EXPIRES_IN = 3600
 35
 36

[docs]
 37class CloudFrontURLSigner(URLSigner):
 38    """
 39    Generates CloudFront signed URLs using an RSA key pair.
 40
 41    Implements the CloudFront canned-policy signing spec directly so that it
 42    has no dependency on ``botocore`` — only the ``cryptography`` package is
 43    required (RSA-SHA1 / PKCS1v15).
 44    """
 45
 46    _key_pair_id: str
 47    _private_key_path: str
 48    _domain: str
 49    _expires_in: int
 50    _origin_path: str
 51    _origin_prefix: str
 52
 53    def __init__(
 54        self,
 55        *,
 56        key_pair_id: str,
 57        private_key_path: str,
 58        domain: str,
 59        expires_in: int = DEFAULT_CLOUDFRONT_EXPIRES_IN,
 60        origin_path: str = "",
 61        **_kwargs: Any,
 62    ) -> None:
 63        if load_pem_private_key is None:
 64            raise ImportError(
 65                "The 'cryptography' package is required for CloudFront URL signing. "
 66                "Install it with: pip install 'multi-storage-client[cloudfront]'"
 67            )
 68        self._key_pair_id = key_pair_id
 69        self._private_key_path = private_key_path
 70        self._domain = domain.rstrip("/")
 71        self._expires_in = expires_in
 72        self._origin_path = origin_path.strip("/")
 73        self._origin_prefix = self._origin_path + "/" if self._origin_path else ""
 74        self._private_key: Any = None
 75
 76    def _get_private_key(self) -> Any:
 77        if self._private_key is None:
 78            if load_pem_private_key is None:
 79                raise ImportError("cryptography package is required for CloudFront signed URLs")
 80            with open(self._private_key_path, "rb") as f:
 81                self._private_key = load_pem_private_key(f.read(), password=None)
 82        return self._private_key
 83

[docs]
 84    def generate_presigned_url(self, path: str, *, method: str = "GET") -> str:
 85        private_key = self._get_private_key()
 86
 87        effective = path.lstrip("/")
 88        if self._origin_prefix:
 89            if effective.startswith(self._origin_prefix):
 90                effective = effective[len(self._origin_prefix) :]
 91            else:
 92                raise ValueError(
 93                    f"Object path {path} does not start with CloudFront origin path {self._origin_path}. "
 94                    "Ensure the 'origin_path' option matches the distribution's configured origin path."
 95                )
 96
 97        url = f"https://{self._domain}/{effective}"
 98        expiry = datetime.now(timezone.utc) + timedelta(seconds=self._expires_in)
 99        epoch = int(expiry.timestamp())
100
101        policy = json.dumps(
102            {"Statement": [{"Resource": url, "Condition": {"DateLessThan": {"AWS:EpochTime": epoch}}}]},
103            separators=(",", ":"),
104        )
105
106        signature = private_key.sign(policy.encode("utf-8"), PKCS1v15(), SHA1())  # type: ignore[union-attr]
107
108        encoded_sig = _cf_b64encode(signature)
109        separator = "&" if "?" in url else "?"
110        return f"{url}{separator}Expires={epoch}&Signature={encoded_sig}&Key-Pair-Id={self._key_pair_id}"


111
112
113def _cf_b64encode(data: bytes) -> str:
114    """CloudFront URL-safe base64: ``+`` → ``-``, ``=`` → ``_``, ``/`` → ``~``.
115
116    Standard base64 characters ``+``, ``=``, and ``/`` are reserved in URLs;
117    CloudFront requires this substitution for signed URL query parameters.
118    See https://docs.aws.amazon.com/AmazonCloudFront/latest/DeveloperGuide/private-content-creating-signed-url-canned-policy.html
119    """
120    return base64.b64encode(data).decode("ascii").replace("+", "-").replace("=", "_").replace("/", "~")