# Author: Dan Walsh <dwalsh@redhat.com>
|
# Author: Ryan Hallisey <rhallise@redhat.com>
|
# Author: Jason Zaman <perfinion@gentoo.org>
|
|
import errno
|
import selinux
|
import setools
|
import glob
|
import sepolgen.defaults as defaults
|
import sepolgen.interfaces as interfaces
|
import sys
|
import os
|
import re
|
import gzip
|
|
PROGNAME = "policycoreutils"
|
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
|
|
TYPE = 1
|
ROLE = 2
|
ATTRIBUTE = 3
|
PORT = 4
|
USER = 5
|
BOOLEAN = 6
|
TCLASS = 7
|
|
ALLOW = 'allow'
|
AUDITALLOW = 'auditallow'
|
NEVERALLOW = 'neverallow'
|
DONTAUDIT = 'dontaudit'
|
SOURCE = 'source'
|
TARGET = 'target'
|
PERMS = 'permlist'
|
CLASS = 'class'
|
TRANSITION = 'transition'
|
ROLE_ALLOW = 'role_allow'
|
|
|
# Autofill for adding files *************************
|
DEFAULT_DIRS = {}
|
DEFAULT_DIRS["/etc"] = "etc_t"
|
DEFAULT_DIRS["/tmp"] = "tmp_t"
|
DEFAULT_DIRS["/usr/lib/systemd/system"] = "unit_file_t"
|
DEFAULT_DIRS["/lib/systemd/system"] = "unit_file_t"
|
DEFAULT_DIRS["/etc/systemd/system"] = "unit_file_t"
|
DEFAULT_DIRS["/var/cache"] = "var_cache_t"
|
DEFAULT_DIRS["/var/lib"] = "var_lib_t"
|
DEFAULT_DIRS["/var/log"] = "log_t"
|
DEFAULT_DIRS["/var/run"] = "var_run_t"
|
DEFAULT_DIRS["/run"] = "var_run_t"
|
DEFAULT_DIRS["/run/lock"] = "var_lock_t"
|
DEFAULT_DIRS["/var/run/lock"] = "var_lock_t"
|
DEFAULT_DIRS["/var/spool"] = "var_spool_t"
|
DEFAULT_DIRS["/var/www"] = "content_t"
|
|
file_type_str = {}
|
file_type_str["a"] = _("all files")
|
file_type_str["f"] = _("regular file")
|
file_type_str["d"] = _("directory")
|
file_type_str["c"] = _("character device")
|
file_type_str["b"] = _("block device")
|
file_type_str["s"] = _("socket file")
|
file_type_str["l"] = _("symbolic link")
|
file_type_str["p"] = _("named pipe")
|
|
trans_file_type_str = {}
|
trans_file_type_str[""] = "a"
|
trans_file_type_str["--"] = "f"
|
trans_file_type_str["-d"] = "d"
|
trans_file_type_str["-c"] = "c"
|
trans_file_type_str["-b"] = "b"
|
trans_file_type_str["-s"] = "s"
|
trans_file_type_str["-l"] = "l"
|
trans_file_type_str["-p"] = "p"
|
|
# the setools policy handle
|
_pol = None
|
|
# cache the lookup results
|
file_equiv_modified = None
|
file_equiv = None
|
local_files = None
|
fcdict = None
|
methods = []
|
all_types = None
|
all_types_info = None
|
user_types = None
|
role_allows = None
|
portrecs = None
|
portrecsbynum = None
|
all_domains = None
|
roles = None
|
selinux_user_list = None
|
login_mappings = None
|
file_types = None
|
port_types = None
|
bools = None
|
all_attributes = None
|
booleans = None
|
booleans_dict = None
|
all_allow_rules = None
|
all_transitions = None
|
|
|
def policy_sortkey(policy_path):
|
# Parse the extension of a policy path which looks like .../policy/policy.31
|
extension = policy_path.rsplit('/policy.', 1)[1]
|
try:
|
return int(extension), policy_path
|
except ValueError:
|
# Fallback with sorting on the full path
|
return 0, policy_path
|
|
def get_installed_policy(root="/"):
|
try:
|
path = root + selinux.selinux_binary_policy_path()
|
policies = glob.glob("%s.*" % path)
|
policies.sort(key=policy_sortkey)
|
return policies[-1]
|
except:
|
pass
|
raise ValueError(_("No SELinux Policy installed"))
|
|
def get_store_policy(store):
|
"""Get the path to the policy file located in the given store name"""
|
policies = glob.glob("%s%s/policy/policy.*" %
|
(selinux.selinux_path(), store))
|
if not policies:
|
return None
|
# Return the policy with the higher version number
|
policies.sort(key=policy_sortkey)
|
return policies[-1]
|
|
def policy(policy_file):
|
global all_domains
|
global all_attributes
|
global bools
|
global all_types
|
global role_allows
|
global users
|
global roles
|
global file_types
|
global port_types
|
all_domains = None
|
all_attributes = None
|
bools = None
|
all_types = None
|
role_allows = None
|
users = None
|
roles = None
|
file_types = None
|
port_types = None
|
global _pol
|
|
try:
|
_pol = setools.SELinuxPolicy(policy_file)
|
except:
|
raise ValueError(_("Failed to read %s policy file") % policy_file)
|
|
def load_store_policy(store):
|
policy_file = get_store_policy(store)
|
if not policy_file:
|
return None
|
policy(policy_file)
|
|
try:
|
policy_file = get_installed_policy()
|
policy(policy_file)
|
except ValueError as e:
|
if selinux.is_selinux_enabled() == 1:
|
raise e
|
|
|
def info(setype, name=None):
|
if setype == TYPE:
|
q = setools.TypeQuery(_pol)
|
q.name = name
|
results = list(q.results())
|
|
if name and len(results) < 1:
|
# type not found, try alias
|
q.name = None
|
q.alias = name
|
results = list(q.results())
|
|
return ({
|
'aliases': list(map(str, x.aliases())),
|
'name': str(x),
|
'permissive': bool(x.ispermissive),
|
'attributes': list(map(str, x.attributes()))
|
} for x in results)
|
|
elif setype == ROLE:
|
q = setools.RoleQuery(_pol)
|
if name:
|
q.name = name
|
|
return ({
|
'name': str(x),
|
'roles': list(map(str, x.expand())),
|
'types': list(map(str, x.types())),
|
} for x in q.results())
|
|
elif setype == ATTRIBUTE:
|
q = setools.TypeAttributeQuery(_pol)
|
if name:
|
q.name = name
|
|
return ({
|
'name': str(x),
|
'types': list(map(str, x.expand())),
|
} for x in q.results())
|
|
elif setype == PORT:
|
q = setools.PortconQuery(_pol)
|
if name:
|
ports = [int(i) for i in name.split("-")]
|
if len(ports) == 2:
|
q.ports = ports
|
elif len(ports) == 1:
|
q.ports = (ports[0], ports[0])
|
|
if _pol.mls:
|
return ({
|
'high': x.ports.high,
|
'protocol': str(x.protocol),
|
'range': str(x.context.range_),
|
'type': str(x.context.type_),
|
'low': x.ports.low,
|
} for x in q.results())
|
return ({
|
'high': x.ports.high,
|
'protocol': str(x.protocol),
|
'type': str(x.context.type_),
|
'low': x.ports.low,
|
} for x in q.results())
|
|
elif setype == USER:
|
q = setools.UserQuery(_pol)
|
if name:
|
q.name = name
|
|
if _pol.mls:
|
return ({
|
'range': str(x.mls_range),
|
'name': str(x),
|
'roles': list(map(str, x.roles)),
|
'level': str(x.mls_level),
|
} for x in q.results())
|
return ({
|
'name': str(x),
|
'roles': list(map(str, x.roles)),
|
} for x in q.results())
|
|
elif setype == BOOLEAN:
|
q = setools.BoolQuery(_pol)
|
if name:
|
q.name = name
|
|
return ({
|
'name': str(x),
|
'state': x.state,
|
} for x in q.results())
|
|
elif setype == TCLASS:
|
q = setools.ObjClassQuery(_pol)
|
if name:
|
q.name = name
|
|
return ({
|
'name': str(x),
|
'permlist': list(x.perms),
|
} for x in q.results())
|
|
else:
|
raise ValueError("Invalid type")
|
|
|
def _setools_rule_to_dict(rule):
|
d = {
|
'type': str(rule.ruletype),
|
'source': str(rule.source),
|
'target': str(rule.target),
|
'class': str(rule.tclass),
|
}
|
|
# Evaluate boolean expression associated with given rule (if there is any)
|
try:
|
# Get state of all booleans in the conditional expression
|
boolstate = {}
|
for boolean in rule.conditional.booleans:
|
boolstate[str(boolean)] = boolean.state
|
# evaluate if the rule is enabled
|
enabled = rule.conditional.evaluate(**boolstate) == rule.conditional_block
|
except AttributeError:
|
# non-conditional rules are always enabled
|
enabled = True
|
|
d['enabled'] = enabled
|
|
try:
|
d['permlist'] = list(map(str, rule.perms))
|
except AttributeError:
|
pass
|
|
try:
|
d['transtype'] = str(rule.default)
|
except AttributeError:
|
pass
|
|
try:
|
d['boolean'] = [(str(rule.conditional), enabled)]
|
except AttributeError:
|
pass
|
|
try:
|
d['filename'] = rule.filename
|
except AttributeError:
|
pass
|
|
return d
|
|
|
def search(types, seinfo=None):
|
if not seinfo:
|
seinfo = {}
|
valid_types = set([ALLOW, AUDITALLOW, NEVERALLOW, DONTAUDIT, TRANSITION, ROLE_ALLOW])
|
for setype in types:
|
if setype not in valid_types:
|
raise ValueError("Type has to be in %s" % " ".join(valid_types))
|
|
source = None
|
if SOURCE in seinfo:
|
source = str(seinfo[SOURCE])
|
|
target = None
|
if TARGET in seinfo:
|
target = str(seinfo[TARGET])
|
|
tclass = None
|
if CLASS in seinfo:
|
tclass = str(seinfo[CLASS]).split(',')
|
|
toret = []
|
|
tertypes = []
|
if ALLOW in types:
|
tertypes.append(ALLOW)
|
if NEVERALLOW in types:
|
tertypes.append(NEVERALLOW)
|
if AUDITALLOW in types:
|
tertypes.append(AUDITALLOW)
|
if DONTAUDIT in types:
|
tertypes.append(DONTAUDIT)
|
|
if len(tertypes) > 0:
|
q = setools.TERuleQuery(_pol,
|
ruletype=tertypes,
|
source=source,
|
target=target,
|
tclass=tclass)
|
|
if PERMS in seinfo:
|
q.perms = seinfo[PERMS]
|
|
toret += [_setools_rule_to_dict(x) for x in q.results()]
|
|
if TRANSITION in types:
|
rtypes = ['type_transition', 'type_change', 'type_member']
|
q = setools.TERuleQuery(_pol,
|
ruletype=rtypes,
|
source=source,
|
target=target,
|
tclass=tclass)
|
|
if PERMS in seinfo:
|
q.perms = seinfo[PERMS]
|
|
toret += [_setools_rule_to_dict(x) for x in q.results()]
|
|
if ROLE_ALLOW in types:
|
ratypes = ['allow']
|
q = setools.RBACRuleQuery(_pol,
|
ruletype=ratypes,
|
source=source,
|
target=target,
|
tclass=tclass)
|
|
for r in q.results():
|
toret.append({'source': str(r.source),
|
'target': str(r.target)})
|
|
return toret
|
|
|
def get_conditionals(src, dest, tclass, perm):
|
tdict = {}
|
tlist = []
|
src_list = [src]
|
dest_list = [dest]
|
# add assigned attributes
|
try:
|
src_list += list(filter(lambda x: x['name'] == src, get_all_types_info()))[0]['attributes']
|
except:
|
pass
|
try:
|
dest_list += list(filter(lambda x: x['name'] == dest, get_all_types_info()))[0]['attributes']
|
except:
|
pass
|
allows = map(lambda y: y, filter(lambda x:
|
x['source'] in src_list and
|
x['target'] in dest_list and
|
set(perm).issubset(x[PERMS]) and
|
'boolean' in x,
|
get_all_allow_rules()))
|
|
try:
|
for i in allows:
|
tdict.update({'source': i['source'], 'boolean': i['boolean']})
|
if tdict not in tlist:
|
tlist.append(tdict)
|
tdict = {}
|
except KeyError:
|
return(tlist)
|
|
return (tlist)
|
|
|
def get_conditionals_format_text(cond):
|
|
enabled = False
|
for x in cond:
|
if x['boolean'][0][1]:
|
enabled = True
|
break
|
return _("-- Allowed %s [ %s ]") % (enabled, " || ".join(set(map(lambda x: "%s=%d" % (x['boolean'][0][0], x['boolean'][0][1]), cond))))
|
|
|
def get_types_from_attribute(attribute):
|
return list(info(ATTRIBUTE, attribute))[0]["types"]
|
|
|
def get_file_types(setype):
|
flist = []
|
mpaths = {}
|
for f in get_all_file_types():
|
if f.startswith(gen_short_name(setype)):
|
flist.append(f)
|
fcdict = get_fcdict()
|
for f in flist:
|
try:
|
mpaths[f] = (fcdict[f]["regex"], file_type_str[fcdict[f]["ftype"]])
|
except KeyError:
|
mpaths[f] = []
|
return mpaths
|
|
|
def get_real_type_name(name):
|
"""Return the real name of a type
|
|
* If 'name' refers to a type alias, return the corresponding type name.
|
* Otherwise return the original name (even if the type does not exist).
|
"""
|
if not name:
|
return name
|
|
try:
|
return next(info(TYPE, name))["name"]
|
except (RuntimeError, StopIteration):
|
return name
|
|
def get_writable_files(setype):
|
file_types = get_all_file_types()
|
all_writes = []
|
mpaths = {}
|
permlist = search([ALLOW], {'source': setype, 'permlist': ['open', 'write'], 'class': 'file'})
|
if permlist is None or len(permlist) == 0:
|
return mpaths
|
|
fcdict = get_fcdict()
|
|
attributes = ["proc_type", "sysctl_type"]
|
for i in permlist:
|
if i['target'] in attributes:
|
continue
|
if "enabled" in i:
|
if not i["enabled"]:
|
continue
|
if i['target'].endswith("_t"):
|
if i['target'] not in file_types:
|
continue
|
if i['target'] not in all_writes:
|
if i['target'] != setype:
|
all_writes.append(i['target'])
|
else:
|
for t in get_types_from_attribute(i['target']):
|
if t not in all_writes:
|
all_writes.append(t)
|
|
for f in all_writes:
|
try:
|
mpaths[f] = (fcdict[f]["regex"], file_type_str[fcdict[f]["ftype"]])
|
except KeyError:
|
mpaths[f] = [] # {"regex":[],"paths":[]}
|
return mpaths
|
|
|
def find_file(reg):
|
if os.path.exists(reg):
|
return [reg]
|
try:
|
pat = re.compile(r"%s$" % reg)
|
except:
|
print("bad reg:", reg)
|
return []
|
p = reg
|
if p.endswith("(/.*)?"):
|
p = p[:-6] + "/"
|
|
path = os.path.dirname(p)
|
|
try: # Bug fix: when "all files on system"
|
if path[-1] != "/": # is pass in it breaks without try block
|
path += "/"
|
except IndexError:
|
print("try failed got an IndexError")
|
pass
|
|
try:
|
pat = re.compile(r"%s$" % reg)
|
return [x for x in map(lambda x: path + x, os.listdir(path)) if pat.match(x)]
|
except:
|
return []
|
|
|
def find_all_files(domain, exclude_list=[]):
|
executable_files = get_entrypoints(domain)
|
for exe in executable_files.keys():
|
if exe.endswith("_exec_t") and exe not in exclude_list:
|
for path in executable_files[exe]:
|
for f in find_file(path):
|
return f
|
return None
|
|
|
def find_entrypoint_path(exe, exclude_list=[]):
|
fcdict = get_fcdict()
|
try:
|
if exe.endswith("_exec_t") and exe not in exclude_list:
|
for path in fcdict[exe]["regex"]:
|
for f in find_file(path):
|
return f
|
except KeyError:
|
pass
|
return None
|
|
|
def read_file_equiv(edict, fc_path, modify):
|
try:
|
with open(fc_path, "r") as fd:
|
for e in fd:
|
f = e.split()
|
if f and not f[0].startswith('#'):
|
edict[f[0]] = {"equiv": f[1], "modify": modify}
|
except OSError as e:
|
if e.errno != errno.ENOENT:
|
raise
|
return edict
|
|
|
def get_file_equiv_modified(fc_path=selinux.selinux_file_context_path()):
|
global file_equiv_modified
|
if file_equiv_modified:
|
return file_equiv_modified
|
file_equiv_modified = {}
|
file_equiv_modified = read_file_equiv(file_equiv_modified, fc_path + ".subs", modify=True)
|
return file_equiv_modified
|
|
|
def get_file_equiv(fc_path=selinux.selinux_file_context_path()):
|
global file_equiv
|
if file_equiv:
|
return file_equiv
|
file_equiv = get_file_equiv_modified(fc_path)
|
file_equiv = read_file_equiv(file_equiv, fc_path + ".subs_dist", modify=False)
|
return file_equiv
|
|
|
def get_local_file_paths(fc_path=selinux.selinux_file_context_path()):
|
global local_files
|
if local_files:
|
return local_files
|
local_files = []
|
try:
|
with open(fc_path + ".local", "r") as fd:
|
fc = fd.readlines()
|
except OSError as e:
|
if e.errno != errno.ENOENT:
|
raise
|
return []
|
for i in fc:
|
rec = i.split()
|
if len(rec) == 0:
|
continue
|
try:
|
if len(rec) > 2:
|
ftype = trans_file_type_str[rec[1]]
|
else:
|
ftype = "a"
|
|
local_files.append((rec[0], ftype))
|
except KeyError:
|
pass
|
return local_files
|
|
|
def get_fcdict(fc_path=selinux.selinux_file_context_path()):
|
global fcdict
|
if fcdict:
|
return fcdict
|
fd = open(fc_path, "r")
|
fc = fd.readlines()
|
fd.close()
|
fd = open(fc_path + ".homedirs", "r")
|
fc += fd.readlines()
|
fd.close()
|
fcdict = {}
|
try:
|
with open(fc_path + ".local", "r") as fd:
|
fc += fd.readlines()
|
except OSError as e:
|
if e.errno != errno.ENOENT:
|
raise
|
|
for i in fc:
|
rec = i.split()
|
try:
|
if len(rec) > 2:
|
ftype = trans_file_type_str[rec[1]]
|
else:
|
ftype = "a"
|
|
t = rec[-1].split(":")[2]
|
if t in fcdict:
|
fcdict[t]["regex"].append(rec[0])
|
else:
|
fcdict[t] = {"regex": [rec[0]], "ftype": ftype}
|
except:
|
pass
|
|
fcdict["logfile"] = {"regex": ["all log files"]}
|
fcdict["user_tmp_type"] = {"regex": ["all user tmp files"]}
|
fcdict["user_home_type"] = {"regex": ["all user home files"]}
|
fcdict["virt_image_type"] = {"regex": ["all virtual image files"]}
|
fcdict["noxattrfs"] = {"regex": ["all files on file systems which do not support extended attributes"]}
|
fcdict["sandbox_tmpfs_type"] = {"regex": ["all sandbox content in tmpfs file systems"]}
|
fcdict["user_tmpfs_type"] = {"regex": ["all user content in tmpfs file systems"]}
|
fcdict["file_type"] = {"regex": ["all files on the system"]}
|
fcdict["samba_share_t"] = {"regex": ["use this label for random content that will be shared using samba"]}
|
return fcdict
|
|
|
def get_transitions_into(setype):
|
try:
|
return [x for x in search([TRANSITION], {'class': 'process'}) if x["transtype"] == setype]
|
except (TypeError, AttributeError):
|
pass
|
return None
|
|
|
def get_transitions(setype):
|
try:
|
return search([TRANSITION], {'source': setype, 'class': 'process'})
|
except (TypeError, AttributeError):
|
pass
|
return None
|
|
|
def get_file_transitions(setype):
|
try:
|
return [x for x in search([TRANSITION], {'source': setype}) if x['class'] != "process"]
|
except (TypeError, AttributeError):
|
pass
|
return None
|
|
|
def get_boolean_rules(setype, boolean):
|
boollist = []
|
permlist = search([ALLOW], {'source': setype})
|
for p in permlist:
|
if "boolean" in p:
|
try:
|
for b in p["boolean"]:
|
if boolean in b:
|
boollist.append(p)
|
except:
|
pass
|
return boollist
|
|
|
def get_all_entrypoints():
|
return get_types_from_attribute("entry_type")
|
|
|
def get_entrypoint_types(setype):
|
q = setools.TERuleQuery(_pol,
|
ruletype=[ALLOW],
|
source=setype,
|
tclass=["file"],
|
perms=["entrypoint"])
|
return [str(x.target) for x in q.results() if x.source == setype]
|
|
|
def get_init_transtype(path):
|
entrypoint = selinux.getfilecon(path)[1].split(":")[2]
|
try:
|
entrypoints = list(filter(lambda x: x['target'] == entrypoint, search([TRANSITION], {'source': "init_t", 'class': 'process'})))
|
return entrypoints[0]["transtype"]
|
except (TypeError, AttributeError, IndexError):
|
pass
|
return None
|
|
|
def get_init_entrypoint(transtype):
|
q = setools.TERuleQuery(_pol,
|
ruletype=["type_transition"],
|
source="init_t",
|
tclass=["process"])
|
entrypoints = []
|
for i in q.results():
|
try:
|
if i.default == transtype:
|
entrypoints.append(i.target)
|
except AttributeError:
|
continue
|
|
return entrypoints
|
|
def get_init_entrypoints_str():
|
q = setools.TERuleQuery(_pol,
|
ruletype=["type_transition"],
|
source="init_t",
|
tclass=["process"])
|
entrypoints = {}
|
for i in q.results():
|
try:
|
transtype = str(i.default)
|
if transtype in entrypoints:
|
entrypoints[transtype].append(str(i.target))
|
else:
|
entrypoints[transtype] = [str(i.target)]
|
except AttributeError:
|
continue
|
|
return entrypoints
|
|
def get_init_entrypoint_target(entrypoint):
|
try:
|
entrypoints = map(lambda x: x['transtype'], search([TRANSITION], {'source': "init_t", 'target': entrypoint, 'class': 'process'}))
|
return list(entrypoints)[0]
|
except (TypeError, IndexError):
|
pass
|
return None
|
|
|
def get_entrypoints(setype):
|
fcdict = get_fcdict()
|
mpaths = {}
|
for f in get_entrypoint_types(setype):
|
try:
|
mpaths[f] = (fcdict[f]["regex"], file_type_str[fcdict[f]["ftype"]])
|
except KeyError:
|
mpaths[f] = []
|
return mpaths
|
|
|
def get_methods():
|
global methods
|
if len(methods) > 0:
|
return methods
|
gen_interfaces()
|
fn = defaults.interface_info()
|
try:
|
fd = open(fn)
|
# List of per_role_template interfaces
|
ifs = interfaces.InterfaceSet()
|
ifs.from_file(fd)
|
methods = list(ifs.interfaces.keys())
|
fd.close()
|
except:
|
sys.stderr.write("could not open interface info [%s]\n" % fn)
|
sys.exit(1)
|
|
methods.sort()
|
return methods
|
|
|
def get_all_types():
|
global all_types
|
if all_types is None:
|
all_types = [x['name'] for x in info(TYPE)]
|
return all_types
|
|
def get_all_types_info():
|
global all_types_info
|
if all_types_info is None:
|
all_types_info = list(info(TYPE))
|
return all_types_info
|
|
def get_user_types():
|
global user_types
|
if user_types is None:
|
user_types = list(list(info(ATTRIBUTE, "userdomain"))[0]["types"])
|
return user_types
|
|
|
def get_all_role_allows():
|
global role_allows
|
if role_allows:
|
return role_allows
|
role_allows = {}
|
|
q = setools.RBACRuleQuery(_pol, ruletype=[ALLOW])
|
for r in q.results():
|
src = str(r.source)
|
tgt = str(r.target)
|
if src == "system_r" or tgt == "system_r":
|
continue
|
if src in role_allows:
|
role_allows[src].append(tgt)
|
else:
|
role_allows[src] = [tgt]
|
|
return role_allows
|
|
|
def get_all_entrypoint_domains():
|
import re
|
all_domains = []
|
types = sorted(get_all_types())
|
for i in types:
|
m = re.findall("(.*)%s" % "_exec_t$", i)
|
if len(m) > 0:
|
if len(re.findall("(.*)%s" % "_initrc$", m[0])) == 0 and m[0] not in all_domains:
|
all_domains.append(m[0])
|
return all_domains
|
|
|
def gen_interfaces():
|
try:
|
from commands import getstatusoutput
|
except ImportError:
|
from subprocess import getstatusoutput
|
ifile = defaults.interface_info()
|
headers = defaults.headers()
|
try:
|
if os.stat(headers).st_mtime <= os.stat(ifile).st_mtime:
|
return
|
except OSError:
|
pass
|
|
if os.getuid() != 0:
|
raise ValueError(_("You must regenerate interface info by running /usr/bin/sepolgen-ifgen"))
|
print(getstatusoutput("/usr/bin/sepolgen-ifgen")[1])
|
|
|
def gen_port_dict():
|
global portrecs
|
global portrecsbynum
|
if portrecs:
|
return (portrecs, portrecsbynum)
|
portrecsbynum = {}
|
portrecs = {}
|
for i in info(PORT):
|
if i['low'] == i['high']:
|
port = str(i['low'])
|
else:
|
port = "%s-%s" % (str(i['low']), str(i['high']))
|
|
if (i['type'], i['protocol']) in portrecs:
|
portrecs[(i['type'], i['protocol'])].append(port)
|
else:
|
portrecs[(i['type'], i['protocol'])] = [port]
|
|
if 'range' in i:
|
portrecsbynum[(i['low'], i['high'], i['protocol'])] = (i['type'], i['range'])
|
else:
|
portrecsbynum[(i['low'], i['high'], i['protocol'])] = (i['type'])
|
|
return (portrecs, portrecsbynum)
|
|
|
def get_all_domains():
|
global all_domains
|
if not all_domains:
|
all_domains = list(list(info(ATTRIBUTE, "domain"))[0]["types"])
|
return all_domains
|
|
|
def get_all_roles():
|
global roles
|
if roles:
|
return roles
|
|
q = setools.RoleQuery(_pol)
|
roles = [str(x) for x in q.results() if str(x) != "object_r"]
|
return roles
|
|
|
def get_selinux_users():
|
global selinux_user_list
|
if not selinux_user_list:
|
selinux_user_list = list(info(USER))
|
if _pol.mls:
|
for x in selinux_user_list:
|
x['range'] = "".join(x['range'].split(" "))
|
return selinux_user_list
|
|
|
def get_login_mappings():
|
global login_mappings
|
if login_mappings:
|
return login_mappings
|
|
fd = open(selinux.selinux_usersconf_path(), "r")
|
buf = fd.read()
|
fd.close()
|
login_mappings = []
|
for b in buf.split("\n"):
|
b = b.strip()
|
if len(b) == 0 or b.startswith("#"):
|
continue
|
x = b.split(":")
|
login_mappings.append({"name": x[0], "seuser": x[1], "mls": ":".join(x[2:])})
|
return login_mappings
|
|
|
def get_all_users():
|
return sorted(map(lambda x: x['name'], get_selinux_users()))
|
|
|
def get_all_file_types():
|
global file_types
|
if file_types:
|
return file_types
|
file_types = list(sorted(info(ATTRIBUTE, "file_type"))[0]["types"])
|
return file_types
|
|
|
def get_all_port_types():
|
global port_types
|
if port_types:
|
return port_types
|
port_types = list(sorted(info(ATTRIBUTE, "port_type"))[0]["types"])
|
return port_types
|
|
|
def get_all_bools():
|
global bools
|
if not bools:
|
bools = list(info(BOOLEAN))
|
return bools
|
|
|
def prettyprint(f, trim):
|
return " ".join(f[:-len(trim)].split("_"))
|
|
|
def markup(f):
|
return f
|
|
|
def get_description(f, markup=markup):
|
|
txt = "Set files with the %s type, if you want to " % markup(f)
|
|
if f.endswith("_var_run_t"):
|
return txt + "store the %s files under the /run or /var/run directory." % prettyprint(f, "_var_run_t")
|
if f.endswith("_pid_t"):
|
return txt + "store the %s files under the /run directory." % prettyprint(f, "_pid_t")
|
if f.endswith("_var_lib_t"):
|
return txt + "store the %s files under the /var/lib directory." % prettyprint(f, "_var_lib_t")
|
if f.endswith("_var_t"):
|
return txt + "store the %s files under the /var directory." % prettyprint(f, "_var_lib_t")
|
if f.endswith("_var_spool_t"):
|
return txt + "store the %s files under the /var/spool directory." % prettyprint(f, "_spool_t")
|
if f.endswith("_spool_t"):
|
return txt + "store the %s files under the /var/spool directory." % prettyprint(f, "_spool_t")
|
if f.endswith("_cache_t") or f.endswith("_var_cache_t"):
|
return txt + "store the files under the /var/cache directory."
|
if f.endswith("_keytab_t"):
|
return txt + "treat the files as kerberos keytab files."
|
if f.endswith("_lock_t"):
|
return txt + "treat the files as %s lock data, stored under the /var/lock directory" % prettyprint(f, "_lock_t")
|
if f.endswith("_log_t"):
|
return txt + "treat the data as %s log data, usually stored under the /var/log directory." % prettyprint(f, "_log_t")
|
if f.endswith("_config_t"):
|
return txt + "treat the files as %s configuration data, usually stored under the /etc directory." % prettyprint(f, "_config_t")
|
if f.endswith("_conf_t"):
|
return txt + "treat the files as %s configuration data, usually stored under the /etc directory." % prettyprint(f, "_conf_t")
|
if f.endswith("_exec_t"):
|
return txt + "transition an executable to the %s_t domain." % f[:-len("_exec_t")]
|
if f.endswith("_cgi_content_t"):
|
return txt + "treat the files as %s cgi content." % prettyprint(f, "_cgi_content_t")
|
if f.endswith("_rw_content_t"):
|
return txt + "treat the files as %s read/write content." % prettyprint(f, "_rw_content_t")
|
if f.endswith("_rw_t"):
|
return txt + "treat the files as %s read/write content." % prettyprint(f, "_rw_t")
|
if f.endswith("_write_t"):
|
return txt + "treat the files as %s read/write content." % prettyprint(f, "_write_t")
|
if f.endswith("_db_t"):
|
return txt + "treat the files as %s database content." % prettyprint(f, "_db_t")
|
if f.endswith("_ra_content_t"):
|
return txt + "treat the files as %s read/append content." % prettyprint(f, "_ra_content_t")
|
if f.endswith("_cert_t"):
|
return txt + "treat the files as %s certificate data." % prettyprint(f, "_cert_t")
|
if f.endswith("_key_t"):
|
return txt + "treat the files as %s key data." % prettyprint(f, "_key_t")
|
|
if f.endswith("_secret_t"):
|
return txt + "treat the files as %s secret data." % prettyprint(f, "_key_t")
|
|
if f.endswith("_ra_t"):
|
return txt + "treat the files as %s read/append content." % prettyprint(f, "_ra_t")
|
|
if f.endswith("_ro_t"):
|
return txt + "treat the files as %s read/only content." % prettyprint(f, "_ro_t")
|
|
if f.endswith("_modules_t"):
|
return txt + "treat the files as %s modules." % prettyprint(f, "_modules_t")
|
|
if f.endswith("_content_t"):
|
return txt + "treat the files as %s content." % prettyprint(f, "_content_t")
|
|
if f.endswith("_state_t"):
|
return txt + "treat the files as %s state data." % prettyprint(f, "_state_t")
|
|
if f.endswith("_files_t"):
|
return txt + "treat the files as %s content." % prettyprint(f, "_files_t")
|
|
if f.endswith("_file_t"):
|
return txt + "treat the files as %s content." % prettyprint(f, "_file_t")
|
|
if f.endswith("_data_t"):
|
return txt + "treat the files as %s content." % prettyprint(f, "_data_t")
|
|
if f.endswith("_file_t"):
|
return txt + "treat the data as %s content." % prettyprint(f, "_file_t")
|
|
if f.endswith("_tmp_t"):
|
return txt + "store %s temporary files in the /tmp directories." % prettyprint(f, "_tmp_t")
|
if f.endswith("_etc_t"):
|
return txt + "store %s files in the /etc directories." % prettyprint(f, "_tmp_t")
|
if f.endswith("_home_t"):
|
return txt + "store %s files in the users home directory." % prettyprint(f, "_home_t")
|
if f.endswith("_tmpfs_t"):
|
return txt + "store %s files on a tmpfs file system." % prettyprint(f, "_tmpfs_t")
|
if f.endswith("_unit_file_t"):
|
return txt + "treat files as a systemd unit file."
|
if f.endswith("_htaccess_t"):
|
return txt + "treat the file as a %s access file." % prettyprint(f, "_htaccess_t")
|
|
return txt + "treat the files as %s data." % prettyprint(f, "_t")
|
|
|
def get_all_attributes():
|
global all_attributes
|
if not all_attributes:
|
all_attributes = list(sorted(map(lambda x: x['name'], info(ATTRIBUTE))))
|
return all_attributes
|
|
|
def _dict_has_perms(dict, perms):
|
for perm in perms:
|
if perm not in dict[PERMS]:
|
return False
|
return True
|
|
|
def gen_short_name(setype):
|
all_domains = get_all_domains()
|
if setype.endswith("_t"):
|
# replace aliases with corresponding types
|
setype = get_real_type_name(setype)
|
domainname = setype[:-2]
|
else:
|
domainname = setype
|
if domainname + "_t" not in all_domains:
|
raise ValueError("domain %s_t does not exist" % domainname)
|
if domainname[-1] == 'd':
|
short_name = domainname[:-1] + "_"
|
else:
|
short_name = domainname + "_"
|
return (domainname, short_name)
|
|
def get_all_allow_rules():
|
global all_allow_rules
|
if not all_allow_rules:
|
all_allow_rules = search([ALLOW])
|
return all_allow_rules
|
|
def get_all_transitions():
|
global all_transitions
|
if not all_transitions:
|
all_transitions = list(search([TRANSITION]))
|
return all_transitions
|
|
def get_bools(setype):
|
bools = []
|
domainbools = []
|
domainname, short_name = gen_short_name(setype)
|
for i in map(lambda x: x['boolean'], filter(lambda x: 'boolean' in x and x['source'] == setype, get_all_allow_rules())):
|
for b in i:
|
if not isinstance(b, tuple):
|
continue
|
try:
|
enabled = selinux.security_get_boolean_active(b[0])
|
except OSError:
|
enabled = b[1]
|
if b[0].startswith(short_name) or b[0].startswith(domainname):
|
if (b[0], enabled) not in domainbools and (b[0], not enabled) not in domainbools:
|
domainbools.append((b[0], enabled))
|
else:
|
if (b[0], enabled) not in bools and (b[0], not enabled) not in bools:
|
bools.append((b[0], enabled))
|
return (domainbools, bools)
|
|
|
def get_all_booleans():
|
global booleans
|
if not booleans:
|
booleans = selinux.security_get_boolean_names()[1]
|
return booleans
|
|
|
def policy_xml(path="/usr/share/selinux/devel/policy.xml"):
|
try:
|
fd = gzip.open(path)
|
buf = fd.read()
|
fd.close()
|
except IOError:
|
fd = open(path)
|
buf = fd.read()
|
fd.close()
|
return buf
|
|
|
def gen_bool_dict(path="/usr/share/selinux/devel/policy.xml"):
|
global booleans_dict
|
if booleans_dict:
|
return booleans_dict
|
import xml.etree.ElementTree
|
booleans_dict = {}
|
try:
|
tree = xml.etree.ElementTree.fromstring(policy_xml(path))
|
for l in tree.findall("layer"):
|
for m in l.findall("module"):
|
for b in m.findall("tunable"):
|
desc = b.find("desc").find("p").text.strip("\n")
|
desc = re.sub("\n", " ", desc)
|
booleans_dict[b.get('name')] = (m.get("name"), b.get('dftval'), desc)
|
for b in m.findall("bool"):
|
desc = b.find("desc").find("p").text.strip("\n")
|
desc = re.sub("\n", " ", desc)
|
booleans_dict[b.get('name')] = (m.get("name"), b.get('dftval'), desc)
|
for i in tree.findall("bool"):
|
desc = i.find("desc").find("p").text.strip("\n")
|
desc = re.sub("\n", " ", desc)
|
booleans_dict[i.get('name')] = ("global", i.get('dftval'), desc)
|
for i in tree.findall("tunable"):
|
desc = i.find("desc").find("p").text.strip("\n")
|
desc = re.sub("\n", " ", desc)
|
booleans_dict[i.get('name')] = ("global", i.get('dftval'), desc)
|
except IOError:
|
pass
|
return booleans_dict
|
|
|
def boolean_category(boolean):
|
booleans_dict = gen_bool_dict()
|
if boolean in booleans_dict:
|
return _(booleans_dict[boolean][0])
|
else:
|
return _("unknown")
|
|
|
def boolean_desc(boolean):
|
booleans_dict = gen_bool_dict()
|
if boolean in booleans_dict:
|
return _(booleans_dict[boolean][2])
|
else:
|
desc = boolean.split("_")
|
return "Allow %s to %s" % (desc[0], " ".join(desc[1:]))
|
|
|
def get_os_version():
|
os_version = ""
|
pkg_name = "selinux-policy"
|
try:
|
try:
|
from commands import getstatusoutput
|
except ImportError:
|
from subprocess import getstatusoutput
|
rc, output = getstatusoutput("rpm -q '%s'" % pkg_name)
|
if rc == 0:
|
os_version = output.split(".")[-2]
|
except:
|
os_version = ""
|
|
if os_version[0:2] == "fc":
|
os_version = "Fedora" + os_version[2:]
|
elif os_version[0:2] == "el":
|
os_version = "RHEL" + os_version[2:]
|
else:
|
os_version = ""
|
|
return os_version
|
|
|
def reinit():
|
global all_attributes
|
global all_domains
|
global all_types
|
global booleans
|
global booleans_dict
|
global bools
|
global fcdict
|
global file_types
|
global local_files
|
global methods
|
global methods
|
global portrecs
|
global portrecsbynum
|
global port_types
|
global role_allows
|
global roles
|
global login_mappings
|
global selinux_user_list
|
global user_types
|
all_attributes = None
|
all_domains = None
|
all_types = None
|
booleans = None
|
booleans_dict = None
|
bools = None
|
fcdict = None
|
file_types = None
|
local_files = None
|
methods = None
|
methods = None
|
portrecs = None
|
portrecsbynum = None
|
port_types = None
|
role_allows = None
|
roles = None
|
user_types = None
|
login_mappings = None
|
selinux_user_list = None
|
