How to use uses_host_addressing method in localstack

Best Python code snippet using localstack_python

s3_utils.py

Source:s3_utils.py

...71    returns True if the host matches website regex72    returns False if the host does not matches website regex73    """74    return bool(re.match(S3_STATIC_WEBSITE_HOST_REGEX, headers.get("host", "")))75def uses_host_addressing(headers):76    """77    Determines if the bucket is using host based addressing style or path based78    """79    # we can assume that the host header we are receiving here is actually the header we originally received80    # from the client (because the edge service is forwarding the request in memory)81    match = re.match(S3_VIRTUAL_HOSTNAME_REGEX, headers.get("host", ""))82    # checks whether there is a bucket name. This is sort of hacky83    return True if match and match.group(3) else False84def extract_bucket_name(headers, path):85    """86    Extract the bucket name87    if using host based addressing it's extracted from host header88    if using path based addressing it's extracted form the path89    """90    bucket_name = None91    if uses_host_addressing(headers):92        pattern = re.compile(S3_VIRTUAL_HOSTNAME_REGEX)93        match = pattern.match(headers.get("host", ""))94        if match and match.group(3):95            bucket_name = match.group(3)96    else:97        bucket_name = path.split("/", 2)[1]98    return bucket_name if bucket_name else None99def extract_key_name(headers, path):100    """101    Extract the key name from the path depending on addressing_style102    """103    key_name = None104    path = path.split("?")[0]  # strip off query params from path105    if uses_host_addressing(headers):106        split = path.split("/", 1)107        if len(split) > 1:108            key_name = split[1]109    else:110        split = path.split("/", 2)111        if len(split) > 2:112            key_name = split[2]113    return key_name if key_name else None114def extract_bucket_and_key_name(headers, path):115    return extract_bucket_name(headers, path), extract_key_name(headers, path)116def normalize_bucket_name(bucket_name):117    bucket_name = bucket_name or ""118    bucket_name = bucket_name.lower()119    return bucket_name120def validate_bucket_name(bucket_name):121    """122    Validate s3 bucket name based on the documentation123    ref. https://docs.aws.amazon.com/AmazonS3/latest/userguide/bucketnamingrules.html124    """125    return True if re.match(BUCKET_NAME_REGEX, bucket_name) else False126def get_bucket_hostname(bucket_name):127    """128    Get bucket name for addressing style host129    """130    return "%s.%s:%s" % (bucket_name, S3_VIRTUAL_HOSTNAME, config.EDGE_PORT)131def get_bucket_website_hostname(bucket_name):132    """133    Get bucket name for addressing style host for website hosting134    """135    return "%s.%s:%s" % (bucket_name, S3_STATIC_WEBSITE_HOSTNAME, config.EDGE_PORT)136def get_forwarded_for_host(headers):137    x_forwarded_header = re.split(r",\s?", headers.get("X-Forwarded-For", ""))138    host = x_forwarded_header[len(x_forwarded_header) - 1]139    return host140def is_real_s3_url(url):141    return re.match(r".*s3(\-website)?\.([^\.]+\.)?amazonaws.com.*", url or "")142def is_expired(expiry_datetime):143    now_datetime = datetime.datetime.now(tz=expiry_datetime.tzinfo)144    return now_datetime > expiry_datetime145def authenticate_presign_url(method, path, headers, data=None):146    url = "{}{}".format(config.get_edge_url(), path)147    parsed = urlparse.urlparse(url)148    query_params = parse_qs(parsed.query)149    forwarded_for = get_forwarded_for_host(headers)150    if forwarded_for:151        url = re.sub("://[^/]+", "://%s" % forwarded_for, url)152    LOGGER.debug("Received presign S3 URL: %s" % url)153    sign_headers = {}154    query_string = {}155    is_v2 = all([p in query_params for p in SIGNATURE_V2_PARAMS])156    is_v4 = all([p in query_params for p in SIGNATURE_V4_PARAMS])157    # Add overrided headers to the query string params158    for param_name, header_name in ALLOWED_HEADER_OVERRIDES.items():159        if param_name in query_params:160            query_string[param_name] = query_params[param_name][0]161    # Request's headers are more essentials than the query parameters in the request.162    # Different values of header in the header of the request and in the query parameter of the163    # request URL will fail the signature calulation. As per the AWS behaviour164    # Add valid headers into the sign_header. Skip the overrided headers165    # and the headers which have been sent in the query string param166    presign_params_lower = (167        [p.lower() for p in SIGNATURE_V4_PARAMS]168        if is_v4169        else [p.lower() for p in SIGNATURE_V2_PARAMS]170    )171    params_header_override = [172        param_name for param_name, header_name in ALLOWED_HEADER_OVERRIDES.items()173    ]174    if len(query_params) > 2:175        for key in query_params:176            key_lower = key.lower()177            if key_lower not in presign_params_lower:178                if (179                    key_lower not in (header[0].lower() for header in headers)180                    and key_lower not in params_header_override181                ):182                    if key_lower in (183                        allowed_param.lower() for allowed_param in ALLOWED_QUERY_PARAMS184                    ):185                        query_string[key] = query_params[key][0]186                    elif key_lower in (187                        blacklisted_header.lower() for blacklisted_header in BLACKLISTED_HEADERS188                    ):189                        pass190                    else:191                        query_string[key] = query_params[key][0]192    for header_name, header_value in headers.items():193        header_name_lower = header_name.lower()194        if header_name_lower.startswith("x-amz-") or header_name_lower.startswith("content-"):195            if is_v2 and header_name_lower in query_params:196                sign_headers[header_name] = header_value197            if is_v4 and header_name_lower in query_params["X-Amz-SignedHeaders"][0]:198                sign_headers[header_name] = header_value199    # Preparnig dictionary of request to build AWSRequest's object of the botocore200    request_url = "{}://{}{}".format(parsed.scheme, parsed.netloc, parsed.path)201    # Fix https://github.com/localstack/localstack/issues/3912202    # urlencode method replaces white spaces with plus sign cause signature calculation to fail203    request_url = (204        "%s?%s" % (request_url, urlencode(query_string, quote_via=urlparse.quote, safe=" "))205        if query_string206        else request_url207    )208    if forwarded_for:209        request_url = re.sub("://[^/]+", "://%s" % forwarded_for, request_url)210    bucket_name = extract_bucket_name(headers, parsed.path)211    request_dict = {212        "url_path": parsed.path,213        "query_string": query_string,214        "method": method,215        "headers": sign_headers,216        "body": b"",217        "url": request_url,218        "context": {219            "is_presign_request": True,220            "use_global_endpoint": True,221            "signing": {"bucket": bucket_name},222        },223    }224    # Support for virtual host addressing style in signature version 2225    # We don't need to do this in v4 as we already concerting it to the virtual addressing style.226    # v2 require path base styled request_dict and v4 require virtual styled request_dict227    if uses_host_addressing(headers) and is_v2:228        request_dict["url_path"] = "/{}{}".format(bucket_name, request_dict["url_path"])229        parsed_url = urlparse.urlparse(request_url)230        request_dict["url"] = "{}://{}:{}{}".format(231            parsed_url.scheme,232            S3_VIRTUAL_HOSTNAME,233            config.EDGE_PORT,234            request_dict["url_path"],235        )236        request_dict["url"] = (237            "%s?%s" % (request_dict["url"], urlencode(query_string))238            if query_string239            else request_dict["url"]240        )241    if not is_v2 and any([p in query_params for p in SIGNATURE_V2_PARAMS]):...

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.