keystone/keystone/notifications.py

# Copyright 2013 IBM Corp.
#
#   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.

"""Notifications module for OpenStack Identity Service resources"""

import collections
import functools
import inspect
import logging
import socket

from oslo_config import cfg
from oslo_log import log
from oslo_log import versionutils
import oslo_messaging
import pycadf
from pycadf import cadftaxonomy as taxonomy
from pycadf import cadftype
from pycadf import credential
from pycadf import eventfactory
from pycadf import resource

from keystone.i18n import _, _LE


notifier_opts = [
    cfg.StrOpt('default_publisher_id',
               help='Default publisher_id for outgoing notifications'),
    cfg.StrOpt('notification_format', default='basic',
               choices=['basic', 'cadf'],
               help='Define the notification format for Identity Service '
                    'events. A "basic" notification has information about '
                    'the resource being operated on. A "cadf" notification '
                    'has the same information, as well as information about '
                    'the initiator of the event.'),
]

config_section = None
list_opts = lambda: [(config_section, notifier_opts), ]

LOG = log.getLogger(__name__)
# NOTE(gyee): actions that can be notified. One must update this list whenever
# a new action is supported.
_ACTIONS = collections.namedtuple(
    'NotificationActions',
    'created, deleted, disabled, updated, internal')
ACTIONS = _ACTIONS(created='created', deleted='deleted', disabled='disabled',
                   updated='updated', internal='internal')
"""The actions on resources."""

CADF_TYPE_MAP = {
    'group': taxonomy.SECURITY_GROUP,
    'project': taxonomy.SECURITY_PROJECT,
    'role': taxonomy.SECURITY_ROLE,
    'user': taxonomy.SECURITY_ACCOUNT_USER,
    'domain': taxonomy.SECURITY_DOMAIN,
    'region': taxonomy.SECURITY_REGION,
    'endpoint': taxonomy.SECURITY_ENDPOINT,
    'service': taxonomy.SECURITY_SERVICE,
    'policy': taxonomy.SECURITY_POLICY,
    'OS-TRUST:trust': taxonomy.SECURITY_TRUST,
    'OS-OAUTH1:access_token': taxonomy.SECURITY_CREDENTIAL,
    'OS-OAUTH1:request_token': taxonomy.SECURITY_CREDENTIAL,
    'OS-OAUTH1:consumer': taxonomy.SECURITY_ACCOUNT,
}

SAML_AUDIT_TYPE = 'http://docs.oasis-open.org/security/saml/v2.0'
# resource types that can be notified
_SUBSCRIBERS = {}
_notifier = None
SERVICE = 'identity'


CONF = cfg.CONF
CONF.register_opts(notifier_opts)

# NOTE(morganfainberg): Special case notifications that are only used
# internally for handling token persistence token deletions
INVALIDATE_USER_TOKEN_PERSISTENCE = 'invalidate_user_tokens'
INVALIDATE_USER_PROJECT_TOKEN_PERSISTENCE = 'invalidate_user_project_tokens'
INVALIDATE_USER_OAUTH_CONSUMER_TOKENS = 'invalidate_user_consumer_tokens'


class Audit(object):
    """Namespace for audit notification functions.

    This is a namespace object to contain all of the direct notification
    functions utilized for ``Manager`` methods.
    """

    @classmethod
    def _emit(cls, operation, resource_type, resource_id, initiator, public):
        """Directly send an event notification.

        :param operation: one of the values from ACTIONS
        :param resource_type: type of resource being affected
        :param resource_id: ID of the resource affected
        :param initiator: CADF representation of the user that created the
                          request
        :param public: If True (default), the event will be sent to the
                       notifier API.  If False, the event will only be sent via
                       notify_event_callbacks to in process listeners
        """
        # NOTE(stevemar): the _send_notification function is
        # overloaded, it's used to register callbacks and to actually
        # send the notification externally. Thus, we should check
        # the desired notification format in the function instead
        # of before it.
        _send_notification(
            operation,
            resource_type,
            resource_id,
            public=public)

        if CONF.notification_format == 'cadf' and public:
            outcome = taxonomy.OUTCOME_SUCCESS
            _create_cadf_payload(operation, resource_type, resource_id,
                                 outcome, initiator)

    @classmethod
    def created(cls, resource_type, resource_id, initiator=None,
                public=True):
        cls._emit(ACTIONS.created, resource_type, resource_id, initiator,
                  public)

    @classmethod
    def updated(cls, resource_type, resource_id, initiator=None,
                public=True):
        cls._emit(ACTIONS.updated, resource_type, resource_id, initiator,
                  public)

    @classmethod
    def disabled(cls, resource_type, resource_id, initiator=None,
                 public=True):
        cls._emit(ACTIONS.disabled, resource_type, resource_id, initiator,
                  public)

    @classmethod
    def deleted(cls, resource_type, resource_id, initiator=None,
                public=True):
        cls._emit(ACTIONS.deleted, resource_type, resource_id, initiator,
                  public)


class ManagerNotificationWrapper(object):
    """Send event notifications for ``Manager`` methods.

    Sends a notification if the wrapped Manager method does not raise an
    ``Exception`` (such as ``keystone.exception.NotFound``).

    :param operation:  one of the values from ACTIONS
    :param resource_type: type of resource being affected
    :param public:  If True (default), the event will be sent to the notifier
                API.  If False, the event will only be sent via
                notify_event_callbacks to in process listeners

    """
    def __init__(self, operation, resource_type, public=True,
                 resource_id_arg_index=1, result_id_arg_attr=None):
        self.operation = operation
        self.resource_type = resource_type
        self.public = public
        self.resource_id_arg_index = resource_id_arg_index
        self.result_id_arg_attr = result_id_arg_attr

    def __call__(self, f):
        def wrapper(*args, **kwargs):
            """Send a notification if the wrapped callable is successful."""
            try:
                result = f(*args, **kwargs)
            except Exception:
                raise
            else:
                if self.result_id_arg_attr is not None:
                    resource_id = result[self.result_id_arg_attr]
                else:
                    resource_id = args[self.resource_id_arg_index]

                # NOTE(stevemar): the _send_notification function is
                # overloaded, it's used to register callbacks and to actually
                # send the notification externally. Thus, we should check
                # the desired notification format in the function instead
                # of before it.
                _send_notification(
                    self.operation,
                    self.resource_type,
                    resource_id,
                    public=self.public)

                # Only emit CADF notifications for public events
                if CONF.notification_format == 'cadf' and self.public:
                    outcome = taxonomy.OUTCOME_SUCCESS
                    # NOTE(morganfainberg): The decorator form will always use
                    # a 'None' initiator, since we do not pass context around
                    # in a manner that allows the decorator to inspect context
                    # and extract the needed information.
                    initiator = None
                    _create_cadf_payload(self.operation, self.resource_type,
                                         resource_id, outcome, initiator)
            return result

        return wrapper


def created(*args, **kwargs):
    """Decorator to send notifications for ``Manager.create_*`` methods."""
    return ManagerNotificationWrapper(ACTIONS.created, *args, **kwargs)


def updated(*args, **kwargs):
    """Decorator to send notifications for ``Manager.update_*`` methods."""
    return ManagerNotificationWrapper(ACTIONS.updated, *args, **kwargs)


def disabled(*args, **kwargs):
    """Decorator to send notifications when an object is disabled."""
    return ManagerNotificationWrapper(ACTIONS.disabled, *args, **kwargs)


def deleted(*args, **kwargs):
    """Decorator to send notifications for ``Manager.delete_*`` methods."""
    return ManagerNotificationWrapper(ACTIONS.deleted, *args, **kwargs)


def internal(*args, **kwargs):
    """Decorator to send notifications for internal notifications only."""
    kwargs['public'] = False
    return ManagerNotificationWrapper(ACTIONS.internal, *args, **kwargs)


def _get_callback_info(callback):
    """Return list containing callback's module and name.

    If the callback is a bound instance method also return the class name.

    :param callback: Function to call
    :type callback: function
    :returns: List containing parent module, (optional class,) function name
    :rtype: list
    """
    module_name = getattr(callback, '__module__', None)
    func_name = callback.__name__
    if inspect.ismethod(callback):
        class_name = callback.__self__.__class__.__name__
        return [module_name, class_name, func_name]
    else:
        return [module_name, func_name]


def register_event_callback(event, resource_type, callbacks):
    """Register each callback with the event.

    :param event: Action being registered
    :type event: keystone.notifications.ACTIONS
    :param resource_type: Type of resource being operated on
    :type resource_type: str
    :param callbacks: Callback items to be registered with event
    :type callbacks: list
    :raises ValueError: If event is not a valid ACTION
    :raises TypeError: If callback is not callable
    """
    if event not in ACTIONS:
        raise ValueError(_('%(event)s is not a valid notification event, must '
                           'be one of: %(actions)s') %
                         {'event': event, 'actions': ', '.join(ACTIONS)})

    if not hasattr(callbacks, '__iter__'):
        callbacks = [callbacks]

    for callback in callbacks:
        if not callable(callback):
            msg = _('Method not callable: %s') % callback
            LOG.error(msg)
            raise TypeError(msg)
        _SUBSCRIBERS.setdefault(event, {}).setdefault(resource_type, set())
        _SUBSCRIBERS[event][resource_type].add(callback)

        if LOG.logger.getEffectiveLevel() <= logging.DEBUG:
            # Do this only if its going to appear in the logs.
            msg = 'Callback: `%(callback)s` subscribed to event `%(event)s`.'
            callback_info = _get_callback_info(callback)
            callback_str = '.'.join(i for i in callback_info if i is not None)
            event_str = '.'.join(['identity', resource_type, event])
            LOG.debug(msg, {'callback': callback_str, 'event': event_str})


def listener(cls):
    """A class decorator to declare a class to be a notification listener.

    A notification listener must specify the event(s) it is interested in by
    defining a ``event_callbacks`` attribute or property. ``event_callbacks``
    is a dictionary where the key is the type of event and the value is a
    dictionary containing a mapping of resource types to callback(s).

    :data:`.ACTIONS` contains constants for the currently
    supported events. There is currently no single place to find constants for
    the resource types.

    Example::

        @listener
        class Something(object):

            def __init__(self):
                self.event_callbacks = {
                    notifications.ACTIONS.created: {
                        'user': self._user_created_callback,
                    },
                    notifications.ACTIONS.deleted: {
                        'project': [
                            self._project_deleted_callback,
                            self._do_cleanup,
                        ]
                    },
                }

    """

    def init_wrapper(init):
        @functools.wraps(init)
        def __new_init__(self, *args, **kwargs):
            init(self, *args, **kwargs)
            _register_event_callbacks(self)
        return __new_init__

    def _register_event_callbacks(self):
        for event, resource_types in self.event_callbacks.items():
            for resource_type, callbacks in resource_types.items():
                register_event_callback(event, resource_type, callbacks)

    cls.__init__ = init_wrapper(cls.__init__)
    return cls


def notify_event_callbacks(service, resource_type, operation, payload):
    """Sends a notification to registered extensions."""
    if operation in _SUBSCRIBERS:
        if resource_type in _SUBSCRIBERS[operation]:
            for cb in _SUBSCRIBERS[operation][resource_type]:
                subst_dict = {'cb_name': cb.__name__,
                              'service': service,
                              'resource_type': resource_type,
                              'operation': operation,
                              'payload': payload}
                LOG.debug('Invoking callback %(cb_name)s for event '
                          '%(service)s %(resource_type)s %(operation)s for'
                          '%(payload)s', subst_dict)
                cb(service, resource_type, operation, payload)


def _get_notifier():
    """Return a notifier object.

    If _notifier is None it means that a notifier object has not been set.
    If _notifier is False it means that a notifier has previously failed to
    construct.
    Otherwise it is a constructed Notifier object.
    """
    global _notifier

    if _notifier is None:
        host = CONF.default_publisher_id or socket.gethostname()
        try:
            transport = oslo_messaging.get_transport(CONF)
            _notifier = oslo_messaging.Notifier(transport,
                                                "identity.%s" % host)
        except Exception:
            LOG.exception(_LE("Failed to construct notifier"))
            _notifier = False

    return _notifier


def clear_subscribers():
    """Empty subscribers dictionary.

    This effectively stops notifications since there will be no subscribers
    to publish to.
    """
    _SUBSCRIBERS.clear()


def reset_notifier():
    """Reset the notifications internal state.

    This is used only for testing purposes.

    """
    global _notifier
    _notifier = None


def _create_cadf_payload(operation, resource_type, resource_id,
                         outcome, initiator):
    """Prepare data for CADF audit notifier.

    Transform the arguments into content to be consumed by the function that
    emits CADF events (_send_audit_notification). Specifically the
    ``resource_type`` (role, user, etc) must be transformed into a CADF
    keyword, such as: ``data/security/role``. The ``resource_id`` is added as a
    top level value for the ``resource_info`` key. Lastly, the ``operation`` is
    used to create the CADF ``action``, and the ``event_type`` name.

    As per the CADF specification, the ``action`` must start with create,
    update, delete, etc... i.e.: created.user or deleted.role

    However the ``event_type`` is an OpenStack-ism that is typically of the
    form project.resource.operation. i.e.: identity.project.updated

    :param operation: operation being performed (created, updated, or deleted)
    :param resource_type: type of resource being operated on (role, user, etc)
    :param resource_id: ID of resource being operated on
    :param outcome: outcomes of the operation (SUCCESS, FAILURE, etc)
    :param initiator: CADF representation of the user that created the request
    """

    if resource_type not in CADF_TYPE_MAP:
        target_uri = taxonomy.UNKNOWN
    else:
        target_uri = CADF_TYPE_MAP.get(resource_type)
    target = resource.Resource(typeURI=target_uri,
                               id=resource_id)

    audit_kwargs = {'resource_info': resource_id}
    cadf_action = '%s.%s' % (operation, resource_type)
    event_type = '%s.%s.%s' % (SERVICE, resource_type, operation)

    _send_audit_notification(cadf_action, initiator, outcome,
                             target, event_type, **audit_kwargs)


def _send_notification(operation, resource_type, resource_id, public=True):
    """Send notification to inform observers about the affected resource.

    This method doesn't raise an exception when sending the notification fails.

    :param operation: operation being performed (created, updated, or deleted)
    :param resource_type: type of resource being operated on
    :param resource_id: ID of resource being operated on
    :param public:  if True (default), the event will be sent
                    to the notifier API.
                    if False, the event will only be sent via
                    notify_event_callbacks to in process listeners.
    """
    payload = {'resource_info': resource_id}

    notify_event_callbacks(SERVICE, resource_type, operation, payload)

    # Only send this notification if the 'basic' format is used, otherwise
    # let the CADF functions handle sending the notification. But we check
    # here so as to not disrupt the notify_event_callbacks function.
    if public and CONF.notification_format == 'basic':
        notifier = _get_notifier()
        if notifier:
            context = {}
            event_type = '%(service)s.%(resource_type)s.%(operation)s' % {
                'service': SERVICE,
                'resource_type': resource_type,
                'operation': operation}
            try:
                notifier.info(context, event_type, payload)
            except Exception:
                LOG.exception(_LE(
                    'Failed to send %(res_id)s %(event_type)s notification'),
                    {'res_id': resource_id, 'event_type': event_type})


def _get_request_audit_info(context, user_id=None):
    """Collect audit information about the request used for CADF.

    :param context: Request context
    :param user_id: Optional user ID, alternatively collected from context
    :returns: Auditing data about the request
    :rtype: :class:`pycadf.Resource`
    """

    remote_addr = None
    http_user_agent = None
    project_id = None
    domain_id = None

    if context and 'environment' in context and context['environment']:
        environment = context['environment']
        remote_addr = environment.get('REMOTE_ADDR')
        http_user_agent = environment.get('HTTP_USER_AGENT')
        if not user_id:
            user_id = environment.get('KEYSTONE_AUTH_CONTEXT',
                                      {}).get('user_id')
        project_id = environment.get('KEYSTONE_AUTH_CONTEXT',
                                     {}).get('project_id')
        domain_id = environment.get('KEYSTONE_AUTH_CONTEXT',
                                    {}).get('domain_id')

    host = pycadf.host.Host(address=remote_addr, agent=http_user_agent)
    initiator = resource.Resource(typeURI=taxonomy.ACCOUNT_USER,
                                  id=user_id, host=host)
    if project_id:
        initiator.project_id = project_id
    if domain_id:
        initiator.domain_id = domain_id

    return initiator


class CadfNotificationWrapper(object):
    """Send CADF event notifications for various methods.

    This function is only used for Authentication events. Its ``action`` and
    ``event_type`` are dictated below.

    - action: authenticate
    - event_type: identity.authenticate

    Sends CADF notifications for events such as whether an authentication was
    successful or not.

    :param operation: The authentication related action being performed

    """

    def __init__(self, operation):
        self.action = operation
        self.event_type = '%s.%s' % (SERVICE, operation)

    def __call__(self, f):
        def wrapper(wrapped_self, context, user_id, *args, **kwargs):
            """Always send a notification."""

            initiator = _get_request_audit_info(context, user_id)
            target = resource.Resource(typeURI=taxonomy.ACCOUNT_USER)
            try:
                result = f(wrapped_self, context, user_id, *args, **kwargs)
            except Exception:
                # For authentication failure send a cadf event as well
                _send_audit_notification(self.action, initiator,
                                         taxonomy.OUTCOME_FAILURE,
                                         target, self.event_type)
                raise
            else:
                _send_audit_notification(self.action, initiator,
                                         taxonomy.OUTCOME_SUCCESS,
                                         target, self.event_type)
                return result

        return wrapper


class CadfRoleAssignmentNotificationWrapper(object):
    """Send CADF notifications for ``role_assignment`` methods.

    This function is only used for role assignment events. Its ``action`` and
    ``event_type`` are dictated below.

    - action: created.role_assignment or deleted.role_assignment
    - event_type: identity.role_assignment.created or
        identity.role_assignment.deleted

    Sends a CADF notification if the wrapped method does not raise an
    ``Exception`` (such as ``keystone.exception.NotFound``).

    :param operation: one of the values from ACTIONS (create or delete)
    """

    ROLE_ASSIGNMENT = 'role_assignment'

    def __init__(self, operation):
        self.action = '%s.%s' % (operation, self.ROLE_ASSIGNMENT)
        self.deprecated_event_type = '%s.%s.%s' % (SERVICE, operation,
                                                   self.ROLE_ASSIGNMENT)
        self.event_type = '%s.%s.%s' % (SERVICE, self.ROLE_ASSIGNMENT,
                                        operation)

    def __call__(self, f):
        def wrapper(wrapped_self, role_id, *args, **kwargs):
            """Send a notification if the wrapped callable is successful."""

            """ NOTE(stevemar): The reason we go through checking kwargs
            and args for possible target and actor values is because the
            create_grant() (and delete_grant()) method are called
            differently in various tests.
            Using named arguments, i.e.:
                create_grant(user_id=user['id'], domain_id=domain['id'],
                             role_id=role['id'])

            Or, using positional arguments, i.e.:
                create_grant(role_id['id'], user['id'], None,
                             domain_id=domain['id'], None)

            Or, both, i.e.:
                create_grant(role_id['id'], user_id=user['id'],
                             domain_id=domain['id'])

            Checking the values for kwargs is easy enough, since it comes
            in as a dictionary

            The actual method signature is
                create_grant(role_id, user_id=None, group_id=None,
                             domain_id=None, project_id=None,
                             inherited_to_projects=False)

            So, if the values of actor or target are still None after
            checking kwargs, we can check the positional arguments,
            based on the method signature.
            """
            call_args = inspect.getcallargs(
                f, wrapped_self, role_id, *args, **kwargs)
            inherited = call_args['inherited_to_projects']
            context = call_args['context']

            initiator = _get_request_audit_info(context)
            target = resource.Resource(typeURI=taxonomy.ACCOUNT_USER)

            audit_kwargs = {}
            if call_args['project_id']:
                audit_kwargs['project'] = call_args['project_id']
            elif call_args['domain_id']:
                audit_kwargs['domain'] = call_args['domain_id']

            if call_args['user_id']:
                audit_kwargs['user'] = call_args['user_id']
            elif call_args['group_id']:
                audit_kwargs['group'] = call_args['group_id']

            audit_kwargs['inherited_to_projects'] = inherited
            audit_kwargs['role'] = role_id

            # For backward compatibility, send both old and new event_type.
            # Deprecate old format and remove it in the next release.
            event_types = [self.deprecated_event_type, self.event_type]
            versionutils.deprecated(
                as_of=versionutils.deprecated.KILO,
                remove_in=+1,
                what=('sending duplicate %s notification event type' %
                      self.deprecated_event_type),
                in_favor_of='%s notification event type' % self.event_type)
            try:
                result = f(wrapped_self, role_id, *args, **kwargs)
            except Exception:
                for event_type in event_types:
                    _send_audit_notification(self.action, initiator,
                                             taxonomy.OUTCOME_FAILURE,
                                             target, event_type,
                                             **audit_kwargs)
                raise
            else:
                for event_type in event_types:
                    _send_audit_notification(self.action, initiator,
                                             taxonomy.OUTCOME_SUCCESS,
                                             target, event_type,
                                             **audit_kwargs)
                return result

        return wrapper


def send_saml_audit_notification(action, context, user_id, group_ids,
                                 identity_provider, protocol, token_id,
                                 outcome):
    """Send notification to inform observers about SAML events.

    :param action: Action being audited
    :type action: str
    :param context: Current request context to collect request info from
    :type context: dict
    :param user_id: User ID from Keystone token
    :type user_id: str
    :param group_ids: List of Group IDs from Keystone token
    :type group_ids: list
    :param identity_provider: ID of the IdP from the Keystone token
    :type identity_provider: str or None
    :param protocol: Protocol ID for IdP from the Keystone token
    :type protocol: str
    :param token_id: audit_id from Keystone token
    :type token_id: str or None
    :param outcome: One of :class:`pycadf.cadftaxonomy`
    :type outcome: str
    """

    initiator = _get_request_audit_info(context)
    target = resource.Resource(typeURI=taxonomy.ACCOUNT_USER)
    audit_type = SAML_AUDIT_TYPE
    user_id = user_id or taxonomy.UNKNOWN
    token_id = token_id or taxonomy.UNKNOWN
    group_ids = group_ids or []
    cred = credential.FederatedCredential(token=token_id, type=audit_type,
                                          identity_provider=identity_provider,
                                          user=user_id, groups=group_ids)
    initiator.credential = cred
    event_type = '%s.%s' % (SERVICE, action)
    _send_audit_notification(action, initiator, outcome, target, event_type)


def _send_audit_notification(action, initiator, outcome, target,
                             event_type, **kwargs):
    """Send CADF notification to inform observers about the affected resource.

    This method logs an exception when sending the notification fails.

    :param action: CADF action being audited (e.g., 'authenticate')
    :param initiator: CADF resource representing the initiator
    :param outcome: The CADF outcome (taxonomy.OUTCOME_PENDING,
        taxonomy.OUTCOME_SUCCESS, taxonomy.OUTCOME_FAILURE)
    :param target: CADF resource representing the target
    :param event_type: An OpenStack-ism, typically this is the meter name that
        Ceilometer uses to poll events.
    :param kwargs: Any additional arguments passed in will be added as
        key-value pairs to the CADF event.

    """

    event = eventfactory.EventFactory().new_event(
        eventType=cadftype.EVENTTYPE_ACTIVITY,
        outcome=outcome,
        action=action,
        initiator=initiator,
        target=target,
        observer=resource.Resource(typeURI=taxonomy.SERVICE_SECURITY))

    for key, value in kwargs.items():
        setattr(event, key, value)

    context = {}
    payload = event.as_dict()
    notifier = _get_notifier()

    if notifier:
        try:
            notifier.info(context, event_type, payload)
        except Exception:
            # diaper defense: any exception that occurs while emitting the
            # notification should not interfere with the API request
            LOG.exception(_LE(
                'Failed to send %(action)s %(event_type)s notification'),
                {'action': action, 'event_type': event_type})


emit_event = CadfNotificationWrapper


role_assignment = CadfRoleAssignmentNotificationWrapper