|
1 | | -"""AWS S3 Signature Version 4 utility for generating authentication headers. |
| 1 | +"""AWS S3 Signature V4 utility for generating authentication headers. |
2 | 2 |
|
3 | 3 | This module provides functions to calculate AWS Signature V4 signatures |
4 | 4 | and build headers for S3 upload requests. |
|
9 | 9 | import hashlib |
10 | 10 | import hmac |
11 | 11 | from datetime import datetime, timezone |
12 | | -from typing import Optional |
| 12 | +from typing import Literal, Optional, Union |
13 | 13 | from urllib.parse import quote |
14 | 14 | from ..api.models import UploadInfo |
15 | 15 |
|
@@ -69,7 +69,7 @@ def _create_canonical_request( |
69 | 69 | canonical_query_string: str, |
70 | 70 | canonical_headers: str, |
71 | 71 | signed_headers: str, |
72 | | - payload_hash: str, |
| 72 | + payload_hash: Union[Literal["UNSIGNED-PAYLOAD"], str], |
73 | 73 | ) -> str: |
74 | 74 | """Create the canonical request string. |
75 | 75 |
|
@@ -126,16 +126,16 @@ def calculate_aws_s3_v4_signature( |
126 | 126 | session_token: str, |
127 | 127 | region: str, |
128 | 128 | timestamp: str, |
129 | | - payload_hash: str = "UNSIGNED-PAYLOAD", |
| 129 | + payload_hash: Union[Literal["UNSIGNED-PAYLOAD"], str] = "UNSIGNED-PAYLOAD", |
130 | 130 | query_params: Optional[dict[str, str]] = None, |
131 | 131 | extra_headers: Optional[dict[str, str]] = None, |
132 | 132 | ) -> tuple[str, str, str]: |
133 | 133 | """Calculate AWS S3 V4 signature. |
134 | 134 |
|
135 | 135 | Args: |
136 | 136 | method: HTTP method (e.g., 'PUT', 'GET') |
137 | | - host: Host header value (e.g., 'streamables-upload.s3.amazonaws.com') |
138 | | - path: Request path (e.g., '/upload/y3vwnh') |
| 137 | + host: Host header value (e.g., 'some-bucket.s3.amazonaws.com') |
| 138 | + path: Request path (e.g., '/some/key') |
139 | 139 | access_key: AWS access key ID |
140 | 140 | secret_key: AWS secret access key |
141 | 141 | session_token: AWS session token |
@@ -223,16 +223,14 @@ def calculate_aws_s3_v4_signature( |
223 | 223 | def build_s3_upload_headers( |
224 | 224 | upload_info: UploadInfo, |
225 | 225 | content_length: int, |
226 | | - content_type: str = "application/octet-stream", |
227 | 226 | use_current_timestamp: bool = True, |
228 | 227 | ) -> dict[str, str]: |
229 | 228 | """Build headers for S3 upload request from UploadInfo model. |
230 | 229 |
|
231 | 230 | Args: |
232 | 231 | upload_info: UploadInfo pydantic model instance |
233 | 232 | content_length: Size of the file being uploaded in bytes |
234 | | - content_type: MIME type of the content (default: 'application/octet-stream') |
235 | | - use_current_timestamp: If True, generate a new timestamp (recommended for actual uploads) |
| 233 | + use_current_timestamp: If True, generate a new timestamp (must use for actual uploads) |
236 | 234 |
|
237 | 235 | Returns: |
238 | 236 | Dictionary of headers ready for the PUT request |
@@ -281,7 +279,7 @@ def build_s3_upload_headers( |
281 | 279 | headers: dict[str, str] = { |
282 | 280 | "Host": host, |
283 | 281 | "Authorization": auth_header, |
284 | | - "Content-Type": content_type, |
| 282 | + "Content-Type": "application/octet-stream", |
285 | 283 | "Content-Length": str(content_length), |
286 | 284 | "x-amz-content-sha256": "UNSIGNED-PAYLOAD", |
287 | 285 | "x-amz-date": timestamp, |
|
0 commit comments