✏️ Editing: signature.py Read Only

from __future__ import absolute_import
from __future__ import print_function
# Authors: John Dennis <jdennis@redhat.com>
#          Thomas Liu <tliu@redhat.com>
#          Dan Walsh <dwalsh@redhat.com>
#
# Copyright (C) 2006-2010 Red Hat, Inc.
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed 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 General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
#

import six
import syslog
from subprocess import *
from six.moves import range
from functools import cmp_to_key
import gettext
from setroubleshoot.config import parse_config_setting, get_config

translation = gettext.translation(domain=get_config('general', 'i18n_text_domain'),
                                  localedir=get_config('general', 'i18n_locale_dir'),
                                  fallback=True)

try:
    _ = translation.ugettext  # Unicode version of gettext for Py2
except AttributeError:
    _ = translation.gettext  # Python3 (uses unicode by default)

__all__ = [
    'SignatureMatch',
    'SEFilter',
    'SEFaultSignature',
    'SEFaultSignatureInfo',
    'SEFaultSignatureSet',
    'SEFaultSignatureUser',
    'SEEnvironment',
    'SEDatabaseProperties',
    'SEFaultUserInfo',
    'SEFaultUserSet',
    'SEPlugin',
    'SEEmailRecipient',
    'SEEmailRecipientSet',

'FILTER_NEVER',
    'FILTER_ALWAYS',
    'FILTER_AFTER_FIRST',
    'filter_text'
]

if __name__ == "__main__":
    gettext.install(domain=get_config('general', 'i18n_text_domain'),
                    localedir=get_config('general', 'i18n_locale_dir'))

from gettext import ngettext as P_
from setroubleshoot.config import get_config
from setroubleshoot.errcode import *
from setroubleshoot.util import *
from setroubleshoot.xml_serialize import *
from setroubleshoot.html_util import *
import setroubleshoot.uuid as uuid
from setroubleshoot.audit_data import *
import hashlib
from types import *
from string import Template
import re
import os

cmp = lambda x, y: (x > y) - (x < y)

# Don't reuse the numeric values!
FILTER_NEVER = 0
FILTER_ALWAYS = 4
FILTER_AFTER_FIRST = 8

filter_text = {
    FILTER_NEVER: _("Never Ignore"),
    FILTER_ALWAYS: _("Ignore Always"),
    FILTER_AFTER_FIRST: _("Ignore After First Alert"),
}

map_filter_value_to_name = {
    FILTER_NEVER: 'never',
    FILTER_ALWAYS: 'always',
    FILTER_AFTER_FIRST: 'after_first',
}

map_filter_name_to_value = {
    'never': FILTER_NEVER,
    'always': FILTER_ALWAYS,
    'after_first': FILTER_AFTER_FIRST,
}

#------------------------------------------------------------------------

class SignatureMatch(object):

def __init__(self, siginfo, score):
        self.siginfo = siginfo
        self.score = score

class SEEnvironment(XmlSerialize):
    _xml_info = {
        'version': {'XMLForm': 'attribute', 'default': lambda: '1.0'},
        'platform': {'XMLForm': 'element'},
        'kernel': {'XMLForm': 'element'},
        'policy_type': {'XMLForm': 'element'},
        'policy_rpm': {'XMLForm': 'element'},
        'local_policy_rpm': {'XMLForm': 'element'},
        'enforce': {'XMLForm': 'element'},
        'selinux_enabled': {'XMLForm': 'element', 'import_typecast': boolean, },
        'selinux_mls_enabled': {'XMLForm': 'element', 'import_typecast': boolean, },
        'policyvers': {'XMLForm': 'element'},
        'hostname': {'XMLForm': 'element'},
        'uname': {'XMLForm': 'element'},
    }

def __init__(self):
        super(SEEnvironment, self).__init__()
        self.update()

def update(self):
        import platform
        import selinux
        # security_getenforce is the same as the getenforce command.
        # selinux_getenforcemode tells you what is set in /etc/selinux/config

self.platform, self.kernel = get_os_environment()
        self.policy_type = selinux.selinux_getpolicytype()[1]
        self.policy_rpm = get_package_nvr_by_file_path("/etc/selinux/%s" % self.policy_type)
        self.local_policy_rpm = self.policy_rpm
        self.policyvers = str(selinux.security_policyvers())
        enforce = selinux.security_getenforce()
        if enforce == 0:
            self.enforce = "Permissive"
        else:
            self.enforce = "Enforcing"

self.selinux_enabled = bool(selinux.is_selinux_enabled())
        self.selinux_mls_enabled = bool(selinux.is_selinux_mls_enabled())
        self.hostname = platform.node()
        self.uname = " ".join(platform.uname())

def __ne__(self, other):
        return not self.__eq__(other)

def __eq__(self, other):
        for name in list(self._xml_info.keys()):
            if getattr(self, name) != getattr(other, name):
                return False
        return True

class SEFilter(XmlSerialize):
    _xml_info = {
        'filter_type': {'XMLForm': 'element', 'import_typecast': int, 'default': lambda: FILTER_NEVER},
        'count': {'XMLForm': 'element', 'import_typecast': int, 'default': lambda: 0},
    }

def __init__(self, filter_type=FILTER_NEVER):
        super(SEFilter, self).__init__()
        self.filter_type = filter_type

class SEFaultSignatureUser(XmlSerialize):
    _xml_info = {
        'username': {'XMLForm': 'attribute'},
        'seen_flag': {'XMLForm': 'attribute', 'import_typecast': boolean, 'default': lambda: False},
        'delete_flag': {'XMLForm': 'attribute', 'import_typecast': boolean, 'default': lambda: False},
        'filter': {'XMLForm': 'element', 'import_typecast': SEFilter, 'default': lambda: SEFilter()},
    }

def __init__(self, username):
        super(SEFaultSignatureUser, self).__init__()
        self.username = username

def update_item(self, item, data):
        if not item in self._names:
            raise ProgramError(ERR_NOT_MEMBER, 'item (%s) is not a defined member' % item)

if item == 'username':
            raise ProgramError(ERR_ILLEGAL_USER_CHANGE, 'changing the username is illegal')

setattr(self, item, data)

def update_filter(self, filter_type, data=None):
        log_debug("update_filter: filter_type=%s data=%s" % (map_filter_value_to_name.get(filter_type, 'unknown'), data))
        if filter_type == FILTER_NEVER or \
           filter_type == FILTER_AFTER_FIRST or \
           filter_type == FILTER_ALWAYS:
            log_debug("update_filter: !!!")
            self.filter = SEFilter(filter_type=filter_type)
            return True
        else:
            raise ValueError("Bad filter_type (%s)" % filter_type)

class_dict = {}
class_dict['dir'] = _("directory")
class_dict['sem'] = _("semaphore")
class_dict['shm'] = _("shared memory")
class_dict['msgq'] = _("message queue")
class_dict['msg'] = _("message")
class_dict['file'] = _("file")
class_dict['socket'] = _("socket")
class_dict['process'] = _("process")
class_dict['process2'] = _("process2")
class_dict['filesystem'] = _("filesystem")
class_dict['node'] = _("node")
class_dict['capability'] = _("capability")
class_dict['capability2'] = _("capability2")

def translate_class(tclass):
    if tclass in list(class_dict.keys()):
        return class_dict[tclass]
    return tclass

# --

class AttributeValueDictionary(XmlSerialize):
    _xml_info = 'unstructured'

def __init__(self):
        super(AttributeValueDictionary, self).__init__()

class SEFaultSignature(XmlSerialize):
    _xml_info = {
        'version': {'XMLForm': 'attribute', 'default': lambda: '4.0', },
        'host': {'XMLForm': 'element', },
        'access': {'XMLForm': 'element', 'list': 'operation', },
        'scontext': {'XMLForm': 'element', 'import_typecast': AvcContext},
        'tcontext': {'XMLForm': 'element', 'import_typecast': AvcContext},
        'tclass': {'XMLForm': 'element', },
        'port': {'XMLForm': 'element', 'import_typecast': int, },
    }

def __init__(self, **kwds):
        super(SEFaultSignature, self).__init__()
        for k, v in list(kwds.items()):
            setattr(self, k, v)

class SEPlugin(XmlSerialize):
    _xml_info = {
        'analysis_id': {'XMLForm': 'element'},
        'args': {'XMLForm': 'element', 'list': 'arg', },
    }

def __init__(self, analysis_id, args):
        super(SEPlugin, self).__init__()
        self.analysis_id = analysis_id
        self.args = args

def __str__(self):
        return str((self.analysis_id, self.args))

class SEFaultSignatureInfo(XmlSerialize):
    _xml_info = {
        'plugin_list': {'XMLForm': 'element', 'list': 'plugin', 'import_typecast': SEPlugin},
        'audit_event': {'XMLForm': 'element', 'import_typecast': AuditEvent},
        'source': {'XMLForm': 'element'},
        'spath': {'XMLForm': 'element'},
        'tpath': {'XMLForm': 'element'},
        'src_rpm_list': {'XMLForm': 'element', 'list': 'rpm', },
        'tgt_rpm_list': {'XMLForm': 'element', 'list': 'rpm', },
        'scontext': {'XMLForm': 'element', 'import_typecast': AvcContext},
        'tcontext': {'XMLForm': 'element', 'import_typecast': AvcContext},
        'tclass': {'XMLForm': 'element', },
        'port': {'XMLForm': 'element', 'import_typecast': int, },

'sig': {'XMLForm': 'element', 'import_typecast': SEFaultSignature},
        'if_text': {'XMLForm': 'element'},
        'then_text': {'XMLForm': 'element'},
        'do_text': {'XMLForm': 'element'},
        'environment': {'XMLForm': 'element', 'import_typecast': SEEnvironment},

'first_seen_date': {'XMLForm': 'element', 'import_typecast': TimeStamp},
        'last_seen_date': {'XMLForm': 'element', 'import_typecast': TimeStamp},
        'report_count': {'XMLForm': 'element', 'import_typecast': int, 'default': lambda: 0},
        'local_id': {'XMLForm': 'element'},
        'users': {'XMLForm': 'element', 'list': 'user', 'import_typecast': SEFaultSignatureUser, },
        'level': {'XMLForm': 'element'},
        'fixable': {'XMLForm': 'element'},
        'button_text': {'XMLForm': 'element'},
    }

merge_include = ['audit_event', 'tpath', 'src_rpm_list', 'tgt_rpm_list',
                     'scontext', 'tcontext', 'tclass', 'port',
                     'environment',
                     'last_seen_date'
                     ]

def __init__(self, **kwds):
        super(SEFaultSignatureInfo, self).__init__()
        for k, v in list(kwds.items()):
            setattr(self, k, v)
        self.report_count = 1
        self.plugin_list = []

use_dbus=True
        if os.getuid() == 0:
            # root doesn't need to use dbus
            use_dbus=False
        try:
            self.environment.local_policy_rpm = get_rpm_nvr_by_scontext(self.scontext, use_dbus=use_dbus)
        except:
            # leave it as it is
            pass

def update_merge(self, siginfo):
        if siginfo.last_seen_date != self.last_seen_date:
            self.last_seen_date = siginfo.last_seen_date
            self.report_count += 1

for name in self.merge_include:
            setattr(self, name, getattr(siginfo, name))

# older databases can have an uninitialized level
        if self.level is None:
            self.level = siginfo.level

def get_policy_rpm(self):
        return self.environment.policy_rpm

def get_hash_str(self):
        return "%s,%s,%s,%s,%s" % (self.source, self.scontext.type, self.tcontext.type, self.tclass, ",".join(self.sig.access))

def get_hash(self):
        hash = hashlib.sha256(self.get_hash_str().encode('utf-8'))
        return hash.hexdigest()

def get_user_data(self, username):
        for user in self.users:
            if user.username == username:
                return user
        log_debug("new SEFaultSignatureUser for %s" % username)
        user = SEFaultSignatureUser(username)
        self.users.append(user)
        return user

def find_filter_by_username(self, username):
        log_debug("find_filter_by_username %s" % username)

filter = None
        user_data = self.get_user_data(username)
        if user_data is not None:
            filter = user_data.filter
        return filter

def update_user_filter(self, username, filter_type, data=None):
        user_data = self.get_user_data(username)
        user_data.update_filter(filter_type, data)

def evaluate_filter_for_user(self, username, filter_type=None):
        action = 'display'
        f = self.find_filter_by_username(username)
        log_debug("evaluate_filter_for_user: found %s user's filter = %s" % (username, f))
        if f is not None:
            if filter_type is not None:
                f.filter_type = filter_type
            action = self.evaluate_filter(f)
            log_debug("evaluate_filter_for_user: found filter for %s: %s\n%s" % (username, action, f))
        return action

def evaluate_filter(self, filter):
        filter_type = filter.filter_type

action = 'display'

if filter_type == FILTER_NEVER:
            action = 'display'
        elif filter_type == FILTER_AFTER_FIRST:
            if filter.count == 0:
                action = 'display'
            else:
                action = 'ignore'
        elif filter_type == FILTER_ALWAYS:
            action = 'ignore'
        else:
            raise ValueError("unknown filter_type (%s)" % (filter_type))
        filter.count += 1
        return action

def format_rpm_list(self, rpm_list):
        if isinstance(rpm_list, list):
            if len(rpm_list) > 0:
                return " ".join(rpm_list)
            else:
                return ""
        else:
            return default_text(None)

def format_target_object(self):
        return "%s [ %s ]" % (self.tpath, self.tclass)

def description_adjusted_for_permissive(self):
        permissive_msg = None
        syscall_record = self.audit_event.get_record_of_type('SYSCALL')
        if syscall_record != None and syscall_record.get_field('success') == 'yes':
            permissive_msg = _("%s has a permissive type (%s). This access was not denied.") % (self.source, self.scontext.type)

if self.environment.enforce == "Permissive":
            permissive_msg = _("SELinux is in permissive mode. This access was not denied.")

def update_derived_template_substitutions(self):
        self.template_substitutions = {}
        self.template_substitutions["SOURCE_TYPE"] = self.scontext.type
        self.template_substitutions["TARGET_TYPE"] = self.tcontext.type
        self.template_substitutions["SOURCE"] = self.source
        self.template_substitutions["SOURCE_PATH"] = self.spath
        self.template_substitutions["SOURCE_BASE_PATH"] = os.path.basename(self.spath)
        self.template_substitutions["MODULE_NAME"] = re.sub('[^a-zA-Z0-9]', '', self.source)
        if self.spath:
            self.template_substitutions["FIX_SOURCE_PATH"] = re.sub(" ", ".", self.spath)
        else:
            self.spath = _("N/A")

self.template_substitutions["TARGET_PATH"] = self.tpath
        self.template_substitutions["TARGET_BASE_PATH"] = os.path.basename(self.tpath)
        if self.tpath:
            self.template_substitutions["FIX_TARGET_PATH"] = re.sub(" ", ".", self.tpath)

if self.tpath is None:
            self.template_substitutions["TARGET_DIR"] = None
        else:
            if self.tclass == 'dir':
                self.template_substitutions["TARGET_DIR"] = self.tpath
            elif self.tclass == 'file':
                self.template_substitutions["TARGET_DIR"] = os.path.dirname(self.tpath)
            else:
                self.template_substitutions["TARGET_DIR"] = None

if self.tclass == "dir":
            self.template_substitutions["TARGET_CLASS"] = "directory"
        else:
            self.template_substitutions["TARGET_CLASS"] = self.tclass

if self.sig.access is None:
            self.template_substitutions["ACCESS"] = None
        else:
            self.template_substitutions["ACCESS"] = ' '.join(self.sig.access)

if len(self.src_rpm_list) > 0:
            self.template_substitutions["SOURCE_PACKAGE"] = self.src_rpm_list[0]
        self.template_substitutions["PORT_NUMBER"] = self.port

# validate, replace any None values with friendly string
        for key, value in list(self.template_substitutions.items()):
            if value is None:
                self.template_substitutions[key] = default_text(value)

def priority_sort(self, x, y):
        return cmp(y[0].priority, x[0].priority)

def summary(self):
        if self.tclass in ["process", "process2"]:
            return P_(_("SELinux is preventing %s from using the %s access on a process."), _("SELinux is preventing %s from using the '%s' accesses on a process."), len(self.sig.access)) % (self.spath, ", ".join(self.sig.access))

if self.tclass in ["capability", "capability2"]:
            return P_(_("SELinux is preventing %s from using the %s capability."), _("SELinux is preventing %s from using the '%s' capabilities."), len(self.sig.access)) % (self.spath, ", ".join(self.sig.access))
        if self.tpath in ["(null)", "Unknown"] :
            return P_(_("SELinux is preventing %s from %s access on the %s labeled %s."), _("SELinux is preventing %s from '%s' accesses on the %s labeled %s."), len(self.sig.access)) % (self.spath, ", ".join(self.sig.access), translate_class(self.tclass), self.tcontext.type)
        return P_(_("SELinux is preventing %s from %s access on the %s %s."), _("SELinux is preventing %s from '%s' accesses on the %s %s."), len(self.sig.access)) % (self.spath, ", ".join(self.sig.access), translate_class(self.tclass), self.tpath)

def get_plugins(self, all=False):
        self.plugins = load_plugins()
        plugins = []
        total_priority = 0
        if all:
            for p in self.plugins:
                total_priority += p.priority
                plugins.append((p, ("allow_ypbind", "1")))
        else:
            for solution in self.plugin_list:
                for p in self.plugins:
                    if solution.analysis_id == p.analysis_id:
                        total_priority += p.priority
                        p.init_args(tuple(solution.args))
                        plugins.append((p, tuple(solution.args)))
                        break

plugins.sort(key=cmp_to_key(self.priority_sort))

# do not show "report bug" button if switching one of the following booleans was suggested
        noreport_booleans = ["mozilla_read_content", "mozilla_plugin_can_network_connect",
                             "mozilla_plugin_use_bluejeans", "unconfined_mozilla_plugin_transition"]
        # suggested commands (from all plugins)
        do_texts = " ".join([p.get_do_text(self.audit_event.records, a) for p, a in plugins])

for b in noreport_booleans:
            if b in do_texts:
                # remove "report bug" button from all plugins
                for p, a in plugins:
                    p.report_bug = False
                break

return total_priority, plugins

def substitute(self, txt):
        return Template(txt).safe_substitute(self.template_substitutions)

def substitute_array(self, args):
        return [self.substitute(txt) for txt in args]

def format_details(self, replace=False):
        env = self.environment

text = _("Additional Information:\n")
        text += format_2_column_name_value(_("Source Context"), self.scontext.format())
        text += format_2_column_name_value(_("Target Context"), self.tcontext.format())
        text += format_2_column_name_value(_("Target Objects"), self.format_target_object())
        text += format_2_column_name_value(_("Source"), default_text(self.source))
        text += format_2_column_name_value(_("Source Path"), default_text(self.spath))
        text += format_2_column_name_value(_("Port"), default_text(self.port))
        if (replace):
            text += format_2_column_name_value(_("Host"), "(removed)")
        else:
            text += format_2_column_name_value(_("Host"), default_text(self.sig.host))
        text += format_2_column_name_value(_("Source RPM Packages"), default_text(self.format_rpm_list(self.src_rpm_list)))
        text += format_2_column_name_value(_("Target RPM Packages"), default_text(self.format_rpm_list(self.tgt_rpm_list)))
        text += format_2_column_name_value(_("SELinux Policy RPM"), default_text(env.policy_rpm))
        text += format_2_column_name_value(_("Local Policy RPM"), default_text(env.local_policy_rpm))
        text += format_2_column_name_value(_("Selinux Enabled"), default_text(env.selinux_enabled))
        text += format_2_column_name_value(_("Policy Type"), default_text(env.policy_type))
        text += format_2_column_name_value(_("Enforcing Mode"), default_text(env.enforce))
        if replace:
            text += format_2_column_name_value(_("Host Name"), "(removed)")
        else:
            text += format_2_column_name_value(_("Host Name"), default_text(env.hostname))

if replace:
            uname = env.uname.split()
            uname[1] = "(removed)"
            text += format_2_column_name_value(_("Platform"), default_text(" ".join(uname)))
        else:
            text += format_2_column_name_value(_("Platform"), default_text(env.uname))
        text += format_2_column_name_value(_("Alert Count"), default_text(self.report_count))
        date_format = "%Y-%m-%d %H:%M:%S %Z"
        text += format_2_column_name_value(_("First Seen"), self.first_seen_date.format(date_format))
        text += format_2_column_name_value(_("Last Seen"), self.last_seen_date.format(date_format))
        text += format_2_column_name_value(_("Local ID"), default_text(self.local_id))

text += '\n' + _("Raw Audit Messages")
        avcbuf = ""
        for audit_record in self.audit_event.records:
            if audit_record.record_type == 'AVC':
                avcbuf += "\n" + audit_record.to_text() + "\n"
            else:
                avcbuf += "\ntype=%s msg=%s: " % (audit_record.record_type, audit_record.event_id)
                avcbuf += ' '.join(["%s=%s" % (k, audit_record.fields[k]) for k in audit_record.fields_ord]) + "\n"

avcbuf += "\nHash: " + self.get_hash_str()

try:
            audit2allow = "/usr/bin/audit2allow"
            if os.path.exist(audit2allow):
                newbuf = "\n\naudit2allow"
                p = Popen([audit2allow], stdin=PIPE, stdout=PIPE)
                newbuf += p.communicate(avcbuf)[0]
                if os.path.exists("/var/lib/sepolgen/interface_info"):
                    newbuf += "\naudit2allow -R"
                    p = Popen([audit2allow, "-R"], stdin=PIPE, stdout=PIPE)
                    newbuf += p.communicate(avcbuf)[0]
                avcbuf += newbuf
        except:
            pass

text += avcbuf + '\n'

return text

def untranslated(self, func, *args, **kwargs):
        r'define.*untranslated\(.*\n'
        # Call the parameter function with the translations turned off
        # This function is not thread safe, since it manipulates globals

global P_, _
        saved_translateP_ = P_
        saved_translate_ = _

try:
            P_ = lambda x, y, z: x if z > 1 else y
            _ = lambda x: x
            return func(*args, **kwargs)
        finally:
            P_ = saved_translateP_
            _ = saved_translate_

def format_text(self, all=False, replace=False):
        self.update_derived_template_substitutions()

text = self.summary()

total_priority, plugins = self.get_plugins(all)

for p, args in plugins:
            title = _("\n\n*****  Plugin %s (%.4s confidence) suggests   ") % (p.analysis_id, ((float(p.priority) / float(total_priority)) * 100 + .5))
            text += title
            for i in range(len(title), 80):
                text += _("*")
            text += _("\n")
            txt = self.substitute(p.get_if_text(self.audit_event.records, args))
            text += _("\n") + txt
            txt = self.substitute(p.get_then_text(self.audit_event.records, args))
            text += _("\nThen ") + txt[0].lower() + txt[1:]

txt = self.substitute(p.get_do_text(self.audit_event.records, args))
            text += _("\nDo\n") + txt[0].lower() + txt[1:]

text += _('\n\n')
        return text

class SEFaultUserInfo(XmlSerialize):
    _xml_info = {
        'version': {'XMLForm': 'attribute', 'default': lambda: '1.0'},
        'username': {'XMLForm': 'attribute'},
        'email_alert': {'XMLForm': 'element', 'import_typecast': boolean, 'default': lambda: False},
        'email_address_list': {'XMLForm': 'element', 'list': 'email_address', },
    }

def __init__(self, username):
        super(SEFaultUserInfo, self).__init__()
        self.username = username

def add_email_address(self, email_address):
        if not email_address in self.email_address_list:
            self.email_address_list.append(email_address)

class SEFaultUserSet(XmlSerialize):
    _xml_info = {
        'version': {'XMLForm': 'attribute', 'default': lambda: '1.0'},
        'user_list': {'XMLForm': 'element', 'list': 'user', 'import_typecast': SEFaultUserInfo, },
    }

def __init__(self):
        super(SEFaultUserSet, self).__init__()

def get_user(self, username):
        for user in self.user_list:
            if username == user.username:
                return user
        return None

def add_user(self, username):
        if self.get_user(username) is not None:
            return
        user = SEFaultUserInfo(username)
        self.user_list.append(user)
        return user

class SEFaultSignatureSet(XmlSerialize):
    _xml_info = {
        'version': {'XMLForm': 'attribute', 'default': lambda: '%d.%d' % (DATABASE_MAJOR_VERSION, DATABASE_MINOR_VERSION)},
        'users': {'XMLForm': 'element', 'import_typecast': SEFaultUserSet, 'default': lambda: SEFaultUserSet()},
        'signature_list': {'XMLForm': 'element', 'list': 'siginfo', 'import_typecast': SEFaultSignatureInfo, },
    }

def __init__(self):
        super(SEFaultSignatureSet, self).__init__()

def siginfos(self):
        for siginfo in self.signature_list:
            yield siginfo

def add_siginfo(self, siginfo):
        self.signature_list.append(siginfo)
        return siginfo

def remove_siginfo(self, siginfo):
        self.signature_list.remove(siginfo)

def clear(self):
        self.signature_list = []

def generate_local_id(self):
        return str(uuid.uuid4())

def lookup_local_id(self, local_id):
        if local_id is None:
            return None

for siginfo in self.signature_list:
            if siginfo.local_id == local_id:
                return siginfo

return None

def match_signatures(self, pat, criteria='exact', xml_info=SEFaultSignature._xml_info):
        match_targets = list(xml_info.keys())
        exact = False
        if criteria == 'exact':
            exact = True
        elif isinstance(criteria, float):
            num_match_targets = len(match_targets)
            score_per_match_target = 1.0 / num_match_targets
        else:
            raise ValueError("unknown criteria = %s" % criteria)

matches = []
        for siginfo in self.signature_list:
            score = 0.0
            sig = siginfo.sig
            for name in match_targets:
                if getattr(pat, name) == getattr(sig, name):
                    if exact:
                        score = 1.0
                    else:
                        score += score_per_match_target
                else:
                    if exact:
                        score = 0.0
                        break
            if exact:
                if score == 1.0:
                    matches.append(SignatureMatch(siginfo, score))
            else:
                if score >= criteria:
                    matches.append(SignatureMatch(siginfo, score))
        matches.sort(key=cmp_to_key(lambda a, b: cmp(b.score, a.score)))
        return matches

class SEDatabaseProperties(XmlSerialize):
    _xml_info = {
        'name': {'XMLForm': 'element'},
        'friendly_name': {'XMLForm': 'element'},
        'filepath': {'XMLForm': 'element'},
    }

def __init__(self, name=None, friendly_name=None, filepath=None):
        super(SEDatabaseProperties, self).__init__()
        if name is not None:
            self.name = name
        if friendly_name is not None:
            self.friendly_name = friendly_name
        if filepath is not None:
            self.filepath = filepath

class SEEmailRecipient(XmlSerialize):
    _xml_info = {
        'address': {'XMLForm': 'element'},
        'filter_type': {'XMLForm': 'element', 'import_typecast': int, 'default': lambda: FILTER_AFTER_FIRST},
    }

def __init__(self, address, filter_type=None):
        super(SEEmailRecipient, self).__init__()
        self.address = address
        if filter_type is not None:
            self.filter_type = filter_type

def __str__(self):
        return "%s:%s" % (self.address, map_filter_value_to_name.get(self.filter_type, 'unknown'))

class SEEmailRecipientSet(XmlSerialize):
    _xml_info = {
        'version': {'XMLForm': 'attribute', 'default': lambda: '1'},
        'recipient_list': {'XMLForm': 'element', 'list': 'recipient', 'import_typecast': SEEmailRecipient, },
    }

def __init__(self, recipient_list=None):
        super(SEEmailRecipientSet, self).__init__()
        if recipient_list is not None:
            self.recipient_list = recipient_list

def __str__(self):
        return ','.join([str(x) for x in self.recipient_list])

def find_address(self, address):
        address = address.strip()
        for recipient in self.recipient_list:
            if address == recipient.address:
                return recipient
        return None

def add_address(self, address, filter_type=FILTER_AFTER_FIRST):
        address = address.strip()
        if not valid_email_address(address):
            raise ProgramError(ERR_INVALID_EMAIL_ADDR, detail="address='%s'" % address)
            return

recipient = self.find_address(address)
        if recipient is not None:
            return
        self.recipient_list.append(SEEmailRecipient(address, filter_type))

def clear_recipient_list(self):
        self.recipient_list = []

def parse_recipient_file(self, filepath):
        import re
        comment_re = re.compile('#.*')
        entry_re = re.compile(r'(\S+)(\s+(.+))?')
        key_value_re = re.compile(r"(\w+)\s*=\s*(\S+)")

map_boolean = {'enabled': True,
                       'true': True,
                       'yes': True,
                       'on': True,
                       'disabled': False,
                       'false': False,
                       'no': False,
                       'off': False,
                       }

try:
            f = open(filepath)
        except IOError as e:
            raise ProgramError(ERR_FILE_OPEN, detail="%s, %s" % (filepath, e.strerror))

self.clear_recipient_list()

for line in f.readlines():
            line = comment_re.sub('', line)
            line = line.strip()
            if line:
                match = entry_re.search(line)
                if match:
                    address = match.group(1)
                    options = match.group(3)
                    filter_type = None

if options:
                        for match in key_value_re.finditer(options):
                            option = match.group(1)
                            value = match.group(2)

if option == 'filter_type':
                                filter_type = map_filter_name_to_value.get(value.lower(), None)
                                if filter_type is None:
                                    log_debug("unknown email filter (%s) for address %s" % (option, address))

else:
                                log_debug("unknown email option (%s) for address %s" % (option, address))

try:
                        self.add_address(address, filter_type)
                    except ProgramError as e:
                        if e.errno == ERR_INVALID_EMAIL_ADDR:
                            log_debug(e.strerror)
                        else:
                            raise e

f.close()

def write_recipient_file(self, filepath):
        try:
            f = open(filepath, 'w')
        except IOError as e:
            raise ProgramError(ERR_FILE_OPEN, detail="%s, %s" % (filepath, e.strerror))

for recipient in self.recipient_list:
            filter_type = map_filter_value_to_name[recipient.filter_type]
            f.write("%-40s filter_type=%s\n" % (recipient.address, filter_type))

f.close()

#------------------------------------------------------------------------

if __name__ == '__main__':
    import libxml2
    #memory debug specific
    libxml2.debugMemory(1)

xml_file = 'audit_listener_database.xml'

sigs = SEFaultSignatureSet()
    sigs.read_xml_file(xml_file, 'sigs')
    siginfo = sigs.signature_list[0]
    record = siginfo.audit_event.records[0]
    print((record.record_type))
    print(("siginfo.audit_event=%s" % siginfo.audit_event))
    print(sigs)

#memory debug specific
    libxml2.cleanupParser()
    if libxml2.debugMemory(1) == 0:
        print("Memory OK")
    else:
        print(("Memory leak %d bytes" % (libxml2.debugMemory(1))))
        libxml2.dumpMemory()

🔐 Sid Gifari File Manager Pro