refactor: add shared utilities and remove code duplication

- Add s3proxy/utils.py with shared helper functions:
  - get_query_param() for safe query parameter extraction
  - format_http_date() for HTTP date formatting (RFC 7231)
  - format_iso8601() for S3 API datetime formatting
- Add crypto.ENCRYPTION_OVERHEAD constant (replaces magic number 28)
- Remove duplicate format_http_date() from handlers/objects.py
- Use format_iso8601() for consistent datetime formatting

Author: ServerSideHannes

s3proxy/crypto.py

@@ -16,6 +16,7 @@
 # Constants
 NONCE_SIZE = 12  # 96 bits for AES-GCM
 TAG_SIZE = 16  # 128 bits authentication tag
+ENCRYPTION_OVERHEAD = NONCE_SIZE + TAG_SIZE  # 28 bytes added per encrypted chunk
 DEK_SIZE = 32  # 256 bits for AES-256
 PART_SIZE = 64 * 1024 * 1024  # 64 MB default part size for internal multipart uploads
 MIN_PART_SIZE = 5 * 1024 * 1024  # 5 MB minimum (S3 requirement for all parts except last)

s3proxy/handlers/multipart_ops.py

@@ -26,6 +26,7 @@
     save_multipart_metadata,
 )
 from ..s3client import S3Credentials
+from ..utils import format_iso8601
 from ..xml_utils import find_elements, get_element_text
 from .base import BaseHandler
 from .objects import decode_aws_chunked, decode_aws_chunked_stream
@@ -946,7 +947,7 @@ async def handle_upload_part_copy(self, request: Request, creds: S3Credentials)
             )
 
             # Return CopyPartResult
-            last_modified = datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%S.000Z")
+            last_modified = format_iso8601(datetime.now(UTC))
 
             return Response(
                 content=xml_responses.upload_part_copy_result(

s3proxy/handlers/objects.py

@@ -26,6 +26,7 @@
     save_multipart_metadata,
 )
 from ..s3client import S3Client, S3Credentials
+from ..utils import format_http_date, format_iso8601
 from ..xml_utils import find_element, find_elements
 from .base import BaseHandler
 
@@ -108,11 +109,6 @@ def chunked(data: bytes, size: int) -> Iterator[tuple[int, bytes]]:
         yield i // size + 1, data[i : i + size]
 
 
-def format_http_date(dt: datetime | None) -> str | None:
-    """Format datetime as HTTP date string."""
-    return dt.strftime("%a, %d %b %Y %H:%M:%S GMT") if dt else None
-
-
 class ObjectHandlerMixin(BaseHandler):
     """Mixin for object operations."""
 
@@ -339,7 +335,7 @@ async def stream():
                                 ciphertext = await resp["Body"].read()
 
                                 expected_size = internal_ct_end - internal_ct_offset + 1
-                                if len(ciphertext) < 28 or len(ciphertext) != expected_size:
+                                if len(ciphertext) < crypto.ENCRYPTION_OVERHEAD or len(ciphertext) != expected_size:
                                     logger.error(
                                         "Invalid ciphertext size - metadata mismatch",
                                         bucket=bucket,
@@ -946,7 +942,7 @@ async def handle_copy_object(self, request: Request, creds: S3Credentials) -> Re
             )
 
             # Return CopyObjectResult
-            last_modified = datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%S.000Z")
+            last_modified = format_iso8601(datetime.now(UTC))
 
             return Response(
                 content=xml_responses.copy_object_result(etag, last_modified),

s3proxy/utils.py

@@ -0,0 +1,50 @@
+"""Shared utilities for S3Proxy."""
+
+from datetime import datetime
+from urllib.parse import parse_qs
+
+# HTTP date format per RFC 7231
+HTTP_DATE_FORMAT = "%a, %d %b %Y %H:%M:%S GMT"
+
+# ISO 8601 format for S3 API responses
+ISO8601_FORMAT = "%Y-%m-%dT%H:%M:%S.000Z"
+
+
+def get_query_param(query: str | dict[str, list[str]], key: str, default: str = "") -> str:
+    """Get a single query parameter value with safe default.
+
+    Handles both raw query strings and pre-parsed dicts from parse_qs().
+    """
+    if isinstance(query, str):
+        query = parse_qs(query, keep_blank_values=True)
+    values = query.get(key, [default])
+    return values[0] if values else default
+
+
+def get_query_param_int(query: str | dict[str, list[str]], key: str, default: int) -> int:
+    """Get a query parameter as integer with safe default."""
+    value = get_query_param(query, key, "")
+    if not value:
+        return default
+    try:
+        return int(value)
+    except ValueError:
+        return default
+
+
+def format_http_date(dt: datetime | str | None) -> str | None:
+    """Format datetime as HTTP date string (RFC 7231)."""
+    if dt is None:
+        return None
+    if isinstance(dt, str):
+        return dt
+    if hasattr(dt, "strftime"):
+        return dt.strftime(HTTP_DATE_FORMAT)
+    return str(dt)
+
+
+def format_iso8601(dt: datetime | None) -> str:
+    """Format datetime as ISO 8601 for S3 API responses."""
+    if dt is None:
+        return datetime.utcnow().strftime(ISO8601_FORMAT)
+    return dt.strftime(ISO8601_FORMAT)