ACIL FM

Nama	Ukuran	Aksi
__pycache__	-	Open
access_control.py	5.86 MB	View DL Edit
analyze.py	25.87 MB	View DL Edit
audit_data.py	40.52 MB	View DL Edit
avc_audit.py	14.94 MB	View DL Edit
config.py	20.7 MB	View DL Edit
email_alert.py	5.81 MB	View DL Edit
errcode.py	3.47 MB	View DL Edit
html_util.py	2.03 MB	View DL Edit
Plugin.py	5.18 MB	View DL Edit
rpc.py	37.01 MB	View DL Edit
rpc_interfaces.py	4.68 MB	View DL Edit
server.py	33.67 MB	View DL Edit
serverconnection.py	6.14 MB	View DL Edit
signature.py	33.26 MB	View DL Edit
util.py	34.57 MB	View DL Edit
uuid.py	16.52 MB	View DL Edit
xml_serialize.py	15.83 MB	View DL Edit
__init__.py	741 B	View DL Edit

Edit file: /usr/lib/python3.9/site-packages/setroubleshoot/server.py

from __future__ import absolute_import
# 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.
#

__all__ = ['RunFaultServer',
           'ClientConnectionHandler',
           'get_host_database',
           'send_alert_notification',
           'ConnectionPool',
           ]

from gi.repository import GObject, GLib
import dbus
import dbus.service
import dbus.glib
import gettext
#import errno as Errno
import os
#import pwd
import signal
import atexit
#from stat import *
import sys
import syslog
import systemd.journal
#import threading
#from types import *

from setroubleshoot.config import parse_config_setting, get_config

domain = get_config('general', 'i18n_text_domain')
localedir = get_config('general', 'i18n_locale_dir')
kwargs = {}
if sys.version_info < (3,):
    kwargs['unicode'] = True
gettext.install(domain=domain,
                localedir=localedir,
                **kwargs)

translation = gettext.translation(domain=domain,
                                  localedir=localedir,
                                  fallback=True)

try:
    _ = translation.ugettext  # This raises exception in Python3, succ. in Py2
except AttributeError:
    _ = translation.gettext  # Python3

from setroubleshoot.access_control import ServerAccess
from setroubleshoot.analyze import (PluginReportReceiver,
                                    SETroubleshootDatabase,
                                    TestPluginReportReceiver,
                                    AnalyzeThread,
                                    )
from setroubleshoot.avc_audit import *
from setroubleshoot.config import get_config
from setroubleshoot.avc_audit import AuditRecordReceiver

from setroubleshoot.errcode import (ProgramError,
                                    ERR_NOT_AUTHENTICATED,
                                    ERR_USER_LOOKUP,
                                    ERR_USER_PROHIBITED,
                                    ERR_USER_PERMISSION,
                                    ERR_FILE_OPEN,
                                    ERR_DATABASE_NOT_FOUND,
                                    )

from setroubleshoot.rpc import (RpcChannel,
                                ConnectionState,
                                get_socket_list_from_config,
                                ListeningServer,
                                )

from setroubleshoot.rpc_interfaces import (SETroubleshootServerInterface,
                                           SETroubleshootDatabaseNotifyInterface,
                                           SEAlertInterface,
                                           )
from setroubleshoot.util import (get_hostname,
                                 make_database_filepath,
                                 assure_file_ownership_permissions,
                                 get_identity, log_init, log_debug, syslog_trace
                                 )

log_init('setroubleshootd_log')

#------------------------------ Utility Functions -----------------------------

def sighandler(signum, frame):
    log_debug("received signal=%s" % signum)
    import setroubleshoot.config as config
    if signum == signal.SIGHUP:
        log_debug("reloading configuration file")
        config.config_init()
        return

def polling_failed_handler(signum, frame):
    log_debug("received signal=%s" % signum)
    syslog.syslog(syslog.LOG_ERR, "/sys/fs/selinux/policy is in use by another process. Exiting!")
    os._exit(1)
    # TODO: change to sys.exit(1) when the bug in audti2why is fixed

def make_instance_id():
    import time
    hostname = get_hostname()
    pid = os.getpid()
    stamp = str(time.time())
    return '%s:%s:%s' % (hostname, pid, stamp)

def get_host_database():
    return host_database

dbus_system_bus_name = get_config('system_dbus', 'bus_name')
dbus_system_object_path = get_config('system_dbus', 'object_path')
dbus_system_interface = get_config('system_dbus', 'interface')

system_bus = dbus.SystemBus()
system_bus.request_name(dbus_system_bus_name)

# FIXME: this should be part of ClientNotifier

def send_alert_notification(siginfo):
    alert = dbus.lowlevel.SignalMessage(dbus_system_object_path, dbus_system_interface, "alert")
    alert.append(siginfo.level)
    alert.append(siginfo.local_id)
    system_bus.send_message(alert)

for client in connection_pool.clients('sealert'):
        client.alert(siginfo)

#------------------------------ Variables -------------------------------

host_database = None
analysis_queue = None
email_recipients = None
email_recipients_filepath = get_config('email', 'recipients_filepath')
pkg_name = get_config('general', 'pkg_name')
pkg_version = get_config('general', 'pkg_version')
rpc_version = get_config('general', 'rpc_version')
instance_id = make_instance_id()

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

class ConnectionPool(object):

def __init__(self):
        self.client_pool = {}

def add_client(self, handler):
        if handler in self.client_pool:
            log_debug("add_client: client (%s) already in client pool" % handler)
            return
        self.client_pool[handler] = None

def remove_client(self, handler):
        if handler not in self.client_pool:
            log_debug("remove_client: client (%s) not in client pool" % handler)
            return
        del(self.client_pool[handler])

def clients(self, channel_type=None):
        for client in self.client_pool:
            if channel_type is None:
                yield client
            elif client.channel_type == channel_type:
                yield client

def close_all(self, channel_type=None):
        for client in self.client_pool:
            if client.channel_type == channel_type:
                client.close_connection()

connection_pool = ConnectionPool()

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

class AlertPluginReportReceiver(PluginReportReceiver):

def __init__(self, database):
        super(AlertPluginReportReceiver, self).__init__(database)

def report_problem(self, siginfo):
        siginfo = super(AlertPluginReportReceiver, self).report_problem(siginfo)

if email_recipients is not None:
            to_addrs = []
            for recipient in email_recipients.recipient_list:
                username = "email:%s" % recipient.address
                action = siginfo.evaluate_filter_for_user(username, recipient.filter_type)
                if action != "ignore":
                    log_debug("Email: siginfo.sig=%s" % siginfo.sig)
                    to_addrs.append(recipient.address)

if len(to_addrs):
                from setroubleshoot.email_alert import email_alert
                email_alert(siginfo, to_addrs)
            self.database.mark_modified()

log_debug("sending alert to all clients")

from setroubleshoot.html_util import html_to_text
        syslog.syslog(syslog.LOG_ERR, siginfo.summary() + _(" For complete SELinux messages run: sealert -l %s") % siginfo.local_id)
        for audit_record in siginfo.audit_event.records:
            if audit_record.record_type == 'AVC':
                pid = audit_record.fields["pid"]
                break
        if get_config('setroubleshootd_log', 'log_full_report', bool):
            global pkg_name
            systemd.journal.send(siginfo.format_text(), OBJECT_PID=pid, SYSLOG_IDENTIFIER=pkg_name)

for u in siginfo.users:
            if u.username[0:6] == "email:":
                # skip email users - they were evaluated before
                continue
            action = siginfo.evaluate_filter_for_user(u.username)
            if action == "ignore":
                return siginfo

send_alert_notification(siginfo)
        return siginfo

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

class ClientConnectionHandler(RpcChannel):

def __init__(self, socket_address):
        RpcChannel.__init__(self, 'sealert')
        self.socket_address = socket_address.copy()
        self.connection_state.connect('changed', self.on_connection_state_change)

def on_connection_state_change(self, connection_state, flags, flags_added, flags_removed):
        log_debug("%s.on_connection_state_change: connection_state=%s flags_added=%s flags_removed=%s address=%s" % (self.__class__.__name__, connection_state, connection_state.flags_to_string(flags_added), connection_state.flags_to_string(flags_removed), self.socket_address))

if flags_removed & ConnectionState.OPEN:
            connection_pool.remove_client(self)

if flags_added & ConnectionState.OPEN:
            connection_pool.add_client(self)

def open(self, socket, socket_address):
        if self.connection_state.flags & ConnectionState.OPEN:
            return True
        self.socket_address.socket = socket
        self.connection_state.update(ConnectionState.OPEN)
        self.io_watch_add(self.handle_client_io)

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

class SetroubleshootdClientConnectionHandler(ClientConnectionHandler,
                                             SETroubleshootServerInterface,
                                             SETroubleshootDatabaseNotifyInterface,
                                             SEAlertInterface,
                                             ):

def __init__(self, socket_address):
        ClientConnectionHandler.__init__(self, socket_address)

self.database = get_host_database()

self.connect_rpc_interface('SETroubleshootServer', self)
        self.connect_rpc_interface('SETroubleshootDatabase', self)

self.access = ServerAccess()
        self.username = None
        self.uid = None
        self.gid = None

if flags_removed & ConnectionState.OPEN:
            connection_pool.remove_client(self)

if flags_added & ConnectionState.OPEN:
            self.uid, self.gid = self.access.get_credentials(self.socket_address.socket)
            log_debug("%s.on_connection_state_change: open, socket credentials: uid=%s gid=%s" % (self.__class__.__name__, self.uid, self.gid))
            connection_pool.add_client(self)

# ---- SETroubleshootServerInterface Methods ----

def database_bind(self, database_name):
        if not (self.connection_state.flags & ConnectionState.AUTHENTICATED):
            raise ProgramError(ERR_NOT_AUTHENTICATED)

host_database = get_host_database()
        if host_database.properties.name == database_name:
            return [host_database.properties]
        raise ProgramError(ERR_DATABASE_NOT_FOUND, "database (%s) not found" % database_name)

def logon(self, type, username, password):
        log_debug("logon(%s) type=%s username=%s" % (self, type, username))

if username != get_identity(self.uid):
            raise ProgramError(ERR_USER_LOOKUP, detail="uid=%s does not match logon username (%s)" % (self.uid, username))

if type == 'sealert':
            privilege = 'client'
        else:
            privilege = None

if not self.access.user_allowed(privilege, username):
            raise ProgramError(ERR_USER_PROHIBITED)

self.channel_type = type
        self.channel_name = username
        self.username = username
        self.user = self.database.get_user(username)
        if self.user is None:
            self.database.add_user(username)

self.connection_state.update(ConnectionState.AUTHENTICATED)
        return [pkg_version, rpc_version]

def query_email_recipients(self):
        if not (self.connection_state.flags & ConnectionState.AUTHENTICATED):
            raise ProgramError(ERR_NOT_AUTHENTICATED)

return [email_recipients]

def set_email_recipients(self, recipients):
        global email_recipients

log_debug("set_email_recipients: %s" % recipients)