Source code for privacyidea.lib.event

# -*- coding: utf-8 -*-
#  2016-05-04 Cornelius Kölbel <>
#             Initial writup
# License:  AGPLv3
# (c) 2016. Cornelius Kölbel
# This code is free software; you can redistribute it and/or
# modify it under the terms of the GNU AFFERO GENERAL PUBLIC LICENSE
# License as published by the Free Software Foundation; either
# version 3 of the License, or any later version.
# This code is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# You should have received a copy of the GNU Affero General Public
# License along with this program.  If not, see <>.
from privacyidea.models import EventHandler, EventHandlerOption, db
from privacyidea.lib.error import ParameterError
from privacyidea.lib.audit import getAudit
import functools
import logging
log = logging.getLogger(__name__)


[docs]class event(object): """ This is the event decorator that calls the event handler in the handler module. This event decorator can be used at any API call """ def __init__(self, eventname, request, g): self.eventname = eventname if not eventname in AVAILABLE_EVENTS: AVAILABLE_EVENTS.append(eventname) self.request = request self.g = g def __call__(self, func): """ Returns a wrapper that wraps func. The wrapper will evaluate the event handling definitions and call the defined action. :param func: The function that is decorated :return: function """ @functools.wraps(func) def event_wrapper(*args, **kwds): # here we have to evaluate the event configuration from the # DB table eventhandler and based on the self.eventname etc... # TODO: do Pre-Event Handling f_result = func(*args, **kwds) # Post-Event Handling e_handles = self.g.event_config.get_handled_events(self.eventname) for e_handler_def in e_handles: log.debug("Handling event {eventname} with " "{eventDef}".format(eventname=self.eventname, eventDef=e_handler_def)) event_handler_name = e_handler_def.get("handlermodule") event_handler = get_handler_object(event_handler_name) # The "action is determined by the event configuration # In the options we can pass the mailserver configuration options = {"request": self.request, "g": self.g, "response": f_result, "handler_def": e_handler_def} if event_handler.check_condition(options=options): log.debug("Handling event {eventname} with options" "{options}".format(eventname=self.eventname, options=options)) # create a new audit object event_audit = getAudit(self.g.audit_object.config) # copy all values from the originial audit entry event_audit_data = dict(self.g.audit_object.audit_data) event_audit_data["action"] = "EVENT {trigger}>>" \ "{handler}:{action}".format( trigger=self.eventname, handler=e_handler_def.get("handlermodule"), action=e_handler_def.get("action")) event_audit_data["action_detail"] = "{0!s}".format( e_handler_def.get("options")) event_audit_data["info"] = e_handler_def.get("name") event_audit.log(event_audit_data)"action"), options=options) # In case the handler has modified the response f_result = options.get("response") # set audit object to success event_audit.log({"success": True}) event_audit.finalize_log() return f_result return event_wrapper
[docs]def get_handler_object(handlername): """ Return an event handler object based on the Name of the event handler class :param handlername: The identifier of the Handler Class :type hanldername: basestring :return: """ # TODO: beautify and make this work with several different handlers from privacyidea.lib.eventhandler.usernotification import \ UserNotificationEventHandler from privacyidea.lib.eventhandler.tokenhandler import TokenEventHandler from privacyidea.lib.eventhandler.scripthandler import ScriptEventHandler from privacyidea.lib.eventhandler.federationhandler import \ FederationEventHandler from privacyidea.lib.eventhandler.counterhandler import CounterEventHandler h_obj = None if handlername == "UserNotification": h_obj = UserNotificationEventHandler() if handlername == "Token": h_obj = TokenEventHandler() if handlername == "Script": h_obj = ScriptEventHandler() if handlername == "Federation": h_obj = FederationEventHandler() if handlername == "Counter": h_obj = CounterEventHandler() return h_obj
[docs]def enable_event(event_id, enable=True): """ Enable or disable the and event :param event_id: ID of the event :return: """ ev = EventHandler.query.filter_by(id=event_id).first() if not ev: raise ParameterError("The event with id '{0!s}' does not " "exist".format(event_id)) # Update the event = enable r = return r
[docs]def set_event(name, event, handlermodule, action, conditions=None, ordering=0, options=None, id=None, active=True): """ Set an event handling configuration. This writes an entry to the database eventhandler. :param name: The name of the event definition :param event: The name of the event to react on. Can be a single event or a comma separated list. :type event: basestring :param handlermodule: The identifier of the event handler module. This is an identifier string like "UserNotification" :type handlermodule: basestring :param action: The action to perform. This is an action defined by the handler module :type action: basestring :param conditions: A condition. Only if this condition is met, the action is performed. :type conditions: dict :param ordering: An optional ordering of the event definitions. :type ordering: integer :param options: Additional options, that are needed as parameters for the action :type options: dict :param id: The DB id of the event. If the id is given, the event is updated. Otherwise a new entry is generated. :type id: int :return: The id of the event. """ if type(event) == list: event = ",".join(event) conditions = conditions or {} if id: id = int(id) event = EventHandler(name, event, handlermodule, action, conditions=conditions, ordering=ordering, options=options, id=id, active=active) return
[docs]def delete_event(event_id): """ Delete the event configuration with this given ID. :param event_id: The database ID of the event. :type event_id: int :return: """ event_id = int(event_id) ev = EventHandler.query.filter_by(id=event_id).first() r = ev.delete() return r
[docs]class EventConfiguration(object): """ This class is supposed to contain the event handling configuration during the Request. It can be read initially (in the init method) an can be accessed later during the request. """ def __init__(self): self.eventlist = [] self._read_events() @property def events(self): return self.eventlist
[docs] def get_handled_events(self, eventname): """ Return a list of the event handling definitions for the given eventname :param eventname: :return: """ eventlist = [e for e in self.eventlist if ( eventname in e.get("event") and e.get("active"))] return eventlist
[docs] def get_event(self, eventid): """ Return the reduced list with the given eventid. This list should only have one element. :param eventid: id of the event :type eventid: int :return: list with one element """ if eventid is not None: eventid = int(eventid) eventlist = [e for e in self.eventlist if e.get("id") == eventid] return eventlist else: return self.eventlist
def _read_events(self): q = EventHandler.query.order_by(EventHandler.ordering) for e in q: self.eventlist.append(e.get())