Fix s3bucket_path (#784)
* check HTTP header for IPv4 or IPv6 addresses and default to path based S3 * improved IPv4 and IPv6 checking with optional ports * typo * subdomain bucket creation with trailing '/' did not work * Use regex for Host field check to determine IPv4/IPv6 * add testcases for trailing slash, IPv4 and IPv6
This commit is contained in:
parent
c54985a39f
commit
5dc8e59fab
@ -5,7 +5,6 @@ import re
|
||||
import six
|
||||
from six.moves.urllib.parse import parse_qs, urlparse
|
||||
|
||||
import socket
|
||||
import xmltodict
|
||||
|
||||
from moto.core.responses import _TemplateEnvironmentMixin
|
||||
@ -57,21 +56,17 @@ class ResponseObject(_TemplateEnvironmentMixin):
|
||||
|
||||
match = re.match(r'^([^\[\]:]+)(:\d+)?$', host)
|
||||
if match:
|
||||
try:
|
||||
socket.inet_pton(socket.AF_INET, match.groups()[0])
|
||||
# For IPv4, default to path-based buckets
|
||||
match = re.match(r'((25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)(\.|$)){4}',
|
||||
match.groups()[0])
|
||||
if match:
|
||||
return False
|
||||
except socket.error:
|
||||
pass
|
||||
|
||||
match = re.match(r'^\[(.+)\](:\d+)?$', host)
|
||||
if match:
|
||||
try:
|
||||
socket.inet_pton(socket.AF_INET6, match.groups()[0])
|
||||
# For IPv6, default to path-based buckets
|
||||
match = re.match(r'^(((?=.*(::))(?!.*\3.+\3))\3?|[\dA-F]{1,4}:)([\dA-F]{1,4}(\3|:\b)|\2){5}(([\dA-F]{1,4}(\3|:\b|$)|\2){2}|(((2[0-4]|1\d|[1-9])?\d|25[0-5])\.?\b){4})\Z',
|
||||
match.groups()[0], re.IGNORECASE)
|
||||
if match:
|
||||
return False
|
||||
except socket.error:
|
||||
pass
|
||||
|
||||
path_based = (host == 's3.amazonaws.com' or re.match(r"s3[\.\-]([^.]*)\.amazonaws\.com", host))
|
||||
return not path_based
|
||||
|
@ -7,13 +7,23 @@ url_bases = [
|
||||
"https?://(?P<bucket_name>[a-zA-Z0-9\-_.]*)\.?s3(.*).amazonaws.com"
|
||||
]
|
||||
|
||||
|
||||
def ambiguous_response1(*args, **kwargs):
|
||||
return S3ResponseInstance.ambiguous_response(*args, **kwargs)
|
||||
|
||||
|
||||
def ambiguous_response2(*args, **kwargs):
|
||||
return S3ResponseInstance.ambiguous_response(*args, **kwargs)
|
||||
|
||||
|
||||
url_paths = {
|
||||
# subdomain bucket
|
||||
'{0}/$': S3ResponseInstance.bucket_response,
|
||||
|
||||
# subdomain key of path-based bucket
|
||||
'{0}/(?P<key_or_bucket_name>[^/]+)/?$': S3ResponseInstance.ambiguous_response,
|
||||
|
||||
'{0}/(?P<key_or_bucket_name>[^/]+)$': ambiguous_response1,
|
||||
# subdomain key of path-based bucket
|
||||
'{0}/(?P<key_or_bucket_name>[^/]+)/$': ambiguous_response2,
|
||||
# path-based bucket + key
|
||||
'{0}/(?P<bucket_name_path>[^/]+)/(?P<key_name>.+)': S3ResponseInstance.key_response,
|
||||
}
|
||||
|
@ -31,6 +31,16 @@ def test_s3_server_bucket_create():
|
||||
res.status_code.should.equal(200)
|
||||
res.data.should.contain(b"ListBucketResult")
|
||||
|
||||
res = test_client.put('/foobar2/', 'http://localhost:5000')
|
||||
res.status_code.should.equal(200)
|
||||
|
||||
res = test_client.get('/')
|
||||
res.data.should.contain(b'<Name>foobar2</Name>')
|
||||
|
||||
res = test_client.get('/foobar2/', 'http://localhost:5000')
|
||||
res.status_code.should.equal(200)
|
||||
res.data.should.contain(b"ListBucketResult")
|
||||
|
||||
res = test_client.get('/missing-bucket', 'http://localhost:5000')
|
||||
res.status_code.should.equal(404)
|
||||
|
||||
@ -57,3 +67,37 @@ def test_s3_server_post_to_bucket():
|
||||
res = test_client.get('/foobar2/the-key', 'http://localhost:5000/')
|
||||
res.status_code.should.equal(200)
|
||||
res.data.should.equal(b"nothing")
|
||||
|
||||
|
||||
def test_s3_server_put_ipv6():
|
||||
backend = server.create_backend_app("s3bucket_path")
|
||||
test_client = backend.test_client()
|
||||
|
||||
res = test_client.put('/foobar2', 'http://[::]:5000/')
|
||||
res.status_code.should.equal(200)
|
||||
|
||||
test_client.post('/foobar2', "https://[::]:5000/", data={
|
||||
'key': 'the-key',
|
||||
'file': 'nothing'
|
||||
})
|
||||
|
||||
res = test_client.get('/foobar2/the-key', 'http://[::]:5000/')
|
||||
res.status_code.should.equal(200)
|
||||
res.data.should.equal(b"nothing")
|
||||
|
||||
|
||||
def test_s3_server_put_ipv4():
|
||||
backend = server.create_backend_app("s3bucket_path")
|
||||
test_client = backend.test_client()
|
||||
|
||||
res = test_client.put('/foobar2', 'http://127.0.0.1:5000/')
|
||||
res.status_code.should.equal(200)
|
||||
|
||||
test_client.post('/foobar2', "https://127.0.0.1:5000/", data={
|
||||
'key': 'the-key',
|
||||
'file': 'nothing'
|
||||
})
|
||||
|
||||
res = test_client.get('/foobar2/the-key', 'http://127.0.0.1:5000/')
|
||||
res.status_code.should.equal(200)
|
||||
res.data.should.equal(b"nothing")
|
||||
|
Loading…
Reference in New Issue
Block a user