File: //lib/python3/dist-packages/firewall/core/logger.py
# SPDX-License-Identifier: GPL-2.0-or-later
#
# Copyright (C) 2005-2007,2012 Red Hat, Inc.
#
# Authors:
# Thomas Woerner <twoerner@redhat.com>
import sys
import types
import time
import inspect
import fnmatch
import syslog
import traceback
import fcntl
import os.path
import os
# ---------------------------------------------------------------------------
# abstract class for logging targets
class LogTarget:
"""Abstract class for logging targets."""
def __init__(self):
self.fd = None
def write(self, data, level, logger, is_debug=0):
raise NotImplementedError("LogTarget.write is an abstract method")
def flush(self):
raise NotImplementedError("LogTarget.flush is an abstract method")
def close(self):
raise NotImplementedError("LogTarget.close is an abstract method")
# ---------------------------------------------------------------------------
# private class for stdout
class _StdoutLog(LogTarget):
def __init__(self):
LogTarget.__init__(self)
self.fd = sys.stdout
def write(self, data, level, logger, is_debug=0):
# ignore level
self.fd.write(data)
self.flush()
def close(self):
self.flush()
def flush(self):
self.fd.flush()
# ---------------------------------------------------------------------------
# private class for stderr
class _StderrLog(_StdoutLog):
def __init__(self):
_StdoutLog.__init__(self)
self.fd = sys.stderr
# ---------------------------------------------------------------------------
# private class for syslog
class _SyslogLog(LogTarget):
def __init__(self):
# Only initialize LogTarget here as fs should be None
LogTarget.__init__(self)
#
# Derived from: https://github.com/canvon/firewalld/commit/af0edfee1cc1891b7b13f302ca5911b24e9b0f13
#
# Work around Python issue 27875, "Syslogs /usr/sbin/foo as /foo
# instead of as foo"
# (but using openlog explicitly might be better anyway)
#
# Set ident to basename, log PID as well, and log to facility "daemon".
syslog.openlog(os.path.basename(sys.argv[0]), syslog.LOG_PID, syslog.LOG_DAEMON)
def write(self, data, level, logger, is_debug=0):
priority = None
if is_debug:
priority = syslog.LOG_DEBUG
else:
if level >= logger.INFO1:
priority = syslog.LOG_INFO
elif level == logger.WARNING:
priority = syslog.LOG_WARNING
elif level == logger.ERROR:
priority = syslog.LOG_ERR
elif level == logger.FATAL:
priority = syslog.LOG_CRIT
if data.endswith("\n"):
data = data[: len(data) - 1]
if len(data) > 0:
if priority is None:
syslog.syslog(data)
else:
syslog.syslog(priority, data)
def close(self):
syslog.closelog()
def flush(self):
pass
# ---------------------------------------------------------------------------
class FileLog(LogTarget):
"""FileLog class.
File will be opened on the first write."""
def __init__(self, filename, mode="w"):
LogTarget.__init__(self)
self.filename = filename
self.mode = mode
def open(self):
if self.fd:
return
flags = os.O_CREAT | os.O_WRONLY
if self.mode.startswith("a"):
flags |= os.O_APPEND
self.fd = os.open(self.filename, flags, 0o640)
# Make sure that existing file has correct perms
os.fchmod(self.fd, 0o640)
# Make it an object
self.fd = os.fdopen(self.fd, self.mode)
fcntl.fcntl(self.fd, fcntl.F_SETFD, fcntl.FD_CLOEXEC)
def write(self, data, level, logger, is_debug=0):
if not self.fd:
self.open()
self.fd.write(data)
self.fd.flush()
def close(self):
if not self.fd:
return
self.fd.close()
self.fd = None
def flush(self):
if not self.fd:
return
self.fd.flush()
# ---------------------------------------------------------------------------
class Logger:
r"""
Format string:
%(class)s Calling class the function belongs to, else empty
%(date)s Date using Logger.date_format, see time module
%(domain)s Full Domain: %(module)s.%(class)s.%(function)s
%(file)s Filename of the module
%(function)s Function name, empty in __main__
%(label)s Label according to log function call from Logger.label
%(level)d Internal logging level
%(line)d Line number in module
%(module)s Module name
%(message)s Log message
Standard levels:
FATAL Fatal error messages
ERROR Error messages
WARNING Warning messages
INFOx, x in [1..5] Information
DEBUGy, y in [1..10] Debug messages
NO_INFO No info output
NO_DEBUG No debug output
INFO_MAX Maximum info level
DEBUG_MAX Maximum debug level
x and y depend on info_max and debug_max from Logger class
initialization. See __init__ function.
Default logging targets:
stdout Logs to stdout
stderr Logs to stderr
syslog Logs to syslog
Additional arguments for logging functions (fatal, error, warning, info
and debug):
nl Disable newline at the end with nl=0, default is nl=1.
fmt Format string for this logging entry, overloads global format
string. Example: fmt="%(file)s:%(line)d %(message)s"
nofmt Only output message with nofmt=1. The nofmt argument wins over
the fmt argument.
Example:
from logger import log
log.setInfoLogLevel(log.INFO1)
log.setDebugLogLevel(log.DEBUG1)
for i in range(1, log.INFO_MAX+1):
log.setInfoLogLabel(i, "INFO%d: " % i)
log.setFormat("%(date)s %(module)s:%(line)d [%(domain)s] %(label)s: "
"%(level)d %(message)s")
log.setDateFormat("%Y-%m-%d %H:%M:%S")
fl = FileLog("/tmp/log", "a")
log.addInfoLogging("*", fl)
log.addDebugLogging("*", fl)
log.addInfoLogging("*", log.syslog, fmt="%(label)s%(message)s")
log.debug3("debug3")
log.debug2("debug2")
log.debug1("debug1")
log.info2("info2")
log.info1("info1")
log.warning("warning\n", nl=0)
log.error("error\n", nl=0)
log.fatal("fatal")
log.info(log.INFO1, "nofmt info", nofmt=1)
"""
ALL = -5
NOTHING = -4
FATAL = -3
TRACEBACK = -2
ERROR = -1
WARNING = 0
# Additional levels are generated in class initilization
stdout = _StdoutLog()
stderr = _StderrLog()
syslog = _SyslogLog()
def __init__(self, info_max=5, debug_max=10):
"""Logger class initialization"""
self._level = {}
self._debug_level = {}
self._format = ""
self._date_format = ""
self._label = {}
self._debug_label = {}
self._logging = {}
self._debug_logging = {}
self._domains = {}
self._debug_domains = {}
# INFO1 is required for standard log level
if info_max < 1:
raise ValueError("Logger: info_max %d is too low" % info_max)
if debug_max < 0:
raise ValueError("Logger: debug_max %d is too low" % debug_max)
self.NO_INFO = self.WARNING # = 0
self.INFO_MAX = info_max
self.NO_DEBUG = 0
self.DEBUG_MAX = debug_max
self.setInfoLogLabel(self.FATAL, "FATAL ERROR: ")
self.setInfoLogLabel(self.TRACEBACK, "")
self.setInfoLogLabel(self.ERROR, "ERROR: ")
self.setInfoLogLabel(self.WARNING, "WARNING: ")
# generate info levels and infox functions
for _level in range(1, self.INFO_MAX + 1):
setattr(self, "INFO%d" % _level, _level)
self.setInfoLogLabel(_level, "")
setattr(
self,
"info%d" % (_level),
(
lambda self, x: lambda message, *args, **kwargs: self.info(
x, message, *args, **kwargs
)
)(self, _level),
) # pylint: disable=E0602
# generate debug levels and debugx functions
for _level in range(1, self.DEBUG_MAX + 1):
setattr(self, "DEBUG%d" % _level, _level)
self.setDebugLogLabel(_level, "DEBUG%d: " % _level)
setattr(
self,
"debug%d" % (_level),
(
lambda self, x: lambda message, *args, **kwargs: self.debug(
x, message, *args, **kwargs
)
)(self, _level),
) # pylint: disable=E0602
# set initial log levels, formats and targets
self.setInfoLogLevel(self.INFO1)
self.setDebugLogLevel(self.NO_DEBUG)
self.setFormat("%(label)s%(message)s")
self.setDateFormat("%d %b %Y %H:%M:%S")
self.setInfoLogging("*", self.stderr, [self.FATAL, self.ERROR, self.WARNING])
self.setInfoLogging(
"*", self.stdout, [i for i in range(self.INFO1, self.INFO_MAX + 1)]
)
self.setDebugLogging(
"*", self.stdout, [i for i in range(1, self.DEBUG_MAX + 1)]
)
def close(self):
"""Close all logging targets"""
for level in range(self.FATAL, self.DEBUG_MAX + 1):
if level not in self._logging:
continue
for dummy, target, dummy in self._logging[level]:
target.close()
def getInfoLogLevel(self, domain="*"):
"""Get info log level."""
self._checkDomain(domain)
if domain in self._level:
return self._level[domain]
return self.NOTHING
def setInfoLogLevel(self, level, domain="*"):
"""Set log level [NOTHING .. INFO_MAX]"""
self._checkDomain(domain)
if level < self.NOTHING:
level = self.NOTHING
if level > self.INFO_MAX:
level = self.INFO_MAX
self._level[domain] = level
def getDebugLogLevel(self, domain="*"):
"""Get debug log level."""
self._checkDomain(domain)
if domain in self._debug_level:
return self._debug_level[domain] + self.NO_DEBUG
return self.NO_DEBUG
def setDebugLogLevel(self, level, domain="*"):
"""Set debug log level [NO_DEBUG .. DEBUG_MAX]"""
self._checkDomain(domain)
if level < 0:
level = 0
if level > self.DEBUG_MAX:
level = self.DEBUG_MAX
self._debug_level[domain] = level - self.NO_DEBUG
def getFormat(self):
return self._format
def setFormat(self, _format):
self._format = _format
def getDateFormat(self):
return self._date_format
def setDateFormat(self, _format):
self._date_format = _format
def setInfoLogLabel(self, level, label):
"""Set log label for level. Level can be a single level or an array
of levels."""
levels = self._getLevels(level)
for level in levels:
self._checkLogLevel(level, min_level=self.FATAL, max_level=self.INFO_MAX)
self._label[level] = label
def setDebugLogLabel(self, level, label):
"""Set log label for level. Level can be a single level or an array
of levels."""
levels = self._getLevels(level, is_debug=1)
for level in levels:
self._checkLogLevel(level, min_level=self.INFO1, max_level=self.DEBUG_MAX)
self._debug_label[level] = label
def setInfoLogging(self, domain, target, level=ALL, fmt=None):
"""Set info log target for domain and level. Level can be a single
level or an array of levels. Use level ALL to set for all levels.
If no format is specified, the default format will be used."""
self._setLogging(domain, target, level, fmt, is_debug=0)
def setDebugLogging(self, domain, target, level=ALL, fmt=None):
"""Set debug log target for domain and level. Level can be a single
level or an array of levels. Use level ALL to set for all levels.
If no format is specified, the default format will be used."""
self._setLogging(domain, target, level, fmt, is_debug=1)
def addInfoLogging(self, domain, target, level=ALL, fmt=None):
"""Add info log target for domain and level. Level can be a single
level or an array of levels. Use level ALL to set for all levels.
If no format is specified, the default format will be used."""
self._addLogging(domain, target, level, fmt, is_debug=0)
def addDebugLogging(self, domain, target, level=ALL, fmt=None):
"""Add debg log target for domain and level. Level can be a single
level or an array of levels. Use level ALL to set for all levels.
If no format is specified, the default format will be used."""
self._addLogging(domain, target, level, fmt, is_debug=1)
def delInfoLogging(self, domain, target, level=ALL, fmt=None):
"""Delete info log target for domain and level. Level can be a single
level or an array of levels. Use level ALL to set for all levels.
If no format is specified, the default format will be used."""
self._delLogging(domain, target, level, fmt, is_debug=0)
def delDebugLogging(self, domain, target, level=ALL, fmt=None):
"""Delete debug log target for domain and level. Level can be a single
level or an array of levels. Use level ALL to set for all levels.
If no format is specified, the default format will be used."""
self._delLogging(domain, target, level, fmt, is_debug=1)
def isInfoLoggingHere(self, level):
"""Is there currently any info logging for this log level (and
domain)?"""
return self._isLoggingHere(level, is_debug=0)
def isDebugLoggingHere(self, level):
"""Is there currently any debug logging for this log level (and
domain)?"""
return self._isLoggingHere(level, is_debug=1)
### log functions
def fatal(self, _format, *args, **kwargs):
"""Fatal error log."""
self._checkKWargs(kwargs)
kwargs["is_debug"] = 0
self._log(self.FATAL, _format, *args, **kwargs)
def error(self, _format, *args, **kwargs):
"""Error log."""
self._checkKWargs(kwargs)
kwargs["is_debug"] = 0
self._log(self.ERROR, _format, *args, **kwargs)
def warning(self, _format, *args, **kwargs):
"""Warning log."""
self._checkKWargs(kwargs)
kwargs["is_debug"] = 0
self._log(self.WARNING, _format, *args, **kwargs)
def info(self, level, _format, *args, **kwargs):
"""Information log using info level [1..info_max].
There are additional infox functions according to info_max from
__init__"""
self._checkLogLevel(level, min_level=1, max_level=self.INFO_MAX)
self._checkKWargs(kwargs)
kwargs["is_debug"] = 0
self._log(level + self.NO_INFO, _format, *args, **kwargs)
def debug(self, level, _format, *args, **kwargs):
"""Debug log using debug level [1..debug_max].
There are additional debugx functions according to debug_max
from __init__"""
self._checkLogLevel(level, min_level=1, max_level=self.DEBUG_MAX)
self._checkKWargs(kwargs)
kwargs["is_debug"] = 1
self._log(level, _format, *args, **kwargs)
def exception(self):
self._log(self.TRACEBACK, traceback.format_exc(), args=[], kwargs={})
### internal functions
def _checkLogLevel(self, level, min_level, max_level):
if level < min_level or level > max_level:
raise ValueError(
"Level %d out of range, should be [%d..%d]."
% (level, min_level, max_level)
)
def _checkKWargs(self, kwargs):
if not kwargs:
return
for key in kwargs.keys():
if key not in ["nl", "fmt", "nofmt"]:
raise ValueError(
"Key '%s' is not allowed as argument for logging." % key
)
def _checkDomain(self, domain):
if not domain or domain == "":
raise ValueError("Domain '%s' is not valid." % domain)
def _getLevels(self, level, is_debug=0):
"""Generate log level array."""
if level != self.ALL:
if isinstance(level, list) or isinstance(level, tuple):
levels = level
else:
levels = [level]
for level in levels:
if is_debug:
self._checkLogLevel(level, min_level=1, max_level=self.DEBUG_MAX)
else:
self._checkLogLevel(
level, min_level=self.FATAL, max_level=self.INFO_MAX
)
else:
if is_debug:
levels = [i for i in range(self.DEBUG1, self.DEBUG_MAX)]
else:
levels = [i for i in range(self.FATAL, self.INFO_MAX)]
return levels
def _getTargets(self, target):
"""Generate target array."""
if isinstance(target, list) or isinstance(target, tuple):
targets = target
else:
targets = [target]
for _target in targets:
if not issubclass(_target.__class__, LogTarget):
raise ValueError(
"'%s' is no valid logging target." % _target.__class__.__name__
)
return targets
def _genDomains(self, is_debug=0):
# private method for self._domains array creation, speeds up
"""Generate dict with domain by level."""
if is_debug:
_domains = self._debug_domains
_logging = self._debug_logging
_range = (1, self.DEBUG_MAX + 1)
else:
_domains = self._domains
_logging = self._logging
_range = (self.FATAL, self.INFO_MAX + 1)
if len(_domains) > 0:
_domains.clear()
for level in range(_range[0], _range[1]):
if level not in _logging:
continue
for domain, dummy, dummy in _logging[level]:
if domain not in _domains:
_domains.setdefault(level, []).append(domain)
def _setLogging(self, domain, target, level=ALL, fmt=None, is_debug=0):
self._checkDomain(domain)
levels = self._getLevels(level, is_debug)
targets = self._getTargets(target)
if is_debug:
_logging = self._debug_logging
else:
_logging = self._logging
for level in levels:
for target in targets:
_logging[level] = [(domain, target, fmt)]
self._genDomains(is_debug)
def _addLogging(self, domain, target, level=ALL, fmt=None, is_debug=0):
self._checkDomain(domain)
levels = self._getLevels(level, is_debug)
targets = self._getTargets(target)
if is_debug:
_logging = self._debug_logging
else:
_logging = self._logging
for level in levels:
for target in targets:
_logging.setdefault(level, []).append((domain, target, fmt))
self._genDomains(is_debug)
def _delLogging(self, domain, target, level=ALL, fmt=None, is_debug=0):
self._checkDomain(domain)
levels = self._getLevels(level, is_debug)
targets = self._getTargets(target)
if is_debug:
_logging = self._debug_logging
else:
_logging = self._logging
for _level in levels:
for target in targets:
if _level not in _logging:
continue
if (domain, target, fmt) in _logging[_level]:
_logging[_level].remove((domain, target, fmt))
if len(_logging[_level]) == 0:
del _logging[_level]
continue
if level != self.ALL:
raise ValueError(
"No matching logging for "
"level %d, domain %s, target %s and format %s."
% (_level, domain, target.__class__.__name__, fmt)
)
self._genDomains(is_debug)
def _isLoggingHere(self, level, is_debug=0):
_dict = self._genDict(level, is_debug)
if not _dict:
return False
point_domain = _dict["domain"] + "."
if is_debug:
_logging = self._debug_logging
else:
_logging = self._logging
# do we need to log?
for domain, dummy, dummy in _logging[level]:
if (
domain == "*"
or point_domain.startswith(domain)
or fnmatch.fnmatchcase(_dict["domain"], domain)
):
return True
return False
def _getClass(self, frame):
"""Function to get calling class. Returns class or None."""
# get class by first function argument, if there are any
if frame.f_code.co_argcount > 0:
selfname = frame.f_code.co_varnames[0]
if selfname in frame.f_locals:
_self = frame.f_locals[selfname]
obj = self._getClass2(_self.__class__, frame.f_code)
if obj:
return obj
module = inspect.getmodule(frame.f_code)
code = frame.f_code
# function in module?
if code.co_name in module.__dict__:
if (
hasattr(module.__dict__[code.co_name], "func_code")
and module.__dict__[code.co_name].__code__ == code
):
return None
# class in module
for dummy, obj in module.__dict__.items():
if isinstance(obj, types.ClassType):
if hasattr(obj, code.co_name):
value = getattr(obj, code.co_name)
if isinstance(value, types.FunctionType):
if value.__code__ == code:
return obj
# nothing found
return None
def _getClass2(self, obj, code):
"""Internal function to get calling class. Returns class or None."""
for value in obj.__dict__.values():
if isinstance(value, types.FunctionType):
if value.__code__ == code:
return obj
for base in obj.__bases__:
_obj = self._getClass2(base, code)
if _obj:
return _obj
return None
# internal log class
def _log(self, level, _format, *args, **kwargs):
is_debug = 0
if "is_debug" in kwargs:
is_debug = kwargs["is_debug"]
nl = 1
if "nl" in kwargs:
nl = kwargs["nl"]
nofmt = 0
if "nofmt" in kwargs:
nofmt = kwargs["nofmt"]
_dict = self._genDict(level, is_debug)
if not _dict:
return
if len(args) > 1:
_dict["message"] = _format % args
elif len(args) == 1: # needed for _format % _dict
_dict["message"] = _format % args[0]
else:
_dict["message"] = _format
point_domain = _dict["domain"] + "."
if is_debug:
_logging = self._debug_logging
else:
_logging = self._logging
used_targets = []
# log to target(s)
for domain, target, _format in _logging[level]:
if target in used_targets:
continue
if (
domain == "*"
or point_domain.startswith(domain + ".")
or fnmatch.fnmatchcase(_dict["domain"], domain)
):
if not _format:
_format = self._format
if "fmt" in kwargs:
_format = kwargs["fmt"]
if nofmt:
target.write(_dict["message"], level, self, is_debug)
else:
target.write(_format % _dict, level, self, is_debug)
if nl: # newline
target.write("\n", level, self, is_debug)
used_targets.append(target)
# internal function to generate the dict, needed for logging
def _genDict(self, level, is_debug=0):
"""Internal function."""
check_domains = []
simple_match = False
if is_debug:
_dict = self._debug_level
_domains = self._debug_domains
_label = self._debug_label
else:
_dict = self._level
_domains = self._domains
_label = self._label
# no debug
for domain in _dict:
if domain == "*":
# '*' matches everything: simple match
if _dict[domain] >= level:
simple_match = True
if len(check_domains) > 0:
check_domains = []
break
else:
if _dict[domain] >= level:
check_domains.append(domain)
if not simple_match and len(check_domains) < 1:
return None
if level not in _domains:
return None
f = inspect.currentframe()
# go outside of logger module as long as there is a lower frame
while f and f.f_back and f.f_globals["__name__"] == self.__module__:
f = f.f_back
if not f:
raise ValueError("Frame information not available.")
# get module name
module_name = f.f_globals["__name__"]
# simple module match test for all entries of check_domain
point_module = module_name + "."
for domain in check_domains:
if point_module.startswith(domain):
# found domain in module name
check_domains = []
break
# get code
co = f.f_code
# optimization: bail out early if domain can not match at all
_len = len(module_name)
for domain in _domains[level]:
i = domain.find("*")
if i == 0:
continue
elif i > 0:
d = domain[:i]
else:
d = domain
if _len >= len(d):
if not module_name.startswith(d):
return None
else:
if not d.startswith(module_name):
return None
# generate _dict for format output
level_str = ""
if level in _label:
level_str = _label[level]
_dict = {
"file": co.co_filename,
"line": f.f_lineno,
"module": module_name,
"class": "",
"function": co.co_name,
"domain": "",
"label": level_str,
"level": level,
"date": time.strftime(self._date_format, time.localtime()),
}
if _dict["function"] == "?":
_dict["function"] = ""
# domain match needed?
domain_needed = False
for domain in _domains[level]:
# standard domain, matches everything
if domain == "*":
continue
# domain is needed
domain_needed = True
break
# do we need to get the class object?
if (
self._format.find("%(domain)") >= 0
or self._format.find("%(class)") >= 0
or domain_needed
or len(check_domains) > 0
):
obj = self._getClass(f)
if obj:
_dict["class"] = obj.__name__
# build domain string
_dict["domain"] = "" + _dict["module"]
if _dict["class"] != "":
_dict["domain"] += "." + _dict["class"]
if _dict["function"] != "":
_dict["domain"] += "." + _dict["function"]
if len(check_domains) < 1:
return _dict
point_domain = _dict["domain"] + "."
for domain in check_domains:
if point_domain.startswith(domain) or fnmatch.fnmatchcase(
_dict["domain"], domain
):
return _dict
return None
# ---------------------------------------------------------------------------
# Global logging object.
log = Logger()
# ---------------------------------------------------------------------------
"""
# Example
if __name__ == '__main__':
log.setInfoLogLevel(log.INFO2)
log.setDebugLogLevel(log.DEBUG5)
for i in range(log.INFO1, log.INFO_MAX+1):
log.setInfoLogLabel(i, "INFO%d: " % i)
for i in range(log.DEBUG1, log.DEBUG_MAX+1):
log.setDebugLogLabel(i, "DEBUG%d: " % i)
log.setFormat("%(date)s %(module)s:%(line)d %(label)s"
"%(message)s")
log.setDateFormat("%Y-%m-%d %H:%M:%S")
fl = FileLog("/tmp/log", "a")
log.addInfoLogging("*", fl)
log.delDebugLogging("*", log.stdout)
log.setDebugLogging("*", log.stdout, [ log.DEBUG1, log.DEBUG2 ] )
log.addDebugLogging("*", fl)
# log.addInfoLogging("*", log.syslog, fmt="%(label)s%(message)s")
# log.addDebugLogging("*", log.syslog, fmt="%(label)s%(message)s")
log.debug10("debug10")
log.debug9("debug9")
log.debug8("debug8")
log.debug7("debug7")
log.debug6("debug6")
log.debug5("debug5")
log.debug4("debug4")
log.debug3("debug3")
log.debug2("debug2", fmt="%(file)s:%(line)d %(message)s")
log.debug1("debug1", nofmt=1)
log.info5("info5")
log.info4("info4")
log.info3("info3")
log.info2("info2")
log.info1("info1")
log.warning("warning\n", nl=0)
log.error("error ", nl=0)
log.error("error", nofmt=1)
log.fatal("fatal")
log.info(log.INFO1, "nofmt info", nofmt=1)
log.info(log.INFO2, "info2 fmt", fmt="%(file)s:%(line)d %(message)s")
try:
a = b
except Exception as e:
log.exception()
"""