HMAC Authentication

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243

import hashlib
import hmac
import urllib.parse
from datetime import datetime, timezone
from typing import Dict, List, Optional
from dataclasses import dataclass
 
@dataclass
class HTTPRequest:
    method: str
    path: str
    query_params: Dict[str, str]
    headers: Dict[str, str]
    body: Optional[bytes] = None
 
class CanonicalRequestBuilder:
    """
    Builds canonical requests following AWS SigV4-style patterns.
    This is the industry-standard approach used by major APIs.
    """
    
    # Headers that should always be signed
    REQUIRED_SIGNED_HEADERS = ['host', 'x-timestamp']
    
    # Headers that should never be signed (ephemeral/transport)
    EXCLUDED_HEADERS = ['authorization', 'x-amz-signature', 'user-agent']
    
    def __init__(self, signed_headers: Optional[List[str]] = None):
        """
        Args:
            signed_headers: Specific headers to include in signature.
                           If None, all headers except excluded ones are signed.
        """
        self.signed_headers = signed_headers
    
    def build(self, request: HTTPRequest) -> str:
        """
        Build the canonical request string.
        
        Returns a deterministic string representation that can be
        reproduced exactly by both client and server.
        """
        # 1. HTTP Method (uppercase)
        method = request.method.upper()
        
        # 2. Canonical URI (path component, URL-encoded)
        canonical_uri = self._canonicalize_uri(request.path)
        
        # 3. Canonical Query String (sorted, URL-encoded)
        canonical_query = self._canonicalize_query_string(request.query_params)
        
        # 4. Canonical Headers (lowercase, sorted, trimmed)
        signed_header_names, canonical_headers = self._canonicalize_headers(
            request.headers
        )
        
        # 5. Signed Headers List (semicolon-separated)
        signed_headers_str = ';'.join(signed_header_names)
        
        # 6. Hashed Payload (SHA-256 of body)
        hashed_payload = self._hash_payload(request.body)
        
        # Combine all components
        canonical_request = '
'.join([
            method,
            canonical_uri,
            canonical_query,
            canonical_headers,
            signed_headers_str,
            hashed_payload,
        ])
        
        return canonical_request
    
    def _canonicalize_uri(self, path: str) -> str:
        """
        Canonicalize the URI path component.
        - Empty path becomes '/'
        - Path segments are URL-encoded
        - Multiple slashes are NOT collapsed (unlike some systems)
        """
        if not path:
            return '/'
        
        # URL-encode each path segment
        segments = path.split('/')
        encoded_segments = [
            urllib.parse.quote(segment, safe='~')
            for segment in segments
        ]
        
        canonical_path = '/'.join(encoded_segments)
        
        # Ensure leading slash
        if not canonical_path.startswith('/'):
            canonical_path = '/' + canonical_path
        
        return canonical_path
    
    def _canonicalize_query_string(
        self, 
        params: Dict[str, str]
    ) -> str:
        """
        Canonicalize query string parameters.
        - Sort by parameter name (code point order)
        - URL-encode both names and values
        - Empty values are preserved as 'key='
        """
        if not params:
            return ''
        
        # Sort by key, then encode
        sorted_params = sorted(params.items())
        
        encoded_pairs = [
            f"{urllib.parse.quote(k, safe='~')}={urllib.parse.quote(v, safe='~')}"
            for k, v in sorted_params
        ]
        
        return '&'.join(encoded_pairs)
    
    def _canonicalize_headers(
        self, 
        headers: Dict[str, str]
    ) -> tuple:
        """
        Canonicalize headers.
        - Convert names to lowercase
        - Trim whitespace from values
        - Collapse multiple spaces to single space
        - Sort by header name
        
        Returns:
            Tuple of (sorted_header_names, canonical_headers_string)
        """
        canonical = {}
        
        for name, value in headers.items():
            lower_name = name.lower()
            
            # Skip excluded headers
            if lower_name in self.EXCLUDED_HEADERS:
                continue
            
            # If specific headers requested, only include those
            if self.signed_headers and lower_name not in self.signed_headers:
                continue
            
            # Normalize value: trim and collapse whitespace
            normalized_value = ' '.join(value.split())
            
            canonical[lower_name] = normalized_value
        
        # Sort by header name
        sorted_names = sorted(canonical.keys())
        
        # Build canonical headers string
        header_lines = [
            f"{name}:{canonical[name]}"
            for name in sorted_names
        ]
        
        # Each header on own line, blank line at end
        canonical_headers = '
'.join(header_lines) + '
'
        
        return sorted_names, canonical_headers
    
    def _hash_payload(self, body: Optional[bytes]) -> str:
        """
        Compute SHA-256 hash of the request body.
        Empty body has a specific hash value.
        """
        if body is None:
            body = b''
        
        return hashlib.sha256(body).hexdigest()
    
    def compute_signature(
        self,
        request: HTTPRequest,
        secret_key: bytes,
        timestamp: datetime,
    ) -> str:
        """
        Compute the full HMAC signature for a request.
        """
        # Build canonical request
        canonical_request = self.build(request)
        
        # Hash the canonical request
        canonical_request_hash = hashlib.sha256(
            canonical_request.encode('utf-8')
        ).hexdigest()
        
        # Build string to sign
        string_to_sign = '
'.join([
            'HMAC-SHA256',  # Algorithm identifier
            timestamp.strftime('%Y%m%dT%H%M%SZ'),  # ISO8601 timestamp
            canonical_request_hash,
        ])
        
        # Compute HMAC signature
        signature = hmac.new(
            secret_key,
            string_to_sign.encode('utf-8'),
            hashlib.sha256
        ).hexdigest()
        
        return signature
 
 
# Example usage
if __name__ == "__main__":
    builder = CanonicalRequestBuilder()
    
    request = HTTPRequest(
        method='POST',
        path='/v1/charges',
        query_params={'expand[]': 'customer'},
        headers={
            'Host': 'api.example.com',
            'Content-Type': 'application/json',
            'X-Timestamp': '2024-01-15T10:30:00Z',
        },
        body=b'{"amount": 1000, "currency": "usd"}',
    )
    
    canonical = builder.build(request)
    print("Canonical Request:")
    print(canonical)
    print()
    
    signature = builder.compute_signature(
        request,
        secret_key=b'sk_live_very_secret_key_here',
        timestamp=datetime.now(timezone.utc),
    )
    print(f"Signature: {signature}")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297

package hmacauth
 
import (
    "bytes"
    "crypto/hmac"
    "crypto/sha256"
    "crypto/subtle"
    "encoding/hex"
    "fmt"
    "io"
    "net/http"
    "sort"
    "strings"
    "time"
)
 
const (
    // Header names
    HeaderSignature = "X-Signature"
    HeaderTimestamp = "X-Timestamp"
    HeaderKeyID     = "X-API-Key-ID"
    
    // Timestamp tolerance (±5 minutes)
    TimestampTolerance = 5 * time.Minute
    
    // Algorithm identifier
    Algorithm = "HMAC-SHA256"
)
 
// SigningError represents an error during signature operations
type SigningError struct {
    Code    string
    Message string
}
 
func (e *SigningError) Error() string {
    return fmt.Sprintf("%s: %s", e.Code, e.Message)
}
 
var (
    ErrMissingSignature = &SigningError{"MISSING_SIGNATURE", "signature header required"}
    ErrMissingTimestamp = &SigningError{"MISSING_TIMESTAMP", "timestamp header required"}
    ErrInvalidTimestamp = &SigningError{"INVALID_TIMESTAMP", "timestamp format invalid"}
    ErrTimestampExpired = &SigningError{"TIMESTAMP_EXPIRED", "request timestamp outside valid window"}
    ErrInvalidSignature = &SigningError{"INVALID_SIGNATURE", "signature verification failed"}
)
 
// Signer handles HMAC request signing
type Signer struct {
    keyID     string
    secretKey []byte
}
 
// NewSigner creates a new HMAC signer
func NewSigner(keyID string, secretKey []byte) *Signer {
    return &Signer{
        keyID:     keyID,
        secretKey: secretKey,
    }
}
 
// SignRequest signs an HTTP request, returning the signature
func (s *Signer) SignRequest(req *http.Request) (string, error) {
    timestamp := time.Now().UTC().Format(time.RFC3339)
    
    // Set timestamp header before signing
    req.Header.Set(HeaderTimestamp, timestamp)
    req.Header.Set(HeaderKeyID, s.keyID)
    
    // Build canonical request
    canonical, err := buildCanonicalRequest(req)
    if err != nil {
        return "", fmt.Errorf("failed to build canonical request: %w", err)
    }
    
    // Build string to sign
    stringToSign := buildStringToSign(canonical, timestamp)
    
    // Compute HMAC
    signature := computeHMAC(s.secretKey, stringToSign)
    
    // Set signature header
    req.Header.Set(HeaderSignature, signature)
    
    return signature, nil
}
 
// Verifier handles HMAC signature verification
type Verifier struct {
    keyLookup func(keyID string) ([]byte, error)
}
 
// NewVerifier creates a new HMAC verifier
func NewVerifier(keyLookup func(keyID string) ([]byte, error)) *Verifier {
    return &Verifier{keyLookup: keyLookup}
}
 
// VerifyRequest verifies the HMAC signature on an incoming request
func (v *Verifier) VerifyRequest(req *http.Request) error {
    // 1. Extract required headers
    signature := req.Header.Get(HeaderSignature)
    if signature == "" {
        return ErrMissingSignature
    }
    
    timestampStr := req.Header.Get(HeaderTimestamp)
    if timestampStr == "" {
        return ErrMissingTimestamp
    }
    
    keyID := req.Header.Get(HeaderKeyID)
    if keyID == "" {
        return &SigningError{"MISSING_KEY_ID", "API key ID header required"}
    }
    
    // 2. Parse and validate timestamp
    timestamp, err := time.Parse(time.RFC3339, timestampStr)
    if err != nil {
        return ErrInvalidTimestamp
    }
    
    now := time.Now().UTC()
    if timestamp.Before(now.Add(-TimestampTolerance)) || 
       timestamp.After(now.Add(TimestampTolerance)) {
        return ErrTimestampExpired
    }
    
    // 3. Look up secret key
    secretKey, err := v.keyLookup(keyID)
    if err != nil {
        return &SigningError{"KEY_LOOKUP_FAILED", "unable to find API key"}
    }
    
    // 4. Build canonical request (server-side)
    canonical, err := buildCanonicalRequest(req)
    if err != nil {
        return fmt.Errorf("failed to build canonical request: %w", err)
    }
    
    // 5. Compute expected signature
    stringToSign := buildStringToSign(canonical, timestampStr)
    expectedSignature := computeHMAC(secretKey, stringToSign)
    
    // 6. Constant-time comparison (critical for security!)
    if subtle.ConstantTimeCompare(
        []byte(signature), 
        []byte(expectedSignature),
    ) != 1 {
        return ErrInvalidSignature
    }
    
    return nil
}
 
// buildCanonicalRequest creates the canonical request string
func buildCanonicalRequest(req *http.Request) (string, error) {
    var buf bytes.Buffer
    
    // 1. HTTP Method
    buf.WriteString(req.Method)
    buf.WriteByte('
')
    
    // 2. Canonical URI
    uri := req.URL.Path
    if uri == "" {
        uri = "/"
    }
    buf.WriteString(uri)
    buf.WriteByte('
')
    
    // 3. Canonical Query String
    buf.WriteString(canonicalQueryString(req.URL.Query()))
    buf.WriteByte('
')
    
    // 4. Canonical Headers
    signedHeaders, headerString := canonicalHeaders(req)
    buf.WriteString(headerString)
    
    // 5. Signed Headers List
    buf.WriteString(signedHeaders)
    buf.WriteByte('
')
    
    // 6. Hashed Payload
    payloadHash, err := hashPayload(req)
    if err != nil {
        return "", err
    }
    buf.WriteString(payloadHash)
    
    return buf.String(), nil
}
 
// canonicalQueryString builds sorted query string
func canonicalQueryString(values map[string][]string) string {
    if len(values) == 0 {
        return ""
    }
    
    keys := make([]string, 0, len(values))
    for k := range values {
        keys = append(keys, k)
    }
    sort.Strings(keys)
    
    var pairs []string
    for _, k := range keys {
        for _, v := range values[k] {
            pairs = append(pairs, fmt.Sprintf("%s=%s", k, v))
        }
    }
    
    return strings.Join(pairs, "&")
}
 
// canonicalHeaders builds the canonical headers string
func canonicalHeaders(req *http.Request) (string, string) {
    // Headers to include in signature
    includeHeaders := []string{
        "content-type",
        "host",
        HeaderTimestamp,
    }
    
    headers := make(map[string]string)
    for _, name := range includeHeaders {
        lowerName := strings.ToLower(name)
        value := req.Header.Get(name)
        if value != "" {
            headers[lowerName] = strings.TrimSpace(value)
        } else if lowerName == "host" && req.Host != "" {
            headers[lowerName] = req.Host
        }
    }
    
    // Sort header names
    names := make([]string, 0, len(headers))
    for name := range headers {
        names = append(names, name)
    }
    sort.Strings(names)
    
    // Build canonical headers string
    var headerBuf bytes.Buffer
    for _, name := range names {
        headerBuf.WriteString(name)
        headerBuf.WriteByte(':')
        headerBuf.WriteString(headers[name])
        headerBuf.WriteByte('
')
    }
    
    return strings.Join(names, ";"), headerBuf.String()
}
 
// hashPayload computes SHA-256 hash of request body
func hashPayload(req *http.Request) (string, error) {
    if req.Body == nil {
        return hashBytes(nil), nil
    }
    
    // Read body (need to restore it after)
    body, err := io.ReadAll(req.Body)
    if err != nil {
        return "", err
    }
    req.Body.Close()
    
    // Restore body for handlers
    req.Body = io.NopCloser(bytes.NewReader(body))
    
    return hashBytes(body), nil
}
 
// hashBytes computes SHA-256 hex digest
func hashBytes(data []byte) string {
    hash := sha256.Sum256(data)
    return hex.EncodeToString(hash[:])
}
 
// buildStringToSign creates the string to sign
func buildStringToSign(canonical string, timestamp string) string {
    canonicalHash := hashBytes([]byte(canonical))
    return fmt.Sprintf("%s
%s
%s", Algorithm, timestamp, canonicalHash)
}
 
// computeHMAC computes HMAC-SHA256
func computeHMAC(key []byte, message string) string {
    h := hmac.New(sha256.New, key)
    h.Write([]byte(message))
    return hex.EncodeToString(h.Sum(nil))
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182

import hashlib
import time
from datetime import datetime, timedelta, timezone
from typing import Optional
import redis
 
class ReplayProtection:
    """
    Hybrid replay protection using timestamp windows and nonce tracking.
    Nonces are only stored for the duration of the timestamp window,
    limiting storage requirements while providing perfect replay prevention.
    """
    
    def __init__(
        self,
        redis_client: redis.Redis,
        timestamp_tolerance_seconds: int = 300,  # ±5 minutes
    ):
        self.redis = redis_client
        self.tolerance = timestamp_tolerance_seconds
        # Store nonces for 2x tolerance to handle edge cases
        self.nonce_ttl = timestamp_tolerance_seconds * 2
    
    def validate_request(
        self,
        timestamp_str: str,
        nonce: str,
        key_id: str,
    ) -> tuple[bool, Optional[str]]:
        """
        Validate request against replay attacks.
        
        Args:
            timestamp_str: ISO8601 timestamp from request
            nonce: Unique request identifier from request
            key_id: API key identifier for namespacing
        
        Returns:
            Tuple of (is_valid, error_message)
        """
        # 1. Parse timestamp
        try:
            timestamp = datetime.fromisoformat(
                timestamp_str.replace('Z', '+00:00')
            )
        except ValueError:
            return False, "invalid timestamp format"
        
        # 2. Check timestamp is within window
        now = datetime.now(timezone.utc)
        min_time = now - timedelta(seconds=self.tolerance)
        max_time = now + timedelta(seconds=self.tolerance)
        
        if timestamp < min_time:
            return False, "timestamp too old"
        
        if timestamp > max_time:
            return False, "timestamp too far in future"
        
        # 3. Check nonce hasn't been used
        nonce_key = self._nonce_key(key_id, nonce)
        
        # SETNX returns 1 if key was set (nonce is new)
        # Returns 0 if key already exists (nonce was used)
        is_new = self.redis.setnx(nonce_key, "1")
        
        if not is_new:
            return False, "nonce already used"
        
        # Set TTL on nonce key
        self.redis.expire(nonce_key, self.nonce_ttl)
        
        return True, None
    
    def _nonce_key(self, key_id: str, nonce: str) -> str:
        """Generate Redis key for nonce tracking."""
        # Hash to prevent key-length issues
        hash_input = f"{key_id}:{nonce}"
        hash_value = hashlib.sha256(hash_input.encode()).hexdigest()[:16]
        return f"nonce:{hash_value}"
    
    def cleanup_expired_nonces(self) -> int:
        """
        Manual cleanup of expired nonces (Redis TTL handles this,
        but can be called for immediate cleanup)
        """
        # Redis automatically cleans up keys with TTL
        # This method exists for explicit control if needed
        pattern = "nonce:*"
        count = 0
        
        for key in self.redis.scan_iter(match=pattern, count=100):
            if self.redis.ttl(key) <= 0:
                self.redis.delete(key)
                count += 1
        
        return count
 
 
class IdempotencyKeyManager:
    """
    Implements idempotency keys for ensuring exactly-once execution.
    Commonly used in payment APIs (Stripe-style).
    """
    
    def __init__(
        self,
        redis_client: redis.Redis,
        key_ttl_hours: int = 24,
    ):
        self.redis = redis_client
        self.key_ttl = key_ttl_hours * 3600
    
    def check_and_store(
        self,
        idempotency_key: str,
        api_key_id: str,
        request_hash: str,
    ) -> tuple[bool, Optional[dict]]:
        """
        Check if request with this idempotency key was already processed.
        
        Args:
            idempotency_key: Client-provided idempotency key
            api_key_id: API key for namespacing
            request_hash: Hash of request body to detect misuse
        
        Returns:
            Tuple of (is_new, cached_response)
            - (True, None) if this is a new request
            - (False, cached_response) if request was already processed
        """
        key = self._cache_key(api_key_id, idempotency_key)
        
        existing = self.redis.hgetall(key)
        
        if existing:
            stored_hash = existing.get(b'request_hash', b'').decode()
            
            # Verify request body matches
            if stored_hash != request_hash:
                raise ValueError(
                    "Idempotency key reused with different request body"
                )
            
            # Return cached response
            import json
            cached_response = json.loads(
                existing.get(b'response', b'{}').decode()
            )
            return False, cached_response
        
        # New request - store placeholder
        self.redis.hset(key, mapping={
            'request_hash': request_hash,
            'status': 'processing',
            'started_at': datetime.now(timezone.utc).isoformat(),
        })
        self.redis.expire(key, self.key_ttl)
        
        return True, None
    
    def store_response(
        self,
        idempotency_key: str,
        api_key_id: str,
        response: dict,
    ) -> None:
        """Store successful response for future lookups."""
        import json
        
        key = self._cache_key(api_key_id, idempotency_key)
        
        self.redis.hset(key, mapping={
            'status': 'complete',
            'response': json.dumps(response),
            'completed_at': datetime.now(timezone.utc).isoformat(),
        })
    
    def _cache_key(self, api_key_id: str, idempotency_key: str) -> str:
        """Generate Redis key for idempotency tracking."""
        return f"idempotency:{api_key_id}:{idempotency_key}"

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136

import hashlib
import hmac
from datetime import datetime, timezone
 
def derive_aws_signing_key(
    secret_key: str,
    date_stamp: str,
    region: str,
    service: str,
) -> bytes:
    """
    Derive the AWS SigV4 signing key.
    
    This multi-step derivation means:
    - Different days get different keys
    - Different regions get different keys
    - Different services get different keys
    
    Even if a signing key is compromised, it's only valid for
    one date/region/service combination.
    """
    def sign(key: bytes, msg: str) -> bytes:
        return hmac.new(key, msg.encode('utf-8'), hashlib.sha256).digest()
    
    # Key derivation chain
    k_date = sign(('AWS4' + secret_key).encode('utf-8'), date_stamp)
    k_region = sign(k_date, region)
    k_service = sign(k_region, service)
    k_signing = sign(k_service, 'aws4_request')
    
    return k_signing
 
 
def create_aws_authorization_header(
    access_key: str,
    secret_key: str,
    method: str,
    canonical_uri: str,
    canonical_querystring: str,
    canonical_headers: str,
    signed_headers: str,
    payload_hash: str,
    region: str = 'us-east-1',
    service: str = 's3',
) -> str:
    """
    Create a complete AWS SigV4 Authorization header.
    
    This demonstrates the full AWS signing process.
    In production, use boto3 or official SDKs.
    """
    # Current time
    t = datetime.now(timezone.utc)
    amz_date = t.strftime('%Y%m%dT%H%M%SZ')
    date_stamp = t.strftime('%Y%m%d')
    
    # Credential scope
    credential_scope = f"{date_stamp}/{region}/{service}/aws4_request"
    
    # Canonical request
    canonical_request = '
'.join([
        method,
        canonical_uri,
        canonical_querystring,
        canonical_headers,
        signed_headers,
        payload_hash,
    ])
    
    # String to sign
    string_to_sign = '
'.join([
        'AWS4-HMAC-SHA256',  # Algorithm
        amz_date,            # Request timestamp
        credential_scope,    # Scope
        hashlib.sha256(canonical_request.encode('utf-8')).hexdigest(),
    ])
    
    # Derive signing key
    signing_key = derive_aws_signing_key(
        secret_key, date_stamp, region, service
    )
    
    # Compute signature
    signature = hmac.new(
        signing_key,
        string_to_sign.encode('utf-8'),
        hashlib.sha256
    ).hexdigest()
    
    # Build authorization header
    authorization = (
        f"AWS4-HMAC-SHA256 "
        f"Credential={access_key}/{credential_scope}, "
        f"SignedHeaders={signed_headers}, "
        f"Signature={signature}"
    )
    
    return authorization
 
 
# Stripe-style HMAC for webhook verification
def verify_stripe_webhook_signature(
    payload: bytes,
    signature_header: str,
    webhook_secret: str,
    tolerance_seconds: int = 300,
) -> bool:
    """
    Verify Stripe webhook signature.
    
    Stripe's format: t=timestamp,v1=signature
    The signed message is: timestamp + '.' + payload
    """
    # Parse signature header
    parts = dict(p.split('=') for p in signature_header.split(','))
    
    timestamp = int(parts.get('t', 0))
    signature = parts.get('v1', '')
    
    # Check timestamp
    now = int(datetime.now(timezone.utc).timestamp())
    if abs(now - timestamp) > tolerance_seconds:
        return False
    
    # Compute expected signature
    signed_payload = f"{timestamp}.{payload.decode('utf-8')}"
    expected = hmac.new(
        webhook_secret.encode('utf-8'),
        signed_payload.encode('utf-8'),
        hashlib.sha256
    ).hexdigest()
    
    # Constant-time comparison
    return hmac.compare_digest(signature, expected)