x/swiftonhpss
Code Issues Proposed changes

Fix bug where admin is able to delete reseller_admin

Browse Source 
Changed the code to only allow the super_admin to delete a reseller_admin.
This follows the same logic of user creation, where only the super_admin
can create a reseller_admin.

Also, took the opportunity to refactor some code and implemented
get_user_detail method to remove duplicated code

Bug 1260239: https://bugs.launchpad.net/gluster-swift/+bug/1260239

Change-Id: I9e4866cd7ad08698f427846be566ab2364ad4850
Signed-off-by: Thiago da Silva <thiago@redhat.com>
Reviewed-on: http://review.gluster.org/6516
Reviewed-by: Luis Pabon <lpabon@redhat.com>
Tested-by: Luis Pabon <lpabon@redhat.com>
This commit is contained in:
Thiago da Silva 2013-12-16 12:00:38 -05:00 committed by Luis Pabon
parent ac55cfb30b
commit 205a6e4aa7
3 changed files with 223 additions and 31 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

90
gluster/swift/common/middleware/gswauth/swauth/middleware.py
View File

@ -908,6 +908,10 @@ class Swauth(object):
return HTTPBadRequest(request=req)
if not self.is_account_admin(req, account):
return self.denied_response(req)
# get information for each user for the specified
# account and create a list of all groups that the users
# are part of
if user == '.groups':
# TODO: This could be very slow for accounts with a really large
# number of users. Speed could be improved by concurrently
@ -932,28 +936,25 @@ class Swauth(object):
break
for obj in sublisting:
if obj['name'][0] != '.':
path = quote('/v1/%s/%s/%s' % (self.auth_account,
account, obj['name']))
resp = self.make_pre_authed_request(
req.environ, 'GET', path).get_response(self.app)
if resp.status_int // 100 != 2:
# get list of groups for each user
user_json = self.get_user_detail(req, account,
obj['name'])
if user_json is None:
raise Exception('Could not retrieve user object: '
'%s %s' % (path, resp.status))
'%s:%s %s' % (account, user, 404))
groups.update(
g['name'] for g in json.loads(resp.body)['groups'])
g['name'] for g in json.loads(user_json)['groups'])
marker = sublisting[-1]['name'].encode('utf-8')
body = json.dumps(
{'groups': [{'name': g} for g in sorted(groups)]})
else:
path = quote('/v1/%s/%s/%s' % (self.auth_account, account, user))
resp = self.make_pre_authed_request(
req.environ, 'GET', path).get_response(self.app)
if resp.status_int == 404:
# get information for specific user,
# if user doesn't exist, return HTTPNotFound
body = self.get_user_detail(req, account, user)
if body is None:
return HTTPNotFound(request=req)
if resp.status_int // 100 != 2:
raise Exception('Could not retrieve user object: %s %s' %
(path, resp.status))
body = resp.body
display_groups = [g['name'] for g in json.loads(body)['groups']]
if ('.admin' in display_groups and
not self.is_reseller_admin(req)) or \
@ -1054,8 +1055,19 @@ class Swauth(object):
if req.path_info or not account or account[0] == '.' or not user or \
user[0] == '.':
return HTTPBadRequest(request=req)
# if user to be deleted is reseller_admin, then requesting
# user must also be a reseller_admin
is_reseller_admin = self.is_user_reseller_admin(req, account, user)
if not is_reseller_admin and not req.credentials_valid:
# if user to be deleted can't be found, return 404
return HTTPNotFound(request=req)
elif is_reseller_admin and not self.is_super_admin(req):
return HTTPForbidden(request=req)
if not self.is_account_admin(req, account):
return self.denied_response(req)
# Delete the user's existing token, if any.
path = quote('/v1/%s/%s/%s' % (self.auth_account, account, user))
resp = self.make_pre_authed_request(
@ -1084,6 +1096,26 @@ class Swauth(object):
(path, resp.status))
return HTTPNoContent(request=req)
def is_user_reseller_admin(self, req, account, user):
"""
Returns True if the user is a .reseller_admin.
:param account: account user is part of
:param user: the user
:returns: True if user .reseller_admin, False
if user is not a reseller_admin and None if the user
doesn't exist.
"""
req.credentials_valid = True
user_json = self.get_user_detail(req, account, user)
if user_json is None:
req.credentials_valid = False
return False
user_detail = json.loads(user_json)
return '.reseller_admin' in (g['name'] for g in user_detail['groups'])
def handle_get_token(self, req):
"""
Handles the various `request for token and service end point(s)` calls.
@ -1391,18 +1423,34 @@ class Swauth(object):
return None
admin_account, admin_user = \
req.headers.get('x-auth-admin-user').split(':', 1)
path = quote('/v1/%s/%s/%s' % (self.auth_account, admin_account,
admin_user))
user_json = self.get_user_detail(req, admin_account, admin_user)
if user_json is None:
return None
admin_detail = json.loads(user_json)
admin_detail['account'] = admin_account
return admin_detail
def get_user_detail(self, req, account, user):
"""
Returns the response body of a GET request for the specified user
The body is in JSON format and contains all user information.
:param req: The swob request
:param account: the account the user is a member of
:param user: the user
:returns: A JSON response with the user detail information, None
if the user doesn't exist
"""
path = quote('/v1/%s/%s/%s' % (self.auth_account, account, user))
resp = self.make_pre_authed_request(
req.environ, 'GET', path).get_response(self.app)
if resp.status_int == 404:
return None
if resp.status_int // 100 != 2:
raise Exception('Could not get admin user object: %s %s' %
raise Exception('Could not get user object: %s %s' %
(path, resp.status))
admin_detail = json.loads(resp.body)
admin_detail['account'] = admin_account
return admin_detail
return resp.body
def credentials_match(self, user_detail, key):
"""

4
test/functional_auth/gswauth/test_gswauth_cli.py
View File

@ -492,12 +492,12 @@ class TestUser(unittest.TestCase):
self.assertEqual('403 Forbidden' in output,True, 're_admin deletion succeeded with re_admin user of other account: '+output)
'''
Utils.addResellerAdminUser('test2', 're_admintobedeletedbyotheraccountusers2', 'testing')
(status,output) = Utils.deleteUser('test2', 're_admintobedeletedbyotherusers2',user='test:admin',key='testing')
(status,output) = Utils.deleteUser('test2', 're_admintobedeletedbyotheraccountusers2',user='test:admin',key='testing')
self.assertNotEqual(status, 0, 're_admin deletion succeeded with admin user of other account: '+output)
self.assertEqual('403 Forbidden' in output,True, 're_admin deletion succeeded with admin user of other account: '+output)
Utils.addResellerAdminUser('test2', 're_admintobedeletedbyotheraccountusers3', 'testing')
(status,output) = Utils.deleteUser('test2', 're_admintobedeletedbyotherusers3',user='test:tester',key='testing')
(status,output) = Utils.deleteUser('test2', 're_admintobedeletedbyotheraccountusers3',user='test:tester',key='testing')
self.assertNotEqual(status, 0, 're_admin deletion succeeded with regular user of other account: '+output)
self.assertEqual('403 Forbidden' in output,True, 're_admin deletion succeeded with user of other account: '+output)

160
test/unit/common/middleware/gswauth/swauth/test_middleware.py
View File

@ -3543,6 +3543,9 @@ class TestAuth(unittest.TestCase):
def test_delete_user_bad_creds(self):
self.test_auth.app = FakeApp(iter([
('200 Ok', {}, json.dumps({"groups": [{"name": "act2:adm"},
{"name": "test"}, {"name": ".admin"}],
"auth": "plaintext:key"})),
# GET of user object (account admin, but wrong
# account)
('200 Ok', {}, json.dumps({"groups": [{"name": "act2:adm"},
@ -3557,9 +3560,11 @@ class TestAuth(unittest.TestCase):
'X-Auth-Admin-Key': 'key'}
).get_response(self.test_auth)
self.assertEquals(resp.status_int, 403)
self.assertEquals(self.test_auth.app.calls, 1)
self.assertEquals(self.test_auth.app.calls, 2)
self.test_auth.app = FakeApp(iter([
('200 Ok', {}, json.dumps({"groups": [{"name": "act:usr"},
{"name": "test"}], "auth": "plaintext:key"})),
# GET of user object (regular user)
('200 Ok', {}, json.dumps({"groups": [{"name": "act:usr"},
{"name": "test"}], "auth": "plaintext:key"}))]))
@ -3572,8 +3577,55 @@ class TestAuth(unittest.TestCase):
'X-Auth-Admin-Key': 'key'}
).get_response(self.test_auth)
self.assertEquals(resp.status_int, 403)
self.assertEquals(self.test_auth.app.calls, 2)
def test_delete_reseller_admin_user_fail(self):
self.test_auth.app = FakeApp(iter([
# is user being deleted a reseller_admin
('200 Ok', {}, json.dumps({"groups": [{"name": "act2:re_adm"},
{"name": "act2"}, {"name": ".admin"},
{"name": ".reseller_admin"}], "auth": "plaintext:key"})),
# GET of user object
('200 Ok', {}, json.dumps({"groups": [{"name": "act2:adm"},
{"name": "act2"}, {"name": ".admin"}],
"auth": "plaintext:key"}))]))
resp = Request.blank('/auth/v2/act2/re_adm',
environ={
'REQUEST_METHOD': 'DELETE'},
headers={
'X-Auth-Admin-User':
'act2:adm',
'X-Auth-Admin-Key': 'key'}
).get_response(self.test_auth)
self.assertEquals(resp.status_int, 403)
self.assertEquals(self.test_auth.app.calls, 1)
def test_delete_reseller_admin_user_success(self):
self.test_auth.app = FakeApp(iter([
# is user being deleted a reseller_admin
('200 Ok', {}, json.dumps({"groups": [{"name": "act2:re_adm"},
{"name": "act2"}, {"name": ".admin"},
{"name": ".reseller_admin"}], "auth": "plaintext:key"})),
# HEAD of user object
('200 Ok',
{'X-Object-Meta-Auth-Token': 'AUTH_tk'}, ''),
# DELETE of token
('204 No Content', {}, ''),
# DELETE of user object
('204 No Content', {}, '')]))
resp = Request.blank('/auth/v2/act2/re_adm',
environ={
'REQUEST_METHOD': 'DELETE'},
headers={
'X-Auth-Admin-User':
'.super_admin',
'X-Auth-Admin-Key': 'supertest'}
).get_response(self.test_auth)
self.assertEquals(resp.status_int, 204)
self.assertEquals(self.test_auth.app.calls, 4)
def test_delete_user_invalid_account(self):
resp = Request.blank('/auth/v2/.invalid/usr',
environ={
@ -3628,6 +3680,9 @@ class TestAuth(unittest.TestCase):
def test_delete_user_fail_delete_token(self):
self.test_auth.app = FakeApp(iter([
# is user reseller_admin
('200 Ok', {}, json.dumps({"groups": [{"name": "act:usr"},
{"name": "test"}], "auth": "plaintext:key"})),
# HEAD of user object
('200 Ok',
{'X-Object-Meta-Auth-Token': 'AUTH_tk'}, ''),
@ -3642,10 +3697,13 @@ class TestAuth(unittest.TestCase):
'X-Auth-Admin-Key': 'supertest'}
).get_response(self.test_auth)
self.assertEquals(resp.status_int, 500)
self.assertEquals(self.test_auth.app.calls, 2)
self.assertEquals(self.test_auth.app.calls, 3)
def test_delete_user_fail_delete_user(self):
self.test_auth.app = FakeApp(iter([
# is user reseller_admin
('200 Ok', {}, json.dumps({"groups": [{"name": "act:usr"},
{"name": "test"}], "auth": "plaintext:key"})),
# HEAD of user object
('200 Ok',
{'X-Object-Meta-Auth-Token': 'AUTH_tk'}, ''),
@ -3662,10 +3720,13 @@ class TestAuth(unittest.TestCase):
'X-Auth-Admin-Key': 'supertest'}
).get_response(self.test_auth)
self.assertEquals(resp.status_int, 500)
self.assertEquals(self.test_auth.app.calls, 3)
self.assertEquals(self.test_auth.app.calls, 4)
def test_delete_user_success(self):
self.test_auth.app = FakeApp(iter([
# is user reseller_admin
('200 Ok', {}, json.dumps({"groups": [{"name": "act:usr"},
{"name": "test"}], "auth": "plaintext:key"})),
# HEAD of user object
('200 Ok',
{'X-Object-Meta-Auth-Token': 'AUTH_tk'}, ''),
@ -3682,10 +3743,13 @@ class TestAuth(unittest.TestCase):
'X-Auth-Admin-Key': 'supertest'}
).get_response(self.test_auth)
self.assertEquals(resp.status_int, 204)
self.assertEquals(self.test_auth.app.calls, 3)
self.assertEquals(self.test_auth.app.calls, 4)
def test_delete_user_success_missing_user_at_end(self):
self.test_auth.app = FakeApp(iter([
# is user reseller_admin
('200 Ok', {}, json.dumps({"groups": [{"name": "act:usr"},
{"name": "test"}], "auth": "plaintext:key"})),
# HEAD of user object
('200 Ok',
{'X-Object-Meta-Auth-Token': 'AUTH_tk'}, ''),
@ -3702,10 +3766,13 @@ class TestAuth(unittest.TestCase):
'X-Auth-Admin-Key': 'supertest'}
).get_response(self.test_auth)
self.assertEquals(resp.status_int, 204)
self.assertEquals(self.test_auth.app.calls, 3)
self.assertEquals(self.test_auth.app.calls, 4)
def test_delete_user_success_missing_token(self):
self.test_auth.app = FakeApp(iter([
# is user reseller_admin
('200 Ok', {}, json.dumps({"groups": [{"name": "act:usr"},
{"name": "test"}], "auth": "plaintext:key"})),
# HEAD of user object
('200 Ok',
{'X-Object-Meta-Auth-Token': 'AUTH_tk'}, ''),
@ -3722,10 +3789,13 @@ class TestAuth(unittest.TestCase):
'X-Auth-Admin-Key': 'supertest'}
).get_response(self.test_auth)
self.assertEquals(resp.status_int, 204)
self.assertEquals(self.test_auth.app.calls, 3)
self.assertEquals(self.test_auth.app.calls, 4)
def test_delete_user_success_no_token(self):
self.test_auth.app = FakeApp(iter([
# is user reseller_admin
('200 Ok', {}, json.dumps({"groups": [{"name": "act:usr"},
{"name": "test"}], "auth": "plaintext:key"})),
# HEAD of user object
('200 Ok', {}, ''),
# DELETE of user object
@ -3739,7 +3809,7 @@ class TestAuth(unittest.TestCase):
'X-Auth-Admin-Key': 'supertest'}
).get_response(self.test_auth)
self.assertEquals(resp.status_int, 204)
self.assertEquals(self.test_auth.app.calls, 2)
self.assertEquals(self.test_auth.app.calls, 3)
def test_validate_token_bad_prefix(self):
resp = Request.blank('/auth/v2/.token/BAD_token'
@ -3934,7 +4004,7 @@ class TestAuth(unittest.TestCase):
headers={'X-Auth-Admin-User': 'act:usr'}))
except Exception as err:
exc = err
self.assertEquals(str(exc), 'Could not get admin user object: '
self.assertEquals(str(exc), 'Could not get user object: '
'/v1/AUTH_gsmetadata/act/usr 503 Service Unavailable')
self.assertEquals(self.test_auth.app.calls, 1)
@ -3954,6 +4024,80 @@ class TestAuth(unittest.TestCase):
'groups': [{'name': 'act:usr'}, {'name': 'act'},
{'name': '.admin'}]})
def test_get_user_detail_success(self):
self.test_auth.app = FakeApp(iter([
('200 Ok', {},
json.dumps({"auth": "plaintext:key",
"groups": [{'name': "act:usr"}, {'name': "act"},
{'name': ".admin"}]}))]))
detail = self.test_auth.get_user_detail(
Request.blank('/',
headers={'X-Auth-Admin-User': 'act:usr'}),
'act', 'usr')
self.assertEquals(self.test_auth.app.calls, 1)
detail_json = json.loads(detail)
self.assertEquals("plaintext:key", detail_json['auth'])
def test_get_user_detail_fail_user_doesnt_exist(self):
self.test_auth.app = FakeApp(
iter([('404 Not Found', {}, '')]))
detail = self.test_auth.get_user_detail(
Request.blank('/',
headers={'X-Auth-Admin-User': 'act:usr'}),
'act', 'usr')
self.assertEquals(self.test_auth.app.calls, 1)
self.assertEquals(detail, None)
def test_get_user_detail_fail_exception(self):
self.test_auth.app = FakeApp(iter([
('503 Service Unavailable', {}, '')]))
exc = None
try:
detail = self.test_auth.get_user_detail(
Request.blank('/',
headers={'X-Auth-Admin-User': 'act:usr'}),
'act', 'usr')
except Exception as err:
exc = err
self.assertEquals(str(exc), 'Could not get user object: '
'/v1/AUTH_gsmetadata/act/usr 503 Service Unavailable')
self.assertEquals(self.test_auth.app.calls, 1)
def test_is_user_reseller_admin_success(self):
self.test_auth.app = FakeApp(iter([
('200 Ok', {},
json.dumps({"auth": "plaintext:key",
"groups": [{'name': "act:usr"}, {'name': "act"},
{'name': ".reseller_admin"}]}))]))
result = self.test_auth.is_user_reseller_admin(
Request.blank('/',
headers={'X-Auth-Admin-User': 'act:usr'}),
'act', 'usr')
self.assertEquals(self.test_auth.app.calls, 1)
self.assertTrue(result)
def test_is_user_reseller_admin_fail(self):
self.test_auth.app = FakeApp(iter([
('200 Ok', {},
json.dumps({"auth": "plaintext:key",
"groups": [{'name': "act:usr"}, {'name': "act"},
{'name': ".admin"}]}))]))
result = self.test_auth.is_user_reseller_admin(
Request.blank('/',
headers={'X-Auth-Admin-User': 'act:usr'}),
'act', 'usr')
self.assertEquals(self.test_auth.app.calls, 1)
self.assertFalse(result)
def test_is_user_reseller_admin_fail_user_doesnt_exist(self):
self.test_auth.app = FakeApp(
iter([('404 Not Found', {}, '')]))
req = Request.blank('/', headers={'X-Auth-Admin-User': 'act:usr'})
result = self.test_auth.is_user_reseller_admin(req, 'act', 'usr')
self.assertEquals(self.test_auth.app.calls, 1)
self.assertFalse(result)
self.assertFalse(req.credentials_valid)
def test_credentials_match_success(self):
self.assert_(self.test_auth.credentials_match(
{'auth': 'plaintext:key'}, 'key'))