openstack/deb-keystone
Code Issues Proposed changes

Merge "remove swift dependency of s3 middleware"

Browse Source
This commit is contained in:
Jenkins
2013-08-06 01:34:12 +00:00
committed by Gerrit Code Review
parent 8f5b048ed7 55938e8b60
commit db9535cfb4
3 changed files with 91 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

56
keystone/middleware/s3_token.py
View File

@@ -34,14 +34,62 @@ This WSGI component:
"""
import httplib
import urllib
import webob
from swift.common import utils as swift_utils
from keystone.common import logging
from keystone.openstack.common import jsonutils
PROTOCOL_NAME = 'S3 Token Authentication'
LOG = logging.getLogger(__name__)
# TODO(kun): remove it after oslo merge this.
def split_path(path, minsegs=1, maxsegs=None, rest_with_last=False):
"""Validate and split the given HTTP request path.
**Examples**::
['a'] = split_path('/a')
['a', None] = split_path('/a', 1, 2)
['a', 'c'] = split_path('/a/c', 1, 2)
['a', 'c', 'o/r'] = split_path('/a/c/o/r', 1, 3, True)
:param path: HTTP Request path to be split
:param minsegs: Minimum number of segments to be extracted
:param maxsegs: Maximum number of segments to be extracted
:param rest_with_last: If True, trailing data will be returned as part
of last segment. If False, and there is
trailing data, raises ValueError.
:returns: list of segments with a length of maxsegs (non-existant
segments will return as None)
:raises: ValueError if given an invalid path
"""
if not maxsegs:
maxsegs = minsegs
if minsegs > maxsegs:
raise ValueError('minsegs > maxsegs: %d > %d' % (minsegs, maxsegs))
if rest_with_last:
segs = path.split('/', maxsegs)
minsegs += 1
maxsegs += 1
count = len(segs)
if (segs[0] or count < minsegs or count > maxsegs or
'' in segs[1:minsegs]):
raise ValueError('Invalid path: %s' % urllib.quote(path))
else:
minsegs += 1
maxsegs += 1
segs = path.split('/', maxsegs)
count = len(segs)
if (segs[0] or count < minsegs or count > maxsegs + 1 or
'' in segs[1:minsegs] or
(count == maxsegs + 1 and segs[maxsegs])):
raise ValueError('Invalid path: %s' % urllib.quote(path))
segs = segs[1:maxsegs]
segs.extend([None] * (maxsegs - 1 - len(segs)))
return segs
class ServiceError(Exception):
@@ -54,7 +102,7 @@ class S3Token(object):
def __init__(self, app, conf):
"""Common initialization code."""
self.app = app
self.logger = swift_utils.get_logger(conf, log_route='s3token')
self.logger = LOG
self.logger.debug('Starting the %s component' % PROTOCOL_NAME)
self.reseller_prefix = conf.get('reseller_prefix', 'AUTH_')
# where to find the auth service (we use this to validate tokens)
@@ -119,7 +167,7 @@ class S3Token(object):
self.logger.debug('Calling S3Token middleware.')
try:
parts = swift_utils.split_path(req.path, 1, 4, True)
parts = split_path(req.path, 1, 4, True)
version, account, container, obj = parts
except ValueError:
msg = 'Not a path query, skipping.'

2
test-requirements.txt
View File

@@ -38,8 +38,6 @@ httplib2
requests>=1.0.0
keyring
# swift_auth test dependencies
http://tarballs.openstack.org/swift/swift-master.tar.gz#egg=swift
netifaces
# For translations processing

58
tests/test_s3_token_middleware.py
View File

@@ -14,32 +14,13 @@
# License for the specific language governing permissions and limitations
# under the License.
import logging
import stubout
import unittest2 as unittest
import webob
from swift.common import utils as swift_utils
from keystone.middleware import s3_token
from keystone.openstack.common import jsonutils
def setUpModule(self):
self.stubs = stubout.StubOutForTesting()
# Stub out swift_utils.get_logger. get_logger tries to configure
# syslogging to '/dev/log', which will fail on OS X.
def fake_get_logger(config, log_route=None):
return logging.getLogger(log_route)
self.stubs.Set(swift_utils, 'get_logger', fake_get_logger)
def tearDownModule(self):
self.stubs.UnsetAll()
class FakeHTTPResponse(object):
def __init__(self, status, body):
self.status = status
@@ -211,3 +192,42 @@ class S3TokenMiddlewareTestBad(S3TokenMiddlewareTestBase):
s3_invalid_req = self.middleware.deny_request('InvalidURI')
self.assertEqual(resp.body, s3_invalid_req.body)
self.assertEqual(resp.status_int, s3_invalid_req.status_int)
class S3TokenMiddlewareTestUtil(unittest.TestCase):
def test_split_path_failed(self):
self.assertRaises(ValueError, s3_token.split_path, '')
self.assertRaises(ValueError, s3_token.split_path, '/')
self.assertRaises(ValueError, s3_token.split_path, '//')
self.assertRaises(ValueError, s3_token.split_path, '//a')
self.assertRaises(ValueError, s3_token.split_path, '/a/c')
self.assertRaises(ValueError, s3_token.split_path, '//c')
self.assertRaises(ValueError, s3_token.split_path, '/a/c/')
self.assertRaises(ValueError, s3_token.split_path, '/a//')
self.assertRaises(ValueError, s3_token.split_path, '/a', 2)
self.assertRaises(ValueError, s3_token.split_path, '/a', 2, 3)
self.assertRaises(ValueError, s3_token.split_path, '/a', 2, 3, True)
self.assertRaises(ValueError, s3_token.split_path, '/a/c/o/r', 3, 3)
self.assertRaises(ValueError, s3_token.split_path, '/a', 5, 4)
def test_split_path_success(self):
self.assertEquals(s3_token.split_path('/a'), ['a'])
self.assertEquals(s3_token.split_path('/a/'), ['a'])
self.assertEquals(s3_token.split_path('/a/c', 2), ['a', 'c'])
self.assertEquals(s3_token.split_path('/a/c/o', 3), ['a', 'c', 'o'])
self.assertEquals(s3_token.split_path('/a/c/o/r', 3, 3, True),
['a', 'c', 'o/r'])
self.assertEquals(s3_token.split_path('/a/c', 2, 3, True),
['a', 'c', None])
self.assertEquals(s3_token.split_path('/a/c/', 2), ['a', 'c'])
self.assertEquals(s3_token.split_path('/a/c/', 2, 3), ['a', 'c', ''])
def test_split_path_invalid_path(self):
try:
s3_token.split_path('o\nn e', 2)
except ValueError, err:
self.assertEquals(str(err), 'Invalid path: o%0An%20e')
try:
s3_token.split_path('o\nn e', 2, 3, True)
except ValueError, err:
self.assertEquals(str(err), 'Invalid path: o%0An%20e')