Merge "Add test to verify staticweb uses auth'd subrequests"
This commit is contained in:
commit
e283718ed0
@ -17,7 +17,7 @@ import json
|
|||||||
import unittest
|
import unittest
|
||||||
import mock
|
import mock
|
||||||
|
|
||||||
from swift.common.swob import Request, Response
|
from swift.common.swob import Request, Response, HTTPUnauthorized
|
||||||
from swift.common.middleware import staticweb
|
from swift.common.middleware import staticweb
|
||||||
|
|
||||||
|
|
||||||
@ -37,7 +37,8 @@ meta_map = {
|
|||||||
'web-error': 'error.html',
|
'web-error': 'error.html',
|
||||||
'web-listings': 't',
|
'web-listings': 't',
|
||||||
'web-listings-css': 'listing.css'}},
|
'web-listings-css': 'listing.css'}},
|
||||||
'c6': {'meta': {'web-listings': 't'}},
|
'c6': {'meta': {'web-listings': 't',
|
||||||
|
'web-error': 'error.html'}},
|
||||||
'c7': {'meta': {'web-listings': 'f'}},
|
'c7': {'meta': {'web-listings': 'f'}},
|
||||||
'c8': {'meta': {'web-error': 'error.html',
|
'c8': {'meta': {'web-error': 'error.html',
|
||||||
'web-listings': 't',
|
'web-listings': 't',
|
||||||
@ -74,6 +75,10 @@ class FakeApp(object):
|
|||||||
|
|
||||||
def __call__(self, env, start_response):
|
def __call__(self, env, start_response):
|
||||||
self.calls += 1
|
self.calls += 1
|
||||||
|
if 'swift.authorize' in env:
|
||||||
|
resp = env['swift.authorize'](Request(env))
|
||||||
|
if resp:
|
||||||
|
return resp(env, start_response)
|
||||||
if env['PATH_INFO'] == '/':
|
if env['PATH_INFO'] == '/':
|
||||||
return Response(status='404 Not Found')(env, start_response)
|
return Response(status='404 Not Found')(env, start_response)
|
||||||
elif env['PATH_INFO'] == '/v1':
|
elif env['PATH_INFO'] == '/v1':
|
||||||
@ -183,6 +188,14 @@ class FakeApp(object):
|
|||||||
return self.listing(env, start_response)
|
return self.listing(env, start_response)
|
||||||
elif env['PATH_INFO'] == '/v1/a/c6/subdir':
|
elif env['PATH_INFO'] == '/v1/a/c6/subdir':
|
||||||
return Response(status='404 Not Found')(env, start_response)
|
return Response(status='404 Not Found')(env, start_response)
|
||||||
|
elif env['PATH_INFO'] == '/v1/a/c6/401error.html':
|
||||||
|
return Response(status='200 Ok', body='''
|
||||||
|
<html>
|
||||||
|
<body style="background: #000000; color: #ffaaaa">
|
||||||
|
<p>Hey, you're not authorized to see this!</p>
|
||||||
|
</body>
|
||||||
|
</html>
|
||||||
|
'''.strip())(env, start_response)
|
||||||
elif env['PATH_INFO'] in ('/v1/a/c7', '/v1/a/c7/'):
|
elif env['PATH_INFO'] in ('/v1/a/c7', '/v1/a/c7/'):
|
||||||
return self.listing(env, start_response)
|
return self.listing(env, start_response)
|
||||||
elif env['PATH_INFO'] in ('/v1/a/c8', '/v1/a/c8/'):
|
elif env['PATH_INFO'] in ('/v1/a/c8', '/v1/a/c8/'):
|
||||||
@ -382,11 +395,19 @@ class FakeApp(object):
|
|||||||
|
|
||||||
class FakeAuthFilter(object):
|
class FakeAuthFilter(object):
|
||||||
|
|
||||||
def __init__(self, app):
|
def __init__(self, app, deny_objects=False, deny_listing=False):
|
||||||
self.app = app
|
self.app = app
|
||||||
|
self.deny_objects = deny_objects
|
||||||
|
self.deny_listing = deny_listing
|
||||||
|
|
||||||
|
def authorize(self, req):
|
||||||
|
path_parts = req.path.strip('/').split('/')
|
||||||
|
if ((self.deny_objects and len(path_parts) > 3)
|
||||||
|
or (self.deny_listing and len(path_parts) == 3)):
|
||||||
|
return HTTPUnauthorized()
|
||||||
|
|
||||||
def __call__(self, env, start_response):
|
def __call__(self, env, start_response):
|
||||||
env['swift.authorize'] = None
|
env['swift.authorize'] = self.authorize
|
||||||
return self.app(env, start_response)
|
return self.app(env, start_response)
|
||||||
|
|
||||||
|
|
||||||
@ -609,6 +630,27 @@ class TestStaticWeb(unittest.TestCase):
|
|||||||
'/v1/a/c6/subdir').get_response(self.test_staticweb)
|
'/v1/a/c6/subdir').get_response(self.test_staticweb)
|
||||||
self.assertEqual(resp.status_int, 301)
|
self.assertEqual(resp.status_int, 301)
|
||||||
|
|
||||||
|
def test_container6listing(self):
|
||||||
|
# container6 has web-listings = t, web-error=error.html
|
||||||
|
resp = Request.blank('/v1/a/c6/').get_response(self.test_staticweb)
|
||||||
|
self.assertEqual(resp.status_int, 200)
|
||||||
|
|
||||||
|
# expect custom 401 if request is not auth'd for listing but is auth'd
|
||||||
|
# to GET objects
|
||||||
|
test_staticweb = FakeAuthFilter(
|
||||||
|
staticweb.filter_factory({})(self.app), deny_listing=True)
|
||||||
|
resp = Request.blank('/v1/a/c6/').get_response(test_staticweb)
|
||||||
|
self.assertEqual(resp.status_int, 401)
|
||||||
|
self.assertIn("Hey, you're not authorized to see this!", resp.body)
|
||||||
|
|
||||||
|
# expect default 401 if request is not auth'd for listing or object GET
|
||||||
|
test_staticweb = FakeAuthFilter(
|
||||||
|
staticweb.filter_factory({})(self.app), deny_listing=True,
|
||||||
|
deny_objects=True)
|
||||||
|
resp = Request.blank('/v1/a/c6/').get_response(test_staticweb)
|
||||||
|
self.assertEqual(resp.status_int, 401)
|
||||||
|
self.assertNotIn("Hey, you're not authorized to see this!", resp.body)
|
||||||
|
|
||||||
def test_container7listing(self):
|
def test_container7listing(self):
|
||||||
resp = Request.blank('/v1/a/c7/').get_response(self.test_staticweb)
|
resp = Request.blank('/v1/a/c7/').get_response(self.test_staticweb)
|
||||||
self.assertEqual(resp.status_int, 404)
|
self.assertEqual(resp.status_int, 404)
|
||||||
|
Loading…
x
Reference in New Issue
Block a user