KeyValuePairController._validate_decrypt_query_parameter() - Code Metrics - Inspection of "Merge pull request #2727 from StackStorm/feature/r..." - StackStorm/st2 - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( f534b1...a82689 )

by Plexxi

created 2017-03-23 10:00 UTC

_validate_decrypt_query_parameter() A

↳ Parent: KeyValuePairController

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	4
c	0
b	0
f	0
dl	0
loc	9
rs	9.2

# 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.

from oslo_config import cfg

import six
from mongoengine import ValidationError

from st2api.controllers.resource import ResourceController
from st2common import log as logging
from st2common.constants.keyvalue import ALL_SCOPE, FULL_SYSTEM_SCOPE
from st2common.constants.keyvalue import FULL_USER_SCOPE, USER_SCOPE, ALLOWED_SCOPES
from st2common.exceptions.db import StackStormDBObjectNotFoundError
from st2common.exceptions.keyvalue import CryptoKeyNotSetupException, InvalidScopeException
from st2common.models.api.keyvalue import KeyValuePairAPI
from st2common.models.db.auth import UserDB
from st2common.persistence.keyvalue import KeyValuePair
from st2common.services import coordination
from st2common.services.keyvalues import get_key_reference
from st2common.util.keyvalue import get_datastore_full_scope
from st2common.exceptions.rbac import AccessDeniedError
from st2common.rbac import utils as rbac_utils
from st2common.router import abort
from st2common.router import Response
from st2common.rbac.utils import assert_user_is_admin_if_user_query_param_is_provided

http_client = six.moves.http_client

LOG = logging.getLogger(__name__)

__all__ = [
    'KeyValuePairController'
]


class KeyValuePairController(ResourceController):
    """
    Implements the REST endpoint for managing the key value store.
    """

    model = KeyValuePairAPI
    access = KeyValuePair
    supported_filters = {
        'prefix': 'name__startswith',
        'scope': 'scope'
    }

    def __init__(self):
        super(KeyValuePairController, self).__init__()
        self._coordinator = coordination.get_coordinator()
        self.get_one_db_method = self._get_by_name

    def get_one(self, name, scope=FULL_SYSTEM_SCOPE, user=None, requester_user=None, decrypt=False):

        """
            List key by name.

            Handle:
                GET /keys/key1
        """
        if not scope:
            scope = FULL_SYSTEM_SCOPE

        if user:
            # Providing a user implies a user scope
            scope = FULL_USER_SCOPE

        if not requester_user:
            requester_user = UserDB(cfg.CONF.system_user.user)

        scope = get_datastore_full_scope(scope)
        self._validate_scope(scope=scope)

        is_admin = rbac_utils.user_is_admin(user_db=requester_user)

        # User needs to be either admin or requesting item for itself
        self._validate_decrypt_query_parameter(decrypt=decrypt, scope=scope, is_admin=is_admin,
                                               requester_user=requester_user)

        user = user or requester_user.name

        # Validate that the authenticated user is admin if user query param is provided
        assert_user_is_admin_if_user_query_param_is_provided(user_db=requester_user,
                                                             user=user)

        key_ref = get_key_reference(scope=scope, name=name, user=user)
        from_model_kwargs = {'mask_secrets': not decrypt}
        kvp_api = self._get_one_by_scope_and_name(
            name=key_ref,
            scope=scope,
            from_model_kwargs=from_model_kwargs
        )

        return kvp_api

    def get_all(self, prefix=None, scope=FULL_SYSTEM_SCOPE, user=None, requester_user=None,
                decrypt=False, **kwargs):
        """
            List all keys.

            Handles requests:
                GET /keys/
        """
        if not scope:
            scope = FULL_SYSTEM_SCOPE

        if user:
            # Providing a user implies a user scope
            scope = FULL_USER_SCOPE

        if not requester_user:
            requester_user = UserDB(cfg.CONF.system_user.user)

        scope = get_datastore_full_scope(scope)
        is_all_scope = (scope == ALL_SCOPE)

        is_admin = rbac_utils.user_is_admin(user_db=requester_user)
        if is_all_scope and not is_admin:
            msg = '"all" scope requires administrator access'
            raise AccessDeniedError(message=msg, user_db=requester_user)

        # User needs to be either admin or requesting items for themselves
        self._validate_decrypt_query_parameter(decrypt=decrypt, scope=scope, is_admin=is_admin,
                                               requester_user=requester_user)

        user = user or requester_user.name

        # Validate that the authenticated user is admin if user query param is provided
        assert_user_is_admin_if_user_query_param_is_provided(user_db=requester_user,
                                                             user=user)

        from_model_kwargs = {'mask_secrets': not decrypt}
        kwargs['prefix'] = prefix

        if scope and scope not in ALL_SCOPE:
            self._validate_scope(scope=scope)
            kwargs['scope'] = scope

        if scope == USER_SCOPE or scope == FULL_USER_SCOPE:
            # Make sure we only returned values scoped to current user
            if kwargs['prefix']:
                kwargs['prefix'] = get_key_reference(name=kwargs['prefix'], scope=scope,
                                                     user=user)
            else:
                kwargs['prefix'] = get_key_reference(name='', scope=scope,
                                                     user=user)

        kvp_apis = super(KeyValuePairController, self)._get_all(from_model_kwargs=from_model_kwargs,
                                                                **kwargs)
        return kvp_apis

    def put(self, kvp, name, requester_user=None, scope=FULL_SYSTEM_SCOPE):
        """
        Create a new entry or update an existing one.
        """
        if not scope:
            scope = FULL_SYSTEM_SCOPE

        if not requester_user:
            requester_user = UserDB(cfg.CONF.system_user.user)

        scope = getattr(kvp, 'scope', scope)
        scope = get_datastore_full_scope(scope)
        self._validate_scope(scope=scope)

        user = getattr(kvp, 'user', requester_user.name) or requester_user.name

        # Validate that the authenticated user is admin if user query param is provided
        assert_user_is_admin_if_user_query_param_is_provided(user_db=requester_user,
                                                             user=user)

        key_ref = get_key_reference(scope=scope, name=name, user=user)
        lock_name = self._get_lock_name_for_key(name=key_ref, scope=scope)
        LOG.debug('PUT scope: %s, name: %s', scope, name)
        # TODO: Custom permission check since the key doesn't need to exist here

        # Note: We use lock to avoid a race
        with self._coordinator.get_lock(lock_name):
            try:
                existing_kvp_api = self._get_one_by_scope_and_name(
                    scope=scope,
                    name=key_ref
                )
            except StackStormDBObjectNotFoundError:
                existing_kvp_api = None

            kvp.name = key_ref
            kvp.scope = scope

            try:
                kvp_db = KeyValuePairAPI.to_model(kvp)

                if existing_kvp_api:
                    kvp_db.id = existing_kvp_api.id

                kvp_db = KeyValuePair.add_or_update(kvp_db)
            except (ValidationError, ValueError) as e:
                LOG.exception('Validation failed for key value data=%s', kvp)
                abort(http_client.BAD_REQUEST, str(e))
                return
            except CryptoKeyNotSetupException as e:
                LOG.exception(str(e))
                abort(http_client.BAD_REQUEST, str(e))
                return
            except InvalidScopeException as e:
                LOG.exception(str(e))
                abort(http_client.BAD_REQUEST, str(e))
                return
        extra = {'kvp_db': kvp_db}
        LOG.audit('KeyValuePair updated. KeyValuePair.id=%s' % (kvp_db.id), extra=extra)

        kvp_api = KeyValuePairAPI.from_model(kvp_db)
        return kvp_api

    def delete(self, name, scope=FULL_SYSTEM_SCOPE, requester_user=None, user=None):
        """
            Delete the key value pair.

            Handles requests:
                DELETE /keys/1
        """
        if not scope:
            scope = FULL_SYSTEM_SCOPE

        if not requester_user:
            requester_user = UserDB(cfg.CONF.system_user.user)

        scope = get_datastore_full_scope(scope)
        self._validate_scope(scope=scope)

        user = user or requester_user.name

        # Validate that the authenticated user is admin if user query param is provided
        assert_user_is_admin_if_user_query_param_is_provided(user_db=requester_user,
                                                             user=user)

        key_ref = get_key_reference(scope=scope, name=name, user=user)
        lock_name = self._get_lock_name_for_key(name=key_ref, scope=scope)

        # Note: We use lock to avoid a race
        with self._coordinator.get_lock(lock_name):
            from_model_kwargs = {'mask_secrets': True}
            kvp_api = self._get_one_by_scope_and_name(
                name=key_ref,
                scope=scope,
                from_model_kwargs=from_model_kwargs
            )

            kvp_db = KeyValuePairAPI.to_model(kvp_api)

            LOG.debug('DELETE /keys/ lookup with scope=%s name=%s found object: %s',
                      scope, name, kvp_db)

            try:
                KeyValuePair.delete(kvp_db)
            except Exception as e:
                LOG.exception('Database delete encountered exception during '
                              'delete of name="%s". ', name)
                abort(http_client.INTERNAL_SERVER_ERROR, str(e))
                return

        extra = {'kvp_db': kvp_db}
        LOG.audit('KeyValuePair deleted. KeyValuePair.id=%s' % (kvp_db.id), extra=extra)

        return Response(status=http_client.NO_CONTENT)

    def _get_lock_name_for_key(self, name, scope=FULL_SYSTEM_SCOPE):
        """
        Retrieve a coordination lock name for the provided datastore item name.

        :param name: Datastore item name (PK).
        :type name: ``str``
        """
        lock_name = 'kvp-crud-%s.%s' % (scope, name)
        return lock_name

    def _validate_decrypt_query_parameter(self, decrypt, scope, is_admin, requester_user):
        """
        Validate that the provider user is either admin or requesting to decrypt value for
        themselves.
        """
        is_user_scope = (scope == USER_SCOPE or scope == FULL_USER_SCOPE)
        if decrypt and (not is_user_scope and not is_admin):
            msg = 'Decrypt option requires administrator access'
            raise AccessDeniedError(message=msg, user_db=requester_user)

    def _validate_scope(self, scope):
        if scope not in ALLOWED_SCOPES:
            msg = 'Scope %s is not in allowed scopes list: %s.' % (scope, ALLOWED_SCOPES)
            raise ValueError(msg)


key_value_pair_controller = KeyValuePairController()


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			from oslo_config import cfg
17
18			import six
19			from mongoengine import ValidationError
20
21			from st2api.controllers.resource import ResourceController
22			from st2common import log as logging
23			from st2common.constants.keyvalue import ALL_SCOPE, FULL_SYSTEM_SCOPE
24			from st2common.constants.keyvalue import FULL_USER_SCOPE, USER_SCOPE, ALLOWED_SCOPES
25			from st2common.exceptions.db import StackStormDBObjectNotFoundError
26			from st2common.exceptions.keyvalue import CryptoKeyNotSetupException, InvalidScopeException
27			from st2common.models.api.keyvalue import KeyValuePairAPI
28			from st2common.models.db.auth import UserDB
29			from st2common.persistence.keyvalue import KeyValuePair
30			from st2common.services import coordination
31			from st2common.services.keyvalues import get_key_reference
32			from st2common.util.keyvalue import get_datastore_full_scope
33			from st2common.exceptions.rbac import AccessDeniedError
34			from st2common.rbac import utils as rbac_utils
35			from st2common.router import abort
36			from st2common.router import Response
37			from st2common.rbac.utils import assert_user_is_admin_if_user_query_param_is_provided
38
39			http_client = six.moves.http_client
40
41			LOG = logging.getLogger(__name__)
42
43			__all__ = [
44			'KeyValuePairController'
45			]
46
47
48			class KeyValuePairController(ResourceController):
49			"""
50			Implements the REST endpoint for managing the key value store.
51			"""
52
53			model = KeyValuePairAPI
54			access = KeyValuePair
55			supported_filters = {
56			'prefix': 'name__startswith',
57			'scope': 'scope'
58			}
59
60			def __init__(self):
61			super(KeyValuePairController, self).__init__()
62			self._coordinator = coordination.get_coordinator()
63			self.get_one_db_method = self._get_by_name
64
65			def get_one(self, name, scope=FULL_SYSTEM_SCOPE, user=None, requester_user=None, decrypt=False):
			0 ignored issues – show Bug introduced 2016-05-05 19:08 UTC by Report Bug Copy Issue Report Arguments number differs from overridden 'get_one' method Loading history...
66			"""
67			List key by name.
68
69			Handle:
70			GET /keys/key1
71			"""
72			if not scope:
73			scope = FULL_SYSTEM_SCOPE
74
75			if user:
76			# Providing a user implies a user scope
77			scope = FULL_USER_SCOPE
78
79			if not requester_user:
80			requester_user = UserDB(cfg.CONF.system_user.user)
81
82			scope = get_datastore_full_scope(scope)
83			self._validate_scope(scope=scope)
84
85			is_admin = rbac_utils.user_is_admin(user_db=requester_user)
86
87			# User needs to be either admin or requesting item for itself
88			self._validate_decrypt_query_parameter(decrypt=decrypt, scope=scope, is_admin=is_admin,
89			requester_user=requester_user)
90
91			user = user or requester_user.name
92
93			# Validate that the authenticated user is admin if user query param is provided
94			assert_user_is_admin_if_user_query_param_is_provided(user_db=requester_user,
95			user=user)
96
97			key_ref = get_key_reference(scope=scope, name=name, user=user)
98			from_model_kwargs = {'mask_secrets': not decrypt}
99			kvp_api = self._get_one_by_scope_and_name(
100			name=key_ref,
101			scope=scope,
102			from_model_kwargs=from_model_kwargs
103			)
104
105			return kvp_api
106
107			def get_all(self, prefix=None, scope=FULL_SYSTEM_SCOPE, user=None, requester_user=None,
108			decrypt=False, **kwargs):
109			"""
110			List all keys.
111
112			Handles requests:
113			GET /keys/
114			"""
115			if not scope:
116			scope = FULL_SYSTEM_SCOPE
117
118			if user:
119			# Providing a user implies a user scope
120			scope = FULL_USER_SCOPE
121
122			if not requester_user:
123			requester_user = UserDB(cfg.CONF.system_user.user)
124
125			scope = get_datastore_full_scope(scope)
126			is_all_scope = (scope == ALL_SCOPE)
127
128			is_admin = rbac_utils.user_is_admin(user_db=requester_user)
129			if is_all_scope and not is_admin:
130			msg = '"all" scope requires administrator access'
131			raise AccessDeniedError(message=msg, user_db=requester_user)
132
133			# User needs to be either admin or requesting items for themselves
134			self._validate_decrypt_query_parameter(decrypt=decrypt, scope=scope, is_admin=is_admin,
135			requester_user=requester_user)
136
137			user = user or requester_user.name
138
139			# Validate that the authenticated user is admin if user query param is provided
140			assert_user_is_admin_if_user_query_param_is_provided(user_db=requester_user,
141			user=user)
142
143			from_model_kwargs = {'mask_secrets': not decrypt}
144			kwargs['prefix'] = prefix
145
146			if scope and scope not in ALL_SCOPE:
147			self._validate_scope(scope=scope)
148			kwargs['scope'] = scope
149
150			if scope == USER_SCOPE or scope == FULL_USER_SCOPE:
151			# Make sure we only returned values scoped to current user
152			if kwargs['prefix']:
153			kwargs['prefix'] = get_key_reference(name=kwargs['prefix'], scope=scope,
154			user=user)
155			else:
156			kwargs['prefix'] = get_key_reference(name='', scope=scope,
157			user=user)
158
159			kvp_apis = super(KeyValuePairController, self)._get_all(from_model_kwargs=from_model_kwargs,
160			**kwargs)
161			return kvp_apis
162
163			def put(self, kvp, name, requester_user=None, scope=FULL_SYSTEM_SCOPE):
164			"""
165			Create a new entry or update an existing one.
166			"""
167			if not scope:
168			scope = FULL_SYSTEM_SCOPE
169
170			if not requester_user:
171			requester_user = UserDB(cfg.CONF.system_user.user)
172
173			scope = getattr(kvp, 'scope', scope)
174			scope = get_datastore_full_scope(scope)
175			self._validate_scope(scope=scope)
176
177			user = getattr(kvp, 'user', requester_user.name) or requester_user.name
178
179			# Validate that the authenticated user is admin if user query param is provided
180			assert_user_is_admin_if_user_query_param_is_provided(user_db=requester_user,
181			user=user)
182
183			key_ref = get_key_reference(scope=scope, name=name, user=user)
184			lock_name = self._get_lock_name_for_key(name=key_ref, scope=scope)
185			LOG.debug('PUT scope: %s, name: %s', scope, name)
186			# TODO: Custom permission check since the key doesn't need to exist here
187
188			# Note: We use lock to avoid a race
189			with self._coordinator.get_lock(lock_name):
190			try:
191			existing_kvp_api = self._get_one_by_scope_and_name(
192			scope=scope,
193			name=key_ref
194			)
195			except StackStormDBObjectNotFoundError:
196			existing_kvp_api = None
197
198			kvp.name = key_ref
199			kvp.scope = scope
200
201			try:
202			kvp_db = KeyValuePairAPI.to_model(kvp)
203
204			if existing_kvp_api:
205			kvp_db.id = existing_kvp_api.id
206
207			kvp_db = KeyValuePair.add_or_update(kvp_db)
208			except (ValidationError, ValueError) as e:
209			LOG.exception('Validation failed for key value data=%s', kvp)
210			abort(http_client.BAD_REQUEST, str(e))
211			return
212			except CryptoKeyNotSetupException as e:
213			LOG.exception(str(e))
214			abort(http_client.BAD_REQUEST, str(e))
215			return
216			except InvalidScopeException as e:
217			LOG.exception(str(e))
218			abort(http_client.BAD_REQUEST, str(e))
219			return
220			extra = {'kvp_db': kvp_db}
221			LOG.audit('KeyValuePair updated. KeyValuePair.id=%s' % (kvp_db.id), extra=extra)
222
223			kvp_api = KeyValuePairAPI.from_model(kvp_db)
224			return kvp_api
225
226			def delete(self, name, scope=FULL_SYSTEM_SCOPE, requester_user=None, user=None):
227			"""
228			Delete the key value pair.
229
230			Handles requests:
231			DELETE /keys/1
232			"""
233			if not scope:
234			scope = FULL_SYSTEM_SCOPE
235
236			if not requester_user:
237			requester_user = UserDB(cfg.CONF.system_user.user)
238
239			scope = get_datastore_full_scope(scope)
240			self._validate_scope(scope=scope)
241
242			user = user or requester_user.name
243
244			# Validate that the authenticated user is admin if user query param is provided
245			assert_user_is_admin_if_user_query_param_is_provided(user_db=requester_user,
246			user=user)
247
248			key_ref = get_key_reference(scope=scope, name=name, user=user)
249			lock_name = self._get_lock_name_for_key(name=key_ref, scope=scope)
250
251			# Note: We use lock to avoid a race
252			with self._coordinator.get_lock(lock_name):
253			from_model_kwargs = {'mask_secrets': True}
254			kvp_api = self._get_one_by_scope_and_name(
255			name=key_ref,
256			scope=scope,
257			from_model_kwargs=from_model_kwargs
258			)
259
260			kvp_db = KeyValuePairAPI.to_model(kvp_api)
261
262			LOG.debug('DELETE /keys/ lookup with scope=%s name=%s found object: %s',
263			scope, name, kvp_db)
264
265			try:
266			KeyValuePair.delete(kvp_db)
267			except Exception as e:
268			LOG.exception('Database delete encountered exception during '
269			'delete of name="%s". ', name)
270			abort(http_client.INTERNAL_SERVER_ERROR, str(e))
271			return
272
273			extra = {'kvp_db': kvp_db}
274			LOG.audit('KeyValuePair deleted. KeyValuePair.id=%s' % (kvp_db.id), extra=extra)
275
276			return Response(status=http_client.NO_CONTENT)
277
278			def _get_lock_name_for_key(self, name, scope=FULL_SYSTEM_SCOPE):
279			"""
280			Retrieve a coordination lock name for the provided datastore item name.
281
282			:param name: Datastore item name (PK).
283			:type name: ``str``
284			"""
285			lock_name = 'kvp-crud-%s.%s' % (scope, name)
286			return lock_name
287
288			def _validate_decrypt_query_parameter(self, decrypt, scope, is_admin, requester_user):
289			"""
290			Validate that the provider user is either admin or requesting to decrypt value for
291			themselves.
292			"""
293			is_user_scope = (scope == USER_SCOPE or scope == FULL_USER_SCOPE)
294			if decrypt and (not is_user_scope and not is_admin):
295			msg = 'Decrypt option requires administrator access'
296			raise AccessDeniedError(message=msg, user_db=requester_user)
297
298			def _validate_scope(self, scope):
299			if scope not in ALLOWED_SCOPES:
300			msg = 'Scope %s is not in allowed scopes list: %s.' % (scope, ALLOWED_SCOPES)
301			raise ValueError(msg)
302
303
304			key_value_pair_controller = KeyValuePairController()
305

StackStorm / st2

Push — master ( f534b1...a82689 )

_validate_decrypt_query_parameter() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like