2869 lines
103 KiB
Python
2869 lines
103 KiB
Python
# Copyright (C) 2005-2013 Red Hat
|
|
# see file 'COPYING' for use and warranty information
|
|
#
|
|
# semanage is a tool for managing SELinux configuration files
|
|
#
|
|
# 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., 59 Temple Place, Suite 330, Boston, MA
|
|
# 02111-1307 USA
|
|
#
|
|
#
|
|
|
|
import pwd
|
|
import grp
|
|
import selinux
|
|
import os
|
|
import re
|
|
import sys
|
|
import stat
|
|
import socket
|
|
from semanage import *
|
|
PROGNAME = "policycoreutils"
|
|
import sepolicy
|
|
import setools
|
|
import ipaddress
|
|
|
|
try:
|
|
import gettext
|
|
kwargs = {}
|
|
if sys.version_info < (3,):
|
|
kwargs['unicode'] = True
|
|
gettext.install(PROGNAME,
|
|
localedir="/usr/share/locale",
|
|
codeset='utf-8',
|
|
**kwargs)
|
|
except:
|
|
try:
|
|
import builtins
|
|
builtins.__dict__['_'] = str
|
|
except ImportError:
|
|
import __builtin__
|
|
__builtin__.__dict__['_'] = unicode
|
|
|
|
import syslog
|
|
|
|
file_types = {}
|
|
file_types[""] = SEMANAGE_FCONTEXT_ALL
|
|
file_types["all files"] = SEMANAGE_FCONTEXT_ALL
|
|
file_types["a"] = SEMANAGE_FCONTEXT_ALL
|
|
file_types["regular file"] = SEMANAGE_FCONTEXT_REG
|
|
file_types["--"] = SEMANAGE_FCONTEXT_REG
|
|
file_types["f"] = SEMANAGE_FCONTEXT_REG
|
|
file_types["-d"] = SEMANAGE_FCONTEXT_DIR
|
|
file_types["directory"] = SEMANAGE_FCONTEXT_DIR
|
|
file_types["d"] = SEMANAGE_FCONTEXT_DIR
|
|
file_types["-c"] = SEMANAGE_FCONTEXT_CHAR
|
|
file_types["character device"] = SEMANAGE_FCONTEXT_CHAR
|
|
file_types["c"] = SEMANAGE_FCONTEXT_CHAR
|
|
file_types["-b"] = SEMANAGE_FCONTEXT_BLOCK
|
|
file_types["block device"] = SEMANAGE_FCONTEXT_BLOCK
|
|
file_types["b"] = SEMANAGE_FCONTEXT_BLOCK
|
|
file_types["-s"] = SEMANAGE_FCONTEXT_SOCK
|
|
file_types["socket"] = SEMANAGE_FCONTEXT_SOCK
|
|
file_types["s"] = SEMANAGE_FCONTEXT_SOCK
|
|
file_types["-l"] = SEMANAGE_FCONTEXT_LINK
|
|
file_types["l"] = SEMANAGE_FCONTEXT_LINK
|
|
file_types["symbolic link"] = SEMANAGE_FCONTEXT_LINK
|
|
file_types["p"] = SEMANAGE_FCONTEXT_PIPE
|
|
file_types["-p"] = SEMANAGE_FCONTEXT_PIPE
|
|
file_types["named pipe"] = SEMANAGE_FCONTEXT_PIPE
|
|
|
|
file_type_str_to_option = {"all files": "a",
|
|
"regular file": "f",
|
|
"directory": "d",
|
|
"character device": "c",
|
|
"block device": "b",
|
|
"socket": "s",
|
|
"symbolic link": "l",
|
|
"named pipe": "p"}
|
|
|
|
ftype_to_audit = {"": "any",
|
|
"a" : "any",
|
|
"b": "block",
|
|
"c": "char",
|
|
"d": "dir",
|
|
"f": "file",
|
|
"l": "symlink",
|
|
"p": "pipe",
|
|
"s": "socket"}
|
|
|
|
try:
|
|
import audit
|
|
#test if audit module is enabled
|
|
audit.audit_close(audit.audit_open())
|
|
|
|
class logger:
|
|
|
|
def __init__(self):
|
|
self.audit_fd = audit.audit_open()
|
|
self.log_list = []
|
|
self.log_change_list = []
|
|
|
|
def log(self, msg, name="", sename="", serole="", serange="", oldsename="", oldserole="", oldserange=""):
|
|
|
|
sep = "-"
|
|
if sename != oldsename:
|
|
msg += sep + "sename"
|
|
sep = ","
|
|
if serole != oldserole:
|
|
msg += sep + "role"
|
|
sep = ","
|
|
if serange != oldserange:
|
|
msg += sep + "range"
|
|
sep = ","
|
|
|
|
self.log_list.append([self.audit_fd, audit.AUDIT_ROLE_ASSIGN, sys.argv[0], str(msg), name, 0, sename, serole, serange, oldsename, oldserole, oldserange, "", "", ""])
|
|
|
|
def log_remove(self, msg, name="", sename="", serole="", serange="", oldsename="", oldserole="", oldserange=""):
|
|
self.log_list.append([self.audit_fd, audit.AUDIT_ROLE_REMOVE, sys.argv[0], str(msg), name, 0, sename, serole, serange, oldsename, oldserole, oldserange, "", "", ""])
|
|
|
|
def log_change(self, msg):
|
|
self.log_change_list.append([self.audit_fd, audit.AUDIT_USER_MAC_CONFIG_CHANGE, str(msg), "semanage", "", "", ""])
|
|
|
|
def commit(self, success):
|
|
for l in self.log_list:
|
|
audit.audit_log_semanage_message(*(l + [success]))
|
|
for l in self.log_change_list:
|
|
audit.audit_log_user_comm_message(*(l + [success]))
|
|
|
|
self.log_list = []
|
|
self.log_change_list = []
|
|
except (OSError, ImportError):
|
|
class logger:
|
|
|
|
def __init__(self):
|
|
self.log_list = []
|
|
|
|
def log(self, msg, name="", sename="", serole="", serange="", oldsename="", oldserole="", oldserange=""):
|
|
message = " %s name=%s" % (msg, name)
|
|
if sename != "":
|
|
message += " sename=" + sename
|
|
if oldsename != "":
|
|
message += " oldsename=" + oldsename
|
|
if serole != "":
|
|
message += " role=" + serole
|
|
if oldserole != "":
|
|
message += " old_role=" + oldserole
|
|
if serange != "" and serange is not None:
|
|
message += " MLSRange=" + serange
|
|
if oldserange != "" and oldserange is not None:
|
|
message += " old_MLSRange=" + oldserange
|
|
self.log_list.append(message)
|
|
|
|
def log_remove(self, msg, name="", sename="", serole="", serange="", oldsename="", oldserole="", oldserange=""):
|
|
self.log(msg, name, sename, serole, serange, oldsename, oldserole, oldserange)
|
|
|
|
def log_change(self, msg):
|
|
self.log_list.append(" %s" % msg)
|
|
|
|
def commit(self, success):
|
|
if success == 1:
|
|
message = "Successful: "
|
|
else:
|
|
message = "Failed: "
|
|
for l in self.log_list:
|
|
syslog.syslog(syslog.LOG_INFO, message + l)
|
|
|
|
|
|
class nulllogger:
|
|
|
|
def log(self, msg, name="", sename="", serole="", serange="", oldsename="", oldserole="", oldserange=""):
|
|
pass
|
|
|
|
def log_remove(self, msg, name="", sename="", serole="", serange="", oldsename="", oldserole="", oldserange=""):
|
|
pass
|
|
|
|
def log_change(self, msg):
|
|
pass
|
|
|
|
def commit(self, success):
|
|
pass
|
|
|
|
|
|
def validate_level(raw):
|
|
sensitivity = "s[0-9]*"
|
|
category = "c[0-9]*"
|
|
cat_range = category + r"(\." + category + ")?"
|
|
categories = cat_range + r"(\," + cat_range + ")*"
|
|
reg = sensitivity + "(-" + sensitivity + ")?" + "(:" + categories + ")?"
|
|
return re.search("^" + reg + "$", raw)
|
|
|
|
|
|
def translate(raw, prepend=1):
|
|
filler = "a:b:c:"
|
|
if prepend == 1:
|
|
context = "%s%s" % (filler, raw)
|
|
else:
|
|
context = raw
|
|
(rc, trans) = selinux.selinux_raw_to_trans_context(context)
|
|
if rc != 0:
|
|
return raw
|
|
if prepend:
|
|
trans = trans[len(filler):]
|
|
if trans == "":
|
|
return raw
|
|
else:
|
|
return trans
|
|
|
|
|
|
def untranslate(trans, prepend=1):
|
|
filler = "a:b:c:"
|
|
if prepend == 1:
|
|
context = "%s%s" % (filler, trans)
|
|
else:
|
|
context = trans
|
|
|
|
(rc, raw) = selinux.selinux_trans_to_raw_context(context)
|
|
if rc != 0:
|
|
return trans
|
|
if prepend:
|
|
raw = raw[len(filler):]
|
|
if raw == "":
|
|
return trans
|
|
else:
|
|
return raw
|
|
|
|
|
|
class semanageRecords:
|
|
transaction = False
|
|
handle = None
|
|
store = None
|
|
args = None
|
|
|
|
def __init__(self, args = None):
|
|
global handle
|
|
if args:
|
|
# legacy code - args was store originally
|
|
if type(args) == str:
|
|
self.store = args
|
|
else:
|
|
self.args = args
|
|
self.noreload = getattr(args, "noreload", False)
|
|
if not self.store:
|
|
self.store = getattr(args, "store", "")
|
|
|
|
self.sh = self.get_handle(self.store)
|
|
|
|
rc, localstore = selinux.selinux_getpolicytype()
|
|
if self.store == "" or self.store == localstore:
|
|
self.mylog = logger()
|
|
else:
|
|
sepolicy.load_store_policy(self.store)
|
|
selinux.selinux_set_policy_root("%s%s" % (selinux.selinux_path(), self.store))
|
|
self.mylog = nulllogger()
|
|
|
|
def set_reload(self, load):
|
|
self.noreload = not load
|
|
|
|
def get_handle(self, store):
|
|
global is_mls_enabled
|
|
|
|
if semanageRecords.handle:
|
|
return semanageRecords.handle
|
|
|
|
handle = semanage_handle_create()
|
|
if not handle:
|
|
raise ValueError(_("Could not create semanage handle"))
|
|
|
|
if not semanageRecords.transaction and store != "":
|
|
semanage_select_store(handle, store, SEMANAGE_CON_DIRECT)
|
|
semanageRecords.store = store
|
|
|
|
if not semanage_is_managed(handle):
|
|
semanage_handle_destroy(handle)
|
|
raise ValueError(_("SELinux policy is not managed or store cannot be accessed."))
|
|
|
|
rc = semanage_access_check(handle)
|
|
if rc < SEMANAGE_CAN_READ:
|
|
semanage_handle_destroy(handle)
|
|
raise ValueError(_("Cannot read policy store."))
|
|
|
|
rc = semanage_connect(handle)
|
|
if rc < 0:
|
|
semanage_handle_destroy(handle)
|
|
raise ValueError(_("Could not establish semanage connection"))
|
|
|
|
is_mls_enabled = semanage_mls_enabled(handle)
|
|
if is_mls_enabled < 0:
|
|
semanage_handle_destroy(handle)
|
|
raise ValueError(_("Could not test MLS enabled status"))
|
|
|
|
semanageRecords.handle = handle
|
|
return semanageRecords.handle
|
|
|
|
def deleteall(self):
|
|
raise ValueError(_("Not yet implemented"))
|
|
|
|
def start(self):
|
|
if semanageRecords.transaction:
|
|
raise ValueError(_("Semanage transaction already in progress"))
|
|
self.begin()
|
|
semanageRecords.transaction = True
|
|
|
|
def begin(self):
|
|
if semanageRecords.transaction:
|
|
return
|
|
rc = semanage_begin_transaction(self.sh)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not start semanage transaction"))
|
|
|
|
def customized(self):
|
|
raise ValueError(_("Not yet implemented"))
|
|
|
|
def commit(self):
|
|
if semanageRecords.transaction:
|
|
return
|
|
|
|
if self.noreload:
|
|
semanage_set_reload(self.sh, 0)
|
|
rc = semanage_commit(self.sh)
|
|
if rc < 0:
|
|
self.mylog.commit(0)
|
|
raise ValueError(_("Could not commit semanage transaction"))
|
|
self.mylog.commit(1)
|
|
|
|
def finish(self):
|
|
if not semanageRecords.transaction:
|
|
raise ValueError(_("Semanage transaction not in progress"))
|
|
semanageRecords.transaction = False
|
|
self.commit()
|
|
|
|
|
|
class moduleRecords(semanageRecords):
|
|
|
|
def __init__(self, args = None):
|
|
semanageRecords.__init__(self, args)
|
|
|
|
def get_all(self):
|
|
l = []
|
|
(rc, mlist, number) = semanage_module_list_all(self.sh)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not list SELinux modules"))
|
|
|
|
for i in range(number):
|
|
mod = semanage_module_list_nth(mlist, i)
|
|
|
|
rc, name = semanage_module_info_get_name(self.sh, mod)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not get module name"))
|
|
|
|
rc, enabled = semanage_module_info_get_enabled(self.sh, mod)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not get module enabled"))
|
|
|
|
rc, priority = semanage_module_info_get_priority(self.sh, mod)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not get module priority"))
|
|
|
|
rc, lang_ext = semanage_module_info_get_lang_ext(self.sh, mod)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not get module lang_ext"))
|
|
|
|
l.append((name, enabled, priority, lang_ext))
|
|
|
|
# sort the list so they are in name order, but with higher priorities coming first
|
|
l.sort(key=lambda t: t[3], reverse=True)
|
|
l.sort(key=lambda t: t[0])
|
|
return l
|
|
|
|
def customized(self):
|
|
all = self.get_all()
|
|
if len(all) == 0:
|
|
return []
|
|
return ["-d %s" % x[0] for x in [t for t in all if t[1] == 0]]
|
|
|
|
def list(self, heading=1, locallist=0):
|
|
all = self.get_all()
|
|
if len(all) == 0:
|
|
return
|
|
|
|
if heading:
|
|
print("\n%-25s %-9s %s\n" % (_("Module Name"), _("Priority"), _("Language")))
|
|
for t in all:
|
|
if t[1] == 0:
|
|
disabled = _("Disabled")
|
|
else:
|
|
if locallist:
|
|
continue
|
|
disabled = ""
|
|
print("%-25s %-9s %-5s %s" % (t[0], t[2], t[3], disabled))
|
|
|
|
def add(self, file, priority):
|
|
if not os.path.exists(file):
|
|
raise ValueError(_("Module does not exist: %s ") % file)
|
|
|
|
rc = semanage_set_default_priority(self.sh, priority)
|
|
if rc < 0:
|
|
raise ValueError(_("Invalid priority %d (needs to be between 1 and 999)") % priority)
|
|
|
|
rc = semanage_module_install_file(self.sh, file)
|
|
if rc >= 0:
|
|
self.commit()
|
|
|
|
def set_enabled(self, module, enable):
|
|
for m in module.split():
|
|
rc, key = semanage_module_key_create(self.sh)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not create module key"))
|
|
|
|
rc = semanage_module_key_set_name(self.sh, key, m)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not set module key name"))
|
|
|
|
rc = semanage_module_set_enabled(self.sh, key, enable)
|
|
if rc < 0:
|
|
if enable:
|
|
raise ValueError(_("Could not enable module %s") % m)
|
|
else:
|
|
raise ValueError(_("Could not disable module %s") % m)
|
|
self.commit()
|
|
|
|
def delete(self, module, priority):
|
|
rc = semanage_set_default_priority(self.sh, priority)
|
|
if rc < 0:
|
|
raise ValueError(_("Invalid priority %d (needs to be between 1 and 999)") % priority)
|
|
|
|
for m in module.split():
|
|
rc = semanage_module_remove(self.sh, m)
|
|
if rc < 0 and rc != -2:
|
|
raise ValueError(_("Could not remove module %s (remove failed)") % m)
|
|
|
|
self.commit()
|
|
|
|
def deleteall(self):
|
|
l = [x[0] for x in [t for t in self.get_all() if t[1] == 0]]
|
|
for m in l:
|
|
self.set_enabled(m, True)
|
|
|
|
|
|
class dontauditClass(semanageRecords):
|
|
|
|
def __init__(self, args = None):
|
|
semanageRecords.__init__(self, args)
|
|
|
|
def toggle(self, dontaudit):
|
|
if dontaudit not in ["on", "off"]:
|
|
raise ValueError(_("dontaudit requires either 'on' or 'off'"))
|
|
self.begin()
|
|
semanage_set_disable_dontaudit(self.sh, dontaudit == "off")
|
|
self.commit()
|
|
|
|
|
|
class permissiveRecords(semanageRecords):
|
|
|
|
def __init__(self, args = None):
|
|
semanageRecords.__init__(self, args)
|
|
|
|
def get_all(self):
|
|
l = []
|
|
(rc, mlist, number) = semanage_module_list(self.sh)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not list SELinux modules"))
|
|
|
|
for i in range(number):
|
|
mod = semanage_module_list_nth(mlist, i)
|
|
name = semanage_module_get_name(mod)
|
|
if name and name.startswith("permissive_"):
|
|
l.append(name.split("permissive_")[1])
|
|
return l
|
|
|
|
def customized(self):
|
|
return ["-a %s" % x for x in sorted(self.get_all())]
|
|
|
|
def list(self, heading=1, locallist=0):
|
|
all = [y["name"] for y in [x for x in sepolicy.info(sepolicy.TYPE) if x["permissive"]]]
|
|
if len(all) == 0:
|
|
return
|
|
|
|
if heading:
|
|
print("\n%-25s\n" % (_("Builtin Permissive Types")))
|
|
customized = self.get_all()
|
|
for t in all:
|
|
if t not in customized:
|
|
print(t)
|
|
|
|
if len(customized) == 0:
|
|
return
|
|
|
|
if heading:
|
|
print("\n%-25s\n" % (_("Customized Permissive Types")))
|
|
for t in customized:
|
|
print(t)
|
|
|
|
def add(self, type):
|
|
try:
|
|
import sepolgen.module as module
|
|
except ImportError:
|
|
raise ValueError(_("The sepolgen python module is required to setup permissive domains.\nIn some distributions it is included in the policycoreutils-devel package.\n# yum install policycoreutils-devel\nOr similar for your distro."))
|
|
|
|
name = "permissive_%s" % type
|
|
modtxt = "(typepermissive %s)" % type
|
|
|
|
rc = semanage_module_install(self.sh, modtxt, len(modtxt), name, "cil")
|
|
if rc >= 0:
|
|
self.commit()
|
|
|
|
if rc < 0:
|
|
raise ValueError(_("Could not set permissive domain %s (module installation failed)") % name)
|
|
|
|
def delete(self, name):
|
|
for n in name.split():
|
|
rc = semanage_module_remove(self.sh, "permissive_%s" % n)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not remove permissive domain %s (remove failed)") % name)
|
|
|
|
self.commit()
|
|
|
|
def deleteall(self):
|
|
l = self.get_all()
|
|
if len(l) > 0:
|
|
all = " ".join(l)
|
|
self.delete(all)
|
|
|
|
|
|
class loginRecords(semanageRecords):
|
|
|
|
def __init__(self, args = None):
|
|
semanageRecords.__init__(self, args)
|
|
self.oldsename = None
|
|
self.oldserange = None
|
|
self.sename = None
|
|
self.serange = None
|
|
|
|
def __add(self, name, sename, serange):
|
|
rec, self.oldsename, self.oldserange = selinux.getseuserbyname(name)
|
|
if sename == "":
|
|
sename = "user_u"
|
|
|
|
userrec = seluserRecords(self.args)
|
|
range, (rc, oldserole) = userrec.get(self.oldsename)
|
|
range, (rc, serole) = userrec.get(sename)
|
|
|
|
if is_mls_enabled == 1:
|
|
if serange != "":
|
|
serange = untranslate(serange)
|
|
else:
|
|
serange = range
|
|
|
|
(rc, k) = semanage_seuser_key_create(self.sh, name)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not create a key for %s") % name)
|
|
|
|
(rc, exists) = semanage_seuser_exists(self.sh, k)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not check if login mapping for %s is defined") % name)
|
|
if exists:
|
|
raise ValueError(_("Login mapping for %s is already defined") % name)
|
|
if name[0] == '%':
|
|
try:
|
|
grp.getgrnam(name[1:])
|
|
except:
|
|
raise ValueError(_("Linux Group %s does not exist") % name[1:])
|
|
else:
|
|
try:
|
|
pwd.getpwnam(name)
|
|
except:
|
|
raise ValueError(_("Linux User %s does not exist") % name)
|
|
|
|
(rc, u) = semanage_seuser_create(self.sh)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not create login mapping for %s") % name)
|
|
|
|
rc = semanage_seuser_set_name(self.sh, u, name)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not set name for %s") % name)
|
|
|
|
if (is_mls_enabled == 1) and (serange != ""):
|
|
rc = semanage_seuser_set_mlsrange(self.sh, u, serange)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not set MLS range for %s") % name)
|
|
|
|
rc = semanage_seuser_set_sename(self.sh, u, sename)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not set SELinux user for %s") % name)
|
|
|
|
rc = semanage_seuser_modify_local(self.sh, k, u)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not add login mapping for %s") % name)
|
|
|
|
semanage_seuser_key_free(k)
|
|
semanage_seuser_free(u)
|
|
|
|
def add(self, name, sename, serange):
|
|
try:
|
|
self.begin()
|
|
self.__add(name, sename, serange)
|
|
self.commit()
|
|
except ValueError as error:
|
|
raise error
|
|
|
|
def __modify(self, name, sename="", serange=""):
|
|
rec, self.oldsename, self.oldserange = selinux.getseuserbyname(name)
|
|
if sename == "" and serange == "":
|
|
raise ValueError(_("Requires seuser or serange"))
|
|
|
|
userrec = seluserRecords(self.args)
|
|
range, (rc, oldserole) = userrec.get(self.oldsename)
|
|
|
|
if sename != "":
|
|
range, (rc, serole) = userrec.get(sename)
|
|
else:
|
|
serole = oldserole
|
|
|
|
if serange != "":
|
|
self.serange = serange
|
|
else:
|
|
self.serange = range
|
|
|
|
(rc, k) = semanage_seuser_key_create(self.sh, name)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not create a key for %s") % name)
|
|
|
|
(rc, exists) = semanage_seuser_exists(self.sh, k)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not check if login mapping for %s is defined") % name)
|
|
if not exists:
|
|
raise ValueError(_("Login mapping for %s is not defined") % name)
|
|
|
|
(rc, u) = semanage_seuser_query(self.sh, k)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not query seuser for %s") % name)
|
|
|
|
self.oldserange = semanage_seuser_get_mlsrange(u)
|
|
self.oldsename = semanage_seuser_get_sename(u)
|
|
if (is_mls_enabled == 1) and (serange != ""):
|
|
semanage_seuser_set_mlsrange(self.sh, u, untranslate(serange))
|
|
|
|
if sename != "":
|
|
semanage_seuser_set_sename(self.sh, u, sename)
|
|
self.sename = sename
|
|
else:
|
|
self.sename = self.oldsename
|
|
|
|
rc = semanage_seuser_modify_local(self.sh, k, u)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not modify login mapping for %s") % name)
|
|
|
|
semanage_seuser_key_free(k)
|
|
semanage_seuser_free(u)
|
|
|
|
def modify(self, name, sename="", serange=""):
|
|
try:
|
|
self.begin()
|
|
self.__modify(name, sename, serange)
|
|
self.commit()
|
|
except ValueError as error:
|
|
raise error
|
|
|
|
def __delete(self, name):
|
|
rec, self.oldsename, self.oldserange = selinux.getseuserbyname(name)
|
|
userrec = seluserRecords(self.args)
|
|
range, (rc, oldserole) = userrec.get(self.oldsename)
|
|
|
|
(rc, k) = semanage_seuser_key_create(self.sh, name)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not create a key for %s") % name)
|
|
|
|
(rc, exists) = semanage_seuser_exists(self.sh, k)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not check if login mapping for %s is defined") % name)
|
|
if not exists:
|
|
raise ValueError(_("Login mapping for %s is not defined") % name)
|
|
|
|
(rc, exists) = semanage_seuser_exists_local(self.sh, k)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not check if login mapping for %s is defined") % name)
|
|
if not exists:
|
|
raise ValueError(_("Login mapping for %s is defined in policy, cannot be deleted") % name)
|
|
|
|
rc = semanage_seuser_del_local(self.sh, k)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not delete login mapping for %s") % name)
|
|
|
|
semanage_seuser_key_free(k)
|
|
|
|
rec, self.sename, self.serange = selinux.getseuserbyname("__default__")
|
|
range, (rc, serole) = userrec.get(self.sename)
|
|
|
|
def delete(self, name):
|
|
try:
|
|
self.begin()
|
|
self.__delete(name)
|
|
self.commit()
|
|
|
|
except ValueError as error:
|
|
raise error
|
|
|
|
def deleteall(self):
|
|
(rc, ulist) = semanage_seuser_list_local(self.sh)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not list login mappings"))
|
|
|
|
try:
|
|
self.begin()
|
|
for u in ulist:
|
|
self.__delete(semanage_seuser_get_name(u))
|
|
self.commit()
|
|
except ValueError as error:
|
|
raise error
|
|
|
|
def get_all_logins(self):
|
|
ddict = {}
|
|
self.logins_path = selinux.selinux_policy_root() + "/logins"
|
|
for path, dirs, files in os.walk(self.logins_path):
|
|
if path == self.logins_path:
|
|
for name in files:
|
|
try:
|
|
fd = open(path + "/" + name)
|
|
rec = fd.read().rstrip().split(":")
|
|
fd.close()
|
|
ddict[name] = (rec[1], rec[2], rec[0])
|
|
except IndexError:
|
|
pass
|
|
return ddict
|
|
|
|
def get_all(self, locallist=0):
|
|
ddict = {}
|
|
if locallist:
|
|
(rc, self.ulist) = semanage_seuser_list_local(self.sh)
|
|
else:
|
|
(rc, self.ulist) = semanage_seuser_list(self.sh)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not list login mappings"))
|
|
|
|
for u in self.ulist:
|
|
name = semanage_seuser_get_name(u)
|
|
ddict[name] = (semanage_seuser_get_sename(u), semanage_seuser_get_mlsrange(u), "*")
|
|
return ddict
|
|
|
|
def customized(self):
|
|
l = []
|
|
ddict = self.get_all(True)
|
|
for k in sorted(ddict.keys()):
|
|
if ddict[k][1]:
|
|
l.append("-a -s %s -r '%s' %s" % (ddict[k][0], ddict[k][1], k))
|
|
else:
|
|
l.append("-a -s %s %s" % (ddict[k][0], k))
|
|
return l
|
|
|
|
def list(self, heading=1, locallist=0):
|
|
ddict = self.get_all(locallist)
|
|
ldict = self.get_all_logins()
|
|
lkeys = sorted(ldict.keys())
|
|
keys = sorted(ddict.keys())
|
|
if len(keys) == 0 and len(lkeys) == 0:
|
|
return
|
|
|
|
if is_mls_enabled == 1:
|
|
if heading:
|
|
print("\n%-20s %-20s %-20s %s\n" % (_("Login Name"), _("SELinux User"), _("MLS/MCS Range"), _("Service")))
|
|
for k in keys:
|
|
u = ddict[k]
|
|
print("%-20s %-20s %-20s %s" % (k, u[0], translate(u[1]), u[2]))
|
|
if len(lkeys):
|
|
print("\nLocal customization in %s" % self.logins_path)
|
|
|
|
for k in lkeys:
|
|
u = ldict[k]
|
|
print("%-20s %-20s %-20s %s" % (k, u[0], translate(u[1]), u[2]))
|
|
else:
|
|
if heading:
|
|
print("\n%-25s %-25s\n" % (_("Login Name"), _("SELinux User")))
|
|
for k in keys:
|
|
print("%-25s %-25s" % (k, ddict[k][0]))
|
|
|
|
|
|
class seluserRecords(semanageRecords):
|
|
|
|
def __init__(self, args = None):
|
|
semanageRecords.__init__(self, args)
|
|
|
|
def get(self, name):
|
|
(rc, k) = semanage_user_key_create(self.sh, name)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not create a key for %s") % name)
|
|
(rc, exists) = semanage_user_exists(self.sh, k)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not check if SELinux user %s is defined") % name)
|
|
(rc, u) = semanage_user_query(self.sh, k)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not query user for %s") % name)
|
|
serange = semanage_user_get_mlsrange(u)
|
|
serole = semanage_user_get_roles(self.sh, u)
|
|
semanage_user_key_free(k)
|
|
semanage_user_free(u)
|
|
return serange, serole
|
|
|
|
def __add(self, name, roles, selevel, serange, prefix):
|
|
if is_mls_enabled == 1:
|
|
if serange == "":
|
|
serange = "s0"
|
|
else:
|
|
serange = untranslate(serange)
|
|
|
|
if selevel == "":
|
|
selevel = "s0"
|
|
else:
|
|
selevel = untranslate(selevel)
|
|
|
|
if len(roles) < 1:
|
|
raise ValueError(_("You must add at least one role for %s") % name)
|
|
|
|
(rc, k) = semanage_user_key_create(self.sh, name)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not create a key for %s") % name)
|
|
|
|
(rc, exists) = semanage_user_exists(self.sh, k)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not check if SELinux user %s is defined") % name)
|
|
if exists:
|
|
raise ValueError(_("SELinux user %s is already defined") % name)
|
|
|
|
(rc, u) = semanage_user_create(self.sh)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not create SELinux user for %s") % name)
|
|
|
|
rc = semanage_user_set_name(self.sh, u, name)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not set name for %s") % name)
|
|
|
|
for r in roles:
|
|
rc = semanage_user_add_role(self.sh, u, r)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not add role %s for %s") % (r, name))
|
|
|
|
if is_mls_enabled == 1:
|
|
rc = semanage_user_set_mlsrange(self.sh, u, serange)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not set MLS range for %s") % name)
|
|
|
|
rc = semanage_user_set_mlslevel(self.sh, u, selevel)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not set MLS level for %s") % name)
|
|
rc = semanage_user_set_prefix(self.sh, u, prefix)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not add prefix %s for %s") % (r, prefix))
|
|
(rc, key) = semanage_user_key_extract(self.sh, u)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not extract key for %s") % name)
|
|
|
|
rc = semanage_user_modify_local(self.sh, k, u)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not add SELinux user %s") % name)
|
|
|
|
semanage_user_key_free(k)
|
|
semanage_user_free(u)
|
|
self.mylog.log("seuser", sename=name, serole=",".join(roles), serange=serange)
|
|
|
|
def add(self, name, roles, selevel, serange, prefix):
|
|
try:
|
|
self.begin()
|
|
self.__add(name, roles, selevel, serange, prefix)
|
|
self.commit()
|
|
except ValueError as error:
|
|
self.mylog.commit(0)
|
|
raise error
|
|
|
|
def __modify(self, name, roles=[], selevel="", serange="", prefix=""):
|
|
oldserole = ""
|
|
oldserange = ""
|
|
newroles = " ".join(roles)
|
|
if prefix == "" and len(roles) == 0 and serange == "" and selevel == "":
|
|
if is_mls_enabled == 1:
|
|
raise ValueError(_("Requires prefix, roles, level or range"))
|
|
else:
|
|
raise ValueError(_("Requires prefix or roles"))
|
|
|
|
(rc, k) = semanage_user_key_create(self.sh, name)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not create a key for %s") % name)
|
|
|
|
(rc, exists) = semanage_user_exists(self.sh, k)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not check if SELinux user %s is defined") % name)
|
|
if not exists:
|
|
raise ValueError(_("SELinux user %s is not defined") % name)
|
|
|
|
(rc, u) = semanage_user_query(self.sh, k)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not query user for %s") % name)
|
|
|
|
oldserange = semanage_user_get_mlsrange(u)
|
|
(rc, rlist) = semanage_user_get_roles(self.sh, u)
|
|
if rc >= 0:
|
|
oldserole = " ".join(rlist)
|
|
|
|
if (is_mls_enabled == 1) and (serange != ""):
|
|
semanage_user_set_mlsrange(self.sh, u, untranslate(serange))
|
|
if (is_mls_enabled == 1) and (selevel != ""):
|
|
semanage_user_set_mlslevel(self.sh, u, untranslate(selevel))
|
|
|
|
if prefix != "":
|
|
semanage_user_set_prefix(self.sh, u, prefix)
|
|
|
|
if len(roles) != 0:
|
|
for r in rlist:
|
|
if r not in roles:
|
|
semanage_user_del_role(u, r)
|
|
for r in roles:
|
|
if r not in rlist:
|
|
semanage_user_add_role(self.sh, u, r)
|
|
|
|
rc = semanage_user_modify_local(self.sh, k, u)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not modify SELinux user %s") % name)
|
|
|
|
semanage_user_key_free(k)
|
|
semanage_user_free(u)
|
|
|
|
role = ",".join(newroles.split())
|
|
oldserole = ",".join(oldserole.split())
|
|
self.mylog.log("seuser", sename=name, oldsename=name, serole=role, serange=serange, oldserole=oldserole, oldserange=oldserange)
|
|
|
|
def modify(self, name, roles=[], selevel="", serange="", prefix=""):
|
|
try:
|
|
self.begin()
|
|
self.__modify(name, roles, selevel, serange, prefix)
|
|
self.commit()
|
|
except ValueError as error:
|
|
self.mylog.commit(0)
|
|
raise error
|
|
|
|
def __delete(self, name):
|
|
(rc, k) = semanage_user_key_create(self.sh, name)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not create a key for %s") % name)
|
|
|
|
(rc, exists) = semanage_user_exists(self.sh, k)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not check if SELinux user %s is defined") % name)
|
|
if not exists:
|
|
raise ValueError(_("SELinux user %s is not defined") % name)
|
|
|
|
(rc, exists) = semanage_user_exists_local(self.sh, k)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not check if SELinux user %s is defined") % name)
|
|
if not exists:
|
|
raise ValueError(_("SELinux user %s is defined in policy, cannot be deleted") % name)
|
|
|
|
(rc, u) = semanage_user_query(self.sh, k)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not query user for %s") % name)
|
|
oldserange = semanage_user_get_mlsrange(u)
|
|
(rc, rlist) = semanage_user_get_roles(self.sh, u)
|
|
oldserole = ",".join(rlist)
|
|
|
|
rc = semanage_user_del_local(self.sh, k)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not delete SELinux user %s") % name)
|
|
|
|
semanage_user_key_free(k)
|
|
semanage_user_free(u)
|
|
|
|
self.mylog.log_remove("seuser", oldsename=name, oldserange=oldserange, oldserole=oldserole)
|
|
|
|
def delete(self, name):
|
|
try:
|
|
self.begin()
|
|
self.__delete(name)
|
|
self.commit()
|
|
|
|
except ValueError as error:
|
|
self.mylog.commit(0)
|
|
raise error
|
|
|
|
def deleteall(self):
|
|
(rc, ulist) = semanage_user_list_local(self.sh)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not list login mappings"))
|
|
|
|
try:
|
|
self.begin()
|
|
for u in ulist:
|
|
self.__delete(semanage_user_get_name(u))
|
|
self.commit()
|
|
except ValueError as error:
|
|
self.mylog.commit(0)
|
|
raise error
|
|
|
|
def get_all(self, locallist=0):
|
|
ddict = {}
|
|
if locallist:
|
|
(rc, self.ulist) = semanage_user_list_local(self.sh)
|
|
else:
|
|
(rc, self.ulist) = semanage_user_list(self.sh)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not list SELinux users"))
|
|
|
|
for u in self.ulist:
|
|
name = semanage_user_get_name(u)
|
|
(rc, rlist) = semanage_user_get_roles(self.sh, u)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not list roles for user %s") % name)
|
|
|
|
roles = " ".join(rlist)
|
|
ddict[semanage_user_get_name(u)] = (semanage_user_get_prefix(u), semanage_user_get_mlslevel(u), semanage_user_get_mlsrange(u), roles)
|
|
|
|
return ddict
|
|
|
|
def customized(self):
|
|
l = []
|
|
ddict = self.get_all(True)
|
|
for k in sorted(ddict.keys()):
|
|
if ddict[k][1] or ddict[k][2]:
|
|
l.append("-a -L %s -r %s -R '%s' %s" % (ddict[k][1], ddict[k][2], ddict[k][3], k))
|
|
else:
|
|
l.append("-a -R '%s' %s" % (ddict[k][3], k))
|
|
return l
|
|
|
|
def list(self, heading=1, locallist=0):
|
|
ddict = self.get_all(locallist)
|
|
if len(ddict) == 0:
|
|
return
|
|
keys = sorted(ddict.keys())
|
|
|
|
if is_mls_enabled == 1:
|
|
if heading:
|
|
print("\n%-15s %-10s %-10s %-30s" % ("", _("Labeling"), _("MLS/"), _("MLS/")))
|
|
print("%-15s %-10s %-10s %-30s %s\n" % (_("SELinux User"), _("Prefix"), _("MCS Level"), _("MCS Range"), _("SELinux Roles")))
|
|
for k in keys:
|
|
print("%-15s %-10s %-10s %-30s %s" % (k, ddict[k][0], translate(ddict[k][1]), translate(ddict[k][2]), ddict[k][3]))
|
|
else:
|
|
if heading:
|
|
print("%-15s %s\n" % (_("SELinux User"), _("SELinux Roles")))
|
|
for k in keys:
|
|
print("%-15s %s" % (k, ddict[k][3]))
|
|
|
|
|
|
class portRecords(semanageRecords):
|
|
|
|
valid_types = []
|
|
|
|
def __init__(self, args = None):
|
|
semanageRecords.__init__(self, args)
|
|
try:
|
|
self.valid_types = list(list(sepolicy.info(sepolicy.ATTRIBUTE, "port_type"))[0]["types"])
|
|
except RuntimeError:
|
|
pass
|
|
|
|
def __genkey(self, port, proto):
|
|
protocols = {"tcp": SEMANAGE_PROTO_TCP,
|
|
"udp": SEMANAGE_PROTO_UDP,
|
|
"sctp": SEMANAGE_PROTO_SCTP,
|
|
"dccp": SEMANAGE_PROTO_DCCP}
|
|
|
|
if proto in protocols.keys():
|
|
proto_d = protocols[proto]
|
|
else:
|
|
raise ValueError(_("Protocol has to be one of udp, tcp, dccp or sctp"))
|
|
if port == "":
|
|
raise ValueError(_("Port is required"))
|
|
|
|
if isinstance(port, str):
|
|
ports = port.split('-', 1)
|
|
else:
|
|
ports = (port,)
|
|
|
|
if len(ports) == 1:
|
|
high = low = int(ports[0])
|
|
else:
|
|
low = int(ports[0])
|
|
high = int(ports[1])
|
|
|
|
if high > 65535:
|
|
raise ValueError(_("Invalid Port"))
|
|
|
|
(rc, k) = semanage_port_key_create(self.sh, low, high, proto_d)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not create a key for %s/%s") % (proto, port))
|
|
return (k, proto_d, low, high)
|
|
|
|
def __add(self, port, proto, serange, type):
|
|
if is_mls_enabled == 1:
|
|
if serange == "":
|
|
serange = "s0"
|
|
else:
|
|
serange = untranslate(serange)
|
|
|
|
if type == "":
|
|
raise ValueError(_("Type is required"))
|
|
|
|
type = sepolicy.get_real_type_name(type)
|
|
|
|
if type not in self.valid_types:
|
|
raise ValueError(_("Type %s is invalid, must be a port type") % type)
|
|
|
|
(k, proto_d, low, high) = self.__genkey(port, proto)
|
|
|
|
(rc, exists) = semanage_port_exists(self.sh, k)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not check if port %s/%s is defined") % (proto, port))
|
|
if exists:
|
|
raise ValueError(_("Port %s/%s already defined") % (proto, port))
|
|
|
|
(rc, p) = semanage_port_create(self.sh)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not create port for %s/%s") % (proto, port))
|
|
|
|
semanage_port_set_proto(p, proto_d)
|
|
semanage_port_set_range(p, low, high)
|
|
(rc, con) = semanage_context_create(self.sh)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not create context for %s/%s") % (proto, port))
|
|
|
|
rc = semanage_context_set_user(self.sh, con, "system_u")
|
|
if rc < 0:
|
|
raise ValueError(_("Could not set user in port context for %s/%s") % (proto, port))
|
|
|
|
rc = semanage_context_set_role(self.sh, con, "object_r")
|
|
if rc < 0:
|
|
raise ValueError(_("Could not set role in port context for %s/%s") % (proto, port))
|
|
|
|
rc = semanage_context_set_type(self.sh, con, type)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not set type in port context for %s/%s") % (proto, port))
|
|
|
|
if (is_mls_enabled == 1) and (serange != ""):
|
|
rc = semanage_context_set_mls(self.sh, con, serange)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not set mls fields in port context for %s/%s") % (proto, port))
|
|
|
|
rc = semanage_port_set_con(self.sh, p, con)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not set port context for %s/%s") % (proto, port))
|
|
|
|
rc = semanage_port_modify_local(self.sh, k, p)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not add port %s/%s") % (proto, port))
|
|
|
|
semanage_context_free(con)
|
|
semanage_port_key_free(k)
|
|
semanage_port_free(p)
|
|
|
|
self.mylog.log_change("resrc=port op=add lport=%s proto=%s tcontext=%s:%s:%s:%s" % (port, socket.getprotobyname(proto), "system_u", "object_r", type, serange))
|
|
|
|
def add(self, port, proto, serange, type):
|
|
self.begin()
|
|
self.__add(port, proto, serange, type)
|
|
self.commit()
|
|
|
|
def __modify(self, port, proto, serange, setype):
|
|
if serange == "" and setype == "":
|
|
if is_mls_enabled == 1:
|
|
raise ValueError(_("Requires setype or serange"))
|
|
else:
|
|
raise ValueError(_("Requires setype"))
|
|
|
|
setype = sepolicy.get_real_type_name(setype)
|
|
if setype and setype not in self.valid_types:
|
|
raise ValueError(_("Type %s is invalid, must be a port type") % setype)
|
|
|
|
(k, proto_d, low, high) = self.__genkey(port, proto)
|
|
|
|
(rc, exists) = semanage_port_exists(self.sh, k)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not check if port %s/%s is defined") % (proto, port))
|
|
if not exists:
|
|
raise ValueError(_("Port %s/%s is not defined") % (proto, port))
|
|
|
|
(rc, p) = semanage_port_query(self.sh, k)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not query port %s/%s") % (proto, port))
|
|
|
|
con = semanage_port_get_con(p)
|
|
|
|
if is_mls_enabled == 1:
|
|
if serange == "":
|
|
serange = "s0"
|
|
else:
|
|
semanage_context_set_mls(self.sh, con, untranslate(serange))
|
|
if setype != "":
|
|
semanage_context_set_type(self.sh, con, setype)
|
|
|
|
rc = semanage_port_modify_local(self.sh, k, p)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not modify port %s/%s") % (proto, port))
|
|
|
|
semanage_port_key_free(k)
|
|
semanage_port_free(p)
|
|
|
|
self.mylog.log_change("resrc=port op=modify lport=%s proto=%s tcontext=%s:%s:%s:%s" % (port, socket.getprotobyname(proto), "system_u", "object_r", setype, serange))
|
|
|
|
def modify(self, port, proto, serange, setype):
|
|
self.begin()
|
|
self.__modify(port, proto, serange, setype)
|
|
self.commit()
|
|
|
|
def deleteall(self):
|
|
(rc, plist) = semanage_port_list_local(self.sh)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not list the ports"))
|
|
|
|
self.begin()
|
|
|
|
for port in plist:
|
|
proto = semanage_port_get_proto(port)
|
|
proto_str = semanage_port_get_proto_str(proto)
|
|
low = semanage_port_get_low(port)
|
|
high = semanage_port_get_high(port)
|
|
port_str = "%s-%s" % (low, high)
|
|
|
|
(k, proto_d, low, high) = self.__genkey(port_str, proto_str)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not create a key for %s") % port_str)
|
|
|
|
rc = semanage_port_del_local(self.sh, k)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not delete the port %s") % port_str)
|
|
semanage_port_key_free(k)
|
|
|
|
if low == high:
|
|
port_str = low
|
|
|
|
self.mylog.log_change("resrc=port op=delete lport=%s proto=%s" % (port_str, socket.getprotobyname(proto_str)))
|
|
|
|
self.commit()
|
|
|
|
def __delete(self, port, proto):
|
|
(k, proto_d, low, high) = self.__genkey(port, proto)
|
|
(rc, exists) = semanage_port_exists(self.sh, k)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not check if port %s/%s is defined") % (proto, port))
|
|
if not exists:
|
|
raise ValueError(_("Port %s/%s is not defined") % (proto, port))
|
|
|
|
(rc, exists) = semanage_port_exists_local(self.sh, k)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not check if port %s/%s is defined") % (proto, port))
|
|
if not exists:
|
|
raise ValueError(_("Port %s/%s is defined in policy, cannot be deleted") % (proto, port))
|
|
|
|
rc = semanage_port_del_local(self.sh, k)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not delete port %s/%s") % (proto, port))
|
|
|
|
semanage_port_key_free(k)
|
|
|
|
self.mylog.log_change("resrc=port op=delete lport=%s proto=%s" % (port, socket.getprotobyname(proto)))
|
|
|
|
def delete(self, port, proto):
|
|
self.begin()
|
|
self.__delete(port, proto)
|
|
self.commit()
|
|
|
|
def get_all(self, locallist=0):
|
|
ddict = {}
|
|
if locallist:
|
|
(rc, self.plist) = semanage_port_list_local(self.sh)
|
|
else:
|
|
(rc, self.plist) = semanage_port_list(self.sh)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not list ports"))
|
|
|
|
for port in self.plist:
|
|
con = semanage_port_get_con(port)
|
|
ctype = semanage_context_get_type(con)
|
|
level = semanage_context_get_mls(con)
|
|
proto = semanage_port_get_proto(port)
|
|
proto_str = semanage_port_get_proto_str(proto)
|
|
low = semanage_port_get_low(port)
|
|
high = semanage_port_get_high(port)
|
|
ddict[(low, high, proto_str)] = (ctype, level)
|
|
return ddict
|
|
|
|
def get_all_by_type(self, locallist=0):
|
|
ddict = {}
|
|
if locallist:
|
|
(rc, self.plist) = semanage_port_list_local(self.sh)
|
|
else:
|
|
(rc, self.plist) = semanage_port_list(self.sh)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not list ports"))
|
|
|
|
for port in self.plist:
|
|
con = semanage_port_get_con(port)
|
|
ctype = semanage_context_get_type(con)
|
|
proto = semanage_port_get_proto(port)
|
|
proto_str = semanage_port_get_proto_str(proto)
|
|
low = semanage_port_get_low(port)
|
|
high = semanage_port_get_high(port)
|
|
if (ctype, proto_str) not in ddict.keys():
|
|
ddict[(ctype, proto_str)] = []
|
|
if low == high:
|
|
ddict[(ctype, proto_str)].append("%d" % low)
|
|
else:
|
|
ddict[(ctype, proto_str)].append("%d-%d" % (low, high))
|
|
return ddict
|
|
|
|
def customized(self):
|
|
l = []
|
|
ddict = self.get_all(True)
|
|
for k in sorted(ddict.keys()):
|
|
port = k[0] if k[0] == k[1] else "%s-%s" % (k[0], k[1])
|
|
if ddict[k][1]:
|
|
l.append("-a -t %s -r '%s' -p %s %s" % (ddict[k][0], ddict[k][1], k[2], port))
|
|
else:
|
|
l.append("-a -t %s -p %s %s" % (ddict[k][0], k[2], port))
|
|
return l
|
|
|
|
def list(self, heading=1, locallist=0):
|
|
ddict = self.get_all_by_type(locallist)
|
|
if len(ddict) == 0:
|
|
return
|
|
keys = sorted(ddict.keys())
|
|
|
|
if heading:
|
|
print("%-30s %-8s %s\n" % (_("SELinux Port Type"), _("Proto"), _("Port Number")))
|
|
for i in keys:
|
|
rec = "%-30s %-8s " % i
|
|
rec += "%s" % ddict[i][0]
|
|
for p in ddict[i][1:]:
|
|
rec += ", %s" % p
|
|
print(rec)
|
|
|
|
class ibpkeyRecords(semanageRecords):
|
|
|
|
valid_types = []
|
|
|
|
def __init__(self, args = None):
|
|
semanageRecords.__init__(self, args)
|
|
try:
|
|
q = setools.TypeQuery(setools.SELinuxPolicy(sepolicy.get_store_policy(self.store)), attrs=["ibpkey_type"])
|
|
self.valid_types = sorted(str(t) for t in q.results())
|
|
except:
|
|
pass
|
|
|
|
def __genkey(self, pkey, subnet_prefix):
|
|
if subnet_prefix == "":
|
|
raise ValueError(_("Subnet Prefix is required"))
|
|
|
|
pkeys = pkey.split("-")
|
|
if len(pkeys) == 1:
|
|
high = low = int(pkeys[0], 0)
|
|
else:
|
|
low = int(pkeys[0], 0)
|
|
high = int(pkeys[1], 0)
|
|
|
|
if high > 65535:
|
|
raise ValueError(_("Invalid Pkey"))
|
|
|
|
(rc, k) = semanage_ibpkey_key_create(self.sh, subnet_prefix, low, high)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not create a key for %s/%s") % (subnet_prefix, pkey))
|
|
return (k, subnet_prefix, low, high)
|
|
|
|
def __add(self, pkey, subnet_prefix, serange, type):
|
|
if is_mls_enabled == 1:
|
|
if serange == "":
|
|
serange = "s0"
|
|
else:
|
|
serange = untranslate(serange)
|
|
|
|
if type == "":
|
|
raise ValueError(_("Type is required"))
|
|
|
|
type = sepolicy.get_real_type_name(type)
|
|
|
|
if type not in self.valid_types:
|
|
raise ValueError(_("Type %s is invalid, must be a ibpkey type") % type)
|
|
|
|
(k, subnet_prefix, low, high) = self.__genkey(pkey, subnet_prefix)
|
|
|
|
(rc, exists) = semanage_ibpkey_exists(self.sh, k)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not check if ibpkey %s/%s is defined") % (subnet_prefix, pkey))
|
|
if exists:
|
|
raise ValueError(_("ibpkey %s/%s already defined") % (subnet_prefix, pkey))
|
|
|
|
(rc, p) = semanage_ibpkey_create(self.sh)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not create ibpkey for %s/%s") % (subnet_prefix, pkey))
|
|
|
|
semanage_ibpkey_set_subnet_prefix(self.sh, p, subnet_prefix)
|
|
semanage_ibpkey_set_range(p, low, high)
|
|
(rc, con) = semanage_context_create(self.sh)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not create context for %s/%s") % (subnet_prefix, pkey))
|
|
|
|
rc = semanage_context_set_user(self.sh, con, "system_u")
|
|
if rc < 0:
|
|
raise ValueError(_("Could not set user in ibpkey context for %s/%s") % (subnet_prefix, pkey))
|
|
|
|
rc = semanage_context_set_role(self.sh, con, "object_r")
|
|
if rc < 0:
|
|
raise ValueError(_("Could not set role in ibpkey context for %s/%s") % (subnet_prefix, pkey))
|
|
|
|
rc = semanage_context_set_type(self.sh, con, type)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not set type in ibpkey context for %s/%s") % (subnet_prefix, pkey))
|
|
|
|
if (is_mls_enabled == 1) and (serange != ""):
|
|
rc = semanage_context_set_mls(self.sh, con, serange)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not set mls fields in ibpkey context for %s/%s") % (subnet_prefix, pkey))
|
|
|
|
rc = semanage_ibpkey_set_con(self.sh, p, con)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not set ibpkey context for %s/%s") % (subnet_prefix, pkey))
|
|
|
|
rc = semanage_ibpkey_modify_local(self.sh, k, p)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not add ibpkey %s/%s") % (subnet_prefix, pkey))
|
|
|
|
semanage_context_free(con)
|
|
semanage_ibpkey_key_free(k)
|
|
semanage_ibpkey_free(p)
|
|
|
|
def add(self, pkey, subnet_prefix, serange, type):
|
|
self.begin()
|
|
self.__add(pkey, subnet_prefix, serange, type)
|
|
self.commit()
|
|
|
|
def __modify(self, pkey, subnet_prefix, serange, setype):
|
|
if serange == "" and setype == "":
|
|
if is_mls_enabled == 1:
|
|
raise ValueError(_("Requires setype or serange"))
|
|
else:
|
|
raise ValueError(_("Requires setype"))
|
|
|
|
setype = sepolicy.get_real_type_name(setype)
|
|
|
|
if setype and setype not in self.valid_types:
|
|
raise ValueError(_("Type %s is invalid, must be a ibpkey type") % setype)
|
|
|
|
(k, subnet_prefix, low, high) = self.__genkey(pkey, subnet_prefix)
|
|
|
|
(rc, exists) = semanage_ibpkey_exists(self.sh, k)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not check if ibpkey %s/%s is defined") % (subnet_prefix, pkey))
|
|
if not exists:
|
|
raise ValueError(_("ibpkey %s/%s is not defined") % (subnet_prefix, pkey))
|
|
|
|
(rc, p) = semanage_ibpkey_query(self.sh, k)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not query ibpkey %s/%s") % (subnet_prefix, pkey))
|
|
|
|
con = semanage_ibpkey_get_con(p)
|
|
|
|
if (is_mls_enabled == 1) and (serange != ""):
|
|
semanage_context_set_mls(self.sh, con, untranslate(serange))
|
|
if setype != "":
|
|
semanage_context_set_type(self.sh, con, setype)
|
|
|
|
rc = semanage_ibpkey_modify_local(self.sh, k, p)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not modify ibpkey %s/%s") % (subnet_prefix, pkey))
|
|
|
|
semanage_ibpkey_key_free(k)
|
|
semanage_ibpkey_free(p)
|
|
|
|
def modify(self, pkey, subnet_prefix, serange, setype):
|
|
self.begin()
|
|
self.__modify(pkey, subnet_prefix, serange, setype)
|
|
self.commit()
|
|
|
|
def deleteall(self):
|
|
(rc, plist) = semanage_ibpkey_list_local(self.sh)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not list the ibpkeys"))
|
|
|
|
self.begin()
|
|
|
|
for ibpkey in plist:
|
|
(rc, subnet_prefix) = semanage_ibpkey_get_subnet_prefix(self.sh, ibpkey)
|
|
low = semanage_ibpkey_get_low(ibpkey)
|
|
high = semanage_ibpkey_get_high(ibpkey)
|
|
pkey_str = "%s-%s" % (low, high)
|
|
(k, subnet_prefix, low, high) = self.__genkey(pkey_str, subnet_prefix)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not create a key for %s") % pkey_str)
|
|
|
|
rc = semanage_ibpkey_del_local(self.sh, k)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not delete the ibpkey %s") % pkey_str)
|
|
semanage_ibpkey_key_free(k)
|
|
|
|
self.commit()
|
|
|
|
def __delete(self, pkey, subnet_prefix):
|
|
(k, subnet_prefix, low, high) = self.__genkey(pkey, subnet_prefix)
|
|
(rc, exists) = semanage_ibpkey_exists(self.sh, k)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not check if ibpkey %s/%s is defined") % (subnet_prefix, pkey))
|
|
if not exists:
|
|
raise ValueError(_("ibpkey %s/%s is not defined") % (subnet_prefix, pkey))
|
|
|
|
(rc, exists) = semanage_ibpkey_exists_local(self.sh, k)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not check if ibpkey %s/%s is defined") % (subnet_prefix, pkey))
|
|
if not exists:
|
|
raise ValueError(_("ibpkey %s/%s is defined in policy, cannot be deleted") % (subnet_prefix, pkey))
|
|
|
|
rc = semanage_ibpkey_del_local(self.sh, k)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not delete ibpkey %s/%s") % (subnet_prefix, pkey))
|
|
|
|
semanage_ibpkey_key_free(k)
|
|
|
|
def delete(self, pkey, subnet_prefix):
|
|
self.begin()
|
|
self.__delete(pkey, subnet_prefix)
|
|
self.commit()
|
|
|
|
def get_all(self, locallist=0):
|
|
ddict = {}
|
|
if locallist:
|
|
(rc, self.plist) = semanage_ibpkey_list_local(self.sh)
|
|
else:
|
|
(rc, self.plist) = semanage_ibpkey_list(self.sh)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not list ibpkeys"))
|
|
|
|
for ibpkey in self.plist:
|
|
con = semanage_ibpkey_get_con(ibpkey)
|
|
ctype = semanage_context_get_type(con)
|
|
if ctype == "reserved_ibpkey_t":
|
|
continue
|
|
level = semanage_context_get_mls(con)
|
|
(rc, subnet_prefix) = semanage_ibpkey_get_subnet_prefix(self.sh, ibpkey)
|
|
low = semanage_ibpkey_get_low(ibpkey)
|
|
high = semanage_ibpkey_get_high(ibpkey)
|
|
ddict[(low, high, subnet_prefix)] = (ctype, level)
|
|
return ddict
|
|
|
|
def get_all_by_type(self, locallist=0):
|
|
ddict = {}
|
|
if locallist:
|
|
(rc, self.plist) = semanage_ibpkey_list_local(self.sh)
|
|
else:
|
|
(rc, self.plist) = semanage_ibpkey_list(self.sh)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not list ibpkeys"))
|
|
|
|
for ibpkey in self.plist:
|
|
con = semanage_ibpkey_get_con(ibpkey)
|
|
ctype = semanage_context_get_type(con)
|
|
(rc, subnet_prefix) = semanage_ibpkey_get_subnet_prefix(self.sh, ibpkey)
|
|
low = semanage_ibpkey_get_low(ibpkey)
|
|
high = semanage_ibpkey_get_high(ibpkey)
|
|
if (ctype, subnet_prefix) not in ddict.keys():
|
|
ddict[(ctype, subnet_prefix)] = []
|
|
if low == high:
|
|
ddict[(ctype, subnet_prefix)].append("0x%x" % low)
|
|
else:
|
|
ddict[(ctype, subnet_prefix)].append("0x%x-0x%x" % (low, high))
|
|
return ddict
|
|
|
|
def customized(self):
|
|
l = []
|
|
ddict = self.get_all(True)
|
|
|
|
for k in sorted(ddict.keys()):
|
|
port = k[0] if k[0] == k[1] else "%s-%s" % (k[0], k[1])
|
|
if ddict[k][1]:
|
|
l.append("-a -t %s -r '%s' -x %s %s" % (ddict[k][0], ddict[k][1], k[2], port))
|
|
else:
|
|
l.append("-a -t %s -x %s %s" % (ddict[k][0], k[2], port))
|
|
return l
|
|
|
|
def list(self, heading=1, locallist=0):
|
|
ddict = self.get_all_by_type(locallist)
|
|
keys = ddict.keys()
|
|
if len(keys) == 0:
|
|
return
|
|
|
|
if heading:
|
|
print("%-30s %-18s %s\n" % (_("SELinux IB Pkey Type"), _("Subnet_Prefix"), _("Pkey Number")))
|
|
for i in sorted(keys):
|
|
rec = "%-30s %-18s " % i
|
|
rec += "%s" % ddict[i][0]
|
|
for p in ddict[i][1:]:
|
|
rec += ", %s" % p
|
|
print(rec)
|
|
|
|
class ibendportRecords(semanageRecords):
|
|
|
|
valid_types = []
|
|
|
|
def __init__(self, args = None):
|
|
semanageRecords.__init__(self, args)
|
|
try:
|
|
q = setools.TypeQuery(setools.SELinuxPolicy(sepolicy.get_store_policy(self.store)), attrs=["ibendport_type"])
|
|
self.valid_types = set(str(t) for t in q.results())
|
|
except:
|
|
pass
|
|
|
|
def __genkey(self, ibendport, ibdev_name):
|
|
if ibdev_name == "":
|
|
raise ValueError(_("IB device name is required"))
|
|
|
|
port = int(ibendport)
|
|
|
|
if port > 255 or port < 1:
|
|
raise ValueError(_("Invalid Port Number"))
|
|
|
|
(rc, k) = semanage_ibendport_key_create(self.sh, ibdev_name, port)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not create a key for ibendport %s/%s") % (ibdev_name, ibendport))
|
|
return (k, ibdev_name, port)
|
|
|
|
def __add(self, ibendport, ibdev_name, serange, type):
|
|
if is_mls_enabled == 1:
|
|
if serange == "":
|
|
serange = "s0"
|
|
else:
|
|
serange = untranslate(serange)
|
|
|
|
if type == "":
|
|
raise ValueError(_("Type is required"))
|
|
|
|
type = sepolicy.get_real_type_name(type)
|
|
|
|
if type not in self.valid_types:
|
|
raise ValueError(_("Type %s is invalid, must be an ibendport type") % type)
|
|
(k, ibendport, port) = self.__genkey(ibendport, ibdev_name)
|
|
|
|
(rc, exists) = semanage_ibendport_exists(self.sh, k)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not check if ibendport %s/%s is defined") % (ibdev_name, port))
|
|
if exists:
|
|
raise ValueError(_("ibendport %s/%s already defined") % (ibdev_name, port))
|
|
|
|
(rc, p) = semanage_ibendport_create(self.sh)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not create ibendport for %s/%s") % (ibdev_name, port))
|
|
|
|
semanage_ibendport_set_ibdev_name(self.sh, p, ibdev_name)
|
|
semanage_ibendport_set_port(p, port)
|
|
(rc, con) = semanage_context_create(self.sh)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not create context for %s/%s") % (ibdev_name, port))
|
|
|
|
rc = semanage_context_set_user(self.sh, con, "system_u")
|
|
if rc < 0:
|
|
raise ValueError(_("Could not set user in ibendport context for %s/%s") % (ibdev_name, port))
|
|
|
|
rc = semanage_context_set_role(self.sh, con, "object_r")
|
|
if rc < 0:
|
|
raise ValueError(_("Could not set role in ibendport context for %s/%s") % (ibdev_name, port))
|
|
|
|
rc = semanage_context_set_type(self.sh, con, type)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not set type in ibendport context for %s/%s") % (ibdev_name, port))
|
|
|
|
if (is_mls_enabled == 1) and (serange != ""):
|
|
rc = semanage_context_set_mls(self.sh, con, serange)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not set mls fields in ibendport context for %s/%s") % (ibdev_name, port))
|
|
|
|
rc = semanage_ibendport_set_con(self.sh, p, con)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not set ibendport context for %s/%s") % (ibdev_name, port))
|
|
|
|
rc = semanage_ibendport_modify_local(self.sh, k, p)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not add ibendport %s/%s") % (ibdev_name, port))
|
|
|
|
semanage_context_free(con)
|
|
semanage_ibendport_key_free(k)
|
|
semanage_ibendport_free(p)
|
|
|
|
def add(self, ibendport, ibdev_name, serange, type):
|
|
self.begin()
|
|
self.__add(ibendport, ibdev_name, serange, type)
|
|
self.commit()
|
|
|
|
def __modify(self, ibendport, ibdev_name, serange, setype):
|
|
if serange == "" and setype == "":
|
|
if is_mls_enabled == 1:
|
|
raise ValueError(_("Requires setype or serange"))
|
|
else:
|
|
raise ValueError(_("Requires setype"))
|
|
|
|
setype = sepolicy.get_real_type_name(setype)
|
|
|
|
if setype and setype not in self.valid_types:
|
|
raise ValueError(_("Type %s is invalid, must be an ibendport type") % setype)
|
|
|
|
(k, ibdev_name, port) = self.__genkey(ibendport, ibdev_name)
|
|
|
|
(rc, exists) = semanage_ibendport_exists(self.sh, k)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not check if ibendport %s/%s is defined") % (ibdev_name, ibendport))
|
|
if not exists:
|
|
raise ValueError(_("ibendport %s/%s is not defined") % (ibdev_name, ibendport))
|
|
|
|
(rc, p) = semanage_ibendport_query(self.sh, k)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not query ibendport %s/%s") % (ibdev_name, ibendport))
|
|
|
|
con = semanage_ibendport_get_con(p)
|
|
|
|
if (is_mls_enabled == 1) and (serange != ""):
|
|
semanage_context_set_mls(self.sh, con, untranslate(serange))
|
|
if setype != "":
|
|
semanage_context_set_type(self.sh, con, setype)
|
|
|
|
rc = semanage_ibendport_modify_local(self.sh, k, p)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not modify ibendport %s/%s") % (ibdev_name, ibendport))
|
|
|
|
semanage_ibendport_key_free(k)
|
|
semanage_ibendport_free(p)
|
|
|
|
def modify(self, ibendport, ibdev_name, serange, setype):
|
|
self.begin()
|
|
self.__modify(ibendport, ibdev_name, serange, setype)
|
|
self.commit()
|
|
|
|
def deleteall(self):
|
|
(rc, plist) = semanage_ibendport_list_local(self.sh)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not list the ibendports"))
|
|
|
|
self.begin()
|
|
|
|
for ibendport in plist:
|
|
(rc, ibdev_name) = semanage_ibendport_get_ibdev_name(self.sh, ibendport)
|
|
port = semanage_ibendport_get_port(ibendport)
|
|
(k, ibdev_name, port) = self.__genkey(str(port), ibdev_name)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not create a key for %s/%d") % (ibdevname, port))
|
|
|
|
rc = semanage_ibendport_del_local(self.sh, k)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not delete the ibendport %s/%d") % (ibdev_name, port))
|
|
semanage_ibendport_key_free(k)
|
|
|
|
self.commit()
|
|
|
|
def __delete(self, ibendport, ibdev_name):
|
|
(k, ibdev_name, port) = self.__genkey(ibendport, ibdev_name)
|
|
(rc, exists) = semanage_ibendport_exists(self.sh, k)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not check if ibendport %s/%s is defined") % (ibdev_name, ibendport))
|
|
if not exists:
|
|
raise ValueError(_("ibendport %s/%s is not defined") % (ibdev_name, ibendport))
|
|
|
|
(rc, exists) = semanage_ibendport_exists_local(self.sh, k)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not check if ibendport %s/%s is defined") % (ibdev_name, ibendport))
|
|
if not exists:
|
|
raise ValueError(_("ibendport %s/%s is defined in policy, cannot be deleted") % (ibdev_name, ibendport))
|
|
|
|
rc = semanage_ibendport_del_local(self.sh, k)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not delete ibendport %s/%s") % (ibdev_name, ibendport))
|
|
|
|
semanage_ibendport_key_free(k)
|
|
|
|
def delete(self, ibendport, ibdev_name):
|
|
self.begin()
|
|
self.__delete(ibendport, ibdev_name)
|
|
self.commit()
|
|
|
|
def get_all(self, locallist=0):
|
|
ddict = {}
|
|
if locallist:
|
|
(rc, self.plist) = semanage_ibendport_list_local(self.sh)
|
|
else:
|
|
(rc, self.plist) = semanage_ibendport_list(self.sh)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not list ibendports"))
|
|
|
|
for ibendport in self.plist:
|
|
con = semanage_ibendport_get_con(ibendport)
|
|
ctype = semanage_context_get_type(con)
|
|
if ctype == "reserved_ibendport_t":
|
|
continue
|
|
level = semanage_context_get_mls(con)
|
|
(rc, ibdev_name) = semanage_ibendport_get_ibdev_name(self.sh, ibendport)
|
|
port = semanage_ibendport_get_port(ibendport)
|
|
ddict[(port, ibdev_name)] = (ctype, level)
|
|
return ddict
|
|
|
|
def get_all_by_type(self, locallist=0):
|
|
ddict = {}
|
|
if locallist:
|
|
(rc, self.plist) = semanage_ibendport_list_local(self.sh)
|
|
else:
|
|
(rc, self.plist) = semanage_ibendport_list(self.sh)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not list ibendports"))
|
|
|
|
for ibendport in self.plist:
|
|
con = semanage_ibendport_get_con(ibendport)
|
|
ctype = semanage_context_get_type(con)
|
|
(rc, ibdev_name) = semanage_ibendport_get_ibdev_name(self.sh, ibendport)
|
|
port = semanage_ibendport_get_port(ibendport)
|
|
if (ctype, ibdev_name) not in ddict.keys():
|
|
ddict[(ctype, ibdev_name)] = []
|
|
ddict[(ctype, ibdev_name)].append("0x%x" % port)
|
|
return ddict
|
|
|
|
def customized(self):
|
|
l = []
|
|
ddict = self.get_all(True)
|
|
|
|
for k in sorted(ddict.keys()):
|
|
if ddict[k][1]:
|
|
l.append("-a -t %s -r '%s' -z %s %s" % (ddict[k][0], ddict[k][1], k[1], k[0]))
|
|
else:
|
|
l.append("-a -t %s -z %s %s" % (ddict[k][0], k[1], k[0]))
|
|
return l
|
|
|
|
def list(self, heading=1, locallist=0):
|
|
ddict = self.get_all_by_type(locallist)
|
|
keys = ddict.keys()
|
|
if len(keys) == 0:
|
|
return
|
|
|
|
if heading:
|
|
print("%-30s %-18s %s\n" % (_("SELinux IB End Port Type"), _("IB Device Name"), _("Port Number")))
|
|
for i in sorted(keys):
|
|
rec = "%-30s %-18s " % i
|
|
rec += "%s" % ddict[i][0]
|
|
for p in ddict[i][1:]:
|
|
rec += ", %s" % p
|
|
print(rec)
|
|
|
|
class nodeRecords(semanageRecords):
|
|
|
|
valid_types = []
|
|
|
|
def __init__(self, args = None):
|
|
semanageRecords.__init__(self, args)
|
|
self.protocol = ["ipv4", "ipv6"]
|
|
try:
|
|
self.valid_types = list(list(sepolicy.info(sepolicy.ATTRIBUTE, "node_type"))[0]["types"])
|
|
except RuntimeError:
|
|
pass
|
|
|
|
def validate(self, addr, mask, protocol):
|
|
newaddr = addr
|
|
newmask = mask
|
|
newprotocol = ""
|
|
|
|
if addr == "":
|
|
raise ValueError(_("Node Address is required"))
|
|
|
|
# verify that (addr, mask) is either a IP address (without a mask) or a valid network mask
|
|
if len(mask) == 0 or mask[0] == "/":
|
|
i = ipaddress.ip_network(addr + mask)
|
|
newaddr = str(i.network_address)
|
|
newmask = str(i.netmask)
|
|
protocol = "ipv%d" % i.version
|
|
|
|
try:
|
|
newprotocol = self.protocol.index(protocol)
|
|
except:
|
|
raise ValueError(_("Unknown or missing protocol"))
|
|
|
|
try:
|
|
audit_protocol = socket.getprotobyname(protocol)
|
|
except:
|
|
# Entry for "ipv4" not found in /etc/protocols on (at
|
|
# least) Debian? To ensure audit log compatibility, let's
|
|
# use the same numeric value as Fedora: 4, which is
|
|
# actually understood by kernel as IP over IP.
|
|
if (protocol == "ipv4"):
|
|
audit_protocol = socket.IPPROTO_IPIP
|
|
else:
|
|
raise ValueError(_("Unknown or missing protocol"))
|
|
|
|
return newaddr, newmask, newprotocol, audit_protocol
|
|
|
|
def __add(self, addr, mask, proto, serange, ctype):
|
|
addr, mask, proto, audit_proto = self.validate(addr, mask, proto)
|
|
|
|
if is_mls_enabled == 1:
|
|
if serange == "":
|
|
serange = "s0"
|
|
else:
|
|
serange = untranslate(serange)
|
|
|
|
if ctype == "":
|
|
raise ValueError(_("SELinux node type is required"))
|
|
|
|
ctype = sepolicy.get_real_type_name(ctype)
|
|
|
|
if ctype not in self.valid_types:
|
|
raise ValueError(_("Type %s is invalid, must be a node type") % ctype)
|
|
|
|
(rc, k) = semanage_node_key_create(self.sh, addr, mask, proto)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not create key for %s") % addr)
|
|
|
|
(rc, exists) = semanage_node_exists(self.sh, k)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not check if addr %s is defined") % addr)
|
|
if exists:
|
|
raise ValueError(_("Addr %s already defined") % addr)
|
|
|
|
(rc, node) = semanage_node_create(self.sh)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not create addr for %s") % addr)
|
|
semanage_node_set_proto(node, proto)
|
|
|
|
rc = semanage_node_set_addr(self.sh, node, proto, addr)
|
|
(rc, con) = semanage_context_create(self.sh)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not create context for %s") % addr)
|
|
|
|
rc = semanage_node_set_mask(self.sh, node, proto, mask)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not set mask for %s") % addr)
|
|
|
|
rc = semanage_context_set_user(self.sh, con, "system_u")
|
|
if rc < 0:
|
|
raise ValueError(_("Could not set user in addr context for %s") % addr)
|
|
|
|
rc = semanage_context_set_role(self.sh, con, "object_r")
|
|
if rc < 0:
|
|
raise ValueError(_("Could not set role in addr context for %s") % addr)
|
|
|
|
rc = semanage_context_set_type(self.sh, con, ctype)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not set type in addr context for %s") % addr)
|
|
|
|
if (is_mls_enabled == 1) and (serange != ""):
|
|
rc = semanage_context_set_mls(self.sh, con, serange)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not set mls fields in addr context for %s") % addr)
|
|
|
|
rc = semanage_node_set_con(self.sh, node, con)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not set addr context for %s") % addr)
|
|
|
|
rc = semanage_node_modify_local(self.sh, k, node)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not add addr %s") % addr)
|
|
|
|
semanage_context_free(con)
|
|
semanage_node_key_free(k)
|
|
semanage_node_free(node)
|
|
|
|
self.mylog.log_change("resrc=node op=add laddr=%s netmask=%s proto=%s tcontext=%s:%s:%s:%s" % (addr, mask, audit_proto, "system_u", "object_r", ctype, serange))
|
|
|
|
def add(self, addr, mask, proto, serange, ctype):
|
|
self.begin()
|
|
self.__add(addr, mask, proto, serange, ctype)
|
|
self.commit()
|
|
|
|
def __modify(self, addr, mask, proto, serange, setype):
|
|
addr, mask, proto, audit_proto = self.validate(addr, mask, proto)
|
|
|
|
if serange == "" and setype == "":
|
|
raise ValueError(_("Requires setype or serange"))
|
|
|
|
setype = sepolicy.get_real_type_name(setype)
|
|
|
|
if setype and setype not in self.valid_types:
|
|
raise ValueError(_("Type %s is invalid, must be a node type") % setype)
|
|
|
|
(rc, k) = semanage_node_key_create(self.sh, addr, mask, proto)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not create key for %s") % addr)
|
|
|
|
(rc, exists) = semanage_node_exists(self.sh, k)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not check if addr %s is defined") % addr)
|
|
if not exists:
|
|
raise ValueError(_("Addr %s is not defined") % addr)
|
|
|
|
(rc, node) = semanage_node_query(self.sh, k)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not query addr %s") % addr)
|
|
|
|
con = semanage_node_get_con(node)
|
|
if (is_mls_enabled == 1) and (serange != ""):
|
|
semanage_context_set_mls(self.sh, con, untranslate(serange))
|
|
if setype != "":
|
|
semanage_context_set_type(self.sh, con, setype)
|
|
|
|
rc = semanage_node_modify_local(self.sh, k, node)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not modify addr %s") % addr)
|
|
|
|
semanage_node_key_free(k)
|
|
semanage_node_free(node)
|
|
|
|
self.mylog.log_change("resrc=node op=modify laddr=%s netmask=%s proto=%s tcontext=%s:%s:%s:%s" % (addr, mask, audit_proto, "system_u", "object_r", setype, serange))
|
|
|
|
def modify(self, addr, mask, proto, serange, setype):
|
|
self.begin()
|
|
self.__modify(addr, mask, proto, serange, setype)
|
|
self.commit()
|
|
|
|
def __delete(self, addr, mask, proto):
|
|
addr, mask, proto, audit_proto = self.validate(addr, mask, proto)
|
|
|
|
(rc, k) = semanage_node_key_create(self.sh, addr, mask, proto)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not create key for %s") % addr)
|
|
|
|
(rc, exists) = semanage_node_exists(self.sh, k)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not check if addr %s is defined") % addr)
|
|
if not exists:
|
|
raise ValueError(_("Addr %s is not defined") % addr)
|
|
|
|
(rc, exists) = semanage_node_exists_local(self.sh, k)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not check if addr %s is defined") % addr)
|
|
if not exists:
|
|
raise ValueError(_("Addr %s is defined in policy, cannot be deleted") % addr)
|
|
|
|
rc = semanage_node_del_local(self.sh, k)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not delete addr %s") % addr)
|
|
|
|
semanage_node_key_free(k)
|
|
|
|
self.mylog.log_change("resrc=node op=delete laddr=%s netmask=%s proto=%s" % (addr, mask, audit_proto))
|
|
|
|
def delete(self, addr, mask, proto):
|
|
self.begin()
|
|
self.__delete(addr, mask, proto)
|
|
self.commit()
|
|
|
|
def deleteall(self):
|
|
(rc, nlist) = semanage_node_list_local(self.sh)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not deleteall node mappings"))
|
|
|
|
self.begin()
|
|
for node in nlist:
|
|
self.__delete(semanage_node_get_addr(self.sh, node)[1], semanage_node_get_mask(self.sh, node)[1], self.protocol[semanage_node_get_proto(node)])
|
|
self.commit()
|
|
|
|
def get_all(self, locallist=0):
|
|
ddict = {}
|
|
if locallist:
|
|
(rc, self.ilist) = semanage_node_list_local(self.sh)
|
|
else:
|
|
(rc, self.ilist) = semanage_node_list(self.sh)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not list addrs"))
|
|
|
|
for node in self.ilist:
|
|
con = semanage_node_get_con(node)
|
|
addr = semanage_node_get_addr(self.sh, node)
|
|
mask = semanage_node_get_mask(self.sh, node)
|
|
proto = self.protocol[semanage_node_get_proto(node)]
|
|
ddict[(addr[1], mask[1], proto)] = (semanage_context_get_user(con), semanage_context_get_role(con), semanage_context_get_type(con), semanage_context_get_mls(con))
|
|
|
|
return ddict
|
|
|
|
def customized(self):
|
|
l = []
|
|
ddict = self.get_all(True)
|
|
for k in sorted(ddict.keys()):
|
|
if ddict[k][3]:
|
|
l.append("-a -M %s -p %s -t %s -r '%s' %s" % (k[1], k[2], ddict[k][2], ddict[k][3], k[0]))
|
|
else:
|
|
l.append("-a -M %s -p %s -t %s %s" % (k[1], k[2], ddict[k][2], k[0]))
|
|
return l
|
|
|
|
def list(self, heading=1, locallist=0):
|
|
ddict = self.get_all(locallist)
|
|
if len(ddict) == 0:
|
|
return
|
|
keys = sorted(ddict.keys())
|
|
|
|
if heading:
|
|
print("%-18s %-18s %-5s %-5s\n" % ("IP Address", "Netmask", "Protocol", "Context"))
|
|
if is_mls_enabled:
|
|
for k in keys:
|
|
val = ''
|
|
for fields in k:
|
|
val = val + '\t' + str(fields)
|
|
print("%-18s %-18s %-5s %s:%s:%s:%s " % (k[0], k[1], k[2], ddict[k][0], ddict[k][1], ddict[k][2], translate(ddict[k][3], False)))
|
|
else:
|
|
for k in keys:
|
|
print("%-18s %-18s %-5s %s:%s:%s " % (k[0], k[1], k[2], ddict[k][0], ddict[k][1], ddict[k][2]))
|
|
|
|
|
|
class interfaceRecords(semanageRecords):
|
|
|
|
def __init__(self, args = None):
|
|
semanageRecords.__init__(self, args)
|
|
|
|
def __add(self, interface, serange, ctype):
|
|
if is_mls_enabled == 1:
|
|
if serange == "":
|
|
serange = "s0"
|
|
else:
|
|
serange = untranslate(serange)
|
|
|
|
if ctype == "":
|
|
raise ValueError(_("SELinux Type is required"))
|
|
|
|
(rc, k) = semanage_iface_key_create(self.sh, interface)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not create key for %s") % interface)
|
|
|
|
(rc, exists) = semanage_iface_exists(self.sh, k)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not check if interface %s is defined") % interface)
|
|
if exists:
|
|
raise ValueError(_("Interface %s already defined") % interface)
|
|
|
|
(rc, iface) = semanage_iface_create(self.sh)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not create interface for %s") % interface)
|
|
|
|
rc = semanage_iface_set_name(self.sh, iface, interface)
|
|
(rc, con) = semanage_context_create(self.sh)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not create context for %s") % interface)
|
|
|
|
rc = semanage_context_set_user(self.sh, con, "system_u")
|
|
if rc < 0:
|
|
raise ValueError(_("Could not set user in interface context for %s") % interface)
|
|
|
|
rc = semanage_context_set_role(self.sh, con, "object_r")
|
|
if rc < 0:
|
|
raise ValueError(_("Could not set role in interface context for %s") % interface)
|
|
|
|
rc = semanage_context_set_type(self.sh, con, ctype)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not set type in interface context for %s") % interface)
|
|
|
|
if (is_mls_enabled == 1) and (serange != ""):
|
|
rc = semanage_context_set_mls(self.sh, con, serange)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not set mls fields in interface context for %s") % interface)
|
|
|
|
rc = semanage_iface_set_ifcon(self.sh, iface, con)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not set interface context for %s") % interface)
|
|
|
|
rc = semanage_iface_set_msgcon(self.sh, iface, con)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not set message context for %s") % interface)
|
|
|
|
rc = semanage_iface_modify_local(self.sh, k, iface)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not add interface %s") % interface)
|
|
|
|
semanage_context_free(con)
|
|
semanage_iface_key_free(k)
|
|
semanage_iface_free(iface)
|
|
|
|
self.mylog.log_change("resrc=interface op=add netif=%s tcontext=%s:%s:%s:%s" % (interface, "system_u", "object_r", ctype, serange))
|
|
|
|
def add(self, interface, serange, ctype):
|
|
self.begin()
|
|
self.__add(interface, serange, ctype)
|
|
self.commit()
|
|
|
|
def __modify(self, interface, serange, setype):
|
|
if serange == "" and setype == "":
|
|
raise ValueError(_("Requires setype or serange"))
|
|
|
|
(rc, k) = semanage_iface_key_create(self.sh, interface)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not create key for %s") % interface)
|
|
|
|
(rc, exists) = semanage_iface_exists(self.sh, k)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not check if interface %s is defined") % interface)
|
|
if not exists:
|
|
raise ValueError(_("Interface %s is not defined") % interface)
|
|
|
|
(rc, iface) = semanage_iface_query(self.sh, k)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not query interface %s") % interface)
|
|
|
|
con = semanage_iface_get_ifcon(iface)
|
|
|
|
if (is_mls_enabled == 1) and (serange != ""):
|
|
semanage_context_set_mls(self.sh, con, untranslate(serange))
|
|
if setype != "":
|
|
semanage_context_set_type(self.sh, con, setype)
|
|
|
|
rc = semanage_iface_modify_local(self.sh, k, iface)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not modify interface %s") % interface)
|
|
|
|
semanage_iface_key_free(k)
|
|
semanage_iface_free(iface)
|
|
|
|
self.mylog.log_change("resrc=interface op=modify netif=%s tcontext=%s:%s:%s:%s" % (interface, "system_u", "object_r", setype, serange))
|
|
|
|
def modify(self, interface, serange, setype):
|
|
self.begin()
|
|
self.__modify(interface, serange, setype)
|
|
self.commit()
|
|
|
|
def __delete(self, interface):
|
|
(rc, k) = semanage_iface_key_create(self.sh, interface)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not create key for %s") % interface)
|
|
|
|
(rc, exists) = semanage_iface_exists(self.sh, k)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not check if interface %s is defined") % interface)
|
|
if not exists:
|
|
raise ValueError(_("Interface %s is not defined") % interface)
|
|
|
|
(rc, exists) = semanage_iface_exists_local(self.sh, k)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not check if interface %s is defined") % interface)
|
|
if not exists:
|
|
raise ValueError(_("Interface %s is defined in policy, cannot be deleted") % interface)
|
|
|
|
rc = semanage_iface_del_local(self.sh, k)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not delete interface %s") % interface)
|
|
|
|
semanage_iface_key_free(k)
|
|
|
|
self.mylog.log_change("resrc=interface op=delete netif=%s" % interface)
|
|
|
|
def delete(self, interface):
|
|
self.begin()
|
|
self.__delete(interface)
|
|
self.commit()
|
|
|
|
def deleteall(self):
|
|
(rc, ulist) = semanage_iface_list_local(self.sh)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not delete all interface mappings"))
|
|
|
|
self.begin()
|
|
for i in ulist:
|
|
self.__delete(semanage_iface_get_name(i))
|
|
self.commit()
|
|
|
|
def get_all(self, locallist=0):
|
|
ddict = {}
|
|
if locallist:
|
|
(rc, self.ilist) = semanage_iface_list_local(self.sh)
|
|
else:
|
|
(rc, self.ilist) = semanage_iface_list(self.sh)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not list interfaces"))
|
|
|
|
for interface in self.ilist:
|
|
con = semanage_iface_get_ifcon(interface)
|
|
ddict[semanage_iface_get_name(interface)] = (semanage_context_get_user(con), semanage_context_get_role(con), semanage_context_get_type(con), semanage_context_get_mls(con))
|
|
|
|
return ddict
|
|
|
|
def customized(self):
|
|
l = []
|
|
ddict = self.get_all(True)
|
|
for k in sorted(ddict.keys()):
|
|
if ddict[k][3]:
|
|
l.append("-a -t %s -r '%s' %s" % (ddict[k][2], ddict[k][3], k))
|
|
else:
|
|
l.append("-a -t %s %s" % (ddict[k][2], k))
|
|
return l
|
|
|
|
def list(self, heading=1, locallist=0):
|
|
ddict = self.get_all(locallist)
|
|
if len(ddict) == 0:
|
|
return
|
|
keys = sorted(ddict.keys())
|
|
|
|
if heading:
|
|
print("%-30s %s\n" % (_("SELinux Interface"), _("Context")))
|
|
if is_mls_enabled:
|
|
for k in keys:
|
|
print("%-30s %s:%s:%s:%s " % (k, ddict[k][0], ddict[k][1], ddict[k][2], translate(ddict[k][3], False)))
|
|
else:
|
|
for k in keys:
|
|
print("%-30s %s:%s:%s " % (k, ddict[k][0], ddict[k][1], ddict[k][2]))
|
|
|
|
|
|
class fcontextRecords(semanageRecords):
|
|
|
|
valid_types = []
|
|
|
|
def __init__(self, args = None):
|
|
semanageRecords.__init__(self, args)
|
|
try:
|
|
self.valid_types = list(list(sepolicy.info(sepolicy.ATTRIBUTE, "file_type"))[0]["types"])
|
|
self.valid_types += list(list(sepolicy.info(sepolicy.ATTRIBUTE, "device_node"))[0]["types"])
|
|
except RuntimeError:
|
|
pass
|
|
|
|
self.equiv = {}
|
|
self.equiv_dist = {}
|
|
self.equal_ind = False
|
|
try:
|
|
fd = open(selinux.selinux_file_context_subs_path(), "r")
|
|
for i in fd.readlines():
|
|
i = i.strip()
|
|
if len(i) == 0:
|
|
continue
|
|
if i.startswith("#"):
|
|
continue
|
|
target, substitute = i.split()
|
|
self.equiv[target] = substitute
|
|
fd.close()
|
|
except IOError:
|
|
pass
|
|
try:
|
|
fd = open(selinux.selinux_file_context_subs_dist_path(), "r")
|
|
for i in fd.readlines():
|
|
i = i.strip()
|
|
if len(i) == 0:
|
|
continue
|
|
if i.startswith("#"):
|
|
continue
|
|
target, substitute = i.split()
|
|
self.equiv_dist[target] = substitute
|
|
fd.close()
|
|
except IOError:
|
|
pass
|
|
|
|
def commit(self):
|
|
if self.equal_ind:
|
|
subs_file = selinux.selinux_file_context_subs_path()
|
|
tmpfile = "%s.tmp" % subs_file
|
|
fd = open(tmpfile, "w")
|
|
for target in self.equiv.keys():
|
|
fd.write("%s %s\n" % (target, self.equiv[target]))
|
|
fd.close()
|
|
try:
|
|
os.chmod(tmpfile, os.stat(subs_file)[stat.ST_MODE])
|
|
except:
|
|
pass
|
|
os.rename(tmpfile, subs_file)
|
|
self.equal_ind = False
|
|
semanageRecords.commit(self)
|
|
|
|
def add_equal(self, target, substitute):
|
|
self.begin()
|
|
if target != "/" and target[-1] == "/":
|
|
raise ValueError(_("Target %s is not valid. Target is not allowed to end with '/'") % target)
|
|
|
|
if substitute != "/" and substitute[-1] == "/":
|
|
raise ValueError(_("Substitute %s is not valid. Substitute is not allowed to end with '/'") % substitute)
|
|
|
|
if target in self.equiv.keys():
|
|
raise ValueError(_("Equivalence class for %s already exists") % target)
|
|
self.validate(target)
|
|
|
|
for fdict in (self.equiv, self.equiv_dist):
|
|
for i in fdict:
|
|
if i.startswith(target + "/"):
|
|
raise ValueError(_("File spec %s conflicts with equivalency rule '%s %s'") % (target, i, fdict[i]))
|
|
|
|
self.mylog.log_change("resrc=fcontext op=add-equal %s %s" % (audit.audit_encode_nv_string("sglob", target, 0), audit.audit_encode_nv_string("tglob", substitute, 0)))
|
|
|
|
self.equiv[target] = substitute
|
|
self.equal_ind = True
|
|
self.commit()
|
|
|
|
def modify_equal(self, target, substitute):
|
|
self.begin()
|
|
if target not in self.equiv.keys():
|
|
raise ValueError(_("Equivalence class for %s does not exist") % target)
|
|
self.equiv[target] = substitute
|
|
self.equal_ind = True
|
|
|
|
self.mylog.log_change("resrc=fcontext op=modify-equal %s %s" % (audit.audit_encode_nv_string("sglob", target, 0), audit.audit_encode_nv_string("tglob", substitute, 0)))
|
|
|
|
self.commit()
|
|
|
|
def createcon(self, target, seuser="system_u"):
|
|
(rc, con) = semanage_context_create(self.sh)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not create context for %s") % target)
|
|
if seuser == "":
|
|
seuser = "system_u"
|
|
|
|
rc = semanage_context_set_user(self.sh, con, seuser)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not set user in file context for %s") % target)
|
|
|
|
rc = semanage_context_set_role(self.sh, con, "object_r")
|
|
if rc < 0:
|
|
raise ValueError(_("Could not set role in file context for %s") % target)
|
|
|
|
if is_mls_enabled == 1:
|
|
rc = semanage_context_set_mls(self.sh, con, "s0")
|
|
if rc < 0:
|
|
raise ValueError(_("Could not set mls fields in file context for %s") % target)
|
|
|
|
return con
|
|
|
|
def validate(self, target):
|
|
if target == "" or target.find("\n") >= 0:
|
|
raise ValueError(_("Invalid file specification"))
|
|
if target.find(" ") != -1:
|
|
raise ValueError(_("File specification can not include spaces"))
|
|
for fdict in (self.equiv, self.equiv_dist):
|
|
for i in fdict:
|
|
if target.startswith(i + "/"):
|
|
t = re.sub(i, fdict[i], target)
|
|
raise ValueError(_("File spec %s conflicts with equivalency rule '%s %s'; Try adding '%s' instead") % (target, i, fdict[i], t))
|
|
|
|
def __add(self, target, type, ftype="", serange="", seuser="system_u"):
|
|
self.validate(target)
|
|
|
|
if is_mls_enabled == 1:
|
|
serange = untranslate(serange)
|
|
|
|
if type == "":
|
|
raise ValueError(_("SELinux Type is required"))
|
|
|
|
if type != "<<none>>":
|
|
type = sepolicy.get_real_type_name(type)
|
|
if type not in self.valid_types:
|
|
raise ValueError(_("Type %s is invalid, must be a file or device type") % type)
|
|
|
|
(rc, k) = semanage_fcontext_key_create(self.sh, target, file_types[ftype])
|
|
if rc < 0:
|
|
raise ValueError(_("Could not create key for %s") % target)
|
|
|
|
(rc, exists) = semanage_fcontext_exists(self.sh, k)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not check if file context for %s is defined") % target)
|
|
|
|
if not exists:
|
|
(rc, exists) = semanage_fcontext_exists_local(self.sh, k)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not check if file context for %s is defined") % target)
|
|
|
|
if exists:
|
|
raise ValueError(_("File context for %s already defined") % target)
|
|
|
|
(rc, fcontext) = semanage_fcontext_create(self.sh)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not create file context for %s") % target)
|
|
|
|
rc = semanage_fcontext_set_expr(self.sh, fcontext, target)
|
|
if type != "<<none>>":
|
|
con = self.createcon(target, seuser)
|
|
|
|
rc = semanage_context_set_type(self.sh, con, type)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not set type in file context for %s") % target)
|
|
|
|
if (is_mls_enabled == 1) and (serange != ""):
|
|
rc = semanage_context_set_mls(self.sh, con, serange)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not set mls fields in file context for %s") % target)
|
|
rc = semanage_fcontext_set_con(self.sh, fcontext, con)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not set file context for %s") % target)
|
|
|
|
semanage_fcontext_set_type(fcontext, file_types[ftype])
|
|
|
|
rc = semanage_fcontext_modify_local(self.sh, k, fcontext)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not add file context for %s") % target)
|
|
|
|
if type != "<<none>>":
|
|
semanage_context_free(con)
|
|
semanage_fcontext_key_free(k)
|
|
semanage_fcontext_free(fcontext)
|
|
|
|
if not seuser:
|
|
seuser = "system_u"
|
|
|
|
self.mylog.log_change("resrc=fcontext op=add %s ftype=%s tcontext=%s:%s:%s:%s" % (audit.audit_encode_nv_string("tglob", target, 0), ftype_to_audit[ftype], seuser, "object_r", type, serange))
|
|
|
|
def add(self, target, type, ftype="", serange="", seuser="system_u"):
|
|
self.begin()
|
|
self.__add(target, type, ftype, serange, seuser)
|
|
self.commit()
|
|
|
|
def __modify(self, target, setype, ftype, serange, seuser):
|
|
if serange == "" and setype == "" and seuser == "":
|
|
raise ValueError(_("Requires setype, serange or seuser"))
|
|
if setype not in ["", "<<none>>"]:
|
|
setype = sepolicy.get_real_type_name(setype)
|
|
if setype not in self.valid_types:
|
|
raise ValueError(_("Type %s is invalid, must be a file or device type") % setype)
|
|
|
|
self.validate(target)
|
|
|
|
(rc, k) = semanage_fcontext_key_create(self.sh, target, file_types[ftype])
|
|
if rc < 0:
|
|
raise ValueError(_("Could not create a key for %s") % target)
|
|
|
|
(rc, exists) = semanage_fcontext_exists(self.sh, k)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not check if file context for %s is defined") % target)
|
|
if not exists:
|
|
(rc, exists) = semanage_fcontext_exists_local(self.sh, k)
|
|
if not exists:
|
|
raise ValueError(_("File context for %s is not defined") % target)
|
|
|
|
try:
|
|
(rc, fcontext) = semanage_fcontext_query_local(self.sh, k)
|
|
except OSError:
|
|
try:
|
|
(rc, fcontext) = semanage_fcontext_query(self.sh, k)
|
|
except OSError:
|
|
raise ValueError(_("Could not query file context for %s") % target)
|
|
|
|
if setype != "<<none>>":
|
|
con = semanage_fcontext_get_con(fcontext)
|
|
|
|
if con is None:
|
|
con = self.createcon(target)
|
|
|
|
if (is_mls_enabled == 1) and (serange != ""):
|
|
semanage_context_set_mls(self.sh, con, untranslate(serange))
|
|
if seuser != "":
|
|
semanage_context_set_user(self.sh, con, seuser)
|
|
|
|
if setype != "":
|
|
semanage_context_set_type(self.sh, con, setype)
|
|
|
|
rc = semanage_fcontext_set_con(self.sh, fcontext, con)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not set file context for %s") % target)
|
|
else:
|
|
rc = semanage_fcontext_set_con(self.sh, fcontext, None)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not set file context for %s") % target)
|
|
|
|
rc = semanage_fcontext_modify_local(self.sh, k, fcontext)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not modify file context for %s") % target)
|
|
|
|
semanage_fcontext_key_free(k)
|
|
semanage_fcontext_free(fcontext)
|
|
|
|
if not seuser:
|
|
seuser = "system_u"
|
|
|
|
self.mylog.log_change("resrc=fcontext op=modify %s ftype=%s tcontext=%s:%s:%s:%s" % (audit.audit_encode_nv_string("tglob", target, 0), ftype_to_audit[ftype], seuser, "object_r", setype, serange))
|
|
|
|
def modify(self, target, setype, ftype, serange, seuser):
|
|
self.begin()
|
|
self.__modify(target, setype, ftype, serange, seuser)
|
|
self.commit()
|
|
|
|
def deleteall(self):
|
|
(rc, flist) = semanage_fcontext_list_local(self.sh)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not list the file contexts"))
|
|
|
|
self.begin()
|
|
|
|
for fcontext in flist:
|
|
target = semanage_fcontext_get_expr(fcontext)
|
|
ftype = semanage_fcontext_get_type(fcontext)
|
|
ftype_str = semanage_fcontext_get_type_str(ftype)
|
|
(rc, k) = semanage_fcontext_key_create(self.sh, target, file_types[ftype_str])
|
|
if rc < 0:
|
|
raise ValueError(_("Could not create a key for %s") % target)
|
|
|
|
rc = semanage_fcontext_del_local(self.sh, k)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not delete the file context %s") % target)
|
|
semanage_fcontext_key_free(k)
|
|
|
|
self.mylog.log_change("resrc=fcontext op=delete %s ftype=%s" % (audit.audit_encode_nv_string("tglob", target, 0), ftype_to_audit[file_type_str_to_option[ftype_str]]))
|
|
|
|
self.equiv = {}
|
|
self.equal_ind = True
|
|
self.commit()
|
|
|
|
def __delete(self, target, ftype):
|
|
if target in self.equiv.keys():
|
|
self.equiv.pop(target)
|
|
self.equal_ind = True
|
|
|
|
self.mylog.log_change("resrc=fcontext op=delete-equal %s" % (audit.audit_encode_nv_string("tglob", target, 0)))
|
|
|
|
return
|
|
|
|
(rc, k) = semanage_fcontext_key_create(self.sh, target, file_types[ftype])
|
|
if rc < 0:
|
|
raise ValueError(_("Could not create a key for %s") % target)
|
|
|
|
(rc, exists) = semanage_fcontext_exists_local(self.sh, k)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not check if file context for %s is defined") % target)
|
|
if not exists:
|
|
(rc, exists) = semanage_fcontext_exists(self.sh, k)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not check if file context for %s is defined") % target)
|
|
if exists:
|
|
raise ValueError(_("File context for %s is defined in policy, cannot be deleted") % target)
|
|
else:
|
|
raise ValueError(_("File context for %s is not defined") % target)
|
|
|
|
rc = semanage_fcontext_del_local(self.sh, k)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not delete file context for %s") % target)
|
|
|
|
semanage_fcontext_key_free(k)
|
|
|
|
self.mylog.log_change("resrc=fcontext op=delete %s ftype=%s" % (audit.audit_encode_nv_string("tglob", target, 0), ftype_to_audit[ftype]))
|
|
|
|
def delete(self, target, ftype):
|
|
self.begin()
|
|
self.__delete(target, ftype)
|
|
self.commit()
|
|
|
|
def get_all(self, locallist=0):
|
|
if locallist:
|
|
(rc, self.flist) = semanage_fcontext_list_local(self.sh)
|
|
else:
|
|
(rc, self.flist) = semanage_fcontext_list(self.sh)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not list file contexts"))
|
|
|
|
(rc, fchomedirs) = semanage_fcontext_list_homedirs(self.sh)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not list file contexts for home directories"))
|
|
|
|
(rc, fclocal) = semanage_fcontext_list_local(self.sh)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not list local file contexts"))
|
|
|
|
self.flist += fchomedirs
|
|
self.flist += fclocal
|
|
|
|
ddict = {}
|
|
for fcontext in self.flist:
|
|
expr = semanage_fcontext_get_expr(fcontext)
|
|
ftype = semanage_fcontext_get_type(fcontext)
|
|
ftype_str = semanage_fcontext_get_type_str(ftype)
|
|
con = semanage_fcontext_get_con(fcontext)
|
|
if con:
|
|
ddict[(expr, ftype_str)] = (semanage_context_get_user(con), semanage_context_get_role(con), semanage_context_get_type(con), semanage_context_get_mls(con))
|
|
else:
|
|
ddict[(expr, ftype_str)] = con
|
|
|
|
return ddict
|
|
|
|
def customized(self):
|
|
l = []
|
|
fcon_dict = self.get_all(True)
|
|
for k in sorted(fcon_dict.keys()):
|
|
if fcon_dict[k]:
|
|
if fcon_dict[k][3]:
|
|
l.append("-a -f %s -t %s -r '%s' '%s'" % (file_type_str_to_option[k[1]], fcon_dict[k][2], fcon_dict[k][3], k[0]))
|
|
else:
|
|
l.append("-a -f %s -t %s '%s'" % (file_type_str_to_option[k[1]], fcon_dict[k][2], k[0]))
|
|
|
|
if len(self.equiv):
|
|
for target in self.equiv.keys():
|
|
l.append("-a -e %s %s" % (self.equiv[target], target))
|
|
return l
|
|
|
|
def list(self, heading=1, locallist=0):
|
|
fcon_dict = self.get_all(locallist)
|
|
if len(fcon_dict) != 0:
|
|
if heading:
|
|
print("%-50s %-18s %s\n" % (_("SELinux fcontext"), _("type"), _("Context")))
|
|
for k in sorted(fcon_dict.keys()):
|
|
if fcon_dict[k]:
|
|
if is_mls_enabled:
|
|
print("%-50s %-18s %s:%s:%s:%s " % (k[0], k[1], fcon_dict[k][0], fcon_dict[k][1], fcon_dict[k][2], translate(fcon_dict[k][3], False)))
|
|
else:
|
|
print("%-50s %-18s %s:%s:%s " % (k[0], k[1], fcon_dict[k][0], fcon_dict[k][1], fcon_dict[k][2]))
|
|
else:
|
|
print("%-50s %-18s <<None>>" % (k[0], k[1]))
|
|
|
|
if len(self.equiv_dist):
|
|
if not locallist:
|
|
if heading:
|
|
print(_("\nSELinux Distribution fcontext Equivalence \n"))
|
|
for target in self.equiv_dist.keys():
|
|
print("%s = %s" % (target, self.equiv_dist[target]))
|
|
if len(self.equiv):
|
|
if heading:
|
|
print(_("\nSELinux Local fcontext Equivalence \n"))
|
|
|
|
for target in self.equiv.keys():
|
|
print("%s = %s" % (target, self.equiv[target]))
|
|
|
|
|
|
class booleanRecords(semanageRecords):
|
|
|
|
def __init__(self, args = None):
|
|
semanageRecords.__init__(self, args)
|
|
self.dict = {}
|
|
self.dict["TRUE"] = 1
|
|
self.dict["FALSE"] = 0
|
|
self.dict["ON"] = 1
|
|
self.dict["OFF"] = 0
|
|
self.dict["1"] = 1
|
|
self.dict["0"] = 0
|
|
|
|
try:
|
|
rc, self.current_booleans = selinux.security_get_boolean_names()
|
|
rc, ptype = selinux.selinux_getpolicytype()
|
|
except:
|
|
self.current_booleans = []
|
|
ptype = None
|
|
|
|
if self.store == "" or self.store == ptype:
|
|
self.modify_local = True
|
|
else:
|
|
self.modify_local = False
|
|
|
|
def __mod(self, name, value):
|
|
name = selinux.selinux_boolean_sub(name)
|
|
|
|
(rc, k) = semanage_bool_key_create(self.sh, name)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not create a key for %s") % name)
|
|
(rc, exists) = semanage_bool_exists(self.sh, k)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not check if boolean %s is defined") % name)
|
|
if not exists:
|
|
raise ValueError(_("Boolean %s is not defined") % name)
|
|
|
|
(rc, b) = semanage_bool_query(self.sh, k)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not query file context %s") % name)
|
|
|
|
if value.upper() in self.dict:
|
|
semanage_bool_set_value(b, self.dict[value.upper()])
|
|
else:
|
|
raise ValueError(_("You must specify one of the following values: %s") % ", ".join(self.dict.keys()))
|
|
|
|
if self.modify_local and name in self.current_booleans:
|
|
rc = semanage_bool_set_active(self.sh, k, b)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not set active value of boolean %s") % name)
|
|
rc = semanage_bool_modify_local(self.sh, k, b)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not modify boolean %s") % name)
|
|
semanage_bool_key_free(k)
|
|
semanage_bool_free(b)
|
|
|
|
def modify(self, name, value=None, use_file=False):
|
|
self.begin()
|
|
if use_file:
|
|
fd = open(name)
|
|
for b in fd.read().split("\n"):
|
|
b = b.strip()
|
|
if len(b) == 0:
|
|
continue
|
|
|
|
try:
|
|
boolname, val = b.split("=")
|
|
except ValueError:
|
|
raise ValueError(_("Bad format %s: Record %s" % (name, b)))
|
|
self.__mod(boolname.strip(), val.strip())
|
|
fd.close()
|
|
else:
|
|
self.__mod(name, value)
|
|
|
|
self.commit()
|
|
|
|
def __delete(self, name):
|
|
name = selinux.selinux_boolean_sub(name)
|
|
|
|
(rc, k) = semanage_bool_key_create(self.sh, name)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not create a key for %s") % name)
|
|
(rc, exists) = semanage_bool_exists(self.sh, k)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not check if boolean %s is defined") % name)
|
|
if not exists:
|
|
raise ValueError(_("Boolean %s is not defined") % name)
|
|
|
|
(rc, exists) = semanage_bool_exists_local(self.sh, k)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not check if boolean %s is defined") % name)
|
|
if not exists:
|
|
raise ValueError(_("Boolean %s is defined in policy, cannot be deleted") % name)
|
|
|
|
rc = semanage_bool_del_local(self.sh, k)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not delete boolean %s") % name)
|
|
|
|
semanage_bool_key_free(k)
|
|
|
|
def delete(self, name):
|
|
self.begin()
|
|
self.__delete(name)
|
|
self.commit()
|
|
|
|
def deleteall(self):
|
|
(rc, self.blist) = semanage_bool_list_local(self.sh)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not list booleans"))
|
|
|
|
self.begin()
|
|
|
|
for boolean in self.blist:
|
|
name = semanage_bool_get_name(boolean)
|
|
self.__delete(name)
|
|
|
|
self.commit()
|
|
|
|
def get_all(self, locallist=0):
|
|
ddict = {}
|
|
if locallist:
|
|
(rc, self.blist) = semanage_bool_list_local(self.sh)
|
|
else:
|
|
(rc, self.blist) = semanage_bool_list(self.sh)
|
|
if rc < 0:
|
|
raise ValueError(_("Could not list booleans"))
|
|
|
|
for boolean in self.blist:
|
|
value = []
|
|
name = semanage_bool_get_name(boolean)
|
|
value.append(semanage_bool_get_value(boolean))
|
|
if self.modify_local and name in self.current_booleans:
|
|
value.append(selinux.security_get_boolean_pending(name))
|
|
value.append(selinux.security_get_boolean_active(name))
|
|
else:
|
|
value.append(value[0])
|
|
value.append(value[0])
|
|
ddict[name] = value
|
|
|
|
return ddict
|
|
|
|
def get_desc(self, name):
|
|
name = selinux.selinux_boolean_sub(name)
|
|
return sepolicy.boolean_desc(name)
|
|
|
|
def get_category(self, name):
|
|
name = selinux.selinux_boolean_sub(name)
|
|
return sepolicy.boolean_category(name)
|
|
|
|
def customized(self):
|
|
l = []
|
|
ddict = self.get_all(True)
|
|
for k in sorted(ddict.keys()):
|
|
if ddict[k]:
|
|
l.append("-m -%s %s" % (ddict[k][2], k))
|
|
return l
|
|
|
|
def list(self, heading=True, locallist=False, use_file=False):
|
|
on_off = (_("off"), _("on"))
|
|
if use_file:
|
|
ddict = self.get_all(locallist)
|
|
for k in sorted(ddict.keys()):
|
|
if ddict[k]:
|
|
print("%s=%s" % (k, ddict[k][2]))
|
|
return
|
|
ddict = self.get_all(locallist)
|
|
if len(ddict) == 0:
|
|
return
|
|
|
|
if heading:
|
|
print("%-30s %s %s %s\n" % (_("SELinux boolean"), _("State"), _("Default"), _("Description")))
|
|
for k in sorted(ddict.keys()):
|
|
if ddict[k]:
|
|
print("%-30s (%-5s,%5s) %s" % (k, on_off[ddict[k][2]], on_off[ddict[k][0]], self.get_desc(k)))
|