Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use append_cors_headers method in localstack

Best Python code snippet using localstack_python

s3_listener.py

Source:s3_listener.py

...126    BUCKET_CORS.pop(bucket_name, {})127    response = Response()128    response.status_code = 200129    return response130def append_cors_headers(bucket_name, request_method, request_headers, response):131    cors = BUCKET_CORS.get(bucket_name)132    if not cors:133        return134    origin = request_headers.get('Origin', '')135    for rule in cors['CORSConfiguration']['CORSRule']:136        allowed_methods = rule.get('AllowedMethod', [])137        if request_method in allowed_methods:138            allowed_origins = rule.get('AllowedOrigin', [])139            for allowed in allowed_origins:140                if origin in allowed or re.match(allowed.replace('*', '.*'), origin):141                    response.headers['Access-Control-Allow-Origin'] = origin142                    break143def strip_chunk_signatures(data):144    # For clients that use streaming v4 authentication, the request contains chunk signatures145    # in the HTTP body (see example below) which we need to strip as moto cannot handle them146    #147    # 17;chunk-signature=6e162122ec4962bea0b18bc624025e6ae4e9322bdc632762d909e87793ac5921148    # <payload data ...>149    # 0;chunk-signature=927ab45acd82fc90a3c210ca7314d59fedc77ce0c914d79095f8cc9563cf2c70150    data_new = re.sub(r'^[0-9a-fA-F]+;chunk-signature=[0-9a-f]{64}', '', data, flags=re.MULTILINE)151    if data_new != data:152        # trim \r (13) or \n (10)153        for i in range(0, 2):154            if ord(data_new[0]) in (10, 13):155                data_new = data_new[1:]156        for i in range(0, 6):157            if ord(data_new[-1]) in (10, 13):158                data_new = data_new[:-1]159    return data_new160def update_s3(method, path, data, headers, response=None, return_forward_info=False):161    if return_forward_info:162        modified_data = None163        # If this request contains streaming v4 authentication signatures, strip them from the message164        # Related isse: https://github.com/atlassian/localstack/issues/98165        # TODO we should evaluate whether to replace moto s3 with scality/S3:166        # https://github.com/scality/S3/issues/237167        if headers.get('x-amz-content-sha256') == 'STREAMING-AWS4-HMAC-SHA256-PAYLOAD':168            modified_data = strip_chunk_signatures(data)169        # persist this API call to disk170        persistence.record('s3', method, path, data, headers)171        parsed = urlparse.urlparse(path)172        query = parsed.query173        path = parsed.path174        bucket = path.split('/')[1]175        query_map = urlparse.parse_qs(query)176        if method == 'PUT' and (query == 'notification' or 'notification' in query_map):177            tree = ET.fromstring(data)178            for dest in ['Queue', 'Topic', 'CloudFunction']:179                config = tree.find('{%s}%sConfiguration' % (XMLNS_S3, dest))180                if config is not None and len(config):181                    S3_NOTIFICATIONS[bucket] = {182                        'Id': get_xml_text(config, 'Id'),183                        'Event': get_xml_text(config, 'Event', ns=XMLNS_S3),184                        dest: get_xml_text(config, dest, ns=XMLNS_S3),185                    }186        if query == 'cors' or 'cors' in query_map:187            if method == 'GET':188                return get_cors(bucket)189            if method == 'PUT':190                return set_cors(bucket, data)191            if method == 'DELETE':192                return delete_cors(bucket)193        if modified_data:194            return Request(data=modified_data, headers=headers, method=method)195        return True196    # get subscribers and send bucket notifications197    if method in ('PUT', 'DELETE') and '/' in path[1:]:198        parts = path[1:].split('/', 1)199        bucket_name = parts[0]200        object_path = '/%s' % parts[1]201        send_notifications(method, bucket_name, object_path)202    # append CORS headers to response203    if response:204        parsed = urlparse.urlparse(path)205        bucket_name = parsed.path.split('/')[0]...

utils.py

Source:utils.py

...47        or mentor in token["mentorIds"]48    )49def create_json_response(status, data, event, headers={}):50    body = {"data": data}51    append_cors_headers(headers, event)52    append_secure_headers(headers)53    response = {"statusCode": status, "body": json.dumps(body), "headers": headers}54    return response55def append_secure_headers(headers):56    secure = {57        "content-security-policy": "upgrade-insecure-requests;",58        "referrer-policy": "no-referrer-when-downgrade",59        "strict-transport-security": "max-age=31536000",60        "x-content-type-options": "nosniff",61        "x-frame-options": "SAMEORIGIN",62        "x-xss-protection": "1; mode=block",63    }64    for h in secure:65        headers[h] = secure[h]66def append_cors_headers(headers, event):67    origin = environ.get("CORS_ORIGIN", "*")68    # TODO specify allowed list of origins and if event["headers"]["origin"] is one of them then allow it69    # if "origin" in event["headers"] and getenv.array('CORS_ORIGIN').includes(event["headers"]["origin"]):70    #     origin = event["headers"]["origin"]71    headers["Access-Control-Allow-Origin"] = origin72    headers["Access-Control-Allow-Origin"] = "*"73    headers["Access-Control-Allow-Headers"] = "GET,PUT,POST,DELETE,OPTIONS"74    headers[75        "Access-Control-Allow-Methods"76    ] = "Authorization,Origin,Accept,Accept-Language,Content-Language,Content-Type"77def props_to_bool(78    name: str, props: Union[Dict[str, Any], _Environ], dft: bool = False79) -> bool:80    if not (props and name in props):...

predict.py

Source:predict.py

...27MODELS_DIR = "/tmp/models"28classifier_dao = Dao(SHARED, MODELS_DIR)29def make_response(status, body, event):30    headers = {}31    append_cors_headers(headers, event)32    append_secure_headers(headers)33    response = {"statusCode": status, "body": json.dumps(body), "headers": headers}34    return response35def handler(event, context):36    log.debug(json.dumps(event))37    if "queryStringParameters" not in event:38        raise Exception("bad request")39    if (40        "mentor" not in event["queryStringParameters"]41        or "query" not in event["queryStringParameters"]42    ):43        raise Exception("bad request")44    mentor = event["queryStringParameters"]["mentor"]45    question = event["queryStringParameters"]["query"]...

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.