← Back
Editing: sealert
#!/usr/bin/python3 -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(modal=True, message_type=Gtk.MessageType.ERROR, buttons=Gtk.ButtonsType.CLOSE, text=message) dlg.set_position(Gtk.WindowPosition.CENTER) 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 is 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 Exception: 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(): """ 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 manager 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 is not 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): from selinux import audit2why audit2why.init() scanner = ScanLogfile(logfile_path) scanner.scan_file() audit2why.finish() #----------------------------------------------------------------------------- def display_terminal_traceback(who): import traceback stacktrace = traceback.format_exc() print(_("Oops, %s hit an error!" % who) + '\n\n' + stacktrace) try: from setroubleshoot.gui_utils import display_traceback except ImportError: 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 setroubleshootd' 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)
Save File
Cancel