Credentials cleanup and documentation

README.md

@@ -64,14 +64,26 @@ The following example assumes you have a [MinIO](https://min.io/) instance runni
 Note that you shouldn't hardcode credentials, this is merely for illustration purposes.
 
 ```python
+from fast_s3_signer import FastS3UrlSigner, Credentials
+
 signer = FastS3UrlSigner(
     bucket_endpoint_url="http://localhost:9000/my-bucket/",
     aws_region="us-east-1",
-    access_key_id="minioadmin",
-    secret_access_key="minioadmin",
+    credentials=Credentials(
+        access_key="minioadmin",
+        secret_key="minioadmin",
+    ),
 )
 ```
 
+### :warning: Note on instances created from boto3 and aiobotocore
+
+The boto3 and aiobotocore clients may, when they are configured with an STS Security Token, refresh their credentials when generating signed URLs. When a `FastS3UrlSigner` is created with these clients, their credentials are refreshed so that the signer has access to the current credentials, **but it won't refresh them after it's created**. The signer doesn't even keep a reference to the client you use to create it!
+
+This means that if your aiobotocore/boto3 clients happen to have expiring tokens in them, URLs created by `FastS3UrlSigner` may suddently stop working. Therefore, the best way to use a `FastS3UrlSigner` created from a client is to create it in a function-local scope, use it, and then have it be destroyed by GC when you no longer need it.
+
+If you create your signer instances with non-expiring credentials, you can safely disregard this warning.
+
 ## Local development
 
 This project uses [Poetry](https://python-poetry.org/) for dependency management. Make sure you have it installed.

fast_s3_url/__init__.py

@@ -1,3 +1,4 @@
 from .s3_signer import FastS3UrlSigner
+from .credentials import Credentials, CredentialsWithToken
 
-__all__ = ["FastS3UrlSigner"]
+__all__ = ["FastS3UrlSigner", "Credentials", "CredentialsWithToken"]

fast_s3_url/credentials.py

@@ -0,0 +1,25 @@
+from dataclasses import dataclass
+
+__all__ = ["Credentials", "CredentialsWithToken"]
+
+
+@dataclass(slots=True, frozen=True)
+class Credentials:
+    access_key: str
+    """
+    The Access Key ID. Usually mapped to the `AWS_ACCESS_KEY_ID` environment variable.
+    """
+    secret_key: str
+    """
+    The Secret Access Key. Usually mapped to the `AWS_SECRET_ACCESS_KEY` environment variable.
+    """
+
+
+@dataclass(slots=True, frozen=True)
+class CredentialsWithToken(Credentials):
+    session_token: str
+    """
+    An STS session token. Usually mapped to the `AWS_SESSION_TOKEN` environment variable.
+
+    Note that these tokens usually expire, so you must refresh them yourself.
+    """

fast_s3_url/s3_signer.py

@@ -1,4 +1,5 @@
 from __future__ import annotations
+
 from datetime import datetime, timezone
 import hashlib
 import hmac
@@ -7,6 +8,12 @@
 
 from typing import TYPE_CHECKING, List, Optional
 
+from fast_s3_url.credentials import Credentials, CredentialsWithToken
+from fast_s3_url.util import (
+    get_credentials_from_aiobotocore_client,
+    get_credentials_from_boto3_client,
+)
+
 
 if TYPE_CHECKING:
     from types_aiobotocore_s3 import S3Client as AioBotocoreS3Client
@@ -15,12 +22,6 @@
 __all__ = ["FastS3UrlSigner"]
 
 
-class _Credentials:
-    access_key: str
-    secret_key: str
-    token: Optional[str]
-
-
 class FastS3UrlSigner:
     """
     Alternative signed URL generator for S3, compliant with the s3v4 format,
@@ -32,19 +33,16 @@ def __init__(
         self,
         *,
         bucket_endpoint_url: str,
-        access_key_id: str,
-        secret_access_key: str,
+        credentials: Credentials,
         aws_region: Optional[str] = None,
-        session_token: Optional[str] = None,
     ) -> None:
         """
         Create an instance of this object with the specified parameters.
 
         :param bucket_endpoint_url: The URL to a bucket. Can be virtual-style, e.g.: 'https://my-bucket.s3.amazonaws.com/' or path-style, e.g.: 'https://s3.amazonaws.com/my-bucket/'.
-        :param access_key_id: The Access Key ID. Usually mapped to the `AWS_ACCESS_KEY_ID` environment variable.
-        :param secret_access_key: The Secret Access Key. Usually mapped to the `AWS_SECRET_ACCESS_KEY` environment variable.
         :param aws_region: The AWS region that will be used to generate URLs with. Defaults to 'us-east-1'.
-        :param session_token: An STS session token. Usually mapped to the `AWS_SESSION_TOKEN` environment variable.
+        :param credentials: Credentials to the S3 bucket.
+        :param session_token:
         """
 
         # Get the components from the bucket's endpoint URL.
@@ -57,9 +55,7 @@ def __init__(
         self.bucket_host = parsed_bucket_url.netloc
 
         self.region = aws_region or "us-east-1"
-        self.secret_key = secret_access_key
-        self.access_key = access_key_id
-        self.session_token = session_token
+        self.credentials = credentials
 
     @classmethod
     def from_boto3_client(
@@ -69,6 +65,14 @@ def from_boto3_client(
     ):
         """
         Create an instance from a boto3 S3 client, using its credentials.
+
+        This method may, depending on how your client is configured, refresh its credentials.
+        Bear in mind that the official clients may refresh their credentials when calling methods on them,
+        but the created signer won't. This means that if the credentials are temporary (e.g., STS tokens),
+        the generated URLs may suddently become invalid.
+
+        For the sake of caution, these instances should, therefore, be short-lived
+        (that is, create, use, and destroy).
         """
 
         # Generate a dummy URL to see how the client does it with the current configuration,
@@ -81,14 +85,12 @@ def from_boto3_client(
         bucket_endpoint_url = dummy_url[0 : dummy_url.index(dummy_key, 0)]
 
         # Get the client's current credentials.
-        credentials: _Credentials = client._request_signer._credentials.get_frozen_credentials()  # type: ignore
+        credentials = get_credentials_from_boto3_client(client)
 
         return cls(
             aws_region=client.meta.region_name,
             bucket_endpoint_url=bucket_endpoint_url,
-            access_key_id=credentials.access_key,
-            secret_access_key=credentials.secret_key,
-            session_token=credentials.token,
+            credentials=credentials,
         )
 
     @classmethod
@@ -99,6 +101,14 @@ async def from_aiobotocore_client(
     ):
         """
         Create an instance from an aiobotocore S3 client, using its credentials.
+
+        This method may, depending on how your client is configured, refresh its credentials.
+        Bear in mind that the official clients may refresh their credentials when calling methods on them,
+        but the created signer won't. This means that if the credentials are temporary (e.g., STS tokens),
+        the generated URLs may suddently become invalid.
+
+        For the sake of caution, these instances should, therefore, be short-lived
+        (that is, create, use, and destroy).
         """
 
         # Generate a dummy URL to see how the client does it with the current configuration,
@@ -111,14 +121,12 @@ async def from_aiobotocore_client(
         bucket_endpoint_url = dummy_url[0 : dummy_url.index(dummy_key, 0)]
 
         # Get the client's current credentials.
-        credentials: _Credentials = await client._request_signer._credentials.get_frozen_credentials()  # type: ignore
+        credentials = await get_credentials_from_aiobotocore_client(client)
 
         return cls(
             aws_region=client.meta.region_name,
             bucket_endpoint_url=bucket_endpoint_url,
-            access_key_id=credentials.access_key,
-            secret_access_key=credentials.secret_key,
-            session_token=credentials.token,
+            credentials=credentials,
         )
 
     def generate_presigned_get_object_urls(
@@ -146,7 +154,7 @@ def generate_presigned_get_object_urls(
         amz_date = now.strftime("%Y%m%dT%H%M%SZ")
 
         signing_key = _derive_signing_key(
-            self.secret_key.encode(), datestamp, self.region, "s3"
+            self.credentials.secret_key.encode(), datestamp, self.region, "s3"
         )
 
         canonical_headers = f"host:{self.bucket_host}\n"
@@ -156,7 +164,9 @@ def generate_presigned_get_object_urls(
 
         credential_scope = f"{datestamp}/{self.region}/s3/aws4_request"
 
-        encoded_credential = quote(f"{self.access_key}/{credential_scope}", safe="~")
+        encoded_credential = quote(
+            f"{self.credentials.access_key}/{credential_scope}", safe="~"
+        )
 
         canonical_querystring_template_parts = [
             f"X-Amz-Algorithm={algorithm}",
@@ -166,9 +176,9 @@ def generate_presigned_get_object_urls(
             f"X-Amz-SignedHeaders={signed_headers}",
         ]
 
-        if self.session_token:
+        if isinstance(self.credentials, CredentialsWithToken):
             canonical_querystring_template_parts.append(
-                f"X-Amz-Security-Token={quote(self.session_token, safe='~')}"
+                f"X-Amz-Security-Token={quote(self.credentials.session_token, safe='~')}"
             )
 
         # The query string parameters must be sorted by their name.
@@ -207,7 +217,7 @@ def generate_presigned_get_object_urls(
                 )
             )
 
-            signature = hmac_hex(signing_key, string_to_sign.encode())
+            signature = _hmac_hex(signing_key, string_to_sign.encode())
 
             qs_with_signature = (
                 f"{canonical_querystring_template}&X-Amz-Signature={signature}"
@@ -221,16 +231,16 @@ def generate_presigned_get_object_urls(
 
 
 def _derive_signing_key(key: bytes, datestamp: str, region: str, service: str) -> bytes:
-    k_date = hmac_bytes(b"AWS4" + key, datestamp.encode("utf-8"))
-    k_region = hmac_bytes(k_date, region.encode("utf-8"))
-    k_service = hmac_bytes(k_region, service.encode("utf-8"))
+    k_date = _hmac_bytes(b"AWS4" + key, datestamp.encode("utf-8"))
+    k_region = _hmac_bytes(k_date, region.encode("utf-8"))
+    k_service = _hmac_bytes(k_region, service.encode("utf-8"))
 
-    return hmac_bytes(k_service, b"aws4_request")
+    return _hmac_bytes(k_service, b"aws4_request")
 
 
-def hmac_bytes(key: bytes, data: bytes) -> bytes:
+def _hmac_bytes(key: bytes, data: bytes) -> bytes:
     return hmac.new(key, data, hashlib.sha256).digest()
 
 
-def hmac_hex(key: bytes, data: bytes) -> str:
+def _hmac_hex(key: bytes, data: bytes) -> str:
     return hmac.new(key, data, hashlib.sha256).hexdigest()

fast_s3_url/util.py

@@ -0,0 +1,62 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Optional, Protocol
+
+from fast_s3_url.credentials import Credentials, CredentialsWithToken
+
+
+if TYPE_CHECKING:
+    from types_aiobotocore_s3 import S3Client as AioBotocoreS3Client
+    from mypy_boto3_s3.client import S3Client as Boto3S3Client
+
+
+__all__ = [
+    "get_credentials_from_boto3_client",
+    "get_credentials_from_aiobotocore_client",
+]
+
+
+def get_credentials_from_boto3_client(client: Boto3S3Client) -> Credentials:
+    """
+    Get credentials from a boto3 client.
+
+    Note: This may cause the client to refresh its own credentials.
+    """
+
+    c: _Credentials = client._request_signer._credentials.get_frozen_credentials()  # type: ignore
+
+    return _to_credentials(c)
+
+
+async def get_credentials_from_aiobotocore_client(
+    client: AioBotocoreS3Client,
+) -> Credentials:
+    """
+    Get credentials from an aiobotocore client.
+
+    Note: This may cause the client to refresh its own credentials.
+    """
+
+    c: _Credentials = await client._request_signer._credentials.get_frozen_credentials()  # type: ignore
+
+    return _to_credentials(c)
+
+
+class _Credentials(Protocol):
+    access_key: str
+    secret_key: str
+    token: Optional[str]
+
+
+def _to_credentials(c: _Credentials) -> Credentials:
+    if c.token:
+        return CredentialsWithToken(
+            access_key=c.access_key,
+            secret_key=c.secret_key,
+            session_token=c.token,
+        )
+
+    return Credentials(
+        access_key=c.access_key,
+        secret_key=c.secret_key,
+    )

test/unit/test_validation.py

@@ -3,7 +3,7 @@
 
 import pytest
 
-from fast_s3_url import FastS3UrlSigner
+from fast_s3_url import FastS3UrlSigner, Credentials
 
 
 @pytest.mark.parametrize(
@@ -17,8 +17,10 @@ def test_validates_parameters(invalid_object_key: str):
     # Arrange
     sig = FastS3UrlSigner(
         bucket_endpoint_url="http://localhost:9000/my-bucket/",
-        access_key_id="minioadmin",
-        secret_access_key="minioadmin",
+        credentials=Credentials(
+            access_key="minioadmin",
+            secret_key="minioadmin",
+        ),
     )
 
     # Act / Assert