|
Server : Apache System : Linux srv.rainic.com 4.18.0-553.47.1.el8_10.x86_64 #1 SMP Wed Apr 2 05:45:37 EDT 2025 x86_64 User : rainic ( 1014) PHP Version : 7.4.33 Disable Function : exec,passthru,shell_exec,system Directory : /bin/ |
Upload File : |
#! /usr/libexec/platform-python -Es
#
# Authors: John Dennis <jdennis@redhat.com>
# Authors: 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.
#
from __future__ import print_function
import syslog
import gi
GTK_VERSION = '3.0'
from gi.repository import GObject, GLib
GLib.set_prgname('setroubleshoot')
import dbus
import dbus.service
import errno as Errno
import gettext
import os
import re
import signal
import selinux
import socket as Socket
import fcntl
import sys
from dbus.mainloop.glib import DBusGMainLoop
DBusGMainLoop(set_as_default=True)
from setroubleshoot.config import parse_config_setting, get_config
domain = get_config('general', 'i18n_text_domain')
kwargs = {}
if sys.version_info < (3,):
kwargs['unicode'] = True
gettext.install(domain=domain,
localedir=get_config('general', 'i18n_locale_dir'),
**kwargs)
syslog.openlog(sys.argv[0])
from setroubleshoot.errcode import *
from setroubleshoot.util import get_identity, load_plugins, log_init, log_debug
from setroubleshoot.rpc import *
#------------------------------------------------------------------------------
invocation_style = None
status_icon = None
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')
dbus_session_bus_name = get_config('session_dbus', 'bus_name')
dbus_session_object_path = get_config('session_dbus', 'object_path')
dbus_session_interface = get_config('session_dbus', 'interface')
app = None
default_status_icon_tooltip = _("SELinux AVC denial, click to view")
log_init("sealert_log")
#------------------------------------------------------------------------------
def sighandler(signum, frame):
log_debug("exiting on signal %s" % signum)
sys.exit()
def get_server_socket_address():
addr_list = get_socket_list_from_config('client_connect_to')
if len(addr_list) == 0:
return None
return addr_list[0]
def setup_sighandlers():
signal.signal(signal.SIGHUP, sighandler)
signal.signal(signal.SIGQUIT, sighandler)
signal.signal(signal.SIGTERM, sighandler)
def run_app(user):
global app
from setroubleshoot.serverconnection import ServerConnectionHandler
GObject.type_register(ServerConnectionHandler)
app = SEAlert(user)
return app.main()
def error(message):
dlg = gtk.MessageDialog(None, 0, gtk.MESSAGE_ERROR,
gtk.BUTTONS_CLOSE,
message)
dlg.set_position(gtk.WIN_POS_MOUSE)
dlg.show_all()
dlg.run()
dlg.destroy()
def run_as_dbus_service(user):
global app
#-----------------------------------------------------------------------------
if not selinux.is_selinux_enabled():
log_debug("SELinux not enabled, sealert will not run on non SELinux systems")
error(_("SELinux not enabled, sealert will not run on non SELinux systems"))
sys.exit(3)
try:
log_debug('starting service')
dbus_service = DBusSessionService(dbus_session_bus_name)
app = SEAlert(user, dbus_service.presentation_manager, watch_setroubleshootd=True)
return app.main()
except dbus.DBusException as e:
syslog.syslog(syslog.LOG_ERR, 'could not start dbus: %s' % str(e))
return False
def ask_dbus_to_show_browser():
try:
bus = dbus.SessionBus()
proxy_obj = bus.get_object(dbus_session_bus_name, dbus_session_object_path)
iface = dbus.Interface(proxy_obj, dbus_session_interface)
return True
except dbus.DBusException as e:
syslog.syslog(syslog.LOG_ERR, 'could not start dbus: %s' % str(e))
return False
def ask_dbus_to_quit_app():
try:
bus = dbus.SessionBus()
proxy_obj = bus.get_object(dbus_session_bus_name, dbus_session_object_path)
iface = dbus.Interface(proxy_obj, dbus_session_interface)
iface.quit_app()
return True
except dbus.DBusException as e:
syslog.syslog(syslog.LOG_ERR, 'could not start dbus: %s' % str(e))
return False
def fix_lookup_id(local_id, analysis_id):
def lookup_local_id():
log_debug("calling server to lookup id (%s)" % local_id)
async_rpc = cl.alert_client.query_alerts(local_id)
async_rpc.add_callback(query_alerts_callback)
async_rpc.add_errback(query_alerts_error)
def query_alerts_callback(sigs):
import subprocess
for siginfo in sigs.signature_list:
for plugin in siginfo.plugin_list:
if analysis_id == plugin.analysis_id:
p = load_plugins(analysis_id)[0]
if p.fixable == False:
print(_("Not fixable."))
cl.main_loop.quit()
return
siginfo.update_derived_template_substitutions()
command = siginfo.substitute_array(p.get_fix_cmd(siginfo.audit_event, plugin.args).split())
try:
output = subprocess.check_output(command, universal_newlines=True)
print(_("Successfully ran %s" % ' '.join(command)))
except subprocess.CalledProcessError as e:
print((e.output))
except:
pass
cl.main_loop.quit()
return
print(_("Plugin %s not valid for %s id") % (analysis_id, local_id))
cl.main_loop.quit()
def query_alerts_error(method, errno, strerror):
print("%s error (%d): %s" % (method, errno, strerror), file=sys.stderr)
cl.main_loop.quit()
cl = SECommandLine(lookup_local_id)
cl.run()
def command_line_lookup_id(local_id, html=False):
global err
err = None
def lookup_local_id():
log_debug("calling server to lookup id (%s)" % local_id)
async_rpc = cl.alert_client.query_alerts(local_id)
async_rpc.add_callback(query_alerts_callback)
async_rpc.add_errback(query_alerts_error)
def query_alerts_callback(sigs):
for siginfo in sigs.signature_list:
print(siginfo.format_text())
print(siginfo.format_details())
cl.main_loop.quit()
def query_alerts_error(method, errno, strerror):
global err
cl.main_loop.quit()
err = "%s error (%d): %s" % (method, errno, strerror)
cl = SECommandLine(lookup_local_id)
cl.run()
if err:
print("Error")
raise ValueError(err)
#-----------------------------------------------------------------------------
class PresentationManager(GObject.GObject):
__gsignals__ = {
'show_browser':
(GObject.SignalFlags.RUN_LAST, GObject.TYPE_NONE, (GObject.TYPE_PYOBJECT,)),
'quit_app':
(GObject.SignalFlags.RUN_LAST, GObject.TYPE_NONE, ()),
}
def __init__(self):
GObject.GObject.__init__(self)
def show_browser(self, data_name=None):
self.emit('show_browser', data_name)
def quit_app(self):
self.emit('quit_app')
GObject.type_register(PresentationManager)
#-----------------------------------------------------------------------------
class SETroubleshootdDBus(GObject.GObject):
__gsignals__ = {
'state_change':
(GObject.SignalFlags.RUN_LAST, GObject.TYPE_NONE, (GObject.TYPE_PYOBJECT,)),
'restart_request':
(GObject.SignalFlags.RUN_LAST, GObject.TYPE_NONE, (GObject.TYPE_PYOBJECT,)),
}
def __init__(self):
GObject.GObject.__init__(self)
self.bus = dbus.SystemBus()
proxy_obj = self.bus.get_object(dbus_system_bus_name, dbus_system_object_path)
self.iface = dbus.Interface(proxy_obj, dbus_system_interface)
s = self.iface.start()
log_debug("starting setroubleshootd service")
self.bus.add_signal_receiver(self.on_dbus_name_owner_change, 'NameOwnerChanged',
'org.freedesktop.DBus', 'org.freedesktop.DBus', '/org/freedesktop/DBus',
arg0=dbus_system_bus_name)
# Note: this signal can be emitted from the command line via
# dbus-send --system $dbus_system_object_path $dbus_system_interface.restart string:"reason"
self.bus.add_signal_receiver(self.do_restart, 'restart', dbus_system_interface, None, dbus_system_object_path)
def finish(self):
log_debug("exiting dbus")
s = self.iface.finish()
def __del__(self):
log_debug("exiting dbus")
s = self.iface.finish()
def on_dbus_name_owner_change(self, name, old_owner, new_owner):
log_debug("on_dbus_name_owner_change: name=%s old_owner=%s new_owner=%s" % (name, old_owner, new_owner))
if not old_owner and new_owner:
log_debug("setroubleshootd: came on line")
self.emit('state_change', 'run')
elif old_owner and not new_owner:
log_debug("setroubleshootd: went off line")
self.emit('state_change', 'stop')
elif not old_owner and not new_owner:
log_debug("setroubleshootd: no connection")
self.emit('state_change', 'stop')
elif old_owner and new_owner:
log_debug("setroubleshootd: owner change")
else:
log_debug("don't know how to parse old_owner(%s) new_owner(%s)" % (old_owner, new_owner))
def do_restart(self, reason):
log_debug("do_restart(%s)" % reason)
self.emit('restart_request', reason)
GObject.type_register(SETroubleshootdDBus)
#-----------------------------------------------------------------------------
class DBusSessionService(dbus.service.Object):
def __init__(self, bus_name):
bus = dbus.SessionBus()
bus_name = dbus.service.BusName(dbus_session_bus_name, bus=bus)
dbus.service.Object.__init__(self, bus_name, dbus_session_object_path)
self.presentation_manager = PresentationManager()
@dbus.service.method(dbus_session_interface)
def start(self):
return _("Started")
@dbus.service.method(dbus_session_interface)
def show_browser(self):
log_debug('dbus iface show_browser() called',)
self.presentation_manager.show_browser()
return ""
@dbus.service.method(dbus_session_interface)
def quit_app(self):
log_debug('quit_app() called')
self.presentation_manager.quit_app()
#-----------------------------------------------------------------------------
class SEAlert(object):
"""
The SEAlert object represents a gui client for setroubleshoot. It
processes alerts and presents the user with an appropriate user
interface for handling the alert. Most of the interface code
is in BrowserApplet and StatusIcon. This class is mainly a central
hub for processing the alerts.
"""
def __init__(self, username, presentation_manager=None, watch_setroubleshootd=False):
from setroubleshoot.serverconnection import ServerConnectionHandler
GObject.type_register(ServerConnectionHandler)
try:
self.username = username
from setroubleshoot.browser import BrowserApplet
if presentation_manager is None:
self.presentation_manager = PresentationManager()
GObject.idle_add(self.show_browser_at_startup)
else:
self.presentation_manager = presentation_manager
self.browser = None
self.alert_siginfo = None
if watch_setroubleshootd:
self.setroubleshootd_dbus = SETroubleshootdDBus()
self.setroubleshootd_dbus.connect('state_change', self.on_setroubleshootd_state_change)
else:
self.setroubleshootd_dbus = None
self.alert_client = ServerConnectionHandler(self.username)
self.alert_client.open(get_server_socket_address())
self.browser = BrowserApplet(self.username, self.alert_client, domain=domain)
self.presentation_manager.connect('show_browser', self.on_show_browser)
self.presentation_manager.connect('quit_app', self.on_quit)
self.alert_client.connect('alert', self.alert)
# If there is no presentation mananger make sure when the
# user closes the window the whole application exits. When running
# in "alert" mode we want the application to persist in the background
self.browser.window_delete_hides = False
except ProgramError as e:
syslog.syslog(syslog.LOG_ERR, e.strerror)
sys.exit(1)
def finish(self):
if self.setroubleshootd_dbus != None:
self.setroubleshootd_dbus.finish()
def main(self):
log_debug('creating main GUI application')
try:
Gtk.main()
self.finish()
except IOError as e:
syslog.syslog(syslog.LOG_ERR, str(e))
print(e, file=sys.stderr)
sys.exit(1)
except KeyboardInterrupt as e:
log_debug("got KeyboardInterrupt, exiting ...")
self.alert_client.close_connection(ConnectionState.HUP)
self.finish()
sys.exit()
def alert(self, alert_client, siginfo):
log_debug("evaluating alert")
def alert_filter_result(result):
pass
self.alert_siginfo = siginfo
async_rpc = self.browser.server.evaluate_alert_filter(siginfo.sig, self.username)
async_rpc.add_callback(alert_filter_result)
def show_browser_at_startup(self):
self.presentation_manager.show_browser()
return False
def show_browser(self, data_name):
if data_name is not None:
self.browser.do_visit(data_name)
log_debug("SEAlert.show_browser(): data_name=%s" % data_name)
self.browser.show()
return True
def on_quit(self, widget):
if self.alert_client is not None:
self.alert_client.close_connection(ConnectionState.HUP)
gtk.main_quit()
def on_show_browser(self, widget, data_name):
self.show_browser(data_name)
def on_setroubleshootd_state_change(self, setroubleshootd_dbus, state):
log_debug('setroubleshootd state change (%s)' % state)
if state == 'run':
self.alert_client.open()
def on_restart_request(self, setroubleshootd_dbus, reason):
log_debug('restart request: reason = %s' % reason)
do_restart()
return False
def do_restart():
log_debug("restarting application...)")
window_state = None
geometry = None
if app is not None:
if app.browser is not None:
window_state, visibility_state = app.browser.get_window_state()
geometry = app.browser.get_geometry()
os.environ['SEALERT_WINDOW_STATE'] = window_state
os.environ['SEALERT_WINDOW_GEOMETRY'] = geometry
log_debug("restarting %s: args=%s window_state=%s geometry=%s" % (sys.argv[0], sys.argv[1:], window_state, geometry))
os.execv(sys.argv[0], sys.argv)
#-----------------------------------------------------------------------------
class SECommandLine(object):
def __init__(self, func):
from setroubleshoot.serverconnection import ServerConnectionHandler
self.username = get_identity()
self.func = func
self.alert_client = ServerConnectionHandler(self.username)
self.alert_client.connect('connection_state_changed', self.on_connection_state_change)
self.main_loop = GLib.MainLoop()
def async_error_callback(self, method, errno, strerror):
print("%s error (%d): %s" % (method, errno, strerror), file=sys.stderr)
sys.exit(1)
def do_logon(self):
def logon_callback(pkg_version, rpc_version):
log_debug("logon_callback(): pkg_version=%s rpc_version=%s" % (pkg_version, rpc_version))
self.alert_client.connection_state.update(ConnectionState.AUTHENTICATED)
log_debug("logon: %s" % self.username)
self.alert_client.channel_name = self.username
async_rpc = self.alert_client.logon(self.alert_client.channel_type, self.username, 'passwd')
async_rpc.add_callback(logon_callback)
async_rpc.add_errback(self.async_error_callback)
def on_connection_state_change(self, connection, 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), connection.socket_address))
if (flags_added & (ConnectionState.ERROR | ConnectionState.HUP)) or (flags_removed & ConnectionState.OPEN):
errno, strerror = connection_state.get_result()
print("failed to connect to server: %s" % (strerror), file=sys.stderr)
sys.exit(1)
if flags_added & ConnectionState.OPEN:
self.do_logon()
if flags_added & ConnectionState.AUTHENTICATED:
self.func()
def run(self):
log_debug('executing command line application')
self.alert_client.open(get_server_socket_address())
try:
self.main_loop.run()
except KeyboardInterrupt as e:
sys.exit()
#-----------------------------------------------------------------------------
class ScanLogfile:
def __init__(self, logfile_path):
from setroubleshoot.analyze import LogfileAnalyzer
self.analyzer = LogfileAnalyzer(logfile_path)
self.main_loop = GLib.MainLoop()
if sys.stdout.isatty():
self.analyzer.connect('progress', self.on_progress)
self.analyzer.connect('state-changed', self.on_analyzer_state_change)
def on_progress(self, analyzer, progress):
output = "\r%3d%% done" % (progress * 100)
sys.stdout.write(output)
sys.stdout.flush()
def on_analyzer_state_change(self, analyzer, state):
if state == 'stopped':
if analyzer.strerror:
print("ERROR: %s" % analyzer.strerror, file=sys.stderr)
log_debug("analyzer stopped")
self.main_loop.quit()
self.output_results()
def scan_file(self):
self.analyzer.cancelled = False
self.analyzer.open()
self.analyzer.run()
self.main_loop.run()
def output_results(self):
sigs = self.analyzer.database.query_alerts('*')
separator = '-' * 80 + '\n'
print("\nfound %d alerts in %s" % (len(sigs.signature_list), self.analyzer.logfile_path))
for siginfo in sigs.signature_list:
print(separator)
print(siginfo.format_text())
print(siginfo.format_details())
def do_analyze_logfile(logfile_path):
import selinux.audit2why as audit2why
audit2why.init()
scanner = ScanLogfile(logfile_path)
scanner.scan_file()
audit2why.finish()
#-----------------------------------------------------------------------------
def display_terminal_traceback(who):
import traceback
stacktrace = traceback.format_exc()
print(_("Opps, %s hit an error!" % who) + '\n\n' + stacktrace)
try:
from setroubleshoot.gui_utils import display_traceback
except:
def display_traceback(who):
display_terminal_traceback(who)
# -- Main --
if __name__ == '__main__':
use_terminal = False
setup_sighandlers()
log_debug("main() args=%s" % sys.argv)
# Exit if selinux is disabled - setroubleshootd cannot start
if not selinux.is_selinux_enabled():
log_debug("SELinux not enabled, sealert will not run on non SELinux systems")
print("SELinux not enabled, sealert will not run on non SELinux systems", file=sys.stderr)
sys.exit(3)
def validate_invocation_style(opt, opts_instance, conflict_opts):
global invocation_style
conflict_opts.remove(opt)
invocation_style = opt
for i in conflict_opts:
if (getattr(opts_instance, i)):
print("cannot run as %s with other option" % (opt), file=sys.stderr)
sys.exit(3)
try:
from optparse import OptionParser
parser = OptionParser()
parser.add_option("-b", "--browser", action="store_true", dest="browser", default=False,
help="Launch the browser")
parser.add_option("-s", "--service", action="store_true", dest="service", default=False,
help="Start sealert as a dbus service")
parser.add_option("-S", "--noservice", action="store_true", dest="noservice", default=False,
help="Start sealert without dbus service as stand alone app")
parser.add_option("-l", "--lookupid", dest="lookupid", default=False,
help="Lookup alert by id, id may be wildcard * to lookup all alerts")
parser.add_option("-a", "--analyze", dest="analyze", default=False,
help="Scan a log file, analyze its AVCs", metavar="FILE")
parser.add_option("-u", "--user", dest="user", default=False,
help="logon user name")
parser.add_option("-p", "--password", dest="password", default=False,
help="logon user password")
parser.add_option("-P", "--plugin", dest="plugin", default=False,
help="Plugin Name Required for -f")
parser.add_option("-f", "--fix", dest="fix", default=False,
help="fix avc with the given uuid, requires plugin")
conflict_opts = ["analyze", "browser", "fix", "lookupid", "noservice", "service"]
username = None
password = None
plugin_name = None
(options, args) = parser.parse_args()
if options.user:
username = options.user
else:
username = get_identity()
if options.password:
password = options.password
else:
password = 'passwd'
if options.browser:
validate_invocation_style("browser", options, conflict_opts)
if options.service:
validate_invocation_style("service", options, conflict_opts)
if options.noservice:
validate_invocation_style("noservice", options, conflict_opts)
if options.lookupid:
validate_invocation_style("lookupid", options, conflict_opts)
local_id = options.lookupid
if options.analyze:
validate_invocation_style("analyze", options, conflict_opts)
use_terminal = True
logfile = options.analyze
if options.plugin:
plugin_name = options.plugin
if options.fix:
validate_invocation_style("fix", options, conflict_opts)
local_id = options.fix
if options.plugin:
plugin_name = options.plugin
else:
print("-P plugin_name required for fix", file=sys.stderr)
sys.exit(3)
# Attempt to communicate with the service. DBus should start it if it is not
# running, otherwise we will become the service
if invocation_style is None:
log_debug("invocation style not set, asking dbus to start us")
try:
bus = dbus.SessionBus()
proxy_obj = bus.get_object(dbus_session_bus_name, dbus_session_object_path)
iface = dbus.Interface(proxy_obj, dbus_session_interface)
s = iface.start()
except dbus.DBusException:
print("could not attach to desktop process", file=sys.stderr)
sys.exit(1)
sys.exit()
log_debug("invocation style = '%s'" % invocation_style)
if invocation_style == 'browser':
ask_dbus_to_show_browser()
sys.exit()
elif invocation_style == 'service':
try:
# This import must come before importing gtk to silence warnings
gi.require_version('Gtk', GTK_VERSION)
from gi.repository import Gtk
run_as_dbus_service(username)
except RuntimeError as e:
print(e)
sys.exit(3)
sys.exit()
elif invocation_style == 'noservice':
# This import must come before importing gtk to silence warnings
gi.require_version('Gtk', GTK_VERSION)
from gi.repository import Gtk
run_app(username)
sys.exit()
elif invocation_style == 'lookupid':
use_terminal = True
from setroubleshoot.signature import *
try:
# make sure setroubleshoot is running
bus = dbus.SystemBus()
proxy_obj = bus.get_object(dbus_system_bus_name, dbus_system_object_path)
iface = dbus.Interface(proxy_obj, dbus_system_interface)
iface.start()
command_line_lookup_id(local_id)
iface.finish()
except ValueError as e:
print(e, file=sys.stderr)
iface.finish()
sys.exit(3)
except dbus.exceptions.DBusException as e:
print("Unable to establish connection to setroubleshoot daemon!\n" +
"Check output of 'journalctl -t setroubleshoot' for more details.")
sys.exit(3)
sys.exit()
elif invocation_style == 'analyze':
use_terminal = True
try:
do_analyze_logfile(logfile)
except OSError as e:
print(e)
sys.exit(3)
except ProgramError as e:
print(e.strerror, file=sys.stderr)
sys.exit(3)
except Exception as e:
print("SELinux is disabled or we can't open a policy file")
sys.exit(3)
elif invocation_style == 'fix':
use_terminal = True
from setroubleshoot.signature import *
try:
bus = dbus.SystemBus()
proxy_obj = bus.get_object(dbus_system_bus_name, dbus_system_object_path)
iface = dbus.Interface(proxy_obj, dbus_system_interface)
iface.start()
fix_lookup_id(local_id, plugin_name)
iface.finish()
except dbus.DBusException as e:
print("could not attach to desktop process", file=sys.stderr)
sys.exit(3)
sys.exit()
else:
print("unknown invocation style (%s)" % invocation_style, file=sys.stderr)
sys.exit(3)
except Exception as e:
syslog.syslog(syslog.LOG_ERR, "exception %s: %s" % (e.__class__.__name__, str(e)))
if use_terminal:
display_terminal_traceback('sealert')
else:
display_traceback('sealert')
sys.exit(3)