Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/botocore/auth.py: 20%

1# Copyright (c) 2012-2013 Mitch Garnaat http://garnaat.org/ 

2# Copyright 2012-2014 Amazon.com, Inc. or its affiliates. All Rights Reserved. 

3# 

4# Licensed under the Apache License, Version 2.0 (the "License"). You 

5# may not use this file except in compliance with the License. A copy of 

6# the License is located at 

7# 

8# http://aws.amazon.com/apache2.0/ 

9# 

10# or in the "license" file accompanying this file. This file is 

11# distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF 

12# ANY KIND, either express or implied. See the License for the specific 

13# language governing permissions and limitations under the License. 

14import base64 

15import calendar 

16import datetime 

17import functools 

18import hmac 

19import json 

20import logging 

21import time 

22from collections.abc import Mapping 

23from email.utils import formatdate 

24from hashlib import sha1, sha256 

25from operator import itemgetter 

26 

27from botocore.compat import ( 

28 HAS_CRT, 

29 MD5_AVAILABLE, # noqa: F401 

30 HTTPHeaders, 

31 encodebytes, 

32 ensure_unicode, 

33 get_current_datetime, 

34 parse_qs, 

35 quote, 

36 unquote, 

37 urlsplit, 

38 urlunsplit, 

39) 

40from botocore.exceptions import ( 

41 NoAuthTokenError, 

42 NoCredentialsError, 

43 UnknownSignatureVersionError, 

44 UnsupportedSignatureVersionError, 

45) 

46from botocore.utils import ( 

47 is_valid_ipv6_endpoint_url, 

48 normalize_url_path, 

49 percent_encode_sequence, 

50) 

51 

52logger = logging.getLogger(__name__) 

53 

54 

55EMPTY_SHA256_HASH = ( 

56 'e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855' 

57) 

58# This is the buffer size used when calculating sha256 checksums. 

59# Experimenting with various buffer sizes showed that this value generally 

60# gave the best result (in terms of performance). 

61PAYLOAD_BUFFER = 1024 * 1024 

62ISO8601 = '%Y-%m-%dT%H:%M:%SZ' 

63SIGV4_TIMESTAMP = '%Y%m%dT%H%M%SZ' 

64SIGNED_HEADERS_BLACKLIST = [ 

65 'baggage', 

66 'connection', 

67 'expect', 

68 'keep-alive', 

69 'proxy-authenticate', 

70 'proxy-authorization', 

71 'te', 

72 'traceparent', 

73 'tracestate', 

74 'trailer', 

75 'transfer-encoding', 

76 'upgrade', 

77 'user-agent', 

78 'x-amzn-trace-id', 

79] 

80UNSIGNED_PAYLOAD = 'UNSIGNED-PAYLOAD' 

81STREAMING_UNSIGNED_PAYLOAD_TRAILER = 'STREAMING-UNSIGNED-PAYLOAD-TRAILER' 

82 

83 

84def _host_from_url(url): 

85 # Given URL, derive value for host header. Ensure that value: 

86 # 1) is lowercase 

87 # 2) excludes port, if it was the default port 

88 # 3) excludes userinfo 

89 url_parts = urlsplit(url) 

90 host = url_parts.hostname # urlsplit's hostname is always lowercase 

91 if is_valid_ipv6_endpoint_url(url): 

92 host = f'[{host}]' 

93 default_ports = { 

94 'http': 80, 

95 'https': 443, 

96 } 

97 if url_parts.port is not None: 

98 if url_parts.port != default_ports.get(url_parts.scheme): 

99 host = f'{host}:{url_parts.port}' 

100 return host 

101 

102 

103def _get_body_as_dict(request): 

104 # For query services, request.data is form-encoded and is already a 

105 # dict, but for other services such as rest-json it could be a json 

106 # string or bytes. In those cases we attempt to load the data as a 

107 # dict. 

108 data = request.data 

109 if isinstance(data, bytes): 

110 data = json.loads(data.decode('utf-8')) 

111 elif isinstance(data, str): 

112 data = json.loads(data) 

113 return data 

114 

115 

116class BaseSigner: 

117 REQUIRES_REGION = False 

118 REQUIRES_TOKEN = False 

119 

120 def add_auth(self, request): 

121 raise NotImplementedError("add_auth") 

122 

123 

124class TokenSigner(BaseSigner): 

125 REQUIRES_TOKEN = True 

126 """ 

127 Signers that expect an authorization token to perform the authorization 

128 """ 

129 

130 def __init__(self, auth_token): 

131 self.auth_token = auth_token 

132 

133 

134class SigV2Auth(BaseSigner): 

135 """ 

136 Sign a request with Signature V2. 

137 """ 

138 

139 def __init__(self, credentials): 

140 self.credentials = credentials 

141 

142 def calc_signature(self, request, params): 

143 logger.debug("Calculating signature using v2 auth.") 

144 split = urlsplit(request.url) 

145 path = split.path 

146 if len(path) == 0: 

147 path = '/' 

148 string_to_sign = f"{request.method}\n{split.netloc}\n{path}\n" 

149 lhmac = hmac.new( 

150 self.credentials.secret_key.encode("utf-8"), digestmod=sha256 

151 ) 

152 pairs = [] 

153 for key in sorted(params): 

154 # Any previous signature should not be a part of this 

155 # one, so we skip that particular key. This prevents 

156 # issues during retries. 

157 if key == 'Signature': 

158 continue 

159 value = str(params[key]) 

160 quoted_key = quote(key.encode('utf-8'), safe='') 

161 quoted_value = quote(value.encode('utf-8'), safe='-_~') 

162 pairs.append(f'{quoted_key}={quoted_value}') 

163 qs = '&'.join(pairs) 

164 string_to_sign += qs 

165 logger.debug('String to sign: %s', string_to_sign) 

166 lhmac.update(string_to_sign.encode('utf-8')) 

167 b64 = base64.b64encode(lhmac.digest()).strip().decode('utf-8') 

168 return (qs, b64) 

169 

170 def add_auth(self, request): 

171 # The auth handler is the last thing called in the 

172 # preparation phase of a prepared request. 

173 # Because of this we have to parse the query params 

174 # from the request body so we can update them with 

175 # the sigv2 auth params. 

176 if self.credentials is None: 

177 raise NoCredentialsError() 

178 if request.data: 

179 # POST 

180 params = request.data 

181 else: 

182 # GET 

183 params = request.params 

184 params['AWSAccessKeyId'] = self.credentials.access_key 

185 params['SignatureVersion'] = '2' 

186 params['SignatureMethod'] = 'HmacSHA256' 

187 params['Timestamp'] = time.strftime(ISO8601, time.gmtime()) 

188 if self.credentials.token: 

189 params['SecurityToken'] = self.credentials.token 

190 qs, signature = self.calc_signature(request, params) 

191 params['Signature'] = signature 

192 return request 

193 

194 

195class SigV3Auth(BaseSigner): 

196 def __init__(self, credentials): 

197 self.credentials = credentials 

198 

199 def add_auth(self, request): 

200 if self.credentials is None: 

201 raise NoCredentialsError() 

202 if 'Date' in request.headers: 

203 del request.headers['Date'] 

204 request.headers['Date'] = formatdate(usegmt=True) 

205 if self.credentials.token: 

206 if 'X-Amz-Security-Token' in request.headers: 

207 del request.headers['X-Amz-Security-Token'] 

208 request.headers['X-Amz-Security-Token'] = self.credentials.token 

209 new_hmac = hmac.new( 

210 self.credentials.secret_key.encode('utf-8'), digestmod=sha256 

211 ) 

212 new_hmac.update(request.headers['Date'].encode('utf-8')) 

213 encoded_signature = encodebytes(new_hmac.digest()).strip() 

214 signature = ( 

215 f"AWS3-HTTPS AWSAccessKeyId={self.credentials.access_key}," 

216 f"Algorithm=HmacSHA256,Signature={encoded_signature.decode('utf-8')}" 

217 ) 

218 if 'X-Amzn-Authorization' in request.headers: 

219 del request.headers['X-Amzn-Authorization'] 

220 request.headers['X-Amzn-Authorization'] = signature 

221 

222 

223class SigV4Auth(BaseSigner): 

224 """ 

225 Sign a request with Signature V4. 

226 """ 

227 

228 REQUIRES_REGION = True 

229 

230 def __init__(self, credentials, service_name, region_name): 

231 self.credentials = credentials 

232 # We initialize these value here so the unit tests can have 

233 # valid values. But these will get overriden in ``add_auth`` 

234 # later for real requests. 

235 self._region_name = region_name 

236 self._service_name = service_name 

237 

238 def _sign(self, key, msg, hex=False): 

239 if hex: 

240 sig = hmac.new(key, msg.encode('utf-8'), sha256).hexdigest() 

241 else: 

242 sig = hmac.new(key, msg.encode('utf-8'), sha256).digest() 

243 return sig 

244 

245 def headers_to_sign(self, request): 

246 """ 

247 Select the headers from the request that need to be included 

248 in the StringToSign. 

249 """ 

250 header_map = HTTPHeaders() 

251 for name, value in request.headers.items(): 

252 lname = name.lower() 

253 if lname not in SIGNED_HEADERS_BLACKLIST: 

254 header_map[lname] = value 

255 if 'host' not in header_map: 

256 # TODO: We should set the host ourselves, instead of relying on our 

257 # HTTP client to set it for us. 

258 header_map['host'] = _host_from_url(request.url) 

259 return header_map 

260 

261 def canonical_query_string(self, request): 

262 # The query string can come from two parts. One is the 

263 # params attribute of the request. The other is from the request 

264 # url (in which case we have to re-split the url into its components 

265 # and parse out the query string component). 

266 if request.params: 

267 return self._canonical_query_string_params(request.params) 

268 else: 

269 return self._canonical_query_string_url(urlsplit(request.url)) 

270 

271 def _canonical_query_string_params(self, params): 

272 # [(key, value), (key2, value2)] 

273 key_val_pairs = [] 

274 if isinstance(params, Mapping): 

275 params = params.items() 

276 for key, value in params: 

277 key_val_pairs.append( 

278 (quote(key, safe='-_.~'), quote(str(value), safe='-_.~')) 

279 ) 

280 sorted_key_vals = [] 

281 # Sort by the URI-encoded key names, and in the case of 

282 # repeated keys, then sort by the value. 

283 for key, value in sorted(key_val_pairs): 

284 sorted_key_vals.append(f'{key}={value}') 

285 canonical_query_string = '&'.join(sorted_key_vals) 

286 return canonical_query_string 

287 

288 def _canonical_query_string_url(self, parts): 

289 canonical_query_string = '' 

290 if parts.query: 

291 # [(key, value), (key2, value2)] 

292 key_val_pairs = [] 

293 for pair in parts.query.split('&'): 

294 key, _, value = pair.partition('=') 

295 key_val_pairs.append((key, value)) 

296 sorted_key_vals = [] 

297 # Sort by the URI-encoded key names, and in the case of 

298 # repeated keys, then sort by the value. 

299 for key, value in sorted(key_val_pairs): 

300 sorted_key_vals.append(f'{key}={value}') 

301 canonical_query_string = '&'.join(sorted_key_vals) 

302 return canonical_query_string 

303 

304 def canonical_headers(self, headers_to_sign): 

305 """ 

306 Return the headers that need to be included in the StringToSign 

307 in their canonical form by converting all header keys to lower 

308 case, sorting them in alphabetical order and then joining 

309 them into a string, separated by newlines. 

310 """ 

311 headers = [] 

312 sorted_header_names = sorted(set(headers_to_sign)) 

313 for key in sorted_header_names: 

314 value = ','.join( 

315 self._header_value(v) for v in headers_to_sign.get_all(key) 

316 ) 

317 headers.append(f'{key}:{ensure_unicode(value)}') 

318 return '\n'.join(headers) 

319 

320 def _header_value(self, value): 

321 # From the sigv4 docs: 

322 # Lowercase(HeaderName) + ':' + Trimall(HeaderValue) 

323 # 

324 # The Trimall function removes excess white space before and after 

325 # values, and converts sequential spaces to a single space. 

326 return ' '.join(value.split()) 

327 

328 def signed_headers(self, headers_to_sign): 

329 headers = sorted(n.lower().strip() for n in set(headers_to_sign)) 

330 return ';'.join(headers) 

331 

332 def _is_streaming_checksum_payload(self, request): 

333 checksum_context = request.context.get('checksum', {}) 

334 algorithm = checksum_context.get('request_algorithm') 

335 return isinstance(algorithm, dict) and algorithm.get('in') == 'trailer' 

336 

337 def payload(self, request): 

338 if self._is_streaming_checksum_payload(request): 

339 return STREAMING_UNSIGNED_PAYLOAD_TRAILER 

340 elif not self._should_sha256_sign_payload(request): 

341 # When payload signing is disabled, we use this static string in 

342 # place of the payload checksum. 

343 return UNSIGNED_PAYLOAD 

344 request_body = request.body 

345 if request_body and hasattr(request_body, 'seek'): 

346 position = request_body.tell() 

347 read_chunksize = functools.partial( 

348 request_body.read, PAYLOAD_BUFFER 

349 ) 

350 checksum = sha256() 

351 for chunk in iter(read_chunksize, b''): 

352 checksum.update(chunk) 

353 hex_checksum = checksum.hexdigest() 

354 request_body.seek(position) 

355 return hex_checksum 

356 elif request_body: 

357 # The request serialization has ensured that 

358 # request.body is a bytes() type. 

359 return sha256(request_body).hexdigest() 

360 else: 

361 return EMPTY_SHA256_HASH 

362 

363 def _should_sha256_sign_payload(self, request): 

364 # Payloads will always be signed over insecure connections. 

365 if not request.url.startswith('https'): 

366 return True 

367 

368 # Certain operations may have payload signing disabled by default. 

369 # Since we don't have access to the operation model, we pass in this 

370 # bit of metadata through the request context. 

371 return request.context.get('payload_signing_enabled', True) 

372 

373 def canonical_request(self, request): 

374 cr = [request.method.upper()] 

375 path = self._normalize_url_path(urlsplit(request.url).path) 

376 cr.append(path) 

377 cr.append(self.canonical_query_string(request)) 

378 headers_to_sign = self.headers_to_sign(request) 

379 cr.append(self.canonical_headers(headers_to_sign) + '\n') 

380 cr.append(self.signed_headers(headers_to_sign)) 

381 if 'X-Amz-Content-SHA256' in request.headers: 

382 body_checksum = request.headers['X-Amz-Content-SHA256'] 

383 else: 

384 body_checksum = self.payload(request) 

385 cr.append(body_checksum) 

386 return '\n'.join(cr) 

387 

388 def _normalize_url_path(self, path): 

389 normalized_path = quote(normalize_url_path(path), safe='/~') 

390 return normalized_path 

391 

392 def scope(self, request): 

393 scope = [self.credentials.access_key] 

394 scope.append(request.context['timestamp'][0:8]) 

395 scope.append(self._region_name) 

396 scope.append(self._service_name) 

397 scope.append('aws4_request') 

398 return '/'.join(scope) 

399 

400 def credential_scope(self, request): 

401 scope = [] 

402 scope.append(request.context['timestamp'][0:8]) 

403 scope.append(self._region_name) 

404 scope.append(self._service_name) 

405 scope.append('aws4_request') 

406 return '/'.join(scope) 

407 

408 def string_to_sign(self, request, canonical_request): 

409 """ 

410 Return the canonical StringToSign as well as a dict 

411 containing the original version of all headers that 

412 were included in the StringToSign. 

413 """ 

414 sts = ['AWS4-HMAC-SHA256'] 

415 sts.append(request.context['timestamp']) 

416 sts.append(self.credential_scope(request)) 

417 sts.append(sha256(canonical_request.encode('utf-8')).hexdigest()) 

418 return '\n'.join(sts) 

419 

420 def signature(self, string_to_sign, request): 

421 key = self.credentials.secret_key 

422 k_date = self._sign( 

423 (f"AWS4{key}").encode(), request.context["timestamp"][0:8] 

424 ) 

425 k_region = self._sign(k_date, self._region_name) 

426 k_service = self._sign(k_region, self._service_name) 

427 k_signing = self._sign(k_service, 'aws4_request') 

428 return self._sign(k_signing, string_to_sign, hex=True) 

429 

430 def add_auth(self, request): 

431 if self.credentials is None: 

432 raise NoCredentialsError() 

433 datetime_now = get_current_datetime() 

434 request.context['timestamp'] = datetime_now.strftime(SIGV4_TIMESTAMP) 

435 # This could be a retry. Make sure the previous 

436 # authorization header is removed first. 

437 self._modify_request_before_signing(request) 

438 canonical_request = self.canonical_request(request) 

439 logger.debug("Calculating signature using v4 auth.") 

440 logger.debug('CanonicalRequest:\n%s', canonical_request) 

441 string_to_sign = self.string_to_sign(request, canonical_request) 

442 logger.debug('StringToSign:\n%s', string_to_sign) 

443 signature = self.signature(string_to_sign, request) 

444 logger.debug('Signature:\n%s', signature) 

445 

446 self._inject_signature_to_request(request, signature) 

447 

448 def _inject_signature_to_request(self, request, signature): 

449 auth_str = [f'AWS4-HMAC-SHA256 Credential={self.scope(request)}'] 

450 headers_to_sign = self.headers_to_sign(request) 

451 auth_str.append( 

452 f"SignedHeaders={self.signed_headers(headers_to_sign)}" 

453 ) 

454 auth_str.append(f'Signature={signature}') 

455 request.headers['Authorization'] = ', '.join(auth_str) 

456 return request 

457 

458 def _modify_request_before_signing(self, request): 

459 if 'Authorization' in request.headers: 

460 del request.headers['Authorization'] 

461 self._set_necessary_date_headers(request) 

462 if self.credentials.token: 

463 if 'X-Amz-Security-Token' in request.headers: 

464 del request.headers['X-Amz-Security-Token'] 

465 request.headers['X-Amz-Security-Token'] = self.credentials.token 

466 

467 if not request.context.get('payload_signing_enabled', True): 

468 if 'X-Amz-Content-SHA256' in request.headers: 

469 del request.headers['X-Amz-Content-SHA256'] 

470 request.headers['X-Amz-Content-SHA256'] = UNSIGNED_PAYLOAD 

471 

472 def _set_necessary_date_headers(self, request): 

473 # The spec allows for either the Date _or_ the X-Amz-Date value to be 

474 # used so we check both. If there's a Date header, we use the date 

475 # header. Otherwise we use the X-Amz-Date header. 

476 if 'Date' in request.headers: 

477 del request.headers['Date'] 

478 datetime_timestamp = datetime.datetime.strptime( 

479 request.context['timestamp'], SIGV4_TIMESTAMP 

480 ) 

481 request.headers['Date'] = formatdate( 

482 int(calendar.timegm(datetime_timestamp.timetuple())) 

483 ) 

484 if 'X-Amz-Date' in request.headers: 

485 del request.headers['X-Amz-Date'] 

486 else: 

487 if 'X-Amz-Date' in request.headers: 

488 del request.headers['X-Amz-Date'] 

489 request.headers['X-Amz-Date'] = request.context['timestamp'] 

490 

491 

492class S3SigV4Auth(SigV4Auth): 

493 def _modify_request_before_signing(self, request): 

494 super()._modify_request_before_signing(request) 

495 if 'X-Amz-Content-SHA256' in request.headers: 

496 del request.headers['X-Amz-Content-SHA256'] 

497 

498 request.headers['X-Amz-Content-SHA256'] = self.payload(request) 

499 

500 def _should_sha256_sign_payload(self, request): 

501 # S3 allows optional body signing, so to minimize the performance 

502 # impact, we opt to not SHA256 sign the body on streaming uploads, 

503 # provided that we're on https. 

504 client_config = request.context.get('client_config') 

505 s3_config = getattr(client_config, 's3', None) 

506 

507 # The config could be None if it isn't set, or if the customer sets it 

508 # to None. 

509 if s3_config is None: 

510 s3_config = {} 

511 

512 # The explicit configuration takes precedence over any implicit 

513 # configuration. 

514 sign_payload = s3_config.get('payload_signing_enabled', None) 

515 if sign_payload is not None: 

516 return sign_payload 

517 

518 # We require that both a checksum be present and https be enabled 

519 # to implicitly disable body signing. The combination of TLS and 

520 # a checksum is sufficiently secure and durable for us to be 

521 # confident in the request without body signing. 

522 checksum_header = 'Content-MD5' 

523 checksum_context = request.context.get('checksum', {}) 

524 algorithm = checksum_context.get('request_algorithm') 

525 if isinstance(algorithm, dict) and algorithm.get('in') == 'header': 

526 checksum_header = algorithm['name'] 

527 if ( 

528 not request.url.startswith("https") 

529 or checksum_header not in request.headers 

530 ): 

531 return True 

532 

533 # If the input is streaming we disable body signing by default. 

534 if request.context.get('has_streaming_input', False): 

535 return False 

536 

537 # If the S3-specific checks had no results, delegate to the generic 

538 # checks. 

539 return super()._should_sha256_sign_payload(request) 

540 

541 def _normalize_url_path(self, path): 

542 # For S3, we do not normalize the path. 

543 return path 

544 

545 

546class S3ExpressAuth(S3SigV4Auth): 

547 REQUIRES_IDENTITY_CACHE = True 

548 

549 def __init__( 

550 self, credentials, service_name, region_name, *, identity_cache 

551 ): 

552 super().__init__(credentials, service_name, region_name) 

553 self._identity_cache = identity_cache 

554 

555 def add_auth(self, request): 

556 super().add_auth(request) 

557 

558 def _modify_request_before_signing(self, request): 

559 super()._modify_request_before_signing(request) 

560 if 'x-amz-s3session-token' not in request.headers: 

561 request.headers['x-amz-s3session-token'] = self.credentials.token 

562 # S3Express does not support STS' X-Amz-Security-Token 

563 if 'X-Amz-Security-Token' in request.headers: 

564 del request.headers['X-Amz-Security-Token'] 

565 

566 

567class S3ExpressPostAuth(S3ExpressAuth): 

568 REQUIRES_IDENTITY_CACHE = True 

569 

570 def add_auth(self, request): 

571 datetime_now = get_current_datetime() 

572 request.context['timestamp'] = datetime_now.strftime(SIGV4_TIMESTAMP) 

573 

574 fields = {} 

575 if request.context.get('s3-presign-post-fields', None) is not None: 

576 fields = request.context['s3-presign-post-fields'] 

577 

578 policy = {} 

579 conditions = [] 

580 if request.context.get('s3-presign-post-policy', None) is not None: 

581 policy = request.context['s3-presign-post-policy'] 

582 if policy.get('conditions', None) is not None: 

583 conditions = policy['conditions'] 

584 

585 policy['conditions'] = conditions 

586 

587 fields['x-amz-algorithm'] = 'AWS4-HMAC-SHA256' 

588 fields['x-amz-credential'] = self.scope(request) 

589 fields['x-amz-date'] = request.context['timestamp'] 

590 

591 conditions.append({'x-amz-algorithm': 'AWS4-HMAC-SHA256'}) 

592 conditions.append({'x-amz-credential': self.scope(request)}) 

593 conditions.append({'x-amz-date': request.context['timestamp']}) 

594 

595 if self.credentials.token is not None: 

596 fields['X-Amz-S3session-Token'] = self.credentials.token 

597 conditions.append( 

598 {'X-Amz-S3session-Token': self.credentials.token} 

599 ) 

600 

601 # Dump the base64 encoded policy into the fields dictionary. 

602 fields['policy'] = base64.b64encode( 

603 json.dumps(policy).encode('utf-8') 

604 ).decode('utf-8') 

605 

606 fields['x-amz-signature'] = self.signature(fields['policy'], request) 

607 

608 request.context['s3-presign-post-fields'] = fields 

609 request.context['s3-presign-post-policy'] = policy 

610 

611 

612class S3ExpressQueryAuth(S3ExpressAuth): 

613 DEFAULT_EXPIRES = 300 

614 REQUIRES_IDENTITY_CACHE = True 

615 

616 def __init__( 

617 self, 

618 credentials, 

619 service_name, 

620 region_name, 

621 *, 

622 identity_cache, 

623 expires=DEFAULT_EXPIRES, 

624 ): 

625 super().__init__( 

626 credentials, 

627 service_name, 

628 region_name, 

629 identity_cache=identity_cache, 

630 ) 

631 self._expires = expires 

632 

633 def _modify_request_before_signing(self, request): 

634 # We automatically set this header, so if it's the auto-set value we 

635 # want to get rid of it since it doesn't make sense for presigned urls. 

636 content_type = request.headers.get('content-type') 

637 blocklisted_content_type = ( 

638 'application/x-www-form-urlencoded; charset=utf-8' 

639 ) 

640 if content_type == blocklisted_content_type: 

641 del request.headers['content-type'] 

642 

643 # Note that we're not including X-Amz-Signature. 

644 # From the docs: "The Canonical Query String must include all the query 

645 # parameters from the preceding table except for X-Amz-Signature. 

646 signed_headers = self.signed_headers(self.headers_to_sign(request)) 

647 

648 auth_params = { 

649 'X-Amz-Algorithm': 'AWS4-HMAC-SHA256', 

650 'X-Amz-Credential': self.scope(request), 

651 'X-Amz-Date': request.context['timestamp'], 

652 'X-Amz-Expires': self._expires, 

653 'X-Amz-SignedHeaders': signed_headers, 

654 } 

655 if self.credentials.token is not None: 

656 auth_params['X-Amz-S3session-Token'] = self.credentials.token 

657 # Now parse the original query string to a dict, inject our new query 

658 # params, and serialize back to a query string. 

659 url_parts = urlsplit(request.url) 

660 # parse_qs makes each value a list, but in our case we know we won't 

661 # have repeated keys so we know we have single element lists which we 

662 # can convert back to scalar values. 

663 query_string_parts = parse_qs(url_parts.query, keep_blank_values=True) 

664 query_dict = {k: v[0] for k, v in query_string_parts.items()} 

665 

666 if request.params: 

667 query_dict.update(request.params) 

668 request.params = {} 

669 # The spec is particular about this. It *has* to be: 

670 # https://<endpoint>?<operation params>&<auth params> 

671 # You can't mix the two types of params together, i.e just keep doing 

672 # new_query_params.update(op_params) 

673 # new_query_params.update(auth_params) 

674 # percent_encode_sequence(new_query_params) 

675 operation_params = '' 

676 if request.data: 

677 # We also need to move the body params into the query string. To 

678 # do this, we first have to convert it to a dict. 

679 query_dict.update(_get_body_as_dict(request)) 

680 request.data = '' 

681 if query_dict: 

682 operation_params = percent_encode_sequence(query_dict) + '&' 

683 new_query_string = ( 

684 f"{operation_params}{percent_encode_sequence(auth_params)}" 

685 ) 

686 # url_parts is a tuple (and therefore immutable) so we need to create 

687 # a new url_parts with the new query string. 

688 # <part> - <index> 

689 # scheme - 0 

690 # netloc - 1 

691 # path - 2 

692 # query - 3 <-- we're replacing this. 

693 # fragment - 4 

694 p = url_parts 

695 new_url_parts = (p[0], p[1], p[2], new_query_string, p[4]) 

696 request.url = urlunsplit(new_url_parts) 

697 

698 def _inject_signature_to_request(self, request, signature): 

699 # Rather than calculating an "Authorization" header, for the query 

700 # param quth, we just append an 'X-Amz-Signature' param to the end 

701 # of the query string. 

702 request.url += f'&X-Amz-Signature={signature}' 

703 

704 def _normalize_url_path(self, path): 

705 # For S3, we do not normalize the path. 

706 return path 

707 

708 def payload(self, request): 

709 # From the doc link above: 

710 # "You don't include a payload hash in the Canonical Request, because 

711 # when you create a presigned URL, you don't know anything about the 

712 # payload. Instead, you use a constant string "UNSIGNED-PAYLOAD". 

713 return UNSIGNED_PAYLOAD 

714 

715 

716class SigV4QueryAuth(SigV4Auth): 

717 DEFAULT_EXPIRES = 3600 

718 

719 def __init__( 

720 self, credentials, service_name, region_name, expires=DEFAULT_EXPIRES 

721 ): 

722 super().__init__(credentials, service_name, region_name) 

723 self._expires = expires 

724 

725 def _modify_request_before_signing(self, request): 

726 # We automatically set this header, so if it's the auto-set value we 

727 # want to get rid of it since it doesn't make sense for presigned urls. 

728 content_type = request.headers.get('content-type') 

729 blacklisted_content_type = ( 

730 'application/x-www-form-urlencoded; charset=utf-8' 

731 ) 

732 if content_type == blacklisted_content_type: 

733 del request.headers['content-type'] 

734 

735 # Note that we're not including X-Amz-Signature. 

736 # From the docs: "The Canonical Query String must include all the query 

737 # parameters from the preceding table except for X-Amz-Signature. 

738 signed_headers = self.signed_headers(self.headers_to_sign(request)) 

739 

740 auth_params = { 

741 'X-Amz-Algorithm': 'AWS4-HMAC-SHA256', 

742 'X-Amz-Credential': self.scope(request), 

743 'X-Amz-Date': request.context['timestamp'], 

744 'X-Amz-Expires': self._expires, 

745 'X-Amz-SignedHeaders': signed_headers, 

746 } 

747 if self.credentials.token is not None: 

748 auth_params['X-Amz-Security-Token'] = self.credentials.token 

749 # Now parse the original query string to a dict, inject our new query 

750 # params, and serialize back to a query string. 

751 url_parts = urlsplit(request.url) 

752 # parse_qs makes each value a list, but in our case we know we won't 

753 # have repeated keys so we know we have single element lists which we 

754 # can convert back to scalar values. 

755 query_string_parts = parse_qs(url_parts.query, keep_blank_values=True) 

756 query_dict = {k: v[0] for k, v in query_string_parts.items()} 

757 

758 if request.params: 

759 query_dict.update(request.params) 

760 request.params = {} 

761 # The spec is particular about this. It *has* to be: 

762 # https://<endpoint>?<operation params>&<auth params> 

763 # You can't mix the two types of params together, i.e just keep doing 

764 # new_query_params.update(op_params) 

765 # new_query_params.update(auth_params) 

766 # percent_encode_sequence(new_query_params) 

767 operation_params = '' 

768 if request.data: 

769 # We also need to move the body params into the query string. To 

770 # do this, we first have to convert it to a dict. 

771 query_dict.update(_get_body_as_dict(request)) 

772 request.data = '' 

773 if query_dict: 

774 operation_params = percent_encode_sequence(query_dict) + '&' 

775 new_query_string = ( 

776 f"{operation_params}{percent_encode_sequence(auth_params)}" 

777 ) 

778 # url_parts is a tuple (and therefore immutable) so we need to create 

779 # a new url_parts with the new query string. 

780 # <part> - <index> 

781 # scheme - 0 

782 # netloc - 1 

783 # path - 2 

784 # query - 3 <-- we're replacing this. 

785 # fragment - 4 

786 p = url_parts 

787 new_url_parts = (p[0], p[1], p[2], new_query_string, p[4]) 

788 request.url = urlunsplit(new_url_parts) 

789 

790 def _inject_signature_to_request(self, request, signature): 

791 # Rather than calculating an "Authorization" header, for the query 

792 # param quth, we just append an 'X-Amz-Signature' param to the end 

793 # of the query string. 

794 request.url += f'&X-Amz-Signature={signature}' 

795 

796 

797class S3SigV4QueryAuth(SigV4QueryAuth): 

798 """S3 SigV4 auth using query parameters. 

799 

800 This signer will sign a request using query parameters and signature 

801 version 4, i.e a "presigned url" signer. 

802 

803 Based off of: 

804 

805 http://docs.aws.amazon.com/AmazonS3/latest/API/sigv4-query-string-auth.html 

806 

807 """ 

808 

809 def _normalize_url_path(self, path): 

810 # For S3, we do not normalize the path. 

811 return path 

812 

813 def payload(self, request): 

814 # From the doc link above: 

815 # "You don't include a payload hash in the Canonical Request, because 

816 # when you create a presigned URL, you don't know anything about the 

817 # payload. Instead, you use a constant string "UNSIGNED-PAYLOAD". 

818 return UNSIGNED_PAYLOAD 

819 

820 

821class S3SigV4PostAuth(SigV4Auth): 

822 """ 

823 Presigns a s3 post 

824 

825 Implementation doc here: 

826 http://docs.aws.amazon.com/AmazonS3/latest/API/sigv4-UsingHTTPPOST.html 

827 """ 

828 

829 def add_auth(self, request): 

830 datetime_now = get_current_datetime() 

831 request.context['timestamp'] = datetime_now.strftime(SIGV4_TIMESTAMP) 

832 

833 fields = {} 

834 if request.context.get('s3-presign-post-fields', None) is not None: 

835 fields = request.context['s3-presign-post-fields'] 

836 

837 policy = {} 

838 conditions = [] 

839 if request.context.get('s3-presign-post-policy', None) is not None: 

840 policy = request.context['s3-presign-post-policy'] 

841 if policy.get('conditions', None) is not None: