diff options
| -rw-r--r-- | swift/common/middleware/formpost.py | 10 | ||||
| -rw-r--r-- | test/unit/common/middleware/test_formpost.py | 124 |
2 files changed, 129 insertions, 5 deletions
diff --git a/swift/common/middleware/formpost.py b/swift/common/middleware/formpost.py index 89c1618b0..9172aa82d 100644 --- a/swift/common/middleware/formpost.py +++ b/swift/common/middleware/formpost.py @@ -129,13 +129,14 @@ from time import time import six from six.moves.urllib.parse import quote +from swift.common.constraints import valid_api_version from swift.common.exceptions import MimeInvalid from swift.common.middleware.tempurl import get_tempurl_keys_from_metadata from swift.common.utils import streq_const_time, register_swift_info, \ parse_content_disposition, parse_mime_headers, \ iter_multipart_mime_documents, reiterate, close_if_possible from swift.common.wsgi import make_pre_authed_env -from swift.common.swob import HTTPUnauthorized, wsgi_to_str +from swift.common.swob import HTTPUnauthorized, wsgi_to_str, str_to_wsgi from swift.proxy.controllers.base import get_account_info, get_container_info @@ -365,7 +366,8 @@ class FormPost(object): if not subenv['PATH_INFO'].endswith('/') and \ subenv['PATH_INFO'].count('/') < 4: subenv['PATH_INFO'] += '/' - subenv['PATH_INFO'] += attributes['filename'] or 'filename' + subenv['PATH_INFO'] += str_to_wsgi( + attributes['filename'] or 'filename') if 'x_delete_at' in attributes: try: subenv['HTTP_X_DELETE_AT'] = int(attributes['x_delete_at']) @@ -442,8 +444,8 @@ class FormPost(object): :returns: list of tempurl keys """ parts = env['PATH_INFO'].split('/', 4) - if len(parts) < 4 or parts[0] or parts[1] != 'v1' or not parts[2] or \ - not parts[3]: + if len(parts) < 4 or parts[0] or not valid_api_version(parts[1]) \ + or not parts[2] or not parts[3]: return [] account_info = get_account_info(env, self.app, swift_source='FP') diff --git a/test/unit/common/middleware/test_formpost.py b/test/unit/common/middleware/test_formpost.py index 75a4a6d55..c2d983cc5 100644 --- a/test/unit/common/middleware/test_formpost.py +++ b/test/unit/common/middleware/test_formpost.py @@ -21,7 +21,7 @@ from time import time import six from io import BytesIO -from swift.common.swob import Request, Response +from swift.common.swob import Request, Response, wsgi_quote from swift.common.middleware import tempauth, formpost from swift.common.utils import split_path from swift.proxy.controllers.base import get_cache_key @@ -49,6 +49,8 @@ class FakeApp(object): self.check_no_query_string = check_no_query_string def __call__(self, env, start_response): + # use wsgi_quote to spot check that it really *is* a WSGI string + wsgi_quote(env['PATH_INFO']) try: if self.check_no_query_string and env.get('QUERY_STRING'): raise Exception('Query string %s should have been discarded!' % @@ -756,6 +758,126 @@ class TestFormPost(unittest.TestCase): self.assertEqual(self.app.requests[0].body, b'Test File\nOne\n') self.assertEqual(self.app.requests[1].body, b'Test\nFile\nTwo\n') + def test_curl_with_unicode(self): + key = b'abc' + path = u'/v1/AUTH_test/container/let_it_\N{SNOWFLAKE}/' + if six.PY2: + path = path.encode('utf-8') + redirect = 'http://brim.net' + max_file_size = 1024 + max_file_count = 10 + expires = int(time() + 86400) + sig = hmac.new( + key, + hmac_msg(path, redirect, max_file_size, max_file_count, expires), + sha1).hexdigest() + wsgi_input = '\r\n'.join([ + '--------------------------dea19ac8502ca805', + 'Content-Disposition: form-data; name="redirect"', + '', + redirect, + '--------------------------dea19ac8502ca805', + 'Content-Disposition: form-data; name="max_file_size"', + '', + str(max_file_size), + '--------------------------dea19ac8502ca805', + 'Content-Disposition: form-data; name="max_file_count"', + '', + str(max_file_count), + '--------------------------dea19ac8502ca805', + 'Content-Disposition: form-data; name="expires"', + '', + str(expires), + '--------------------------dea19ac8502ca805', + 'Content-Disposition: form-data; name="signature"', + '', + sig, + '--------------------------dea19ac8502ca805', + 'Content-Disposition: form-data; name="file1"; ' + 'filename="\xe2\x98\x83.txt"', + 'Content-Type: text/plain', + '', + 'Test File\nOne\n', + '--------------------------dea19ac8502ca805', + 'Content-Disposition: form-data; name="file2"; ' + 'filename="testfile2.txt"', + 'Content-Type: text/plain', + '', + 'Test\nFile\nTwo\n', + '--------------------------dea19ac8502ca805', + 'Content-Disposition: form-data; name="file3"; filename=""', + 'Content-Type: application/octet-stream', + '', + '', + '--------------------------dea19ac8502ca805--', + '' + ]) + if not six.PY2: + wsgi_input = wsgi_input.encode('latin1') + wsgi_input = BytesIO(wsgi_input) + wsgi_errors = six.StringIO() + env = { + 'CONTENT_LENGTH': str(len(wsgi_input.getvalue())), + 'CONTENT_TYPE': 'multipart/form-data; ' + 'boundary=------------------------dea19ac8502ca805', + 'HTTP_ACCEPT': '*/*', + 'HTTP_HOST': 'ubuntu:8080', + 'HTTP_USER_AGENT': 'curl/7.58.0', + 'PATH_INFO': '/v1/AUTH_test/container/let_it_\xE2\x9D\x84/', + 'REMOTE_ADDR': '172.16.83.1', + 'REQUEST_METHOD': 'POST', + 'SCRIPT_NAME': '', + 'SERVER_NAME': '172.16.83.128', + 'SERVER_PORT': '8080', + 'SERVER_PROTOCOL': 'HTTP/1.0', + 'swift.infocache': { + get_cache_key('AUTH_test'): self._fake_cache_env( + 'AUTH_test', [key]), + get_cache_key('AUTH_test', 'container'): { + 'meta': {}}}, + 'wsgi.errors': wsgi_errors, + 'wsgi.input': wsgi_input, + 'wsgi.multiprocess': False, + 'wsgi.multithread': True, + 'wsgi.run_once': False, + 'wsgi.url_scheme': 'http', + 'wsgi.version': (1, 0), + } + self.app = FakeApp(iter([('201 Created', {}, b''), + ('201 Created', {}, b'')])) + self.auth = tempauth.filter_factory({})(self.app) + self.formpost = formpost.filter_factory({})(self.auth) + status = [None] + headers = [None] + exc_info = [None] + + def start_response(s, h, e=None): + status[0] = s + headers[0] = h + exc_info[0] = e + + body = b''.join(self.formpost(env, start_response)) + status = status[0] + headers = headers[0] + exc_info = exc_info[0] + self.assertEqual(status, '303 See Other', body) + location = None + for h, v in headers: + if h.lower() == 'location': + location = v + self.assertEqual(location, 'http://brim.net?status=201&message=') + self.assertIsNone(exc_info) + self.assertTrue(b'http://brim.net?status=201&message=' in body) + self.assertEqual(len(self.app.requests), 2) + self.assertEqual( + self.app.requests[0].path, + '/v1/AUTH_test/container/let_it_%E2%9D%84/%E2%98%83.txt') + self.assertEqual(self.app.requests[0].body, b'Test File\nOne\n') + self.assertEqual( + self.app.requests[1].path, + '/v1/AUTH_test/container/let_it_%E2%9D%84/testfile2.txt') + self.assertEqual(self.app.requests[1].body, b'Test\nFile\nTwo\n') + def test_messed_up_start(self): key = b'abc' sig, env, body = self._make_sig_env_body( |