Source code for univention.admindiary

#!/usr/bin/python3
# -*- coding: utf-8 -*-
#
# Copyright 2019-2022 Univention GmbH
#
# https://www.univention.de/
#
# All rights reserved.
#
# The source code of this program is made available
# under the terms of the GNU Affero General Public License version 3
# (GNU AGPL V3) as published by the Free Software Foundation.
#
# Binary versions of this program provided by Univention to you as
# well as other copyrighted, protected or trademarked materials like
# Logos, graphics, fonts, specific documentations and configurations,
# cryptographic keys etc. are subject to a license agreement between
# you and Univention and not subject to the GNU AGPL V3.
#
# In the case you use this program under the terms of the GNU AGPL V3,
# the program is provided in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public
# License with the Debian GNU/Linux or Univention distribution in file
# /usr/share/common-licenses/AGPL-3; if not, see
# <https://www.gnu.org/licenses/>.

import six

from socket import gethostname
from datetime import datetime
import logging
import json
from typing import Dict, List  # noqa: F401

from univention.config_registry import ConfigRegistry

ucr = ConfigRegistry()
ucr.load()

LOG_FILE = '/var/log/univention/admindiary.log'


[docs]def get_events_to_reject(): # type: () -> List[str] ucrv = 'admin/diary/reject' blocked_events = ucr.get(ucrv) if blocked_events: return blocked_events.split() return []
class _ShortNameFormatter(logging.Formatter): shorten = 'univention.admindiary' def format(self, record): # type: (logging.LogRecord) -> str record.short_name = record.name if record.short_name.startswith('%s.' % self.shorten): record.short_name = record.short_name[len(self.shorten) + 1:] return super(_ShortNameFormatter, self).format(record) def _setup_logger(): # type: () -> logging.Logger base_logger = logging.getLogger('univention.admindiary') if not _setup_logger._setup: base_logger.setLevel(logging.INFO) log_format = '%(process)6d %(short_name)-12s %(asctime)s [%(levelname)8s]: %(message)s' log_format_time = '%y-%m-%d %H:%M:%S' formatter = _ShortNameFormatter(log_format, log_format_time) try: handler = logging.FileHandler(LOG_FILE) handler.setFormatter(formatter) base_logger.addHandler(handler) except EnvironmentError: pass _setup_logger._setup = True return base_logger _setup_logger._setup = False
[docs]def get_logger(name): # type: (str) -> logging.Logger base_logger = _setup_logger() logger = base_logger.getChild(name) log_level = ucr.get('admin/diary/logging/%s' % name) if log_level: log_level = logging.getLevelName(log_level) if isinstance(log_level, int): logger.setLevel(log_level) else: logger.warn('Cannot use log level %s. Call ucr set admin/diary/logging/%s=DEBUG (for example)' % (log_level, name)) return logger
[docs]class DiaryEntry(object): def __init__(self, username, message, args, tags, context_id, event_name): # type: (str, str, Dict[str, str], List[str], str, str) -> None self.username = username self.hostname = gethostname() self.message = message self.args = args self.timestamp = datetime.now() self.tags = tags self.context_id = context_id self.event_name = event_name
[docs] def assert_types(self): # type: () -> None if not isinstance(self.username, six.string_types): raise TypeError('DiaryEntry() argument "username" has to be "string", but is: %s (%s)' % (type(self.username), self.username)) if not isinstance(self.hostname, six.string_types): raise TypeError('DiaryEntry().hostname has to be "string", but is: %s (%s)' % (type(self.hostname), self.hostname)) if not isinstance(self.args, dict) or not all(isinstance(key, six.string_types) and isinstance(value, six.string_types) for key, value in self.args.items()): raise TypeError('DiaryEntry() argument "args" has to be "dict of string/string", but is: %s (%s)' % (type(self.args), self.args)) if self.message is not None: if not isinstance(self.message, dict) or not all(isinstance(key, six.string_types) and isinstance(value, six.string_types) for key, value in self.message.items()): raise TypeError('DiaryEntry() argument "message" has to be "dict of string/string", but is: %s (%s)' % (type(self.message), self.message)) for locale, message in self.message.items(): try: message.format(**self.args) except Exception: raise TypeError('DiaryEntry() argument "message" (%s, %r) has wrong format for given args (%r).', locale, message, self.args) if not isinstance(self.timestamp, datetime): raise TypeError('DiaryEntry().timestamp has to be "datetime"') if not isinstance(self.tags, list) or not all(isinstance(tag, six.string_types) for tag in self.tags): raise TypeError('DiaryEntry() argument "tags" have to be "list of string", but is: %s (%s)' % (type(self.tags), self.tags)) if not isinstance(self.context_id, six.string_types): raise TypeError('DiaryEntry() argument "context_id" has to be "string", but is: %s (%s)' % (type(self.context_id), self.context_id)) if not isinstance(self.event_name, six.string_types): raise TypeError('DiaryEntry() argument "event" name has to be "string", but is: %s (%s)' % (type(self.event_name), self.event_name))
[docs] def to_json(self): # type: () -> str attrs = { 'username': self.username, 'hostname': self.hostname, 'message': self.message, 'args': self.args, 'timestamp': self.timestamp.strftime('%Y-%m-%d %H:%M:%S%z'), 'tags': self.tags, 'context_id': self.context_id, 'event': self.event_name, 'type': 'Entry v1', } return json.dumps(attrs)
[docs] @classmethod def from_json(cls, body): # type: (str) -> DiaryEntry json_body = json.loads(body) entry = cls(json_body['username'], json_body['message'], json_body['args'], json_body['tags'], json_body['context_id'], json_body['event']) entry.timestamp = datetime.strptime(json_body['timestamp'], '%Y-%m-%d %H:%M:%S') entry.hostname = json_body['hostname'] entry.assert_types() return entry