|
18 | 18 | from zoneinfo import ZoneInfo
|
19 | 19 |
|
20 | 20 | import boto3 as boto3
|
| 21 | +import botocore |
21 | 22 | import pytest
|
22 | 23 | import requests
|
23 | 24 | import xmltodict
|
@@ -7434,6 +7435,87 @@ def add_content_sha_header(request, **kwargs):
|
7434 | 7435 | resp = requests.put(rewritten_url, data="something", verify=False)
|
7435 | 7436 | assert resp.status_code == 403
|
7436 | 7437 |
|
| 7438 | + @markers.aws.validated |
| 7439 | + def test_pre_signed_url_if_none_match(self, s3_bucket, aws_client, aws_session): |
| 7440 | + # there currently is a bug in Boto3: https://github.com/boto/boto3/issues/4367 |
| 7441 | + # so we need to use botocore directly to allow testing of this, as other SDK like the Java SDK have the correct |
| 7442 | + # behavior |
| 7443 | + object_key = "temp.txt" |
| 7444 | + |
| 7445 | + s3_endpoint_path_style = _endpoint_url() |
| 7446 | + |
| 7447 | + # assert that the regular Boto3 client does not work, and does not sign the parameter as requested |
| 7448 | + client = _s3_client_pre_signed_client( |
| 7449 | + Config(signature_version="s3v4", s3={}), |
| 7450 | + endpoint_url=s3_endpoint_path_style, |
| 7451 | + ) |
| 7452 | + bad_url = client.generate_presigned_url( |
| 7453 | + "put_object", |
| 7454 | + Params={"Bucket": s3_bucket, "Key": object_key, "IfNoneMatch": "*"}, |
| 7455 | + ) |
| 7456 | + assert "if-none-match=%2a" not in bad_url.lower() |
| 7457 | + |
| 7458 | + req = botocore.awsrequest.AWSRequest( |
| 7459 | + method="PUT", |
| 7460 | + url=f"{s3_endpoint_path_style}/{s3_bucket}/{object_key}", |
| 7461 | + data={}, |
| 7462 | + params={ |
| 7463 | + "If-None-Match": "*", |
| 7464 | + }, |
| 7465 | + headers={}, |
| 7466 | + ) |
| 7467 | + |
| 7468 | + botocore.auth.S3SigV4QueryAuth(aws_session.get_credentials(), "s3", "us-east-1").add_auth( |
| 7469 | + req |
| 7470 | + ) |
| 7471 | + |
| 7472 | + assert "if-none-match=%2a" in req.url.lower() |
| 7473 | + |
| 7474 | + response = requests.put(req.url) |
| 7475 | + assert response.status_code == 200 |
| 7476 | + |
| 7477 | + response = requests.put(req.url) |
| 7478 | + # we are now failing because the object already exists |
| 7479 | + assert response.status_code == 412 |
| 7480 | + |
| 7481 | + @markers.aws.validated |
| 7482 | + def test_pre_signed_url_if_match(self, s3_bucket, aws_client, aws_session): |
| 7483 | + key = "test-precondition" |
| 7484 | + aws_client.s3.put_object(Bucket=s3_bucket, Key=key, Body="test") |
| 7485 | + |
| 7486 | + s3_endpoint_path_style = _endpoint_url() |
| 7487 | + # empty object ETag is provided |
| 7488 | + empty_object_etag = "d41d8cd98f00b204e9800998ecf8427e" |
| 7489 | + |
| 7490 | + # assert that the regular Boto3 client does not work, and does not sign the parameter as requested |
| 7491 | + client = _s3_client_pre_signed_client( |
| 7492 | + Config(signature_version="s3v4", s3={}), |
| 7493 | + endpoint_url=s3_endpoint_path_style, |
| 7494 | + ) |
| 7495 | + bad_url = client.generate_presigned_url( |
| 7496 | + "put_object", |
| 7497 | + Params={"Bucket": s3_bucket, "Key": key, "IfMatch": empty_object_etag}, |
| 7498 | + ) |
| 7499 | + assert "if-match=d41d8cd98f00b204e9800998ecf8427e" not in bad_url.lower() |
| 7500 | + |
| 7501 | + req = botocore.awsrequest.AWSRequest( |
| 7502 | + method="PUT", |
| 7503 | + url=f"{s3_endpoint_path_style}/{s3_bucket}/{key}", |
| 7504 | + data={}, |
| 7505 | + params={ |
| 7506 | + "If-Match": empty_object_etag, |
| 7507 | + }, |
| 7508 | + headers={}, |
| 7509 | + ) |
| 7510 | + |
| 7511 | + botocore.auth.S3SigV4QueryAuth(aws_session.get_credentials(), "s3", "us-east-1").add_auth( |
| 7512 | + req |
| 7513 | + ) |
| 7514 | + assert "if-match=d41d8cd98f00b204e9800998ecf8427e" in req.url.lower() |
| 7515 | + |
| 7516 | + response = requests.put(req.url) |
| 7517 | + assert response.status_code == 412 |
| 7518 | + |
7437 | 7519 |
|
7438 | 7520 | class TestS3DeepArchive:
|
7439 | 7521 | """
|
|
0 commit comments