AuthHandlerBase._get_username_for_request() - Code Metrics - Inspection of "Merge pull request #2920 from tonybaloney/oauth" - StackStorm/st2 - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 1c9535...b8d4ca )

by Edward

created 2016-10-06 03:37 UTC

AuthHandlerBase._get_username_for_request() C

↳ Parent: AuthHandlerBase

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	1

Metric	Value
cc	10
c	1
b	0
f	1
dl	0
loc	54
rs	5.1428

How to fix Long Method Complexity

# Licensed to the StackStorm, Inc ('StackStorm') under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You 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.

import base64
from six.moves import http_client
from oslo_config import cfg

from st2common import log as logging
from st2common.exceptions.auth import TTLTooLargeException, UserNotFoundError
from st2common.exceptions.db import StackStormDBObjectNotFoundError
from st2common.exceptions.auth import NoNicknameOriginProvidedError, AmbiguousUserError
from st2common.exceptions.auth import NotServiceUserError
from st2common.persistence.auth import User
from st2common.util.pecan_util import abort_request
from st2common.services.access import create_token
from st2common.models.api.auth import TokenAPI
from st2auth.backends import get_backend_instance

LOG = logging.getLogger(__name__)


class AuthHandlerBase(object):
    def handle_auth(self, request, headers=None, remote_addr=None,
                    remote_user=None, authorization=None, **kwargs):
        raise NotImplementedError()

    def _create_token_for_user(self, username, ttl=None):
        tokendb = create_token(username=username, ttl=ttl)
        return TokenAPI.from_model(tokendb)

    def _get_username_for_request(self, username, request):
        impersonate_user = getattr(request, 'user', None)

        if impersonate_user is not None:
            # check this is a service account
            try:
                if not User.get_by_name(username).is_service:
                    message = "Current user is not a service and cannot " \
                              "request impersonated tokens"
                    abort_request(status_code=http_client.BAD_REQUEST,
                                  message=message)
                    return
                username = impersonate_user
            except (UserNotFoundError, StackStormDBObjectNotFoundError):
                message = "Could not locate user %s" % \
                          (impersonate_user)
                abort_request(status_code=http_client.BAD_REQUEST,
                              message=message)
                return
        else:
            impersonate_user = getattr(request, 'impersonate_user', None)
            nickname_origin = getattr(request, 'nickname_origin', None)
            if impersonate_user is not None:
                try:
                    # check this is a service account
                    if not User.get_by_name(username).is_service:
                        raise NotServiceUserError()
                    username = User.get_by_nickname(impersonate_user,
                                                    nickname_origin).name
                except NotServiceUserError:
                    message = "Current user is not a service and cannot " \
                              "request impersonated tokens"
                    abort_request(status_code=http_client.BAD_REQUEST,
                                  message=message)
                    return
                except (UserNotFoundError, StackStormDBObjectNotFoundError):
                    message = "Could not locate user %s@%s" % \
                              (impersonate_user, nickname_origin)
                    abort_request(status_code=http_client.BAD_REQUEST,
                                  message=message)
                    return
                except NoNicknameOriginProvidedError:
                    message = "Nickname origin is not provided for nickname '%s'" % \
                              impersonate_user
                    abort_request(status_code=http_client.BAD_REQUEST,
                                  message=message)
                    return
                except AmbiguousUserError:
                    message = "%s@%s matched more than one username" % \
                              (impersonate_user, nickname_origin)
                    abort_request(status_code=http_client.BAD_REQUEST,
                                  message=message)
                    return
        return username


class ProxyAuthHandler(AuthHandlerBase):
    def handle_auth(self, request, headers=None, remote_addr=None,
                    remote_user=None, authorization=None, **kwargs):
        remote_addr = headers.get('x-forwarded-for',
                                  remote_addr)
        extra = {'remote_addr': remote_addr}

        if remote_user:
            ttl = getattr(request, 'ttl', None)
            username = self._get_username_for_request(remote_user, request)
            try:
                token = self._create_token_for_user(username=username,
                                                    ttl=ttl)
            except TTLTooLargeException as e:
                abort_request(status_code=http_client.BAD_REQUEST,
                              message=e.message)
            return token

        LOG.audit('Access denied to anonymous user.', extra=extra)
        abort_request()


class StandaloneAuthHandler(AuthHandlerBase):
    def __init__(self, *args, **kwargs):
        self._auth_backend = get_backend_instance(name=cfg.CONF.auth.backend)
        super(StandaloneAuthHandler, self).__init__(*args, **kwargs)

    def handle_auth(self, request, headers=None, remote_addr=None, remote_user=None,
                    authorization=None, **kwargs):
        auth_backend = self._auth_backend.__class__.__name__

        extra = {'auth_backend': auth_backend, 'remote_addr': remote_addr}

        if not authorization:
            LOG.audit('Authorization header not provided', extra=extra)
            abort_request()
            return

        auth_type, auth_value = authorization

        if auth_type.lower() not in ['basic']:
            extra['auth_type'] = auth_type
            LOG.audit('Unsupported authorization type: %s' % (auth_type), extra=extra)
            abort_request()
            return

        try:
            auth_value = base64.b64decode(auth_value)
        except Exception:
            LOG.audit('Invalid authorization header', extra=extra)
            abort_request()
            return

        split = auth_value.split(':')
        if len(split) != 2:
            LOG.audit('Invalid authorization header', extra=extra)
            abort_request()
            return

        username, password = split
        result = self._auth_backend

        result = self._auth_backend.authenticate(username=username, password=password)
        if result is True:
            ttl = getattr(request, 'ttl', None)
            username = self._get_username_for_request(username, request)
            try:
                token = self._create_token_for_user(
                    username=username, ttl=ttl)
                return token
            except TTLTooLargeException as e:
                abort_request(status_code=http_client.BAD_REQUEST,
                              message=e.message)
                return

        LOG.audit('Invalid credentials provided', extra=extra)
        abort_request()


1			# Licensed to the StackStorm, Inc ('StackStorm') under one or more
2			# contributor license agreements. See the NOTICE file distributed with
3			# this work for additional information regarding copyright ownership.
4			# The ASF licenses this file to You under the Apache License, Version 2.0
5			# (the "License"); you may not use this file except in compliance with
6			# the License. You may obtain a copy of the License at
7			#
8			# http://www.apache.org/licenses/LICENSE-2.0
9			#
10			# Unless required by applicable law or agreed to in writing, software
11			# distributed under the License is distributed on an "AS IS" BASIS,
12			# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13			# See the License for the specific language governing permissions and
14			# limitations under the License.
15
16			import base64
17			from six.moves import http_client
18			from oslo_config import cfg
19
20			from st2common import log as logging
21			from st2common.exceptions.auth import TTLTooLargeException, UserNotFoundError
22			from st2common.exceptions.db import StackStormDBObjectNotFoundError
23			from st2common.exceptions.auth import NoNicknameOriginProvidedError, AmbiguousUserError
24			from st2common.exceptions.auth import NotServiceUserError
25			from st2common.persistence.auth import User
26			from st2common.util.pecan_util import abort_request
27			from st2common.services.access import create_token
28			from st2common.models.api.auth import TokenAPI
29			from st2auth.backends import get_backend_instance
30
31			LOG = logging.getLogger(__name__)
32
33
34			class AuthHandlerBase(object):
35			def handle_auth(self, request, headers=None, remote_addr=None,
36			remote_user=None, authorization=None, **kwargs):
37			raise NotImplementedError()
38
39			def _create_token_for_user(self, username, ttl=None):
40			tokendb = create_token(username=username, ttl=ttl)
41			return TokenAPI.from_model(tokendb)
42
43			def _get_username_for_request(self, username, request):
44			impersonate_user = getattr(request, 'user', None)
45
46			if impersonate_user is not None:
47			# check this is a service account
48			try:
49			if not User.get_by_name(username).is_service:
50			message = "Current user is not a service and cannot " \
51			"request impersonated tokens"
52			abort_request(status_code=http_client.BAD_REQUEST,
53			message=message)
54			return
55			username = impersonate_user
56			except (UserNotFoundError, StackStormDBObjectNotFoundError):
57			message = "Could not locate user %s" % \
58			(impersonate_user)
59			abort_request(status_code=http_client.BAD_REQUEST,
60			message=message)
61			return
62			else:
63			impersonate_user = getattr(request, 'impersonate_user', None)
64			nickname_origin = getattr(request, 'nickname_origin', None)
65			if impersonate_user is not None:
66			try:
67			# check this is a service account
68			if not User.get_by_name(username).is_service:
69			raise NotServiceUserError()
70			username = User.get_by_nickname(impersonate_user,
71			nickname_origin).name
72			except NotServiceUserError:
73			message = "Current user is not a service and cannot " \
74			"request impersonated tokens"
75			abort_request(status_code=http_client.BAD_REQUEST,
76			message=message)
77			return
78			except (UserNotFoundError, StackStormDBObjectNotFoundError):
79			message = "Could not locate user %s@%s" % \
80			(impersonate_user, nickname_origin)
81			abort_request(status_code=http_client.BAD_REQUEST,
82			message=message)
83			return
84			except NoNicknameOriginProvidedError:
85			message = "Nickname origin is not provided for nickname '%s'" % \
86			impersonate_user
87			abort_request(status_code=http_client.BAD_REQUEST,
88			message=message)
89			return
90			except AmbiguousUserError:
91			message = "%s@%s matched more than one username" % \
92			(impersonate_user, nickname_origin)
93			abort_request(status_code=http_client.BAD_REQUEST,
94			message=message)
95			return
96			return username
97
98
99			class ProxyAuthHandler(AuthHandlerBase):
100			def handle_auth(self, request, headers=None, remote_addr=None,
101			remote_user=None, authorization=None, **kwargs):
102			remote_addr = headers.get('x-forwarded-for',
103			remote_addr)
104			extra = {'remote_addr': remote_addr}
105
106			if remote_user:
107			ttl = getattr(request, 'ttl', None)
108			username = self._get_username_for_request(remote_user, request)
109			try:
110			token = self._create_token_for_user(username=username,
111			ttl=ttl)
112			except TTLTooLargeException as e:
113			abort_request(status_code=http_client.BAD_REQUEST,
114			message=e.message)
115			return token
116
117			LOG.audit('Access denied to anonymous user.', extra=extra)
118			abort_request()
119
120
121			class StandaloneAuthHandler(AuthHandlerBase):
122			def __init__(self, args, *kwargs):
123			self._auth_backend = get_backend_instance(name=cfg.CONF.auth.backend)
124			super(StandaloneAuthHandler, self).__init__(args, *kwargs)
125
126			def handle_auth(self, request, headers=None, remote_addr=None, remote_user=None,
127			authorization=None, **kwargs):
128			auth_backend = self._auth_backend.__class__.__name__
129
130			extra = {'auth_backend': auth_backend, 'remote_addr': remote_addr}
131
132			if not authorization:
133			LOG.audit('Authorization header not provided', extra=extra)
134			abort_request()
135			return
136
137			auth_type, auth_value = authorization
			0 ignored issues – show Bug Best Practice introduced 2016-09-30 04:26 UTC by Report Bug Copy Issue Report It seems like you are trying to unpack a non-sequence, which was defined at line 127. Loading history...
138			if auth_type.lower() not in ['basic']:
139			extra['auth_type'] = auth_type
140			LOG.audit('Unsupported authorization type: %s' % (auth_type), extra=extra)
141			abort_request()
142			return
143
144			try:
145			auth_value = base64.b64decode(auth_value)
146			except Exception:
147			LOG.audit('Invalid authorization header', extra=extra)
148			abort_request()
149			return
150
151			split = auth_value.split(':')
152			if len(split) != 2:
153			LOG.audit('Invalid authorization header', extra=extra)
154			abort_request()
155			return
156
157			username, password = split
158			result = self._auth_backend
159
160			result = self._auth_backend.authenticate(username=username, password=password)
161			if result is True:
162			ttl = getattr(request, 'ttl', None)
163			username = self._get_username_for_request(username, request)
164			try:
165			token = self._create_token_for_user(
166			username=username, ttl=ttl)
167			return token
168			except TTLTooLargeException as e:
169			abort_request(status_code=http_client.BAD_REQUEST,
170			message=e.message)
171			return
172
173			LOG.audit('Invalid credentials provided', extra=extra)
174			abort_request()
175

StackStorm / st2

Push — master ( 1c9535...b8d4ca )

AuthHandlerBase._get_username_for_request() C

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like