Edit File: sandbox
#! /usr/bin/python -Es # Authors: Dan Walsh <dwalsh@redhat.com> # Authors: Thomas Liu <tliu@fedoraproject.org> # Authors: Josh Cogliati # # Copyright (C) 2009,2010 Red Hat # see file 'COPYING' for use and warranty information # # 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; version 2 only # # 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., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA # import os import stat import sys import socket import random import fcntl import shutil import re import subprocess import selinux import signal from tempfile import mkdtemp import pwd import sepolicy PROGNAME = "policycoreutils" SEUNSHARE = "/usr/sbin/seunshare" SANDBOXSH = "/usr/share/sandbox/sandboxX.sh" import gettext gettext.bindtextdomain(PROGNAME, "/usr/share/locale") gettext.textdomain(PROGNAME) try: gettext.install(PROGNAME, localedir="/usr/share/locale", codeset='utf-8') except IOError: try: import builtins builtins.__dict__['_'] = str except ImportError: import __builtin__ __builtin__.__dict__['_'] = unicode DEFAULT_WINDOWSIZE = "1000x700" DEFAULT_TYPE = "sandbox_t" DEFAULT_X_TYPE = "sandbox_x_t" SAVE_FILES = {} random.seed(None) def sighandler(signum, frame): signal.signal(signum, signal.SIG_IGN) os.kill(0, signum) raise KeyboardInterrupt def setup_sighandlers(): signal.signal(signal.SIGHUP, sighandler) signal.signal(signal.SIGQUIT, sighandler) signal.signal(signal.SIGTERM, sighandler) def error_exit(msg): sys.stderr.write("%s: " % sys.argv[0]) sys.stderr.write("%s\n" % msg) sys.stderr.flush() sys.exit(1) def copyfile(file, srcdir, dest): import re if file.startswith(srcdir): dname = os.path.dirname(file) bname = os.path.basename(file) if dname == srcdir: dest = dest + "/" + bname else: newdir = re.sub(srcdir, dest, dname) if not os.path.exists(newdir): os.makedirs(newdir) dest = newdir + "/" + bname try: if os.path.isdir(file): shutil.copytree(file, dest) else: shutil.copy2(file, dest) except shutil.Error as elist: for e in elist.message: sys.stderr.write(e[2]) SAVE_FILES[file] = (dest, os.path.getmtime(dest)) def savefile(new, orig, X_ind): copy = False if(X_ind): from gi.repository import Gtk dlg = Gtk.MessageDialog(None, 0, Gtk.MessageType.INFO, Gtk.ButtonsType.YES_NO, _("Do you want to save changes to '%s' (Y/N): ") % orig) dlg.set_title(_("Sandbox Message")) dlg.set_position(Gtk.WindowPosition.MOUSE) dlg.show_all() rc = dlg.run() dlg.destroy() if rc == Gtk.ResponseType.YES: copy = True else: try: input = raw_input except NameError: pass ans = input(_("Do you want to save changes to '%s' (y/N): ") % orig) if(re.match(_("[yY]"),ans)): copy = True if(copy): shutil.copy2(new,orig) def reserve(level): sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) sock.bind("\0%s" % level) fcntl.fcntl(sock.fileno(), fcntl.F_SETFD, fcntl.FD_CLOEXEC) def get_range(): try: level = selinux.getcon_raw()[1].split(":")[4] lowc, highc = level.split(".") low = int(lowc[1:]) high = int(highc[1:]) + 1 if high - low == 0: raise IndexError return low, high except IndexError: raise ValueError(_("User account must be setup with an MCS Range")) def gen_mcs(): low, high = get_range() level = None ctr = 0 total = high - low total = (total * (total - 1)) / 2 while ctr < total: ctr += 1 i1 = random.randrange(low, high) i2 = random.randrange(low, high) if i1 == i2: continue if i1 > i2: tmp = i1 i1 = i2 i2 = tmp level = "s0:c%d,c%d" % (i1, i2) try: reserve(level) except socket.error: continue break if level: return level raise ValueError(_("Failed to find any unused category sets. Consider a larger MCS range for this user.")) def fullpath(cmd): for i in ["/", "./", "../"]: if cmd.startswith(i): return cmd for i in os.environ["PATH"].split(':'): f = "%s/%s" % (i, cmd) if os.access(f, os.X_OK): return f return cmd class Sandbox: SYSLOG = "/var/log/messages" def __init__(self): self.setype = DEFAULT_TYPE self.__options = None self.__cmds = None self.__init_files = [] self.__paths = [] self.__mount = False self.__level = None self.__homedir = None self.__tmpdir = None def __validate_mount(self): if self.__options.level: if not self.__options.homedir or not self.__options.tmpdir: self.usage(_("Homedir and tempdir required for level mounts")) if not os.path.exists(SEUNSHARE): raise ValueError(_(""" %s is required for the action you want to perform. """) % SEUNSHARE) def __mount_callback(self, option, opt, value, parser): self.__mount = True def __x_callback(self, option, opt, value, parser): self.__mount = True setattr(parser.values, option.dest, True) if not os.path.exists(SEUNSHARE): raise ValueError(_(""" %s is required for the action you want to perform. """) % SEUNSHARE) if not os.path.exists(SANDBOXSH): raise ValueError(_(""" %s is required for the action you want to perform. """) % SANDBOXSH) def __validdir(self, option, opt, value, parser): if not os.path.isdir(value): raise IOError("Directory " + value + " not found") setattr(parser.values, option.dest, value) self.__mount = True def __include(self, option, opt, value, parser): rp = os.path.realpath(os.path.expanduser(value)) if not os.path.exists(rp): raise IOError(value + " not found") if rp not in self.__init_files: self.__init_files.append(rp) def __includefile(self, option, opt, value, parser): fd = open(value, "r") for i in fd.readlines(): try: self.__include(option, opt, i[:-1], parser) except IOError as e: sys.stderr.write(str(e)) except TypeError as e: sys.stderr.write(str(e)) fd.close() def __copyfiles(self): files = self.__init_files + self.__paths homedir = pwd.getpwuid(os.getuid()).pw_dir for f in files: copyfile(f, homedir, self.__homedir) copyfile(f, "/tmp", self.__tmpdir) copyfile(f, "/var/tmp", self.__tmpdir) def __setup_sandboxrc(self, wm="/usr/bin/matchbox-window-manager"): execfile = self.__homedir + "/.sandboxrc" fd = open(execfile, "w+") if self.__options.session: fd.write("""#!/bin/sh #TITLE: /etc/gdm/Xsession /etc/gdm/Xsession """) else: command = self.__paths[0] + " " for p in self.__paths[1:]: command += "'%s' " % p fd.write("""#! /bin/sh #TITLE: %s /usr/bin/test -r ~/.xmodmap && /usr/bin/xmodmap ~/.xmodmap %s & WM_PID=$! dbus-launch --exit-with-session %s kill -TERM $WM_PID 2> /dev/null """ % (command, wm, command)) fd.close() os.chmod(execfile, 0o700) def usage(self, message=""): error_exit("%s\n%s" % (self.__parser.usage, message)) def __parse_options(self): from optparse import OptionParser types = "" try: types = _(""" Policy defines the following types for use with the -t: \t%s """) % "\n\t".join(sepolicy.info(sepolicy.ATTRIBUTE, "sandbox_type")[0]['types']) except RuntimeError: pass usage = _(""" sandbox [-h] [-l level ] [-[X|M] [-H homedir] [-T tempdir]] [-I includefile ] [-W windowmanager ] [ -w windowsize ] [[-i file ] ...] [ -t type ] command sandbox [-h] [-l level ] [-[X|M] [-H homedir] [-T tempdir]] [-I includefile ] [-W windowmanager ] [ -w windowsize ] [[-i file ] ...] [ -t type ] -S %s """) % types parser = OptionParser(usage=usage) parser.disable_interspersed_args() parser.add_option("-i", "--include", action="callback", callback=self.__include, type="string", help=_("include file in sandbox")) parser.add_option("-I", "--includefile", action="callback", callback=self.__includefile, type="string", help=_("read list of files to include in sandbox from INCLUDEFILE")) parser.add_option("-t", "--type", dest="setype", action="store", default=None, help=_("run sandbox with SELinux type")) parser.add_option("-M", "--mount", action="callback", callback=self.__mount_callback, help=_("mount new home and/or tmp directory")) parser.add_option("-d", "--dpi", dest="dpi", action="store", help=_("dots per inch for X display")) parser.add_option("-S", "--session", action="store_true", dest="session", default=False, help=_("run complete desktop session within sandbox")) parser.add_option("-s", "--shred", action="store_true", dest="shred", default=False, help=_("Shred content before tempory directories are removed")) parser.add_option("-X", dest="X_ind", action="callback", callback=self.__x_callback, default=False, help=_("run X application within a sandbox")) parser.add_option("-H", "--homedir", action="callback", callback=self.__validdir, type="string", dest="homedir", help=_("alternate home directory to use for mounting")) parser.add_option("-T", "--tmpdir", dest="tmpdir", type="string", action="callback", callback=self.__validdir, help=_("alternate /tmp directory to use for mounting")) parser.add_option("-w", "--windowsize", dest="windowsize", type="string", default=DEFAULT_WINDOWSIZE, help="size of the sandbox window") parser.add_option("-W", "--windowmanager", dest="wm", type="string", default="/usr/bin/matchbox-window-manager", help=_("alternate window manager")) parser.add_option("-l", "--level", dest="level", help=_("MCS/MLS level for the sandbox")) parser.add_option("-C", "--capabilities", action="store_true", dest="usecaps", default=False, help="Allow apps requiring capabilities to run within the sandbox.") self.__parser = parser self.__options, cmds = parser.parse_args() if self.__options.X_ind: self.setype = DEFAULT_X_TYPE else: try: sepolicy.info(sepolicy.TYPE, "sandbox_t") except RuntimeError: raise ValueError(_("Sandbox Policy is not currently installed.\nYou need to install the selinux-policy-sandbox package in order to run this command")) if self.__options.setype: self.setype = self.__options.setype if self.__mount: self.__validate_mount() if self.__options.session: if not self.__options.setype: self.setype = selinux.getcon()[1].split(":")[2] if not self.__options.homedir or not self.__options.tmpdir: self.usage(_("You must specify a Homedir and tempdir when setting up a session sandbox")) if len(cmds) > 0: self.usage(_("Commands are not allowed in a session sandbox")) self.__options.X_ind = True self.__homedir = self.__options.homedir self.__tmpdir = self.__options.tmpdir else: if self.__options.level: self.__homedir = self.__options.homedir self.__tmpdir = self.__options.tmpdir if len(cmds) == 0: self.usage(_("Command required")) cmds[0] = fullpath(cmds[0]) if not os.access(cmds[0], os.X_OK): self.usage(_("%s is not an executable") % cmds[0]) self.__cmds = cmds for f in cmds: rp = os.path.realpath(f) if os.path.exists(rp): self.__paths.append(rp) else: self.__paths.append(f) def __gen_context(self): if self.__options.level: level = self.__options.level else: level = gen_mcs() con = selinux.getcon()[1].split(":") self.__execcon = "%s:%s:%s:%s" % (con[0], con[1], self.setype, level) self.__filecon = "%s:object_r:sandbox_file_t:%s" % (con[0], level) def __setup_dir(self): if self.__options.homedir: selinux.chcon(self.__options.homedir, self.__filecon, recursive=True) self.__homedir = self.__options.homedir else: selinux.setfscreatecon(self.__filecon) self.__homedir = mkdtemp(dir="/tmp", prefix=".sandbox_home_") if self.__options.tmpdir: selinux.chcon(self.__options.tmpdir, self.__filecon, recursive=True) self.__tmpdir = self.__options.tmpdir else: selinux.setfscreatecon(self.__filecon) self.__tmpdir = mkdtemp(dir="/tmp", prefix=".sandbox_tmp_") selinux.setfscreatecon(None) self.__copyfiles() def __execute(self): try: cmds = [SEUNSHARE, "-Z", self.__execcon] if self.__options.usecaps: cmds.append('-C') if self.__mount: cmds += ["-t", self.__tmpdir, "-h", self.__homedir] if self.__options.X_ind: if self.__options.dpi: dpi = self.__options.dpi else: import gtk dpi = str(gtk.settings_get_default().props.gtk_xft_dpi / 1024) xmodmapfile = self.__homedir + "/.xmodmap" xd = open(xmodmapfile, "w") subprocess.Popen(["/usr/bin/xmodmap", "-pke"], stdout=xd).wait() xd.close() self.__setup_sandboxrc(self.__options.wm) cmds += ["--", SANDBOXSH, self.__options.windowsize, dpi] else: cmds += ["--"] + self.__paths return subprocess.Popen(cmds).wait() pid = os.fork() if pid == 0: rc = os.setsid() if rc: return rc selinux.setexeccon(self.__execcon) os.execv(self.__cmds[0], self.__cmds) rc = os.waitpid(pid, 0) return os.WEXITSTATUS(rc[1]) finally: for i in self.__paths: if i not in SAVE_FILES: continue (dest, mtime) = SAVE_FILES[i] if os.path.getmtime(dest) > mtime: savefile(dest, i, self.__options.X_ind) if self.__homedir and not self.__options.homedir: if self.__options.shred: self.shred(self.__homedir) shutil.rmtree(self.__homedir) if self.__tmpdir and not self.__options.tmpdir: if self.__options.shred: self.shred(self.__homedir) shutil.rmtree(self.__tmpdir) def shred(self, path): for root, dirs, files in os.walk(path): for f in files: dest = root + "/" + f subprocess.Popen(["/usr/bin/shred", dest]).wait() def main(self): try: self.__parse_options() self.__gen_context() if self.__mount: self.__setup_dir() return self.__execute() except KeyboardInterrupt: sys.exit(0) if __name__ == '__main__': setup_sighandlers() if selinux.is_selinux_enabled() != 1: error_exit("Requires an SELinux enabled system") try: sandbox = Sandbox() rc = sandbox.main() except OSError as error: error_exit(error) except ValueError as error: error_exit(error.args[0]) except KeyError as error: error_exit(_("Invalid value %s") % error.args[0]) except IOError as error: error_exit(error) except KeyboardInterrupt: rc = 0 sys.exit(rc)