create_token() - Code Metrics - Inspection of "[WIP] Extend auth to secure chatops" - StackStorm/st2 - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#2920)

by Anthony

created 2016-09-23 07:13 UTC

create_token() C

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	7
c	1
b	0
f	0
dl	0
loc	51
rs	5.7838

How to fix Long Method

# 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 uuid
import datetime

from oslo_config import cfg

from st2common.util import isotime
from st2common.util import date as date_utils
from st2common.exceptions.auth import TokenNotFoundError, UserNotFoundError
from st2common.exceptions.auth import TTLTooLargeException
from st2common.models.db.auth import TokenDB, UserDB
from st2common.persistence.auth import Token, User
from st2common import log as logging

__all__ = [
    'create_token',
    'delete_token'
]

LOG = logging.getLogger(__name__)


def create_token(username, ttl=None, metadata=None, add_missing_user=True):
    """
    :param username: Username of the user to create the token for. If the account for this user
                     doesn't exist yet it will be created.
    :type username: ``str``

    :param ttl: Token TTL (in seconds).
    :type ttl: ``int``

    :param metadata: Optional metadata to associate with the token.
    :type metadata: ``dict``

    :param add_missing_user: Add the user given by `username` if they don't exist
    :type  add_missing_user: ``bool``
    """

    if ttl:
        if ttl > cfg.CONF.auth.token_ttl:
            msg = 'TTL specified %s is greater than max allowed %s.' % (
                ttl, cfg.CONF.auth.token_ttl
            )
            raise TTLTooLargeException(msg)
    else:
        ttl = cfg.CONF.auth.token_ttl

    if username:
        try:
            User.get_by_name(username)
        except:
            if add_missing_user:
                user = UserDB(name=username)
                User.add_or_update(user)

                extra = {'username': username, 'user': user}
                LOG.audit('Registered new user "%s".' % (username), extra=extra)
            else:
                raise UserNotFoundError()

    token = uuid.uuid4().hex
    expiry = date_utils.get_datetime_utc_now() + datetime.timedelta(seconds=ttl)
    token = TokenDB(user=username, token=token, expiry=expiry, metadata=metadata)
    Token.add_or_update(token)

    username_string = username if username else 'an anonymous user'
    token_expire_string = isotime.format(expiry, offset=False)
    extra = {'username': username, 'token_expiration': token_expire_string}

    LOG.audit('Access granted to "%s" with the token set to expire at "%s".' %
              (username_string, token_expire_string), extra=extra)

    return token


def delete_token(token):
    try:
        token_db = Token.get(token)
        return Token.delete(token_db)
    except TokenNotFoundError:
        pass
    except Exception:
        raise


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 uuid
17			import datetime
18
19			from oslo_config import cfg
20
21			from st2common.util import isotime
22			from st2common.util import date as date_utils
23			from st2common.exceptions.auth import TokenNotFoundError, UserNotFoundError
24			from st2common.exceptions.auth import TTLTooLargeException
25			from st2common.models.db.auth import TokenDB, UserDB
26			from st2common.persistence.auth import Token, User
27			from st2common import log as logging
28
29			__all__ = [
30			'create_token',
31			'delete_token'
32			]
33
34			LOG = logging.getLogger(__name__)
35
36
37			def create_token(username, ttl=None, metadata=None, add_missing_user=True):
38			"""
39			:param username: Username of the user to create the token for. If the account for this user
40			doesn't exist yet it will be created.
41			:type username: ``str``
42
43			:param ttl: Token TTL (in seconds).
44			:type ttl: ``int``
45
46			:param metadata: Optional metadata to associate with the token.
47			:type metadata: ``dict``
48
49			:param add_missing_user: Add the user given by `username` if they don't exist
50			:type add_missing_user: ``bool``
51			"""
52
53			if ttl:
54			if ttl > cfg.CONF.auth.token_ttl:
55			msg = 'TTL specified %s is greater than max allowed %s.' % (
56			ttl, cfg.CONF.auth.token_ttl
57			)
58			raise TTLTooLargeException(msg)
59			else:
60			ttl = cfg.CONF.auth.token_ttl
61
62			if username:
63			try:
64			User.get_by_name(username)
65			except:
66			if add_missing_user:
67			user = UserDB(name=username)
68			User.add_or_update(user)
69
70			extra = {'username': username, 'user': user}
71			LOG.audit('Registered new user "%s".' % (username), extra=extra)
72			else:
73			raise UserNotFoundError()
74
75			token = uuid.uuid4().hex
76			expiry = date_utils.get_datetime_utc_now() + datetime.timedelta(seconds=ttl)
77			token = TokenDB(user=username, token=token, expiry=expiry, metadata=metadata)
78			Token.add_or_update(token)
79
80			username_string = username if username else 'an anonymous user'
81			token_expire_string = isotime.format(expiry, offset=False)
82			extra = {'username': username, 'token_expiration': token_expire_string}
83
84			LOG.audit('Access granted to "%s" with the token set to expire at "%s".' %
85			(username_string, token_expire_string), extra=extra)
86
87			return token
88
89
90			def delete_token(token):
91			try:
92			token_db = Token.get(token)
93			return Token.delete(token_db)
94			except TokenNotFoundError:
95			pass
96			except Exception:
97			raise
98

StackStorm / st2

Pull Request — master (#2920)

create_token() C

Complexity

Size

Duplication

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like