Unlock 30% off on Manual Testing Annual Plans this Holiday Season.

Copied to Clipboard!

How to use _extract_service_indicators method in localstack

Best Python code snippet using localstack_python

service_router.py

Source:service_router.py

...26    # Host field of the HTTP request27    host: Optional[str] = None28    # Path of the HTTP request29    path: Optional[str] = None30def _extract_service_indicators(request: Request) -> _ServiceIndicators:31    """Extracts all different fields that might indicate which service a request is targeting."""32    x_amz_target = request.headers.get("x-amz-target")33    authorization = request.headers.get("authorization")34    signing_name = None35    if authorization:36        try:37            auth_type, auth_info = authorization.split(None, 1)38            auth_type = auth_type.lower().strip()39            if auth_type == "aws4-hmac-sha256":40                values = parse_dict_header(auth_info)41                _, _, _, signing_name, _ = values["Credential"].split("/")42        except (ValueError, KeyError):43            LOG.debug("auth header could not be parsed for service routing: %s", authorization)44            pass45    if x_amz_target:46        if "." in x_amz_target:47            target_prefix, operation = x_amz_target.split(".", 1)48        else:49            target_prefix = None50            operation = x_amz_target51    else:52        target_prefix, operation = None, None53    return _ServiceIndicators(signing_name, target_prefix, operation, request.host, request.path)54signing_name_path_prefix_rules = {55    # custom rules based on URI path prefixes that are not easily generalizable56    "apigateway": {57        "/v2": "apigatewayv2",58    },59    "appconfig": {60        "/configuration": "appconfigdata",61    },62    "execute-api": {63        "/@connections": "apigatewaymanagementapi",64        "/participant": "connectparticipant",65        "*": "iot",66    },67    "ses": {68        "/v2": "sesv2",69        "/v1": "pinpoint-email",70    },71    "greengrass": {"/greengrass/v2/": "greengrassv2"},72    "cloudsearch": {"/2013-01-01": "cloudsearchdomain"},73    "s3": {"/v20180820": "s3control"},74    "iot1click": {75        "/projects": "iot1click-projects",76        "/devices": "iot1click-devices",77    },78    "es": {79        "/2015-01-01": "es",80        "/2021-01-01": "opensearch",81    },82}83def custom_signing_name_rules(signing_name: str, path: str) -> Optional[str]:84    """85    Rules which are based on the signing name (in the auth header) and the request path.86    """87    rules = signing_name_path_prefix_rules.get(signing_name)88    if not rules:89        if signing_name == "servicecatalog":90            if path == "/":91                # servicecatalog uses the protocol json (only uses root-path URIs, i.e. only /)92                return "servicecatalog"93            else:94                # servicecatalog-appregistry uses rest-json (only uses non-root-path request URIs)95                return "servicecatalog-appregistry"96        return97    for prefix, name in rules.items():98        if path.startswith(prefix):99            return name100    return rules.get("*", signing_name)101def custom_host_addressing_rules(host: str) -> Optional[str]:102    """103    Rules based on the host header of the request.104    """105    if ".execute-api." in host:106        return "apigateway"107def custom_path_addressing_rules(path: str) -> Optional[str]:108    """109    Rules which are only based on the request path.110    """111    if is_sqs_queue_url(path):112        return "sqs"113    if path.startswith("/2015-03-31/functions/"):114        return "lambda"115def legacy_rules(request: Request) -> Optional[str]:116    """117    *Legacy* rules which migrate routing logic which will become obsolete with the ASF Gateway.118    All rules which are implemented here should be migrated to the new router once these services are migrated to ASF.119    TODO: These custom rules should become obsolete by migrating these to use the http/router.py120    """121    path = request.path122    method = request.method123    host = hostname_from_url(request.host)124    # API Gateway invocation URLs125    if ("/%s/" % PATH_USER_REQUEST) in request.path or (126        host.endswith(LOCALHOST_HOSTNAME) and "execute-api" in host127    ):128        return "apigateway"129    # DynamoDB shell URLs130    if path.startswith("/shell") or path.startswith("/dynamodb/shell"):131        return "dynamodb"132    # TODO The remaining rules here are special S3 rules - needs to be discussed how these should be handled.133    #      Some are similar to other rules and not that greedy, others are nearly general fallbacks.134    stripped = path.strip("/")135    if method in ["GET", "HEAD"] and stripped:136        # assume that this is an S3 GET request with URL path `/<bucket>/<key ...>`137        return "s3"138    # detect S3 URLs139    if stripped and "/" not in stripped:140        if method == "PUT":141            # assume that this is an S3 PUT bucket request with URL path `/<bucket>`142            return "s3"143        if method == "POST" and "key" in request.values:144            # assume that this is an S3 POST request with form parameters or multipart form in the body145            return "s3"146    # detect S3 requests sent from aws-cli using --no-sign-request option147    if "aws-cli/" in str(request.user_agent):148        return "s3"149    # detect S3 pre-signed URLs150    values = request.values151    if "AWSAccessKeyId" in values or "Signature" in values:152        return "s3"153    # S3 delete object requests154    if method == "POST" and "delete" in values:155        data_bytes = to_bytes(request.get_data())156        if b"<Delete" in data_bytes and b"<Key>" in data_bytes:157            return "s3"158    # Put Object API can have multiple keys159    if stripped.count("/") >= 1 and method == "PUT":160        # assume that this is an S3 PUT bucket object request with URL path `/<bucket>/object`161        # or `/<bucket>/object/object1/+`162        return "s3"163    # detect S3 requests with "AWS id:key" Auth headers164    auth_header = request.headers.get("Authorization") or ""165    if auth_header.startswith("AWS "):166        return "s3"167    if uses_host_addressing(request.headers):168        # Note: This needs to be the last rule (and therefore is not in the host rules), since it is incredibly greedy169        return "s3"170@lru_cache()171def get_service_catalog() -> ServiceCatalog:172    """Loads the ServiceCatalog (which contains all the service specs)."""173    return ServiceCatalog()174def determine_aws_service_name(175    request: Request, services: ServiceCatalog = get_service_catalog()176) -> Optional[str]:177    """178    Tries to determine the name of the AWS service an incoming request is targeting.179    :param request: to determine the target service name of180    :param services: service catalog (can be handed in for caching purposes)181    :return: service name string (or None if the targeting service could not be determined exactly)182    """183    signing_name, target_prefix, operation, host, path = _extract_service_indicators(request)184    candidates = set()185    # 1. check the signing names186    if signing_name:187        signing_name_candidates = services.by_signing_name(signing_name)188        if len(signing_name_candidates) == 1:189            # a unique signing-name -> service name mapping is the case for ~75% of service operations190            return signing_name_candidates[0]191        # try to find a match with the custom signing name rules192        custom_match = custom_signing_name_rules(signing_name, path)193        if custom_match:194            return custom_match195        # still ambiguous - add the services to the list of candidates196        candidates.update(signing_name_candidates)197    # 2. check the target prefix...

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.