Source code for keystone.api.credentials

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

# This file handles all flask-restful resources for /v3/credentials

import hashlib

import flask
import http.client
from oslo_serialization import jsonutils

from keystone.common import provider_api
from keystone.common import rbac_enforcer
from keystone.common import validation
import keystone.conf
from keystone.credential import schema
from keystone import exception
from keystone.i18n import _
from keystone.server import flask as ks_flask

CONF = keystone.conf.CONF
PROVIDERS = provider_api.ProviderAPIs
ENFORCER = rbac_enforcer.RBACEnforcer


def _build_target_enforcement():
    target = {}
    try:
        target['credential'] = PROVIDERS.credential_api.get_credential(
            flask.request.view_args.get('credential_id')
        )
    except exception.NotFound:  # nosec
        # Defer existance in the event the credential doesn't exist, we'll
        # check this later anyway.
        pass

    return target


[docs]class CredentialResource(ks_flask.ResourceBase):
    collection_key = 'credentials'
    member_key = 'credential'

    @staticmethod
    def _blob_to_json(ref):
        # credentials stored via ec2tokens before the fix for #1259584
        # need json_serailzing, as that's the documented API format
        blob = ref.get('blob')
        if isinstance(blob, dict):
            ref = ref.copy()
            ref['blob'] = jsonutils.dumps(blob)
        return ref

    def _validate_blob_json(self, ref):
        try:
            blob = jsonutils.loads(ref.get('blob'))
        except (ValueError, TabError):
            raise exception.ValidationError(
                message=_('Invalid blob in credential'))
        if not blob or not isinstance(blob, dict):
            raise exception.ValidationError(attribute='blob',
                                            target='credential')
        if blob.get('access') is None:
            raise exception.ValidationError(attribute='access',
                                            target='credential')
        return blob

    def _assign_unique_id(
            self, ref, trust_id=None, app_cred_id=None, access_token_id=None):
        # Generates an assigns a unique identifier to a credential reference.
        if ref.get('type', '').lower() == 'ec2':
            blob = self._validate_blob_json(ref)
            ref = ref.copy()
            ref['id'] = hashlib.sha256(
                blob['access'].encode('utf8')).hexdigest()
            # update the blob with the trust_id or app_cred_id, so credentials
            # created with a trust- or app cred-scoped token will result in
            # trust- or app cred-scoped tokens when authentication via
            # ec2tokens happens
            if trust_id is not None:
                blob['trust_id'] = trust_id
                ref['blob'] = jsonutils.dumps(blob)
            if app_cred_id is not None:
                blob['app_cred_id'] = app_cred_id
                ref['blob'] = jsonutils.dumps(blob)
            if access_token_id is not None:
                blob['access_token_id'] = access_token_id
                ref['blob'] = jsonutils.dumps(blob)
            return ref
        else:
            return super(CredentialResource, self)._assign_unique_id(ref)

    def _list_credentials(self):
        filters = ['user_id', 'type']
        if not self.oslo_context.system_scope:
            target = {'credential': {'user_id': self.oslo_context.user_id}}
        else:
            target = None
        ENFORCER.enforce_call(action='identity:list_credentials',
                              filters=filters, target_attr=target)
        hints = self.build_driver_hints(filters)
        refs = PROVIDERS.credential_api.list_credentials(hints)
        # If the request was filtered, make sure to return only the
        # credentials specific to that user. This makes it so that users with
        # roles on projects can't see credentials that aren't theirs.
        filtered_refs = []
        for ref in refs:
            # Check each credential again to make sure the user has access to
            # it, either by owning it, being a project admin with
            # enforce_scope=false, being a system user, or having some other
            # custom policy that allows access.
            try:
                cred = PROVIDERS.credential_api.get_credential(ref['id'])
                ENFORCER.enforce_call(
                    action='identity:get_credential',
                    target_attr={'credential': cred}
                )
                filtered_refs.append(ref)
            except exception.Forbidden:
                pass
        refs = filtered_refs
        refs = [self._blob_to_json(r) for r in refs]
        return self.wrap_collection(refs, hints=hints)

    def _get_credential(self, credential_id):
        ENFORCER.enforce_call(
            action='identity:get_credential',
            build_target=_build_target_enforcement
        )
        credential = PROVIDERS.credential_api.get_credential(credential_id)
        return self.wrap_member(self._blob_to_json(credential))

[docs]    def get(self, credential_id=None):
        # Get Credential or List of credentials.
        if credential_id is None:
            # No Parameter passed means that we're doing a LIST action.
            return self._list_credentials()
        else:
            return self._get_credential(credential_id)

[docs]    def post(self):
        # Create a new credential
        credential = self.request_body_json.get('credential', {})
        target = {}
        target['credential'] = credential
        ENFORCER.enforce_call(
            action='identity:create_credential', target_attr=target
        )
        validation.lazy_validate(schema.credential_create, credential)
        trust_id = getattr(self.oslo_context, 'trust_id', None)
        app_cred_id = getattr(
            self.auth_context['token'], 'application_credential_id', None)
        access_token_id = getattr(
            self.auth_context['token'], 'access_token_id', None)
        ref = self._assign_unique_id(
            self._normalize_dict(credential),
            trust_id=trust_id, app_cred_id=app_cred_id,
            access_token_id=access_token_id)
        ref = PROVIDERS.credential_api.create_credential(
            ref['id'], ref, initiator=self.audit_initiator)
        return self.wrap_member(ref), http.client.CREATED

    def _validate_blob_update_keys(self, credential, ref):
        if credential.get('type', '').lower() == 'ec2':
            new_blob = self._validate_blob_json(ref)
            old_blob = credential.get('blob')
            if isinstance(old_blob, str):
                old_blob = jsonutils.loads(old_blob)
            # if there was a scope set, prevent changing it or unsetting it
            for key in ['trust_id', 'app_cred_id', 'access_token_id',
                        'access_id']:
                if old_blob.get(key) != new_blob.get(key):
                    message = _('%s can not be updated for credential') % key
                    raise exception.ValidationError(message=message)

[docs]    def patch(self, credential_id):
        # Update Credential
        ENFORCER.enforce_call(
            action='identity:update_credential',
            build_target=_build_target_enforcement
        )
        current = PROVIDERS.credential_api.get_credential(credential_id)

        credential = self.request_body_json.get('credential', {})
        validation.lazy_validate(schema.credential_update, credential)
        self._validate_blob_update_keys(current.copy(), credential.copy())
        self._require_matching_id(credential)
        # Check that the user hasn't illegally modified the owner or scope
        target = {'credential': dict(current, **credential)}
        ENFORCER.enforce_call(
            action='identity:update_credential', target_attr=target
        )
        ref = PROVIDERS.credential_api.update_credential(
            credential_id, credential)
        return self.wrap_member(ref)

[docs]    def delete(self, credential_id):
        # Delete credentials
        ENFORCER.enforce_call(
            action='identity:delete_credential',
            build_target=_build_target_enforcement
        )

        return (PROVIDERS.credential_api.delete_credential(
            credential_id, initiator=self.audit_initiator),
            http.client.NO_CONTENT)


[docs]class CredentialAPI(ks_flask.APIBase):

    _name = 'credentials'
    _import_name = __name__
    resource_mapping = []
    resources = [CredentialResource]


APIs = (CredentialAPI,)