a8c84cb2da
- 401(Unauthorized) is to be returned when user credentials are wrong where as 403(Forbidden) is to be returned when user credentials are correct but the user doesn't have the priveleges to carry out the operation. - Also error messages displayed when using swauth-* command line utilities have been updated. Change-Id: I485786896ad14d3263f4325d1857cacc93adab96 Signed-off-by: Prashanth Pai <ppai@redhat.com> Reviewed-on: http://review.gluster.org/6336 Reviewed-by: Luis Pabon <lpabon@redhat.com> Tested-by: Luis Pabon <lpabon@redhat.com> Signed-off-by: Prashanth Pai <ppai@redhat.com> Reviewed-on: http://review.gluster.org/6447 Reviewed-by: Thiago Da Silva <thiago@redhat.com> Tested-by: Thiago Da Silva <thiago@redhat.com>
252 lines
10 KiB
Python
252 lines
10 KiB
Python
#!/usr/bin/python
|
|
|
|
# Copyright (c) 2010-2012 OpenStack Foundation
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License");
|
|
# you may not use this file except in compliance with the License.
|
|
# You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
|
|
# implied.
|
|
# See the License for the specific language governing permissions and
|
|
# limitations under the License.
|
|
|
|
try:
|
|
import simplejson as json
|
|
except ImportError:
|
|
import json
|
|
import unittest
|
|
from nose import SkipTest
|
|
from swift.common.bufferedhttp import http_connect_raw as http_connect
|
|
from test import get_config
|
|
|
|
config = get_config('func_test')
|
|
|
|
class TestGSWauth(unittest.TestCase):
|
|
|
|
def _get_admin_headers(self):
|
|
return {'X-Auth-Admin-User': config['admin_user'],
|
|
'X-Auth-Admin-Key': config['admin_key']}
|
|
|
|
def _check_test_account_is_not_registered(self):
|
|
# check account exists
|
|
path = '%sv2/%s' % (config['auth_prefix'], config['account'])
|
|
|
|
headers = self._get_admin_headers()
|
|
headers.update({'Content-Length': '0'})
|
|
conn = http_connect(config['auth_host'], config['auth_port'], 'GET',
|
|
path, headers)
|
|
resp = conn.getresponse()
|
|
self.assertTrue(resp.status == 404)
|
|
|
|
def _register_test_account(self):
|
|
# create account in swauth (not a swift account)
|
|
# This current version only supports one account per volume
|
|
# and the account name is the same as the volume name
|
|
# still an account must be created with swauth to map
|
|
# swauth accounts with swift accounts
|
|
path = '%sv2/%s' % (config['auth_prefix'], config['account'])
|
|
headers = self._get_admin_headers()
|
|
headers.update({'Content-Length': '0'})
|
|
conn = http_connect(config['auth_host'], config['auth_port'], 'PUT',
|
|
path, headers)
|
|
resp = conn.getresponse()
|
|
self.assertTrue(resp.status == 201)
|
|
|
|
def _deregister_test_account(self):
|
|
# delete account in swauth (not a swift account)
|
|
# @see _register_test_account
|
|
path = '%sv2/%s' % (config['auth_prefix'], config['account'])
|
|
headers = self._get_admin_headers()
|
|
headers.update({'Content-Length': '0'})
|
|
conn = http_connect(config['auth_host'], config['auth_port'],
|
|
'DELETE', path, headers)
|
|
resp = conn.getresponse()
|
|
self.assertTrue(resp.status == 204)
|
|
|
|
def test_register_account(self):
|
|
# check and register account
|
|
self._check_test_account_is_not_registered()
|
|
self._register_test_account()
|
|
|
|
try:
|
|
# list account
|
|
path = '%sv2/%s' % (config['auth_prefix'], config['account'])
|
|
headers = self._get_admin_headers()
|
|
conn = http_connect(config['auth_host'], config['auth_port'],
|
|
'GET', path, headers)
|
|
resp = conn.getresponse()
|
|
body = resp.read()
|
|
info = json.loads(body)
|
|
self.assertEqual(info['account_id'], 'AUTH_test')
|
|
self.assertTrue(resp.status == 200)
|
|
|
|
finally:
|
|
# de-register account
|
|
self._deregister_test_account()
|
|
|
|
def test_add_user(self):
|
|
# check and register account
|
|
self._check_test_account_is_not_registered()
|
|
self._register_test_account()
|
|
|
|
# create user
|
|
path = '%sv2/%s/%s' % (config['auth_prefix'], config['account'],
|
|
config['username'])
|
|
headers = self._get_admin_headers()
|
|
headers.update({'X-Auth-User-Key': config['password'],
|
|
'Content-Length': '0',
|
|
'X-Auth-User-Admin': 'true'})
|
|
conn = http_connect(config['auth_host'], config['auth_port'], 'PUT',
|
|
path, headers)
|
|
resp = conn.getresponse()
|
|
self.assertTrue(resp.status == 201)
|
|
|
|
try:
|
|
# list user
|
|
headers = self._get_admin_headers()
|
|
conn = http_connect(config['auth_host'], config['auth_port'],
|
|
'GET', path, headers)
|
|
resp = conn.getresponse()
|
|
body = resp.read()
|
|
self.assertEqual(body, '{"groups": [{"name": "test:tester"}, {"name":'
|
|
' "test"}, {"name": ".admin"}], "auth": "plaintext:testing"}')
|
|
self.assertTrue(resp.status == 200)
|
|
|
|
finally:
|
|
try:
|
|
# delete user
|
|
headers = self._get_admin_headers()
|
|
conn = http_connect(config['auth_host'], config['auth_port'],
|
|
'DELETE', path, headers)
|
|
resp = conn.getresponse()
|
|
self.assertTrue(resp.status == 204)
|
|
|
|
finally:
|
|
# de-register account
|
|
self._deregister_test_account()
|
|
|
|
def test_register_invalid_account(self):
|
|
# invalid account
|
|
path = '%sv2/%s' % (config['auth_prefix'], '.test')
|
|
headers = self._get_admin_headers()
|
|
headers.update({'Content-Length': '0'})
|
|
conn = http_connect(config['auth_host'], config['auth_port'], 'PUT',
|
|
path, headers)
|
|
resp = conn.getresponse()
|
|
self.assertTrue(resp.status == 400)
|
|
|
|
def test_add_invalid_user(self):
|
|
path = '%sv2/%s/%s' % (config['auth_prefix'], config['account'],
|
|
'.invaliduser')
|
|
headers = self._get_admin_headers()
|
|
headers.update({'X-Auth-User-Key': config['password'],
|
|
'Content-Length': '0',
|
|
'X-Auth-User-Admin': 'true'})
|
|
conn = http_connect(config['auth_host'], config['auth_port'], 'PUT',
|
|
path, headers)
|
|
resp = conn.getresponse()
|
|
self.assertTrue(resp.status == 400)
|
|
|
|
def test_register_account_without_admin_rights(self):
|
|
path = '%sv2/%s' % (config['auth_prefix'], config['account'])
|
|
headers = {'X-Auth-Admin-User': config['admin_user']}
|
|
headers.update({'Content-Length': '0'})
|
|
conn = http_connect(config['auth_host'], config['auth_port'], 'PUT',
|
|
path, headers)
|
|
resp = conn.getresponse()
|
|
self.assertTrue(resp.status == 401)
|
|
|
|
def test_change_user_password(self):
|
|
# check and register account
|
|
self._check_test_account_is_not_registered()
|
|
self._register_test_account()
|
|
|
|
try:
|
|
# create user
|
|
path = '%sv2/%s/%s' % (config['auth_prefix'], config['account'],
|
|
config['username'])
|
|
headers = self._get_admin_headers()
|
|
headers.update({'X-Auth-User-Key': config['password'],
|
|
'Content-Length': '0',
|
|
'X-Auth-User-Admin': 'true'})
|
|
conn = http_connect(config['auth_host'], config['auth_port'], 'PUT',
|
|
path, headers)
|
|
resp = conn.getresponse()
|
|
print "resp creating user %s" % resp.status
|
|
self.assertTrue(resp.status == 201)
|
|
|
|
# change password
|
|
path = '%sv2/%s/%s' % (config['auth_prefix'], config['account'],
|
|
config['username'])
|
|
headers = self._get_admin_headers()
|
|
headers.update({'X-Auth-User-Key': 'newpassword',
|
|
'Content-Length': '0',
|
|
'X-Auth-User-Admin': 'true'})
|
|
conn = http_connect(config['auth_host'], config['auth_port'], 'PUT',
|
|
path, headers)
|
|
resp = conn.getresponse()
|
|
print "resp changing password %s" % resp.status
|
|
self.assertTrue(resp.status == 201)
|
|
finally:
|
|
try:
|
|
# delete user
|
|
headers = self._get_admin_headers()
|
|
conn = http_connect(config['auth_host'], config['auth_port'],
|
|
'DELETE', path, headers)
|
|
resp = conn.getresponse()
|
|
self.assertTrue(resp.status == 204)
|
|
|
|
finally:
|
|
# de-register account
|
|
self._deregister_test_account()
|
|
|
|
def test_change_user_password_without_admin_rights(self):
|
|
# check and register account
|
|
self._check_test_account_is_not_registered()
|
|
self._register_test_account()
|
|
|
|
try:
|
|
# create user
|
|
path = '%sv2/%s/%s' % (config['auth_prefix'], config['account'],
|
|
config['username'])
|
|
headers = self._get_admin_headers()
|
|
headers.update({'X-Auth-User-Key': config['password'],
|
|
'Content-Length': '0',
|
|
'X-Auth-User-Admin': 'true'})
|
|
conn = http_connect(config['auth_host'], config['auth_port'], 'PUT',
|
|
path, headers)
|
|
resp = conn.getresponse()
|
|
print "resp creating user %s" % resp.status
|
|
self.assertTrue(resp.status == 201)
|
|
|
|
# attempt to change password
|
|
path = '%sv2/%s/%s' % (config['auth_prefix'], config['account'],
|
|
config['username'])
|
|
headers = self._get_admin_headers()
|
|
headers.update({'X-Auth-User-Key': 'newpassword',
|
|
'Content-Length': '0',
|
|
'X-Auth-Admin-Key': config['password'],
|
|
'X-Auth-User-Admin': 'true'})
|
|
conn = http_connect(config['auth_host'], config['auth_port'], 'PUT',
|
|
path, headers)
|
|
resp = conn.getresponse()
|
|
self.assertTrue(resp.status == 401)
|
|
|
|
finally:
|
|
try:
|
|
# delete user
|
|
headers = self._get_admin_headers()
|
|
conn = http_connect(config['auth_host'], config['auth_port'],
|
|
'DELETE', path, headers)
|
|
resp = conn.getresponse()
|
|
self.assertTrue(resp.status == 204)
|
|
|
|
finally:
|
|
# de-register account
|
|
self._deregister_test_account()
|