How to use parse_policy_expiration_date method in localstack

Best Python code snippet using localstack_python

s3_listener.py

Source:s3_listener.py

...976            or allowed_query977        ):978            return True979    @staticmethod980    def parse_policy_expiration_date(expiration_string):981        try:982            dt = datetime.datetime.strptime(expiration_string, POLICY_EXPIRATION_FORMAT1)983        except Exception:984            dt = datetime.datetime.strptime(expiration_string, POLICY_EXPIRATION_FORMAT2)985        # both date formats assume a UTC timezone ('Z' suffix), but it's not parsed as tzinfo into the datetime object986        dt = dt.replace(tzinfo=datetime.timezone.utc)987        return dt988    def forward_request(self, method, path, data, headers):989        # Create list of query parameteres from the url990        parsed = urlparse.urlparse("{}{}".format(config.get_edge_url(), path))991        query_params = parse_qs(parsed.query)992        path_orig = path993        path = path.replace(994            "#", "%23"995        )  # support key names containing hashes (e.g., required by Amplify)996        # extracting bucket name from the request997        parsed_path = urlparse.urlparse(path)998        bucket_name = extract_bucket_name(headers, parsed_path.path)999        if method == "PUT" and bucket_name and not re.match(BUCKET_NAME_REGEX, bucket_name):1000            if len(parsed_path.path) <= 1:1001                return error_response(1002                    "Unable to extract valid bucket name. Please ensure that your AWS SDK is "1003                    + "configured to use path style addressing, or send a valid "1004                    + '<Bucket>.s3.localhost.localstack.cloud "Host" header',1005                    "InvalidBucketName",1006                    status_code=400,1007                )1008            return error_response(1009                "The specified bucket is not valid.",1010                "InvalidBucketName",1011                status_code=400,1012            )1013        # Detecting pre-sign url and checking signature1014        if any([p in query_params for p in SIGNATURE_V2_PARAMS]) or any(1015            [p in query_params for p in SIGNATURE_V4_PARAMS]1016        ):1017            response = authenticate_presign_url(1018                method=method, path=path, data=data, headers=headers1019            )1020            if response is not None:1021                return response1022        # handling s3 website hosting requests1023        if is_static_website(headers) and method == "GET":1024            return serve_static_website(headers=headers, path=path, bucket_name=bucket_name)1025        # check content md5 hash integrity if not a copy request or multipart initialization1026        if (1027            "Content-MD5" in headers1028            and not self.is_s3_copy_request(headers, path)1029            and not self.is_create_multipart_request(parsed_path.query)1030        ):1031            response = check_content_md5(data, headers)1032            if response is not None:1033                return response1034        modified_data = None1035        # TODO: For some reason, moto doesn't allow us to put a location constraint on us-east-11036        to_find1 = to_bytes("<LocationConstraint>us-east-1</LocationConstraint>")1037        to_find2 = to_bytes("<CreateBucketConfiguration")1038        if data and data.startswith(to_bytes("<")) and to_find1 in data and to_find2 in data:1039            # Note: with the latest version, <CreateBucketConfiguration> must either1040            # contain a valid <LocationConstraint>, or not be present at all in the body.1041            modified_data = b""1042        # If this request contains streaming v4 authentication signatures, strip them from the message1043        # Related isse: https://github.com/localstack/localstack/issues/981044        # TODO: can potentially be removed after this fix in moto: https://github.com/spulec/moto/pull/42011045        is_streaming_payload = headers.get(CONTENT_SHA256_HEADER) == STREAMING_HMAC_PAYLOAD1046        if is_streaming_payload:1047            modified_data = strip_chunk_signatures(not_none_or(modified_data, data))1048            headers["Content-Length"] = headers.get("x-amz-decoded-content-length")1049            headers.pop(CONTENT_SHA256_HEADER)1050        # POST requests to S3 may include a "${filename}" placeholder in the1051        # key, which should be replaced with an actual file name before storing.1052        if method == "POST":1053            original_data = not_none_or(modified_data, data)1054            expanded_data = multipart_content.expand_multipart_filename(original_data, headers)1055            if expanded_data is not original_data:1056                modified_data = expanded_data1057        # If no content-type is provided, 'binary/octet-stream' should be used1058        # src: https://docs.aws.amazon.com/AmazonS3/latest/API/RESTObjectPUT.html1059        if method == "PUT" and not headers.get("content-type"):1060            headers["content-type"] = "binary/octet-stream"1061        # parse query params1062        query = parsed_path.query1063        path = parsed_path.path1064        query_map = urlparse.parse_qs(query, keep_blank_values=True)1065        # remap metadata query params (not supported in moto) to request headers1066        append_metadata_headers(method, query_map, headers)1067        # apply fixes1068        headers_changed = fix_metadata_key_underscores(request_headers=headers)1069        if query == "notification" or "notification" in query_map:1070            # handle and return response for ?notification request1071            response = handle_notification_request(bucket_name, method, data)1072            return response1073        # if the Expires key in the url is already expired then return error1074        if method == "GET" and "Expires" in query_map:1075            ts = datetime.datetime.fromtimestamp(1076                int(query_map.get("Expires")[0]), tz=datetime.timezone.utc1077            )1078            if is_expired(ts):1079                return token_expired_error(path, headers.get("x-amz-request-id"), 400)1080        # If multipart POST with policy in the params, return error if the policy has expired1081        if method == "POST":1082            policy_key, policy_value = multipart_content.find_multipart_key_value(1083                data, headers, "policy"1084            )1085            if policy_key and policy_value:1086                policy = json.loads(base64.b64decode(policy_value).decode("utf-8"))1087                expiration_string = policy.get("expiration", None)  # Example: 2020-06-05T13:37:12Z1088                if expiration_string:1089                    expiration_datetime = self.parse_policy_expiration_date(expiration_string)1090                    if is_expired(expiration_datetime):1091                        return token_expired_error(path, headers.get("x-amz-request-id"), 400)1092        if query == "cors" or "cors" in query_map:1093            if method == "GET":1094                return get_cors(bucket_name)1095            if method == "PUT":1096                return set_cors(bucket_name, data)1097            if method == "DELETE":1098                return delete_cors(bucket_name)1099        if query == "requestPayment" or "requestPayment" in query_map:1100            if method == "GET":1101                return get_request_payment(bucket_name)1102            if method == "PUT":1103                return set_request_payment(bucket_name, data)...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.