Yasuke/moto
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

S3: Implement CORS headers in OPTIONS requests (#4497)

Browse Source
This commit is contained in:
Vincent Barbaresi 2021-10-30 12:02:30 +02:00 committed by GitHub
parent b0c22c6ac1
commit 03c170e206
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 134 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
moto/s3/models.py
View File

@ -1015,8 +1015,8 @@ class FakeBucket(CloudFormationModel):
assert isinstance(rule.get("AllowedHeader", []), list) or isinstance(
rule.get("AllowedHeader", ""), str
)
assert isinstance(rule.get("ExposedHeader", []), list) or isinstance(
rule.get("ExposedHeader", ""), str
assert isinstance(rule.get("ExposeHeader", []), list) or isinstance(
rule.get("ExposeHeader", ""), str
)
assert isinstance(rule.get("MaxAgeSeconds", "0"), str)
@ -1034,8 +1034,8 @@ class FakeBucket(CloudFormationModel):
rule["AllowedMethod"],
rule["AllowedOrigin"],
rule.get("AllowedHeader"),
rule.get("ExposedHeader"),
rule.get("MaxAgeSecond"),
rule.get("ExposeHeader"),
rule.get("MaxAgeSeconds"),
)
)

75
moto/s3/responses.py
View File

@ -1,6 +1,7 @@
import io
import os
import re
from typing import List, Union
from botocore.awsrequest import AWSPreparedRequest
@ -258,9 +259,10 @@ class ResponseObject(_TemplateEnvironmentMixin, ActionAuthenticatorMixin):
def bucket_response(self, request, full_url, headers):
self.method = request.method
self.path = self._get_path(request)
self.headers = request.headers
if "host" not in self.headers:
self.headers["host"] = urlparse(full_url).netloc
# Make a copy of request.headers because it's immutable
self.headers = dict(request.headers)
if "Host" not in self.headers:
self.headers["Host"] = urlparse(full_url).netloc
try:
response = self._bucket_response(request, full_url, headers)
except S3ClientError as s3error:
@ -315,6 +317,8 @@ class ResponseObject(_TemplateEnvironmentMixin, ActionAuthenticatorMixin):
return self._bucket_response_delete(body, bucket_name, querystring)
elif method == "POST":
return self._bucket_response_post(request, body, bucket_name)
elif method == "OPTIONS":
return self._bucket_response_options(bucket_name)
else:
raise NotImplementedError(
"Method {0} has not been implemented in the S3 backend yet".format(
@ -342,6 +346,64 @@ class ResponseObject(_TemplateEnvironmentMixin, ActionAuthenticatorMixin):
return 404, {}, ""
return 200, {}, ""
def _set_cors_headers(self, bucket):
"""
TODO: smarter way of matching the right CORS rule:
See https://docs.aws.amazon.com/AmazonS3/latest/userguide/cors.html
"When Amazon S3 receives a preflight request from a browser, it evaluates
the CORS configuration for the bucket and uses the first CORSRule rule
that matches the incoming browser request to enable a cross-origin request."
This here just uses all rules and the last rule will override the previous ones
if they are re-defining the same headers.
"""
def _to_string(header: Union[List[str], str]) -> str:
# We allow list and strs in header values. Transform lists in comma-separated strings
if isinstance(header, list):
return ", ".join(header)
return header
for cors_rule in bucket.cors:
if cors_rule.allowed_methods is not None:
self.headers["Access-Control-Allow-Methods"] = _to_string(
cors_rule.allowed_methods
)
if cors_rule.allowed_origins is not None:
self.headers["Access-Control-Allow-Origin"] = _to_string(
cors_rule.allowed_origins
)
if cors_rule.allowed_headers is not None:
self.headers["Access-Control-Allow-Headers"] = _to_string(
cors_rule.allowed_headers
)
if cors_rule.exposed_headers is not None:
self.headers["Access-Control-Expose-Headers"] = _to_string(
cors_rule.exposed_headers
)
if cors_rule.max_age_seconds is not None:
self.headers["Access-Control-Max-Age"] = _to_string(
cors_rule.max_age_seconds
)
return self.headers
def _bucket_response_options(self, bucket_name):
# Return 200 with the headers from the bucket CORS configuration
self._authenticate_and_authorize_s3_action()
try:
bucket = self.backend.head_bucket(bucket_name)
except MissingBucket:
return (
403,
{},
"",
) # AWS S3 seems to return 403 on OPTIONS and 404 on GET/HEAD
self._set_cors_headers(bucket)
return 200, self.headers, ""
def _bucket_response_get(self, bucket_name, querystring):
self._set_action("BUCKET", "GET", querystring)
self._authenticate_and_authorize_s3_action()
@ -1033,9 +1095,10 @@ class ResponseObject(_TemplateEnvironmentMixin, ActionAuthenticatorMixin):
# Key and Control are lumped in because splitting out the regex is too much of a pain :/
self.method = request.method
self.path = self._get_path(request)
self.headers = request.headers
if "host" not in self.headers:
self.headers["host"] = urlparse(full_url).netloc
# Make a copy of request.headers because it's immutable
self.headers = dict(request.headers)
if "Host" not in self.headers:
self.headers["Host"] = urlparse(full_url).netloc
response_headers = {}
try:

2
moto/server.py
View File

@ -18,7 +18,7 @@ import moto.backends as backends
import moto.backend_index as backend_index
from moto.core.utils import convert_flask_to_httpretty_response
HTTP_METHODS = ["GET", "POST", "PUT", "DELETE", "HEAD", "PATCH"]
HTTP_METHODS = ["GET", "POST", "PUT", "DELETE", "HEAD", "PATCH", "OPTIONS"]
DEFAULT_SERVICE_REGION = ("s3", "us-east-1")

61
tests/test_s3/test_server.py
View File

@ -153,7 +153,10 @@ def test_s3_server_post_unicode_bucket_key():
def test_s3_server_post_cors():
"""Test default CORS headers set by flask-cors plugin"""
test_client = authenticated_client()
# Create the bucket
test_client.put("/", "http://tester.localhost:5000/")
preflight_headers = {
"Access-Control-Request-Method": "POST",
@ -167,7 +170,6 @@ def test_s3_server_post_cors():
assert res.status_code in [200, 204]
expected_methods = set(["DELETE", "PATCH", "PUT", "GET", "HEAD", "POST", "OPTIONS"])
assert set(res.headers["Allow"].split(", ")) == expected_methods
assert (
set(res.headers["Access-Control-Allow-Methods"].split(", ")) == expected_methods
)
@ -178,3 +180,60 @@ def test_s3_server_post_cors():
res.headers.should.have.key("Access-Control-Allow-Headers").which.should.equal(
"origin, x-requested-with"
)
def test_s3_server_post_cors_exposed_header():
"""Test that we can override default CORS headers with custom bucket rules"""
# github.com/spulec/moto/issues/4220
cors_config_payload = """<CORSConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
<CORSRule>
<AllowedOrigin>https://example.org</AllowedOrigin>
<AllowedMethod>HEAD</AllowedMethod>
<AllowedMethod>GET</AllowedMethod>
<AllowedMethod>PUT</AllowedMethod>
<AllowedMethod>POST</AllowedMethod>
<AllowedMethod>DELETE</AllowedMethod>
<AllowedHeader>*</AllowedHeader>
<ExposeHeader>ETag</ExposeHeader>
<MaxAgeSeconds>3000</MaxAgeSeconds>
</CORSRule>
</CORSConfiguration>
"""
test_client = authenticated_client()
preflight_headers = {
"Access-Control-Request-Method": "POST",
"Access-Control-Request-Headers": "origin, x-requested-with",
"Origin": "https://localhost:9000",
}
# Returns 403 on non existing bucket
preflight_response = test_client.options(
"/", "http://testcors.localhost:5000/", headers=preflight_headers
)
assert preflight_response.status_code == 403
# Create the bucket
test_client.put("/", "http://testcors.localhost:5000/")
res = test_client.put(
"/?cors", "http://testcors.localhost:5000", data=cors_config_payload
)
assert res.status_code == 200
cors_res = test_client.get("/?cors", "http://testcors.localhost:5000")
assert b"<ExposedHeader>ETag</ExposedHeader>" in cors_res.data
preflight_response = test_client.options(
"/", "http://testcors.localhost:5000/", headers=preflight_headers
)
assert preflight_response.status_code == 200
expected_cors_headers = {
"Access-Control-Allow-Methods": "HEAD, GET, PUT, POST, DELETE",
"Access-Control-Allow-Origin": "https://example.org",
"Access-Control-Allow-Headers": "*",
"Access-Control-Expose-Headers": "ETag",
"Access-Control-Max-Age": "3000",
}
for header_name, header_value in expected_cors_headers.items():
assert header_name in preflight_response.headers
assert preflight_response.headers[header_name] == header_value