openstack/swift
Code Issues Proposed changes

Merge "Make formpost work with multiple tempurl keys."

Browse Source
This commit is contained in:
Jenkins 2013-06-27 20:45:53 +00:00 committed by Gerrit Code Review
commit c6e53721e0
3 changed files with 142 additions and 130 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

60
swift/common/middleware/formpost.py
View File

@ -105,12 +105,13 @@ import hmac
import re import re
import rfc822 import rfc822
from hashlib import sha1 from hashlib import sha1
from StringIO import StringIO
from time import time from time import time
from urllib import quote from urllib import quote
from swift.common.middleware.tempurl import get_tempurl_keys_from_metadata
from swift.common.utils import streq_const_time from swift.common.utils import streq_const_time
from swift.common.wsgi import make_pre_authed_env from swift.common.wsgi import make_pre_authed_env
from swift.proxy.controllers.base import get_account_info
#: The size of data to read from the form at any given time. #: The size of data to read from the form at any given time.
@ -337,7 +338,7 @@ class FormPost(object):
:param boundary: The MIME type boundary to look for. :param boundary: The MIME type boundary to look for.
:returns: status_line, headers_list, body :returns: status_line, headers_list, body
""" """
key = self._get_key(env) keys = self._get_keys(env)
status = message = '' status = message = ''
attributes = {} attributes = {}
file_count = 0 file_count = 0
@ -359,7 +360,7 @@ class FormPost(object):
attributes['content-type'] = \ attributes['content-type'] = \
hdrs['Content-Type'] or 'application/octet-stream' hdrs['Content-Type'] or 'application/octet-stream'
status, message = self._perform_subrequest(env, attributes, fp, status, message = self._perform_subrequest(env, attributes, fp,
key) keys)
if status[:1] != '2': if status[:1] != '2':
break break
else: else:
@ -397,7 +398,7 @@ class FormPost(object):
headers = [('Location', redirect), ('Content-Length', str(len(body)))] headers = [('Location', redirect), ('Content-Length', str(len(body)))]
return '303 See Other', headers, body return '303 See Other', headers, body
def _perform_subrequest(self, orig_env, attributes, fp, key): def _perform_subrequest(self, orig_env, attributes, fp, keys):
""" """
Performs the subrequest and returns the response. Performs the subrequest and returns the response.
@ -405,10 +406,10 @@ class FormPost(object):
to form a new env for the subrequest. to form a new env for the subrequest.
:param attributes: dict of the attributes of the form so far. :param attributes: dict of the attributes of the form so far.
:param fp: The file-like object containing the request body. :param fp: The file-like object containing the request body.
:param key: The account key to validate the signature with. :param keys: The account keys to validate the signature with.
:returns: (status_line, message) :returns: (status_line, message)
""" """
if not key: if not keys:
return '401 Unauthorized', 'invalid signature' return '401 Unauthorized', 'invalid signature'
try: try:
max_file_size = int(attributes.get('max_file_size') or 0) max_file_size = int(attributes.get('max_file_size') or 0)
@ -440,10 +441,16 @@ class FormPost(object):
attributes.get('max_file_size') or '0', attributes.get('max_file_size') or '0',
attributes.get('max_file_count') or '0', attributes.get('max_file_count') or '0',
attributes.get('expires') or '0') attributes.get('expires') or '0')
has_valid_sig = False
for key in keys:
sig = hmac.new(key, hmac_body, sha1).hexdigest() sig = hmac.new(key, hmac_body, sha1).hexdigest()
if not streq_const_time(sig, (attributes.get('signature') or if streq_const_time(sig, (attributes.get('signature') or
'invalid')): 'invalid')):
has_valid_sig = True
if not has_valid_sig:
return '401 Unauthorized', 'invalid signature' return '401 Unauthorized', 'invalid signature'
substatus = [None] substatus = [None]
def _start_response(status, headers, exc_info=None): def _start_response(status, headers, exc_info=None):
@ -456,46 +463,21 @@ class FormPost(object):
pass pass
return substatus[0], '' return substatus[0], ''
def _get_key(self, env): def _get_keys(self, env):
""" """
Returns the X-Account-Meta-Temp-URL-Key header value for the Fetch the tempurl keys for the account. Also validate that the request
account, or None if none is set. path indicates a valid container; if not, no keys will be returned.
:param env: The WSGI environment for the request. :param env: The WSGI environment for the request.
:returns: X-Account-Meta-Temp-URL-Key str value, or None. :returns: list of tempurl keys
""" """
parts = env['PATH_INFO'].split('/', 4) parts = env['PATH_INFO'].split('/', 4)
if len(parts) < 4 or parts[0] or parts[1] != 'v1' or not parts[2] or \ if len(parts) < 4 or parts[0] or parts[1] != 'v1' or not parts[2] or \
not parts[3]: not parts[3]:
return None return []
account = parts[2]
key = None
memcache = env.get('swift.cache')
if memcache:
key = memcache.get('temp-url-key/%s' % account)
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]
def _start_response(status, response_headers, exc_info=None): account_info = get_account_info(env, self.app, swift_source='FP')
for h, v in response_headers: return get_tempurl_keys_from_metadata(account_info['meta'])
if h.lower() == 'x-account-meta-temp-url-key':
key[0] = v
i = iter(self.app(newenv, _start_response))
try:
i.next()
except StopIteration:
pass
key = key[0]
if key and memcache:
memcache.set('temp-url-key/%s' % account, key, time=60)
return key
def filter_factory(global_conf, **local_conf): def filter_factory(global_conf, **local_conf):

18
swift/common/middleware/tempurl.py
View File

@ -123,6 +123,21 @@ DEFAULT_OUTGOING_REMOVE_HEADERS = 'x-object-meta-*'
DEFAULT_OUTGOING_ALLOW_HEADERS = 'x-object-meta-public-*' DEFAULT_OUTGOING_ALLOW_HEADERS = 'x-object-meta-public-*'
def get_tempurl_keys_from_metadata(meta):
"""
Extracts the tempurl keys from metadata.
:param meta: account metadata
:returns: list of keys found (possibly empty if no keys set)
Example:
meta = get_account_info(...)['meta']
keys = get_tempurl_keys_from_metadata(meta)
"""
return [value for key, value in meta.iteritems()
if key.lower() in ('temp-url-key', 'temp-url-key-2')]
class TempURL(object): class TempURL(object):
""" """
WSGI Middleware to grant temporary URLs specific access to Swift WSGI Middleware to grant temporary URLs specific access to Swift
@ -354,8 +369,7 @@ class TempURL(object):
X-Account-Meta-Temp-URL-Key-2 str value if set] X-Account-Meta-Temp-URL-Key-2 str value if set]
""" """
account_info = get_account_info(env, self.app, swift_source='TU') account_info = get_account_info(env, self.app, swift_source='TU')
return [value for key, value in account_info['meta'].iteritems() return get_tempurl_keys_from_metadata(account_info['meta'])
if key.lower() in ('temp-url-key', 'temp-url-key-2')]
def _get_hmacs(self, env, expires, keys, request_method=None): def _get_hmacs(self, env, expires, keys, request_method=None):
""" """

192
test/unit/common/middleware/test_formpost.py
View File

@ -22,34 +22,7 @@ from time import time
from swift.common.swob import Request, Response from swift.common.swob import Request, Response
from swift.common.middleware import tempauth, formpost from swift.common.middleware import tempauth, formpost
from swift.common.utils import split_path
class FakeMemcache(object):
def __init__(self):
self.store = {}
def get(self, key):
return self.store.get(key)
def set(self, key, value, time=0):
self.store[key] = value
return True
def incr(self, key, timeout=0):
self.store[key] = self.store.setdefault(key, 0) + 1
return self.store[key]
@contextmanager
def soft_lock(self, key, timeout=0, retries=5):
yield True
def delete(self, key):
try:
del self.store[key]
except Exception:
pass
return True
class FakeApp(object): class FakeApp(object):
@ -298,11 +271,39 @@ class TestFormPost(unittest.TestCase):
self.auth = tempauth.filter_factory({})(self.app) self.auth = tempauth.filter_factory({})(self.app)
self.formpost = formpost.filter_factory({})(self.auth) self.formpost = formpost.filter_factory({})(self.auth)
def _make_request(self, path, **kwargs): def _make_request(self, path, tempurl_keys=(), **kwargs):
req = Request.blank(path, **kwargs) req = Request.blank(path, **kwargs)
req.environ['swift.cache'] = FakeMemcache()
# Fake out the caching layer so that get_account_info() finds its
# data. Include something that isn't tempurl keys to prove we skip it.
meta = {'user-job-title': 'Personal Trainer',
'user-real-name': 'Jim Shortz'}
for idx, key in enumerate(tempurl_keys):
meta_name = 'temp-url-key' + (("-%d" % (idx + 1) if idx else ""))
if key:
meta[meta_name] = key
_junk, account, _junk, _junk = split_path(path, 2, 4)
req.environ['swift.account/' + account] = self._fake_cache_env(
account, tempurl_keys)
return req return req
def _fake_cache_env(self, account, tempurl_keys=()):
# Fake out the caching layer so that get_account_info() finds its
# data. Include something that isn't tempurl keys to prove we skip it.
meta = {'user-job-title': 'Personal Trainer',
'user-real-name': 'Jim Shortz'}
for idx, key in enumerate(tempurl_keys):
meta_name = 'temp-url-key' + ("-%d" % (idx + 1) if idx else "")
if key:
meta[meta_name] = key
return {'status': 204,
'container_count': '0',
'total_object_count': '0',
'bytes': '0',
'meta': meta}
def _make_sig_env_body(self, path, redirect, max_file_size, max_file_count, def _make_sig_env_body(self, path, redirect, max_file_size, max_file_count,
expires, key, user_agent=True): expires, key, user_agent=True):
sig = hmac.new( sig = hmac.new(
@ -404,8 +405,6 @@ class TestFormPost(unittest.TestCase):
'%s\n%s\n%s\n%s\n%s' % ( '%s\n%s\n%s\n%s\n%s' % (
path, redirect, max_file_size, max_file_count, expires), path, redirect, max_file_size, max_file_count, expires),
sha1).hexdigest() sha1).hexdigest()
memcache = FakeMemcache()
memcache.set('temp-url-key/AUTH_test', key)
wsgi_input = StringIO('\r\n'.join([ wsgi_input = StringIO('\r\n'.join([
'------WebKitFormBoundaryNcxTqxSlX7t4TDkR', '------WebKitFormBoundaryNcxTqxSlX7t4TDkR',
'Content-Disposition: form-data; name="redirect"', 'Content-Disposition: form-data; name="redirect"',
@ -468,7 +467,7 @@ class TestFormPost(unittest.TestCase):
'SERVER_NAME': '172.16.83.128', 'SERVER_NAME': '172.16.83.128',
'SERVER_PORT': '8080', 'SERVER_PORT': '8080',
'SERVER_PROTOCOL': 'HTTP/1.0', 'SERVER_PROTOCOL': 'HTTP/1.0',
'swift.cache': memcache, 'swift.account/AUTH_test': self._fake_cache_env('AUTH_test', [key]),
'wsgi.errors': wsgi_errors, 'wsgi.errors': wsgi_errors,
'wsgi.input': wsgi_input, 'wsgi.input': wsgi_input,
'wsgi.multiprocess': False, 'wsgi.multiprocess': False,
@ -518,8 +517,6 @@ class TestFormPost(unittest.TestCase):
'%s\n%s\n%s\n%s\n%s' % ( '%s\n%s\n%s\n%s\n%s' % (
path, redirect, max_file_size, max_file_count, expires), path, redirect, max_file_size, max_file_count, expires),
sha1).hexdigest() sha1).hexdigest()
memcache = FakeMemcache()
memcache.set('temp-url-key/AUTH_test', key)
wsgi_input = StringIO('\r\n'.join([ wsgi_input = StringIO('\r\n'.join([
'-----------------------------168072824752491622650073', '-----------------------------168072824752491622650073',
'Content-Disposition: form-data; name="redirect"', 'Content-Disposition: form-data; name="redirect"',
@ -581,7 +578,7 @@ class TestFormPost(unittest.TestCase):
'SERVER_NAME': '172.16.83.128', 'SERVER_NAME': '172.16.83.128',
'SERVER_PORT': '8080', 'SERVER_PORT': '8080',
'SERVER_PROTOCOL': 'HTTP/1.0', 'SERVER_PROTOCOL': 'HTTP/1.0',
'swift.cache': memcache, 'swift.account/AUTH_test': self._fake_cache_env('AUTH_test', [key]),
'wsgi.errors': wsgi_errors, 'wsgi.errors': wsgi_errors,
'wsgi.input': wsgi_input, 'wsgi.input': wsgi_input,
'wsgi.multiprocess': False, 'wsgi.multiprocess': False,
@ -631,8 +628,6 @@ class TestFormPost(unittest.TestCase):
'%s\n%s\n%s\n%s\n%s' % ( '%s\n%s\n%s\n%s\n%s' % (
path, redirect, max_file_size, max_file_count, expires), path, redirect, max_file_size, max_file_count, expires),
sha1).hexdigest() sha1).hexdigest()
memcache = FakeMemcache()
memcache.set('temp-url-key/AUTH_test', key)
wsgi_input = StringIO('\r\n'.join([ wsgi_input = StringIO('\r\n'.join([
'------WebKitFormBoundaryq3CFxUjfsDMu8XsA', '------WebKitFormBoundaryq3CFxUjfsDMu8XsA',
'Content-Disposition: form-data; name="redirect"', 'Content-Disposition: form-data; name="redirect"',
@ -697,7 +692,7 @@ class TestFormPost(unittest.TestCase):
'SERVER_NAME': '172.16.83.128', 'SERVER_NAME': '172.16.83.128',
'SERVER_PORT': '8080', 'SERVER_PORT': '8080',
'SERVER_PROTOCOL': 'HTTP/1.0', 'SERVER_PROTOCOL': 'HTTP/1.0',
'swift.cache': memcache, 'swift.account/AUTH_test': self._fake_cache_env('AUTH_test', [key]),
'wsgi.errors': wsgi_errors, 'wsgi.errors': wsgi_errors,
'wsgi.input': wsgi_input, 'wsgi.input': wsgi_input,
'wsgi.multiprocess': False, 'wsgi.multiprocess': False,
@ -747,8 +742,6 @@ class TestFormPost(unittest.TestCase):
'%s\n%s\n%s\n%s\n%s' % ( '%s\n%s\n%s\n%s\n%s' % (
path, redirect, max_file_size, max_file_count, expires), path, redirect, max_file_size, max_file_count, expires),
sha1).hexdigest() sha1).hexdigest()
memcache = FakeMemcache()
memcache.set('temp-url-key/AUTH_test', key)
wsgi_input = StringIO('\r\n'.join([ wsgi_input = StringIO('\r\n'.join([
'-----------------------------7db20d93017c', '-----------------------------7db20d93017c',
'Content-Disposition: form-data; name="redirect"', 'Content-Disposition: form-data; name="redirect"',
@ -809,7 +802,7 @@ class TestFormPost(unittest.TestCase):
'SERVER_NAME': '172.16.83.128', 'SERVER_NAME': '172.16.83.128',
'SERVER_PORT': '8080', 'SERVER_PORT': '8080',
'SERVER_PROTOCOL': 'HTTP/1.0', 'SERVER_PROTOCOL': 'HTTP/1.0',
'swift.cache': memcache, 'swift.account/AUTH_test': self._fake_cache_env('AUTH_test', [key]),
'wsgi.errors': wsgi_errors, 'wsgi.errors': wsgi_errors,
'wsgi.input': wsgi_input, 'wsgi.input': wsgi_input,
'wsgi.multiprocess': False, 'wsgi.multiprocess': False,
@ -853,8 +846,8 @@ class TestFormPost(unittest.TestCase):
'/v1/AUTH_test/container', 'http://brim.net', 5, 10, '/v1/AUTH_test/container', 'http://brim.net', 5, 10,
int(time() + 86400), key) int(time() + 86400), key)
env['wsgi.input'] = StringIO('XX' + '\r\n'.join(body)) env['wsgi.input'] = StringIO('XX' + '\r\n'.join(body))
env['swift.cache'] = FakeMemcache() env['swift.account/AUTH_test'] = self._fake_cache_env(
env['swift.cache'].set('temp-url-key/AUTH_test', key) 'AUTH_test', [key])
self.app = FakeApp(iter([('201 Created', {}, ''), self.app = FakeApp(iter([('201 Created', {}, ''),
('201 Created', {}, '')])) ('201 Created', {}, '')]))
self.auth = tempauth.filter_factory({})(self.app) self.auth = tempauth.filter_factory({})(self.app)
@ -888,8 +881,8 @@ class TestFormPost(unittest.TestCase):
'/v1/AUTH_test/container', 'http://brim.net', 5, 10, '/v1/AUTH_test/container', 'http://brim.net', 5, 10,
int(time() + 86400), key) int(time() + 86400), key)
env['wsgi.input'] = StringIO('\r\n'.join(body)) env['wsgi.input'] = StringIO('\r\n'.join(body))
env['swift.cache'] = FakeMemcache() env['swift.account/AUTH_test'] = self._fake_cache_env(
env['swift.cache'].set('temp-url-key/AUTH_test', key) 'AUTH_test', [key])
self.app = FakeApp(iter([('201 Created', {}, ''), self.app = FakeApp(iter([('201 Created', {}, ''),
('201 Created', {}, '')])) ('201 Created', {}, '')]))
self.auth = tempauth.filter_factory({})(self.app) self.auth = tempauth.filter_factory({})(self.app)
@ -918,8 +911,8 @@ class TestFormPost(unittest.TestCase):
'/v1/AUTH_test/container', 'http://brim.net', 1024, 1, '/v1/AUTH_test/container', 'http://brim.net', 1024, 1,
int(time() + 86400), key) int(time() + 86400), key)
env['wsgi.input'] = StringIO('\r\n'.join(body)) env['wsgi.input'] = StringIO('\r\n'.join(body))
env['swift.cache'] = FakeMemcache() env['swift.account/AUTH_test'] = self._fake_cache_env(
env['swift.cache'].set('temp-url-key/AUTH_test', key) 'AUTH_test', [key])
self.app = FakeApp(iter([('201 Created', {}, ''), self.app = FakeApp(iter([('201 Created', {}, ''),
('201 Created', {}, '')])) ('201 Created', {}, '')]))
self.auth = tempauth.filter_factory({})(self.app) self.auth = tempauth.filter_factory({})(self.app)
@ -958,12 +951,10 @@ class TestFormPost(unittest.TestCase):
'/v1/AUTH_test/container', '', 1024, 10, int(time() + 86400), key) '/v1/AUTH_test/container', '', 1024, 10, int(time() + 86400), key)
env['QUERY_STRING'] = 'this=should&not=get&passed' env['QUERY_STRING'] = 'this=should&not=get&passed'
env['wsgi.input'] = StringIO('\r\n'.join(body)) env['wsgi.input'] = StringIO('\r\n'.join(body))
env['swift.cache'] = FakeMemcache() env['swift.account/AUTH_test'] = self._fake_cache_env(
# We don't cache the key so that it's asked for (and FakeApp verifies 'AUTH_test', [key])
# that no QUERY_STRING got passed).
self.app = FakeApp( self.app = FakeApp(
iter([('200 Ok', {'x-account-meta-temp-url-key': 'abc'}, ''), iter([('201 Created', {}, ''),
('201 Created', {}, ''),
('201 Created', {}, '')]), ('201 Created', {}, '')]),
check_no_query_string=True) check_no_query_string=True)
self.auth = tempauth.filter_factory({})(self.app) self.auth = tempauth.filter_factory({})(self.app)
@ -982,11 +973,11 @@ class TestFormPost(unittest.TestCase):
headers = headers[0] headers = headers[0]
exc_info = exc_info[0] exc_info = exc_info[0]
# Make sure we 201 Created, which means we made the final subrequest # Make sure we 201 Created, which means we made the final subrequest
# (and FakeAp verifies that no QUERY_STRING got passed). # (and FakeApp verifies that no QUERY_STRING got passed).
self.assertEquals(status, '201 Created') self.assertEquals(status, '201 Created')
self.assertEquals(exc_info, None) self.assertEquals(exc_info, None)
self.assertTrue('201 Created' in body) self.assertTrue('201 Created' in body)
self.assertEquals(len(self.app.requests), 3) self.assertEquals(len(self.app.requests), 2)
def test_subrequest_fails(self): def test_subrequest_fails(self):
key = 'abc' key = 'abc'
@ -994,8 +985,8 @@ class TestFormPost(unittest.TestCase):
'/v1/AUTH_test/container', 'http://brim.net', 1024, 10, '/v1/AUTH_test/container', 'http://brim.net', 1024, 10,
int(time() + 86400), key) int(time() + 86400), key)
env['wsgi.input'] = StringIO('\r\n'.join(body)) env['wsgi.input'] = StringIO('\r\n'.join(body))
env['swift.cache'] = FakeMemcache() env['swift.account/AUTH_test'] = self._fake_cache_env(
env['swift.cache'].set('temp-url-key/AUTH_test', key) 'AUTH_test', [key])
self.app = FakeApp(iter([('404 Not Found', {}, ''), self.app = FakeApp(iter([('404 Not Found', {}, ''),
('201 Created', {}, '')])) ('201 Created', {}, '')]))
self.auth = tempauth.filter_factory({})(self.app) self.auth = tempauth.filter_factory({})(self.app)
@ -1076,8 +1067,8 @@ class TestFormPost(unittest.TestCase):
'------WebKitFormBoundaryNcxTqxSlX7t4TDkR--', '------WebKitFormBoundaryNcxTqxSlX7t4TDkR--',
'', '',
])) ]))
env['swift.cache'] = FakeMemcache() env['swift.account/AUTH_test'] = self._fake_cache_env(
env['swift.cache'].set('temp-url-key/AUTH_test', key) 'AUTH_test', [key])
self.app = FakeApp(iter([('201 Created', {}, ''), self.app = FakeApp(iter([('201 Created', {}, ''),
('201 Created', {}, '')])) ('201 Created', {}, '')]))
self.auth = tempauth.filter_factory({})(self.app) self.auth = tempauth.filter_factory({})(self.app)
@ -1143,8 +1134,8 @@ class TestFormPost(unittest.TestCase):
'------WebKitFormBoundaryNcxTqxSlX7t4TDkR--', '------WebKitFormBoundaryNcxTqxSlX7t4TDkR--',
'', '',
])) ]))
env['swift.cache'] = FakeMemcache() env['swift.account/AUTH_test'] = self._fake_cache_env(
env['swift.cache'].set('temp-url-key/AUTH_test', key) 'AUTH_test', [key])
self.app = FakeApp(iter([('201 Created', {}, ''), self.app = FakeApp(iter([('201 Created', {}, ''),
('201 Created', {}, '')])) ('201 Created', {}, '')]))
self.auth = tempauth.filter_factory({})(self.app) self.auth = tempauth.filter_factory({})(self.app)
@ -1182,8 +1173,8 @@ class TestFormPost(unittest.TestCase):
'/v1/AUTH_test/container', 'http://redirect', 1024, 10, '/v1/AUTH_test/container', 'http://redirect', 1024, 10,
int(time() + 86400), key, user_agent=False) int(time() + 86400), key, user_agent=False)
env['wsgi.input'] = StringIO('\r\n'.join(body)) env['wsgi.input'] = StringIO('\r\n'.join(body))
env['swift.cache'] = FakeMemcache() env['swift.account/AUTH_test'] = self._fake_cache_env(
env['swift.cache'].set('temp-url-key/AUTH_test', key) 'AUTH_test', [key])
self.app = FakeApp(iter([('201 Created', {}, ''), self.app = FakeApp(iter([('201 Created', {}, ''),
('201 Created', {}, '')])) ('201 Created', {}, '')]))
self.auth = tempauth.filter_factory({})(self.app) self.auth = tempauth.filter_factory({})(self.app)
@ -1196,14 +1187,40 @@ class TestFormPost(unittest.TestCase):
self.assertEquals(self.app.requests[0].headers['User-Agent'], self.assertEquals(self.app.requests[0].headers['User-Agent'],
'FormPost') 'FormPost')
def test_formpost_with_multiple_keys(self):
key = 'ernie'
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))
# Stick it in X-Account-Meta-Temp-URL-Key-2 and make sure we get it
env['swift.account/AUTH_test'] = self._fake_cache_env(
'AUTH_test', ['bert', 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]
def start_response(s, h, e=None):
status[0] = s
headers[0] = h
body = ''.join(self.formpost(env, start_response))
print repr(headers)
self.assertEqual('303 See Other', status[0])
self.assertEqual(
'http://redirect?status=201&message=',
dict(headers[0]).get('Location'))
def test_redirect(self): def test_redirect(self):
key = 'abc' key = 'abc'
sig, env, body = self._make_sig_env_body( sig, env, body = self._make_sig_env_body(
'/v1/AUTH_test/container', 'http://redirect', 1024, 10, '/v1/AUTH_test/container', 'http://redirect', 1024, 10,
int(time() + 86400), key) int(time() + 86400), key)
env['wsgi.input'] = StringIO('\r\n'.join(body)) env['wsgi.input'] = StringIO('\r\n'.join(body))
env['swift.cache'] = FakeMemcache() env['swift.account/AUTH_test'] = self._fake_cache_env(
env['swift.cache'].set('temp-url-key/AUTH_test', key) 'AUTH_test', [key])
self.app = FakeApp(iter([('201 Created', {}, ''), self.app = FakeApp(iter([('201 Created', {}, ''),
('201 Created', {}, '')])) ('201 Created', {}, '')]))
self.auth = tempauth.filter_factory({})(self.app) self.auth = tempauth.filter_factory({})(self.app)
@ -1239,8 +1256,8 @@ class TestFormPost(unittest.TestCase):
'/v1/AUTH_test/container', 'http://redirect?one=two', 1024, 10, '/v1/AUTH_test/container', 'http://redirect?one=two', 1024, 10,
int(time() + 86400), key) int(time() + 86400), key)
env['wsgi.input'] = StringIO('\r\n'.join(body)) env['wsgi.input'] = StringIO('\r\n'.join(body))
env['swift.cache'] = FakeMemcache() env['swift.account/AUTH_test'] = self._fake_cache_env(
env['swift.cache'].set('temp-url-key/AUTH_test', key) 'AUTH_test', [key])
self.app = FakeApp(iter([('201 Created', {}, ''), self.app = FakeApp(iter([('201 Created', {}, ''),
('201 Created', {}, '')])) ('201 Created', {}, '')]))
self.auth = tempauth.filter_factory({})(self.app) self.auth = tempauth.filter_factory({})(self.app)
@ -1276,8 +1293,8 @@ class TestFormPost(unittest.TestCase):
sig, env, body = self._make_sig_env_body( sig, env, body = self._make_sig_env_body(
'/v1/AUTH_test/container', '', 1024, 10, int(time() + 86400), key) '/v1/AUTH_test/container', '', 1024, 10, int(time() + 86400), key)
env['wsgi.input'] = StringIO('\r\n'.join(body)) env['wsgi.input'] = StringIO('\r\n'.join(body))
env['swift.cache'] = FakeMemcache() env['swift.account/AUTH_test'] = self._fake_cache_env(
env['swift.cache'].set('temp-url-key/AUTH_test', key) 'AUTH_test', [key])
self.app = FakeApp(iter([('201 Created', {}, ''), self.app = FakeApp(iter([('201 Created', {}, ''),
('201 Created', {}, '')])) ('201 Created', {}, '')]))
self.auth = tempauth.filter_factory({})(self.app) self.auth = tempauth.filter_factory({})(self.app)
@ -1312,8 +1329,8 @@ class TestFormPost(unittest.TestCase):
sig, env, body = self._make_sig_env_body( sig, env, body = self._make_sig_env_body(
'/v1/AUTH_test/container', '', 1024, 10, int(time() - 10), key) '/v1/AUTH_test/container', '', 1024, 10, int(time() - 10), key)
env['wsgi.input'] = StringIO('\r\n'.join(body)) env['wsgi.input'] = StringIO('\r\n'.join(body))
env['swift.cache'] = FakeMemcache() env['swift.account/AUTH_test'] = self._fake_cache_env(
env['swift.cache'].set('temp-url-key/AUTH_test', key) 'AUTH_test', [key])
self.app = FakeApp(iter([('201 Created', {}, ''), self.app = FakeApp(iter([('201 Created', {}, ''),
('201 Created', {}, '')])) ('201 Created', {}, '')]))
self.auth = tempauth.filter_factory({})(self.app) self.auth = tempauth.filter_factory({})(self.app)
@ -1345,10 +1362,9 @@ class TestFormPost(unittest.TestCase):
sig, env, body = self._make_sig_env_body( sig, env, body = self._make_sig_env_body(
'/v1/AUTH_test/container', '', 1024, 10, int(time() + 86400), key) '/v1/AUTH_test/container', '', 1024, 10, int(time() + 86400), key)
env['wsgi.input'] = StringIO('\r\n'.join(body)) env['wsgi.input'] = StringIO('\r\n'.join(body))
env['swift.cache'] = FakeMemcache()
# Change key to invalidate sig # Change key to invalidate sig
key = 'def' env['swift.account/AUTH_test'] = self._fake_cache_env(
env['swift.cache'].set('temp-url-key/AUTH_test', key) 'AUTH_test', [key + ' is bogus now'])
self.app = FakeApp(iter([('201 Created', {}, ''), self.app = FakeApp(iter([('201 Created', {}, ''),
('201 Created', {}, '')])) ('201 Created', {}, '')]))
self.auth = tempauth.filter_factory({})(self.app) self.auth = tempauth.filter_factory({})(self.app)
@ -1380,8 +1396,8 @@ class TestFormPost(unittest.TestCase):
sig, env, body = self._make_sig_env_body( sig, env, body = self._make_sig_env_body(
'/v1/AUTH_test/container', '', 1024, 10, int(time() + 86400), key) '/v1/AUTH_test/container', '', 1024, 10, int(time() + 86400), key)
env['wsgi.input'] = StringIO('XX' + '\r\n'.join(body)) env['wsgi.input'] = StringIO('XX' + '\r\n'.join(body))
env['swift.cache'] = FakeMemcache() env['swift.account/AUTH_test'] = self._fake_cache_env(
env['swift.cache'].set('temp-url-key/AUTH_test', key) 'AUTH_test', [key])
self.app = FakeApp(iter([('201 Created', {}, ''), self.app = FakeApp(iter([('201 Created', {}, ''),
('201 Created', {}, '')])) ('201 Created', {}, '')]))
self.auth = tempauth.filter_factory({})(self.app) self.auth = tempauth.filter_factory({})(self.app)
@ -1413,8 +1429,8 @@ class TestFormPost(unittest.TestCase):
sig, env, body = self._make_sig_env_body( sig, env, body = self._make_sig_env_body(
'/v2/AUTH_test/container', '', 1024, 10, int(time() + 86400), key) '/v2/AUTH_test/container', '', 1024, 10, int(time() + 86400), key)
env['wsgi.input'] = StringIO('\r\n'.join(body)) env['wsgi.input'] = StringIO('\r\n'.join(body))
env['swift.cache'] = FakeMemcache() env['swift.account/AUTH_test'] = self._fake_cache_env(
env['swift.cache'].set('temp-url-key/AUTH_test', key) 'AUTH_test', [key])
self.app = FakeApp(iter([('201 Created', {}, ''), self.app = FakeApp(iter([('201 Created', {}, ''),
('201 Created', {}, '')])) ('201 Created', {}, '')]))
self.auth = tempauth.filter_factory({})(self.app) self.auth = tempauth.filter_factory({})(self.app)
@ -1446,8 +1462,8 @@ class TestFormPost(unittest.TestCase):
sig, env, body = self._make_sig_env_body( sig, env, body = self._make_sig_env_body(
'//AUTH_test/container', '', 1024, 10, int(time() + 86400), key) '//AUTH_test/container', '', 1024, 10, int(time() + 86400), key)
env['wsgi.input'] = StringIO('\r\n'.join(body)) env['wsgi.input'] = StringIO('\r\n'.join(body))
env['swift.cache'] = FakeMemcache() env['swift.account/AUTH_test'] = self._fake_cache_env(
env['swift.cache'].set('temp-url-key/AUTH_test', key) 'AUTH_test', [key])
self.app = FakeApp(iter([('201 Created', {}, ''), self.app = FakeApp(iter([('201 Created', {}, ''),
('201 Created', {}, '')])) ('201 Created', {}, '')]))
self.auth = tempauth.filter_factory({})(self.app) self.auth = tempauth.filter_factory({})(self.app)
@ -1479,8 +1495,8 @@ class TestFormPost(unittest.TestCase):
sig, env, body = self._make_sig_env_body( sig, env, body = self._make_sig_env_body(
'/v1//container', '', 1024, 10, int(time() + 86400), key) '/v1//container', '', 1024, 10, int(time() + 86400), key)
env['wsgi.input'] = StringIO('\r\n'.join(body)) env['wsgi.input'] = StringIO('\r\n'.join(body))
env['swift.cache'] = FakeMemcache() env['swift.account/AUTH_test'] = self._fake_cache_env(
env['swift.cache'].set('temp-url-key/AUTH_test', key) 'AUTH_test', [key])
self.app = FakeApp(iter([('201 Created', {}, ''), self.app = FakeApp(iter([('201 Created', {}, ''),
('201 Created', {}, '')])) ('201 Created', {}, '')]))
self.auth = tempauth.filter_factory({})(self.app) self.auth = tempauth.filter_factory({})(self.app)
@ -1512,8 +1528,8 @@ class TestFormPost(unittest.TestCase):
sig, env, body = self._make_sig_env_body( sig, env, body = self._make_sig_env_body(
'/v1/AUTH_tst/container', '', 1024, 10, int(time() + 86400), key) '/v1/AUTH_tst/container', '', 1024, 10, int(time() + 86400), key)
env['wsgi.input'] = StringIO('\r\n'.join(body)) env['wsgi.input'] = StringIO('\r\n'.join(body))
env['swift.cache'] = FakeMemcache() env['swift.account/AUTH_test'] = self._fake_cache_env(
env['swift.cache'].set('temp-url-key/AUTH_test', key) 'AUTH_test', [key])
self.app = FakeApp(iter([ self.app = FakeApp(iter([
('200 Ok', {'x-account-meta-temp-url-key': 'def'}, ''), ('200 Ok', {'x-account-meta-temp-url-key': 'def'}, ''),
('201 Created', {}, ''), ('201 Created', {}, ''),
@ -1547,8 +1563,8 @@ class TestFormPost(unittest.TestCase):
sig, env, body = self._make_sig_env_body( sig, env, body = self._make_sig_env_body(
'/v1/AUTH_test', '', 1024, 10, int(time() + 86400), key) '/v1/AUTH_test', '', 1024, 10, int(time() + 86400), key)
env['wsgi.input'] = StringIO('\r\n'.join(body)) env['wsgi.input'] = StringIO('\r\n'.join(body))
env['swift.cache'] = FakeMemcache() env['swift.account/AUTH_test'] = self._fake_cache_env(
env['swift.cache'].set('temp-url-key/AUTH_test', key) 'AUTH_test', [key])
self.app = FakeApp(iter([('201 Created', {}, ''), self.app = FakeApp(iter([('201 Created', {}, ''),
('201 Created', {}, '')])) ('201 Created', {}, '')]))
self.auth = tempauth.filter_factory({})(self.app) self.auth = tempauth.filter_factory({})(self.app)
@ -1585,8 +1601,8 @@ class TestFormPost(unittest.TestCase):
body[i] = 'badvalue' body[i] = 'badvalue'
break break
env['wsgi.input'] = StringIO('\r\n'.join(body)) env['wsgi.input'] = StringIO('\r\n'.join(body))
env['swift.cache'] = FakeMemcache() env['swift.account/AUTH_test'] = self._fake_cache_env(
env['swift.cache'].set('temp-url-key/AUTH_test', key) 'AUTH_test', [key])
self.app = FakeApp(iter([('201 Created', {}, ''), self.app = FakeApp(iter([('201 Created', {}, ''),
('201 Created', {}, '')])) ('201 Created', {}, '')]))
self.auth = tempauth.filter_factory({})(self.app) self.auth = tempauth.filter_factory({})(self.app)