Add typing in signer.py

minio/signer.py

@@ -26,32 +26,42 @@
 
 """
 
+from __future__ import absolute_import, annotations
+
 import hashlib
 import hmac
 import re
 from collections import OrderedDict
+from datetime import datetime
 from urllib.parse import SplitResult
 
 from . import time
+from .credentials import Credentials
 from .helpers import queryencode, sha256_hash
 
 SIGN_V4_ALGORITHM = 'AWS4-HMAC-SHA256'
 _MULTI_SPACE_REGEX = re.compile(r"( +)")
 
 
-def _hmac_hash(key, data, hexdigest=False):
+def _hmac_hash(
+ key: bytes | bytearray,
+ data: bytes,
+ hexdigest: bool = False,
+) -> bytes | str:
  """Return HMacSHA256 digest of given key and data."""
 
  hasher = hmac.new(key, data, hashlib.sha256)
  return hasher.hexdigest() if hexdigest else hasher.digest()
 
 
-def _get_scope(date, region, service_name):
+def _get_scope(date: datetime, region: str, service_name: str) -> str:
  """Get scope string."""
  return f"{time.to_signer_date(date)}/{region}/{service_name}/aws4_request"
 
 
-def _get_canonical_headers(headers):
+def _get_canonical_headers(
+ headers: dict[str, str | list[str] | tuple[str]],
+) -> tuple[dict[str, str], str]:
  """Get canonical headers."""
 
  canonical_headers = {}
@@ -74,7 +84,7 @@ def _get_canonical_headers(headers):
  return canonical_headers, signed_headers
 
 
-def _get_canonical_query_string(query):
+def _get_canonical_query_string(query: str) -> str:
  """Get canonical query string."""
 
  query = query or ""
@@ -87,7 +97,12 @@ def _get_canonical_query_string(query):
  )
 
 
-def _get_canonical_request_hash(method, url, headers, content_sha256):
+def _get_canonical_request_hash(
+ method: str,
+ url: SplitResult,
+ headers: dict[str, str | list[str] | tuple[str]],
+ content_sha256: str,
+) -> tuple[str, str]:
  """Get canonical request hash."""
  canonical_headers, signed_headers = _get_canonical_headers(headers)
  canonical_query_string = _get_canonical_query_string(url.query)
@@ -110,35 +125,64 @@ def _get_canonical_request_hash(method, url, headers, content_sha256):
  return sha256_hash(canonical_request), signed_headers
 
 
-def _get_string_to_sign(date, scope, canonical_request_hash):
+def _get_string_to_sign(
+ date: datetime,
+ scope: str,
+ canonical_request_hash: str,
+) -> str:
  """Get string-to-sign."""
  return (
  f"AWS4-HMAC-SHA256\n{time.to_amz_date(date)}\n{scope}\n"
  f"{canonical_request_hash}"
  )
 
 
-def _get_signing_key(secret_key, date, region, service_name):
+def _get_signing_key(
+ secret_key: str,
+ date: datetime,
+ region: str,
+ service_name: str,
+) -> bytes:
  """Get signing key."""
 
  date_key = _hmac_hash(
  ("AWS4" + secret_key).encode(),
  time.to_signer_date(date).encode(),
  )
+ if isinstance(date_key, str):
+ date_key = date_key.encode()
+
  date_region_key = _hmac_hash(date_key, region.encode())
+ if isinstance(date_region_key, str):
+ date_region_key = date_region_key.encode()
+
  date_region_service_key = _hmac_hash(
  date_region_key, service_name.encode(),
  )
- return _hmac_hash(date_region_service_key, b"aws4_request")
+ if isinstance(date_region_service_key, str):
+ date_region_service_key = date_region_service_key.encode()
+
+ signing_key = _hmac_hash(date_region_service_key, b"aws4_request")
+ if isinstance(signing_key, str):
+ signing_key = signing_key.encode()
 
+ return signing_key
 
-def _get_signature(signing_key, string_to_sign):
+
+def _get_signature(signing_key: bytes, string_to_sign: str) -> str:
  """Get signature."""
 
- return _hmac_hash(signing_key, string_to_sign.encode(), hexdigest=True)
+ return str(
+ _hmac_hash(signing_key, string_to_sign.encode(), hexdigest=True),
+ )
 
 
-def _get_authorization(access_key, scope, signed_headers, signature):
+def _get_authorization(
+ access_key: str,
+ scope: str,
+ signed_headers: str,
+ signature: str,
+) -> str:
  """Get authorization."""
  return (
  f"AWS4-HMAC-SHA256 Credential={access_key}/{scope}, "
@@ -147,15 +191,15 @@ def _get_authorization(access_key, scope, signed_headers, signature):
 
 
 def _sign_v4(
- service_name,
- method,
- url,
- region,
- headers,
- credentials,
- content_sha256,
- date,
-):
+ service_name: str,
+ method: str,
+ url: SplitResult,
+ region: str,
+ headers: dict[str, str | list[str] | tuple[str]],
+ credentials: Credentials,
+ content_sha256: str,
+ date: datetime,
+) -> dict[str, str | list[str] | tuple[str]]:
  """Do signature V4 of given request for given service name."""
 
  scope = _get_scope(date, region, service_name)
@@ -175,14 +219,14 @@ def _sign_v4(
 
 
 def sign_v4_s3(
- method,
- url,
- region,
- headers,
- credentials,
- content_sha256,
- date,
-):
+ method: str,
+ url: SplitResult,
+ region: str,
+ headers: dict[str, str | list[str] | tuple[str]],
+ credentials: Credentials,
+ content_sha256: str,
+ date: datetime,
+) -> dict[str, str | list[str] | tuple[str]]:
  """Do signature V4 of given request for S3 service."""
  return _sign_v4(
  "s3",
@@ -197,14 +241,14 @@ def sign_v4_s3(
 
 
 def sign_v4_sts(
- method,
- url,
- region,
- headers,
- credentials,
- content_sha256,
- date,
-):
+ method: str,
+ url: SplitResult,
+ region: str,
+ headers: dict[str, str | list[str] | tuple[str]],
+ credentials: Credentials,
+ content_sha256: str,
+ date: datetime,
+) -> dict[str, str | list[str] | tuple[str]]:
  """Do signature V4 of given request for STS service."""
  return _sign_v4(
  "sts",
@@ -219,8 +263,13 @@ def sign_v4_sts(
 
 
 def _get_presign_canonical_request_hash( # pylint: disable=invalid-name
- method, url, access_key, scope, date, expires,
-):
+ method: str,
+ url: SplitResult,
+ access_key: str,
+ scope: str,
+ date: datetime,
+ expires: int,
+) -> tuple[str, SplitResult]:
  """Get canonical request hash for presign request."""
  x_amz_credential = queryencode(access_key + "/" + scope)
  canonical_headers, signed_headers = "host:" + url.netloc, "host"
@@ -258,13 +307,13 @@ def _get_presign_canonical_request_hash( # pylint: disable=invalid-name
 
 
 def presign_v4(
- method,
- url,
- region,
- credentials,
- date,
- expires,
-):
+ method: str,
+ url: SplitResult,
+ region: str,
+ credentials: Credentials,
+ date: datetime,
+ expires: int,
+) -> SplitResult:
  """Do signature V4 of given presign request."""
 
  scope = _get_scope(date, region, "s3")
@@ -281,12 +330,17 @@ def presign_v4(
  return url
 
 
-def get_credential_string(access_key, date, region):
+def get_credential_string(access_key: str, date: datetime, region: str) -> str:
  """Get credential string of given access key, date and region."""
  return f"{access_key}/{time.to_signer_date(date)}/{region}/s3/aws4_request"
 
 
-def post_presign_v4(data, secret_key, date, region):
+def post_presign_v4(
+ data: str,
+ secret_key: str,
+ date: datetime,
+ region: str,
+) -> str:
  """Do signature V4 of given presign POST form-data."""
  return _get_signature(
  _get_signing_key(secret_key, date, region, "s3"),