Merge "Fixed formpost QUERY_STRING bugs."

openstack/swift

Code Issues Proposed changes

Browse Source

This commit is contained in:

Jenkins

2013年03月01日 21:27:58 +00:00

committed by Gerrit Code Review

parent 9e006183f8 ffada71137

commit cc63f8d791

2 changed files with 131 additions and 9 deletions

Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL

Download Patch File Download Diff File Expand all files Collapse all files

22
swift/common/middleware/formpost.py

View File

@@ -377,7 +377,8 @@ class FormPost(object):

if not status:

status = '400 Bad Request'

message = 'no files to process'

if not attributes.get('redirect'):

redirect = attributes.get('redirect')

if not redirect:

body = status

if message:

body = status + '\r\nFormPost: ' + message.title()

@@ -385,13 +386,14 @@ class FormPost(object):

('Content-Length', len(body))]

return status, headers, body

status = status.split('', 1)[0]

body = '<html><body><p><a href="%s?status=%s&message=%s">Click to ' \

'continue...</a></p></body></html>' % \

(attributes['redirect'], quote(status), quote(message))

headers = [

('Location', '%s?status=%s&message=%s' % (

attributes['redirect'], quote(status), quote(message))),

('Content-Length', str(len(body)))]

if '?' in redirect:

redirect += '&'

else:

redirect += '?'

redirect += 'status=%s&message=%s' % (quote(status), quote(message))

body = '<html><body><p><a href="%s">' \

'Click to continue...</a></p></body></html>' % redirect

headers = [('Location', redirect), ('Content-Length', str(len(body)))]

return '303 See Other', headers, body

def _perform_subrequest(self, orig_env, attributes, fp, key):

@@ -413,6 +415,8 @@ class FormPost(object):

raise FormInvalid('max_file_size not an integer')

subenv = make_pre_authed_env(orig_env, 'PUT', agent=None,

swift_source='FP')

if 'QUERY_STRING' in subenv:

del subenv['QUERY_STRING']

subenv['HTTP_TRANSFER_ENCODING'] = 'chunked'

subenv['wsgi.input'] = _CappedFileLikeObject(fp, max_file_size)

if subenv['PATH_INFO'][-1] != '/' and \

@@ -471,6 +475,8 @@ class FormPost(object):

if not key:

newenv = make_pre_authed_env(env, 'HEAD', '/v1/' + account,

agent=None, swift_source='FP')

if 'QUERY_STRING' in newenv:

del newenv['QUERY_STRING']

newenv['CONTENT_LENGTH'] = '0'

newenv['wsgi.input'] = StringIO('')

key = [None]

118
test/unit/common/middleware/test_formpost.py

View File

@@ -54,7 +54,8 @@ class FakeMemcache(object):

class FakeApp(object):

def __init__(self, status_headers_body_iter=None):

def __init__(self, status_headers_body_iter=None,

check_no_query_string=True):

self.status_headers_body_iter = status_headers_body_iter

if not self.status_headers_body_iter:

self.status_headers_body_iter = iter([('404 Not Found', {

@@ -62,8 +63,12 @@ class FakeApp(object):

'x-test-header-two-a': 'value2',

'x-test-header-two-b': 'value3'}, '')])

self.requests = []

self.check_no_query_string = check_no_query_string

def __call__(self, env, start_response):

if self.check_no_query_string and env.get('QUERY_STRING'):

raise Exception('Query string %s should have been discarded!' %

env['QUERY_STRING'])

body = ''

while True:

chunk = env['wsgi.input'].read()

@@ -944,6 +949,42 @@ class TestFormPost(unittest.TestCase):

self.assertEquals(len(self.app.requests), 1)

self.assertEquals(self.app.requests[0].body, 'Test File\nOne\n')

def test_subrequest_does_not_pass_query(self):

key = 'abc'

sig, env, body = self._make_sig_env_body(

'/v1/AUTH_test/container', '', 1024, 10, int(time() + 86400), key)

env['QUERY_STRING'] = 'this=should&not=get&passed'

env['wsgi.input'] = StringIO('\r\n'.join(body))

env['swift.cache'] = FakeMemcache()

# We don't cache the key so that it's asked for (and FakeApp verifies

# that no QUERY_STRING got passed).

self.app = FakeApp(

iter([('200 Ok', {'x-account-meta-temp-url-key': 'abc'}, ''),

('201 Created', {}, ''),

('201 Created', {}, '')]),

check_no_query_string=True)

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 = ''.join(self.formpost(env, start_response))

status = status[0]

headers = headers[0]

exc_info = exc_info[0]

# Make sure we 201 Created, which means we made the final subrequest

# (and FakeAp verifies that no QUERY_STRING got passed).

self.assertEquals(status, '201 Created')

self.assertEquals(exc_info, None)

self.assertTrue('201 Created' in body)

self.assertEquals(len(self.app.requests), 3)

def test_subrequest_fails(self):

key = 'abc'

sig, env, body = self._make_sig_env_body(

@@ -1132,6 +1173,81 @@ class TestFormPost(unittest.TestCase):

in body)

self.assertEquals(len(self.app.requests), 0)

def test_redirect(self):

key = 'abc'

sig, env, body = self._make_sig_env_body(

'/v1/AUTH_test/container', 'http://redirect', 1024, 10,

int(time() + 86400), key)

env['wsgi.input'] = StringIO('\r\n'.join(body))

env['swift.cache'] = FakeMemcache()

env['swift.cache'].set('temp-url-key/AUTH_test', key)

self.app = FakeApp(iter([('201 Created', {}, ''),

('201 Created', {}, '')]))

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 = ''.join(self.formpost(env, start_response))

status = status[0]

headers = headers[0]

exc_info = exc_info[0]

self.assertEquals(status, '303 See Other')

location = None

for h, v in headers:

if h.lower() == 'location':

location = v

self.assertEquals(location, 'http://redirect?status=201&message=')

self.assertEquals(exc_info, None)

self.assertTrue(location in body)

self.assertEquals(len(self.app.requests), 2)

self.assertEquals(self.app.requests[0].body, 'Test File\nOne\n')

self.assertEquals(self.app.requests[1].body, 'Test\nFile\nTwo\n')

def test_redirect_with_query(self):

key = 'abc'

sig, env, body = self._make_sig_env_body(

'/v1/AUTH_test/container', 'http://redirect?one=two', 1024, 10,

int(time() + 86400), key)

env['wsgi.input'] = StringIO('\r\n'.join(body))

env['swift.cache'] = FakeMemcache()

env['swift.cache'].set('temp-url-key/AUTH_test', key)

self.app = FakeApp(iter([('201 Created', {}, ''),

('201 Created', {}, '')]))

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 = ''.join(self.formpost(env, start_response))

status = status[0]

headers = headers[0]

exc_info = exc_info[0]

self.assertEquals(status, '303 See Other')

location = None

for h, v in headers:

if h.lower() == 'location':

location = v

self.assertEquals(location,

'http://redirect?one=two&status=201&message=')

self.assertEquals(exc_info, None)

self.assertTrue(location in body)

self.assertEquals(len(self.app.requests), 2)

self.assertEquals(self.app.requests[0].body, 'Test File\nOne\n')

self.assertEquals(self.app.requests[1].body, 'Test\nFile\nTwo\n')

def test_no_redirect(self):

key = 'abc'

sig, env, body = self._make_sig_env_body(