openstack/swift
Code Issues Proposed changes

Add WWW-Authenticate to 401 responses

Browse Source 
Per http://www.ietf.org/rfc/rfc2616.txt, when a 401 error is returned, the
Www-Authenticate response header MUST also be returned. The format is
described in http://www.ietf.org/rfc/rfc2617.txt.

Swift supports and/or implements a number of authentication schemes
including tempauth, Keystone, tempurl, formpost and container sync. In
this fix, we use a catch-all, "Swift". The realm is the account (where
known) or "unknown" (bad path or where the 401 is returned from code
that does not have the request). Examples:

     Www-Authenticate: Swift realm="AUTH_1234567889"
     Www-Authenticate: Swift realm="unknown"

Fixes bug #1215491

Change-Id: I03362789318dfa156d3733ef9348795062a9cfc4
This commit is contained in:
Donagh McCabe 2013-08-23 15:03:08 +01:00
parent 0c6764ba30
commit 9807a358c6
9 changed files with 287 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -111,6 +111,7 @@ 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.wsgi import make_pre_authed_env
from swift.common.swob import HTTPUnauthorized
from swift.proxy.controllers.base import get_account_info
@ -129,6 +130,10 @@ class FormInvalid(Exception):
pass
class FormUnauthorized(Exception):
pass
def _parse_attrs(header):
"""
Given the value of a header like:
@ -327,6 +332,10 @@ class FormPost(object):
(('Content-Type', 'text/plain'),
('Content-Length', str(len(body)))))
return [body]
except FormUnauthorized as err:
message = 'FormPost: %s' % str(err).title()
return HTTPUnauthorized(body=message)(
env, start_response)
return self.app(env, start_response)
def _translate_form(self, env, boundary):
@ -410,7 +419,7 @@ class FormPost(object):
:returns: (status_line, message)
"""
if not keys:
return '401 Unauthorized', 'invalid signature'
raise FormUnauthorized('invalid signature')
try:
max_file_size = int(attributes.get('max_file_size') or 0)
except ValueError:
@ -432,7 +441,7 @@ class FormPost(object):
del subenv['CONTENT_TYPE']
try:
if int(attributes.get('expires') or 0) < time():
return '401 Unauthorized', 'form expired'
raise FormUnauthorized('form expired')
except ValueError:
raise FormInvalid('expired not an integer')
hmac_body = '%s\n%s\n%s\n%s\n%s' % (
@ -449,7 +458,7 @@ class FormPost(object):
'invalid')):
has_valid_sig = True
if not has_valid_sig:
return '401 Unauthorized', 'invalid signature'
raise FormUnauthorized('invalid signature')
substatus = [None]

34
swift/common/middleware/tempauth.py
View File

@ -156,7 +156,14 @@ class TempAuth(object):
# Because I know I'm the definitive auth for this token, I
# can deny it outright.
self.logger.increment('unauthorized')
return HTTPUnauthorized()(env, start_response)
try:
vrs, realm, rest = split_path(env['PATH_INFO'],
2, 3, True)
except ValueError:
realm = 'unknown'
return HTTPUnauthorized(headers={
'Www-Authenticate': 'Swift realm="%s"' % realm})(
env, start_response)
# Because I'm not certain if I'm the definitive auth for empty
# reseller_prefixed tokens, I won't overwrite swift.authorize.
elif 'swift.authorize' not in env:
@ -408,11 +415,15 @@ class TempAuth(object):
user = req.headers.get('x-auth-user')
if not user or ':' not in user:
self.logger.increment('token_denied')
return HTTPUnauthorized(request=req)
return HTTPUnauthorized(request=req, headers=
{'Www-Authenticate':
'Swift realm="%s"' % account})
account2, user = user.split(':', 1)
if account != account2:
self.logger.increment('token_denied')
return HTTPUnauthorized(request=req)
return HTTPUnauthorized(request=req, headers=
{'Www-Authenticate':
'Swift realm="%s"' % account})
key = req.headers.get('x-storage-pass')
if not key:
key = req.headers.get('x-auth-key')
@ -422,7 +433,9 @@ class TempAuth(object):
user = req.headers.get('x-storage-user')
if not user or ':' not in user:
self.logger.increment('token_denied')
return HTTPUnauthorized(request=req)
return HTTPUnauthorized(request=req, headers=
{'Www-Authenticate':
'Swift realm="unknown"'})
account, user = user.split(':', 1)
key = req.headers.get('x-auth-key')
if not key:
@ -431,15 +444,22 @@ class TempAuth(object):
return HTTPBadRequest(request=req)
if not all((account, user, key)):
self.logger.increment('token_denied')
return HTTPUnauthorized(request=req)
realm = account or 'unknown'
return HTTPUnauthorized(request=req, headers={'Www-Authenticate':
'Swift realm="%s"' %
realm})
# Authenticate user
account_user = account + ':' + user
if account_user not in self.users:
self.logger.increment('token_denied')
return HTTPUnauthorized(request=req)
return HTTPUnauthorized(request=req, headers=
{'Www-Authenticate':
'Swift realm="%s"' % account})
if self.users[account_user]['key'] != key:
self.logger.increment('token_denied')
return HTTPUnauthorized(request=req)
return HTTPUnauthorized(request=req, headers=
{'Www-Authenticate':
'Swift realm="unknown"'})
account_id = self.users[account_user]['url'].rsplit('/', 1)[-1]
# Get memcache client
memcache_client = cache_from_env(req.environ)

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

@ -99,6 +99,7 @@ from urlparse import parse_qs
from swift.proxy.controllers.base import get_account_info
from swift.common.swob import HeaderKeyDict
from swift.common.utils import split_path
from swift.common.swob import HTTPUnauthorized
#: Default headers to remove from incoming requests. Simply a whitespace
@ -412,13 +413,11 @@ class TempURL(object):
:param start_response: The WSGI start_response hook.
:returns: 401 response as per WSGI.
"""
body = '401 Unauthorized: Temp URL invalid\n'
start_response('401 Unauthorized',
[('Content-Type', 'text/plain'),
('Content-Length', str(len(body)))])
if env['REQUEST_METHOD'] == 'HEAD':
return []
return [body]
body = None
else:
body = '401 Unauthorized: Temp URL invalid\n'
return HTTPUnauthorized(body=body)(env, start_response)
def _clean_incoming_headers(self, env):
"""

17
swift/common/swob.py
View File

@ -1042,6 +1042,8 @@ class Response(object):
self.environ = {}
if headers:
self.headers.update(headers)
if self.status_int == 401 and 'www-authenticate' not in self.headers:
self.headers.update({'www-authenticate': self.www_authenticate()})
for key, value in kw.iteritems():
setattr(self, key, value)
# When specifying both 'content_type' and 'charset' in the kwargs,
@ -1152,6 +1154,21 @@ class Response(object):
return self.location
return self.host_url + self.location
def www_authenticate(self):
"""
Construct a suitable value for WWW-Authenticate response header
If we have a request and a valid-looking path, the realm
is the account; otherwise we set it to 'unknown'.
"""
try:
vrs, realm, rest = self.request.split_path(2, 3, True)
if realm in ('v1.0', 'auth'):
realm = 'unknown'
except (AttributeError, ValueError):
realm = 'unknown'
return 'Swift realm="%s"' % realm
@property
def is_success(self):
return self.status_int // 100 == 2

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

@ -392,6 +392,39 @@ class TestFormPost(unittest.TestCase):
self.assertEquals(resp.status_int, 401)
self.assertTrue('FormPost' not in resp.body)
def test_auth_scheme(self):
# FormPost rejects
key = 'abc'
sig, env, body = self._make_sig_env_body(
'/v1/AUTH_test/container', '', 1024, 10, int(time() - 10), key)
env['wsgi.input'] = StringIO('\r\n'.join(body))
env['swift.account/AUTH_test'] = self._fake_cache_env(
'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, '401 Unauthorized')
authenticate_v = None
for h, v in headers:
if h.lower() == 'www-authenticate':
authenticate_v = v
self.assertTrue('FormPost: Form Expired' in body)
self.assertEquals('Swift realm="unknown"', authenticate_v)
def test_safari(self):
key = 'abc'
path = '/v1/AUTH_test/container'

7
test/unit/common/middleware/test_keystoneauth.py
View File

@ -163,6 +163,13 @@ class SwiftAuth(unittest.TestCase):
resp = req.get_response(self._get_successful_middleware())
self.assertEqual(resp.status_int, 200)
def test_auth_scheme(self):
req = self._make_request(path='/v1/BLAH_foo/c/o',
headers={'X_IDENTITY_STATUS': 'Invalid'})
resp = req.get_response(self.test_auth)
self.assertEqual(resp.status_int, 401)
self.assertTrue('Www-Authenticate' in resp.headers)
class TestAuthorize(unittest.TestCase):
def setUp(self):

57
test/unit/common/middleware/test_tempauth.py
View File

@ -143,6 +143,8 @@ class TestAuth(unittest.TestCase):
self.assertEquals(resp.status_int, 401)
self.assertEquals(req.environ['swift.authorize'],
self.test_auth.denied_response)
self.assertEquals(resp.headers.get('Www-Authenticate'),
'Swift realm="unknown"')
def test_anon(self):
req = self._make_request('/v1/AUTH_account')
@ -150,6 +152,15 @@ class TestAuth(unittest.TestCase):
self.assertEquals(resp.status_int, 401)
self.assertEquals(req.environ['swift.authorize'],
self.test_auth.authorize)
self.assertEquals(resp.headers.get('Www-Authenticate'),
'Swift realm="AUTH_account"')
def test_anon_badpath(self):
req = self._make_request('/v1')
resp = req.get_response(self.test_auth)
self.assertEquals(resp.status_int, 401)
self.assertEquals(resp.headers.get('Www-Authenticate'),
'Swift realm="unknown"')
def test_override_asked_for_but_not_allowed(self):
self.test_auth = \
@ -158,6 +169,8 @@ class TestAuth(unittest.TestCase):
environ={'swift.authorize_override': True})
resp = req.get_response(self.test_auth)
self.assertEquals(resp.status_int, 401)
self.assertEquals(resp.headers.get('Www-Authenticate'),
'Swift realm="AUTH_account"')
self.assertEquals(req.environ['swift.authorize'],
self.test_auth.authorize)
@ -182,6 +195,8 @@ class TestAuth(unittest.TestCase):
headers={'X-Auth-Token': 'BLAH_t'})
resp = req.get_response(self.test_auth)
self.assertEquals(resp.status_int, 401)
self.assertEquals(resp.headers.get('Www-Authenticate'),
'Swift realm="BLAH_account"')
self.assertEquals(req.environ['swift.authorize'],
self.test_auth.denied_response)
@ -205,6 +220,8 @@ class TestAuth(unittest.TestCase):
headers={'X-Auth-Token': 't'})
resp = req.get_response(local_auth)
self.assertEquals(resp.status_int, 401)
self.assertEquals(resp.headers.get('Www-Authenticate'),
'Swift realm="account"')
self.assertEquals(local_app.calls, 1)
self.assertEquals(req.environ['swift.authorize'],
local_auth.denied_response)
@ -216,6 +233,8 @@ class TestAuth(unittest.TestCase):
req = self._make_request('/v1/account')
resp = req.get_response(local_auth)
self.assertEquals(resp.status_int, 401)
self.assertEquals(resp.headers.get('Www-Authenticate'),
'Swift realm="account"')
self.assertEquals(req.environ['swift.authorize'],
local_auth.authorize)
# Now make sure we don't override an existing swift.authorize when we
@ -234,11 +253,15 @@ class TestAuth(unittest.TestCase):
'/v1/AUTH_cfa',
headers={'X-Auth-Token': 'AUTH_t'}).get_response(self.test_auth)
self.assertEquals(resp.status_int, 401)
self.assertEquals(resp.headers.get('Www-Authenticate'),
'Swift realm="AUTH_cfa"')
def test_authorize_bad_path(self):
req = self._make_request('/badpath')
resp = self.test_auth.authorize(req)
self.assertEquals(resp.status_int, 401)
self.assertEquals(resp.headers.get('Www-Authenticate'),
'Swift realm="unknown"')
req = self._make_request('/badpath')
req.remote_user = 'act:usr,act,AUTH_cfa'
resp = self.test_auth.authorize(req)
@ -318,6 +341,8 @@ class TestAuth(unittest.TestCase):
req = self._make_request('/v1/AUTH_cfa/c')
resp = self.test_auth.authorize(req)
self.assertEquals(resp.status_int, 401)
self.assertEquals(resp.headers.get('Www-Authenticate'),
'Swift realm="AUTH_cfa"')
req = self._make_request('/v1/AUTH_cfa/c')
req.acl = '.r:*,.rlistings'
self.assertEquals(self.test_auth.authorize(req), None)
@ -325,10 +350,14 @@ class TestAuth(unittest.TestCase):
req.acl = '.r:*' # No listings allowed
resp = self.test_auth.authorize(req)
self.assertEquals(resp.status_int, 401)
self.assertEquals(resp.headers.get('Www-Authenticate'),
'Swift realm="AUTH_cfa"')
req = self._make_request('/v1/AUTH_cfa/c')
req.acl = '.r:.example.com,.rlistings'
resp = self.test_auth.authorize(req)
self.assertEquals(resp.status_int, 401)
self.assertEquals(resp.headers.get('Www-Authenticate'),
'Swift realm="AUTH_cfa"')
req = self._make_request('/v1/AUTH_cfa/c')
req.referer = 'http://www.example.com/index.html'
req.acl = '.r:.example.com,.rlistings'
@ -414,11 +443,16 @@ class TestAuth(unittest.TestCase):
def test_get_token_fail(self):
resp = self._make_request('/auth/v1.0').get_response(self.test_auth)
self.assertEquals(resp.status_int, 401)
self.assertEquals(resp.headers.get('Www-Authenticate'),
'Swift realm="unknown"')
resp = self._make_request(
'/auth/v1.0',
headers={'X-Auth-User': 'act:usr',
'X-Auth-Key': 'key'}).get_response(self.test_auth)
self.assertEquals(resp.status_int, 401)
self.assertTrue('Www-Authenticate' in resp.headers)
self.assertEquals(resp.headers.get('Www-Authenticate'),
'Swift realm="act"')
def test_get_token_fail_invalid_x_auth_user_format(self):
resp = self._make_request(
@ -426,6 +460,8 @@ class TestAuth(unittest.TestCase):
headers={'X-Auth-User': 'usr',
'X-Auth-Key': 'key'}).get_response(self.test_auth)
self.assertEquals(resp.status_int, 401)
self.assertEquals(resp.headers.get('Www-Authenticate'),
'Swift realm="act"')
def test_get_token_fail_non_matching_account_in_request(self):
resp = self._make_request(
@ -433,6 +469,8 @@ class TestAuth(unittest.TestCase):
headers={'X-Auth-User': 'act2:usr',
'X-Auth-Key': 'key'}).get_response(self.test_auth)
self.assertEquals(resp.status_int, 401)
self.assertEquals(resp.headers.get('Www-Authenticate'),
'Swift realm="act"')
def test_get_token_fail_bad_path(self):
resp = self._make_request(
@ -446,6 +484,8 @@ class TestAuth(unittest.TestCase):
'/auth/v1/act/auth',
headers={'X-Auth-User': 'act:usr'}).get_response(self.test_auth)
self.assertEquals(resp.status_int, 401)
self.assertEquals(resp.headers.get('Www-Authenticate'),
'Swift realm="act"')
def test_storage_url_default(self):
self.test_auth = \
@ -619,6 +659,8 @@ class TestAuth(unittest.TestCase):
req.remote_addr = '127.0.0.1'
resp = req.get_response(self.test_auth)
self.assertEquals(resp.status_int, 401)
self.assertEquals(resp.headers.get('Www-Authenticate'),
'Swift realm="AUTH_cfa"')
self.test_auth.app = FakeApp(iter([('204 No Content', {}, '')]),
sync_key='othersecret')
@ -630,6 +672,8 @@ class TestAuth(unittest.TestCase):
req.remote_addr = '127.0.0.1'
resp = req.get_response(self.test_auth)
self.assertEquals(resp.status_int, 401)
self.assertEquals(resp.headers.get('Www-Authenticate'),
'Swift realm="AUTH_cfa"')
self.test_auth.app = FakeApp(iter([('204 No Content', {}, '')]),
sync_key=None)
@ -641,6 +685,8 @@ class TestAuth(unittest.TestCase):
req.remote_addr = '127.0.0.1'
resp = req.get_response(self.test_auth)
self.assertEquals(resp.status_int, 401)
self.assertEquals(resp.headers.get('Www-Authenticate'),
'Swift realm="AUTH_cfa"')
def test_sync_request_fail_no_timestamp(self):
self.test_auth.app = FakeApp(iter([('204 No Content', {}, '')]),
@ -652,6 +698,8 @@ class TestAuth(unittest.TestCase):
req.remote_addr = '127.0.0.1'
resp = req.get_response(self.test_auth)
self.assertEquals(resp.status_int, 401)
self.assertEquals(resp.headers.get('Www-Authenticate'),
'Swift realm="AUTH_cfa"')
def test_sync_request_success_lb_sync_host(self):
self.test_auth.app = FakeApp(iter([('204 No Content', {}, '')]),
@ -696,6 +744,15 @@ class TestAuth(unittest.TestCase):
groups = ath._get_user_groups('test', 'test:tester', 'AUTH_test')
self.assertEquals(groups, 'test,test:tester')
def test_auth_scheme(self):
req = self._make_request('/v1/BLAH_account',
headers={'X-Auth-Token': 'BLAH_t'})
resp = req.get_response(self.test_auth)
self.assertEquals(resp.status_int, 401)
self.assertTrue('Www-Authenticate' in resp.headers)
self.assertEquals(resp.headers.get('Www-Authenticate'),
'Swift realm="BLAH_account"')
class TestParseUserCreation(unittest.TestCase):
def test_parse_user_creation(self):

34
test/unit/common/middleware/test_tempurl.py
View File

@ -214,6 +214,7 @@ class TestTempURL(unittest.TestCase):
resp = req.get_response(self.tempurl)
self.assertEquals(resp.status_int, 401)
self.assertTrue('Temp URL invalid' in resp.body)
self.assertTrue('Www-Authenticate' in resp.headers)
def test_put_valid(self):
method = 'PUT'
@ -246,6 +247,7 @@ class TestTempURL(unittest.TestCase):
resp = req.get_response(self.tempurl)
self.assertEquals(resp.status_int, 401)
self.assertTrue('Temp URL invalid' in resp.body)
self.assertTrue('Www-Authenticate' in resp.headers)
def test_missing_sig(self):
method = 'GET'
@ -260,6 +262,7 @@ class TestTempURL(unittest.TestCase):
resp = req.get_response(self.tempurl)
self.assertEquals(resp.status_int, 401)
self.assertTrue('Temp URL invalid' in resp.body)
self.assertTrue('Www-Authenticate' in resp.headers)
def test_missing_expires(self):
method = 'GET'
@ -274,6 +277,7 @@ class TestTempURL(unittest.TestCase):
resp = req.get_response(self.tempurl)
self.assertEquals(resp.status_int, 401)
self.assertTrue('Temp URL invalid' in resp.body)
self.assertTrue('Www-Authenticate' in resp.headers)
def test_bad_path(self):
method = 'GET'
@ -289,6 +293,7 @@ class TestTempURL(unittest.TestCase):
resp = req.get_response(self.tempurl)
self.assertEquals(resp.status_int, 401)
self.assertTrue('Temp URL invalid' in resp.body)
self.assertTrue('Www-Authenticate' in resp.headers)
def test_no_key(self):
method = 'GET'
@ -304,6 +309,7 @@ class TestTempURL(unittest.TestCase):
resp = req.get_response(self.tempurl)
self.assertEquals(resp.status_int, 401)
self.assertTrue('Temp URL invalid' in resp.body)
self.assertTrue('Www-Authenticate' in resp.headers)
def test_head_allowed_by_get(self):
method = 'GET'
@ -356,6 +362,7 @@ class TestTempURL(unittest.TestCase):
sig, expires)})
resp = req.get_response(self.tempurl)
self.assertEquals(resp.status_int, 401)
self.assertTrue('Www-Authenticate' in resp.headers)
def test_post_not_allowed(self):
method = 'POST'
@ -372,6 +379,7 @@ class TestTempURL(unittest.TestCase):
resp = req.get_response(self.tempurl)
self.assertEquals(resp.status_int, 401)
self.assertTrue('Temp URL invalid' in resp.body)
self.assertTrue('Www-Authenticate' in resp.headers)
def test_delete_not_allowed(self):
method = 'DELETE'
@ -388,6 +396,7 @@ class TestTempURL(unittest.TestCase):
resp = req.get_response(self.tempurl)
self.assertEquals(resp.status_int, 401)
self.assertTrue('Temp URL invalid' in resp.body)
self.assertTrue('Www-Authenticate' in resp.headers)
def test_delete_allowed_with_conf(self):
self.tempurl.methods.append('DELETE')
@ -420,6 +429,7 @@ class TestTempURL(unittest.TestCase):
resp = req.get_response(self.tempurl)
self.assertEquals(resp.status_int, 401)
self.assertTrue('Temp URL invalid' in resp.body)
self.assertTrue('Www-Authenticate' in resp.headers)
def test_changed_path_invalid(self):
method = 'GET'
@ -435,6 +445,7 @@ class TestTempURL(unittest.TestCase):
resp = req.get_response(self.tempurl)
self.assertEquals(resp.status_int, 401)
self.assertTrue('Temp URL invalid' in resp.body)
self.assertTrue('Www-Authenticate' in resp.headers)
def test_changed_sig_invalid(self):
method = 'GET'
@ -454,6 +465,7 @@ class TestTempURL(unittest.TestCase):
resp = req.get_response(self.tempurl)
self.assertEquals(resp.status_int, 401)
self.assertTrue('Temp URL invalid' in resp.body)
self.assertTrue('Www-Authenticate' in resp.headers)
def test_changed_expires_invalid(self):
method = 'GET'
@ -469,6 +481,7 @@ class TestTempURL(unittest.TestCase):
resp = req.get_response(self.tempurl)
self.assertEquals(resp.status_int, 401)
self.assertTrue('Temp URL invalid' in resp.body)
self.assertTrue('Www-Authenticate' in resp.headers)
def test_different_key_invalid(self):
method = 'GET'
@ -484,6 +497,7 @@ class TestTempURL(unittest.TestCase):
resp = req.get_response(self.tempurl)
self.assertEquals(resp.status_int, 401)
self.assertTrue('Temp URL invalid' in resp.body)
self.assertTrue('Www-Authenticate' in resp.headers)
def test_removed_incoming_header(self):
self.tempurl = tempurl.filter_factory({
@ -653,6 +667,26 @@ class TestTempURL(unittest.TestCase):
self.tempurl._invalid({'REQUEST_METHOD': 'HEAD'},
_start_response)))
def test_auth_scheme_value(self):
# Passthrough
environ = {}
resp = self._make_request('/v1/a/c/o', environ=environ).get_response(
self.tempurl)
self.assertEquals(resp.status_int, 401)
self.assertTrue('Temp URL invalid' not in resp.body)
self.assertTrue('Www-Authenticate' in resp.headers)
self.assertTrue('swift.auth_scheme' not in environ)
# Rejected by TempURL
req = self._make_request('/v1/a/c/o', keys=['abc'],
environ={'REQUEST_METHOD': 'PUT',
'QUERY_STRING':
'temp_url_sig=dummy&temp_url_expires=1234'})
resp = req.get_response(self.tempurl)
self.assertEquals(resp.status_int, 401)
self.assertTrue('Temp URL invalid' in resp.body)
self.assert_('Www-Authenticate' in resp.headers)
def test_clean_incoming_headers(self):
irh = ''
iah = ''

95
test/unit/common/test_swob.py
View File

@ -518,6 +518,101 @@ class TestRequest(unittest.TestCase):
self.assertEquals(resp.status_int, 200)
self.assertEquals(resp.body, 'hi')
def test_401_unauthorized(self):
# No request environment
resp = swift.common.swob.HTTPUnauthorized()
self.assertEquals(resp.status_int, 401)
self.assert_('Www-Authenticate' in resp.headers)
# Request environment
req = swift.common.swob.Request.blank('/')
resp = swift.common.swob.HTTPUnauthorized(request=req)
self.assertEquals(resp.status_int, 401)
self.assert_('Www-Authenticate' in resp.headers)
def test_401_valid_account_path(self):
def test_app(environ, start_response):
start_response('401 Unauthorized', [])
return ['hi']
# Request environment contains valid account in path
req = swift.common.swob.Request.blank('/v1/account-name')
resp = req.get_response(test_app)
self.assertEquals(resp.status_int, 401)
self.assert_('Www-Authenticate' in resp.headers)
self.assertEquals('Swift realm="account-name"',
resp.headers['Www-Authenticate'])
# Request environment contains valid account/container in path
req = swift.common.swob.Request.blank('/v1/account-name/c')
resp = req.get_response(test_app)
self.assertEquals(resp.status_int, 401)
self.assert_('Www-Authenticate' in resp.headers)
self.assertEquals('Swift realm="account-name"',
resp.headers['Www-Authenticate'])
def test_401_invalid_path(self):
def test_app(environ, start_response):
start_response('401 Unauthorized', [])
return ['hi']
# Request environment contains bad path
req = swift.common.swob.Request.blank('/random')
resp = req.get_response(test_app)
self.assertEquals(resp.status_int, 401)
self.assert_('Www-Authenticate' in resp.headers)
self.assertEquals('Swift realm="unknown"',
resp.headers['Www-Authenticate'])
def test_401_non_keystone_auth_path(self):
def test_app(environ, start_response):
start_response('401 Unauthorized', [])
return ['no creds in request']
# Request to get token
req = swift.common.swob.Request.blank('/v1.0/auth')
resp = req.get_response(test_app)
self.assertEquals(resp.status_int, 401)
self.assert_('Www-Authenticate' in resp.headers)
self.assertEquals('Swift realm="unknown"',
resp.headers['Www-Authenticate'])
# Other form of path
req = swift.common.swob.Request.blank('/auth/v1.0')
resp = req.get_response(test_app)
self.assertEquals(resp.status_int, 401)
self.assert_('Www-Authenticate' in resp.headers)
self.assertEquals('Swift realm="unknown"',
resp.headers['Www-Authenticate'])
def test_401_www_authenticate_exists(self):
def test_app(environ, start_response):
start_response('401 Unauthorized', {
'Www-Authenticate': 'Me realm="whatever"'})
return ['no creds in request']
# Auth middleware sets own Www-Authenticate
req = swift.common.swob.Request.blank('/auth/v1.0')
resp = req.get_response(test_app)
self.assertEquals(resp.status_int, 401)
self.assert_('Www-Authenticate' in resp.headers)
self.assertEquals('Me realm="whatever"',
resp.headers['Www-Authenticate'])
def test_not_401(self):
# Other status codes should not have WWW-Authenticate in response
def test_app(environ, start_response):
start_response('200 OK', [])
return ['hi']
req = swift.common.swob.Request.blank('/')
resp = req.get_response(test_app)
self.assert_('Www-Authenticate' not in resp.headers)
def test_properties(self):
req = swift.common.swob.Request.blank('/hi/there', body='hi')