Wala shell

Advanced PHP File Manager

Server IP: 127.0.0.1 | PHP Version: 8.3.30 | Domains Hosted: 54

Name	Size	Last Modified	Actions
📁 ..	-	2026-03-21 15:56
📄 __init__.py # Author: Dan Walsh <dwalsh@redhat.com> # Author: Ryan Hallisey <rhallise@redhat.com> # Author: Jason Zaman <perfinion@gentoo.org> import errno import selinux import glob import sepolgen.defaults as defaults import sepolgen.interfaces as interfaces import sys import os import re import gzip from setools.boolquery import BoolQuery from setools.portconquery import PortconQuery from setools.policyrep import SELinuxPolicy from setools.objclassquery import ObjClassQuery from setools.rbacrulequery import RBACRuleQuery from setools.rolequery import RoleQuery from setools.terulequery import TERuleQuery from setools.typeattrquery import TypeAttributeQuery from setools.typequery import TypeQuery from setools.userquery import UserQuery PROGNAME = "selinux-python" try: import gettext kwargs = {} if sys.version_info < (3,): kwargs['unicode'] = True t = gettext.translation(PROGNAME, localedir="/usr/share/locale", kwargs, fallback=True) _ = t.gettext 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_bool_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 = 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) def init_policy(): policy_file = get_installed_policy() policy(policy_file) def info(setype, name=None): global _pol if not _pol: init_policy() if setype == TYPE: q = 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 = 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 = 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 = 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 = 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 = BoolQuery(_pol) if name: q.name = name return ({ 'name': str(x), 'state': x.state, } for x in q.results()) elif setype == TCLASS: q = 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['booleans'] = [(str(b), b.state) for b in rule.conditional.booleans] except AttributeError: pass try: d['conditional'] = str(rule.conditional) except AttributeError: pass try: d['filename'] = rule.filename except AttributeError: pass return d def search(types, seinfo=None): global _pol if not _pol: init_policy() 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 = 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 = 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 = 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 'conditional' in x, get_all_allow_rules())) try: for i in allows: tdict.update({'source': i['source'], 'conditional': (i['conditional'], i['enabled'])}) 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['conditional'][1]: enabled = True break return _("-- Allowed %s [ %s ]") % (enabled, " \|\| ".join(set(map(lambda x: "%s=%d" % (x['conditional'][0], x['conditional'][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") 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 "booleans" in p: try: for b in p["booleans"]: 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 = 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 = 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 = 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 = 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 global _pol if not _pol: init_policy() q = 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, "_secret_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, "_etc_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_bool_rules(): global all_bool_rules if not all_bool_rules: q = TERuleQuery(_pol, boolean=".", boolean_regex=True, ruletype=[ALLOW, DONTAUDIT]) all_bool_rules = [_setools_rule_to_dict(x) for x in q.results()] return all_bool_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['booleans'], filter(lambda x: 'booleans' in x and x['source'] == setype, get_all_bool_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 {subject} to {rest}").format(subject=desc[0], rest=" ".join(desc[1:])) def get_os_version(): import subprocess system_release = "" try: import distro system_release = distro.name(pretty=True) except (ModuleNotFoundError, OSError, IOError, UnicodeError, subprocess.CalledProcessError): system_release = "Misc" return system_release 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	36.94 KB	2023-12-13 14:46
📁 __pycache__	-	2026-03-21 15:45
📄 booleans.py # Copyright (C) 2012 Red Hat # see file 'COPYING' for use and warranty information # # setrans is a tool for analyzing process transitions in SELinux policy # # 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 sepolicy def expand_attribute(attribute): try: return sepolicy.info(sepolicy.ATTRIBUTE, attribute)[0]["types"] except RuntimeError: return [attribute] def get_types(src, tclass, perm): allows = sepolicy.search([sepolicy.ALLOW], {sepolicy.SOURCE: src, sepolicy.CLASS: tclass, sepolicy.PERMS: perm}) if not allows: raise TypeError("The %s type is not allowed to %s any types" % (src, ",".join(perm))) tlist = [] for l in map(lambda y: y[sepolicy.TARGET], filter(lambda x: set(perm).issubset(x[sepolicy.PERMS]), allows)): tlist = tlist + expand_attribute(l) return tlist	1.54 KB	2023-12-13 14:46
📄 communicate.py # Copyright (C) 2012 Red Hat # see file 'COPYING' for use and warranty information # # setrans is a tool for analyzing process transitions in SELinux policy # # 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 sepolicy import sys def usage(parser, msg): parser.print_help() sys.stderr.write("\n%s\n" % msg) sys.stderr.flush() sys.exit(1) def expand_attribute(attribute): try: return list(next(sepolicy.info(sepolicy.ATTRIBUTE, attribute))["types"]) except StopIteration: return [attribute] def get_types(src, tclass, perm): allows = sepolicy.search([sepolicy.ALLOW], {sepolicy.SOURCE: src, sepolicy.CLASS: tclass, sepolicy.PERMS: perm}) if not allows: raise ValueError("The %s type is not allowed to %s any types" % (src, ",".join(perm))) tlist = [] for l in map(lambda y: y[sepolicy.TARGET], filter(lambda x: set(perm).issubset(x[sepolicy.PERMS]), allows)): tlist = tlist + expand_attribute(l) return tlist	1.68 KB	2023-12-13 14:46
📄 generate.py # Copyright (C) 2007-2012 Red Hat # see file 'COPYING' for use and warranty information # # policygentool is a tool for the initial generation of SELinux policy # # 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 os import sys import stat import re import sepolicy from sepolicy import get_all_types, get_all_attributes, get_all_roles import time from .templates import executable from .templates import boolean from .templates import etc_rw from .templates import unit_file from .templates import var_cache from .templates import var_spool from .templates import var_lib from .templates import var_log from .templates import var_run from .templates import tmp from .templates import rw from .templates import network from .templates import script from .templates import spec from .templates import user import sepolgen.interfaces as interfaces import sepolgen.defaults as defaults ## ## I18N ## PROGNAME = "selinux-python" try: import gettext kwargs = {} if sys.version_info < (3,): kwargs['unicode'] = True t = gettext.translation(PROGNAME, localedir="/usr/share/locale", kwargs, fallback=True) _ = t.gettext except: try: import builtins builtins.__dict__['_'] = str except ImportError: import __builtin__ __builtin__.__dict__['_'] = unicode def get_rpm_nvr_from_header(hdr): 'Given an RPM header return the package NVR as a string' name = hdr['name'] version = hdr['version'] release = hdr['release'] release_version = version + "-" + release.split(".")[0] os_version = release.split(".")[1] return [name, release_version, os_version] def get_rpm_nvr_list(package): try: import rpm nvr = None ts = rpm.ts() mi = ts.dbMatch(rpm.RPMTAG_NAME, package) for h in mi: nvr = get_rpm_nvr_from_header(h) break except: print(("Failed to retrieve rpm info for %s") % package) nvr = None return nvr def get_all_ports(): dict = {} for p in sepolicy.info(sepolicy.PORT): if p['type'] == "reserved_port_t" or \ p['type'] == "port_t" or \ p['type'] == "hi_reserved_port_t" or \ p['type'] == "ephemeral_port_t" or \ p['type'] == "unreserved_port_t": continue dict[(p['low'], p['high'], p['protocol'])] = (p['type'], p.get('range')) return dict def get_all_users(): users = [x['name'] for x in sepolicy.info(sepolicy.USER)] users.remove("system_u") users.remove("root") users.sort() return users ALL = 0 RESERVED = 1 UNRESERVED = 2 PORTS = 3 ADMIN_TRANSITION_INTERFACE = "_admin$" USER_TRANSITION_INTERFACE = "_role$" DAEMON = 0 DBUS = 1 INETD = 2 CGI = 3 SANDBOX = 4 USER = 5 EUSER = 6 TUSER = 7 XUSER = 8 LUSER = 9 AUSER = 10 RUSER = 11 NEWTYPE = 12 poltype = {} poltype[DAEMON] = _("Standard Init Daemon") poltype[DBUS] = _("DBUS System Daemon") poltype[INETD] = _("Internet Services Daemon") poltype[CGI] = _("Web Application/Script (CGI)") poltype[SANDBOX] = _("Sandbox") poltype[USER] = _("User Application") poltype[EUSER] = _("Existing Domain Type") poltype[TUSER] = _("Minimal Terminal Login User Role") poltype[XUSER] = _("Minimal X Windows Login User Role") poltype[LUSER] = _("Desktop Login User Role") poltype[AUSER] = _("Administrator Login User Role") poltype[RUSER] = _("Confined Root Administrator Role") poltype[NEWTYPE] = _("Module information for a new type") def get_poltype_desc(): keys = poltype.keys() keys.sort() msg = _("Valid Types:\n") for k in keys: msg += "%2s: %s\n" % (k, poltype[k]) return msg APPLICATIONS = [DAEMON, DBUS, INETD, USER, CGI] USERS = [XUSER, TUSER, LUSER, AUSER, RUSER] def verify_ports(ports): if ports == "": return [] max_port = 2 16 try: temp = [] for a in ports.split(","): r = a.split("-") if len(r) > 2: raise ValueError if len(r) == 1: begin = int(r[0]) end = int(r[0]) else: begin = int(r[0]) end = int(r[1]) if begin > end: raise ValueError for p in range(begin, end + 1): if p < 1 or p > max_port: raise ValueError temp.append(p) return temp except ValueError: raise ValueError(_("Ports must be numbers or ranges of numbers from 1 to %d ") % max_port) class policy: def __init__(self, name, type): self.rpms = [] self.ports = {} self.all_roles = get_all_roles() self.types = [] if type not in poltype: raise ValueError(_("You must enter a valid policy type")) if not name: raise ValueError(_("You must enter a name for your policy module for your '%s'.") % poltype[type]) try: self.ports = get_all_ports() except ValueError as e: print("Can not get port types, must be root for this information") except RuntimeError as e: print("Can not get port types", e) self.symbols = {} self.symbols["openlog"] = "set_use_kerberos(True)" self.symbols["openlog"] = "set_use_kerb_rcache(True)" self.symbols["openlog"] = "set_use_syslog(True)" self.symbols["gethostby"] = "set_use_resolve(True)" self.symbols["getaddrinfo"] = "set_use_resolve(True)" self.symbols["getnameinfo"] = "set_use_resolve(True)" self.symbols["krb"] = "set_use_kerberos(True)" self.symbols["gss_accept_sec_context"] = "set_manage_krb5_rcache(True)" self.symbols["krb5_verify_init_creds"] = "set_manage_krb5_rcache(True)" self.symbols["krb5_rd_req"] = "set_manage_krb5_rcache(True)" self.symbols["__syslog_chk"] = "set_use_syslog(True)" self.symbols["getpwnam"] = "set_use_uid(True)" self.symbols["getpwuid"] = "set_use_uid(True)" self.symbols["dbus_"] = "set_use_dbus(True)" self.symbols["pam_"] = "set_use_pam(True)" self.symbols["pam_"] = "set_use_audit(True)" self.symbols["fork"] = "add_process('fork')" self.symbols["transition"] = "add_process('transition')" self.symbols["sigchld"] = "add_process('sigchld')" self.symbols["sigkill"] = "add_process('sigkill')" self.symbols["sigstop"] = "add_process('sigstop')" self.symbols["signull"] = "add_process('signull')" self.symbols["ptrace"] = "add_process('ptrace')" self.symbols["getsched"] = "add_process('getsched')" self.symbols["setsched"] = "add_process('setsched')" self.symbols["getsession"] = "add_process('getsession')" self.symbols["getpgid"] = "add_process('getpgid')" self.symbols["setpgid"] = "add_process('setpgid')" self.symbols["getcap"] = "add_process('getcap')" self.symbols["setcap"] = "add_process('setcap')" self.symbols["share"] = "add_process('share')" self.symbols["getattr"] = "add_process('getattr')" self.symbols["setexec"] = "add_process('setexec')" self.symbols["setfscreate"] = "add_process('setfscreate')" self.symbols["noatsecure"] = "add_process('noatsecure')" self.symbols["siginh"] = "add_process('siginh')" self.symbols["kill"] = "add_process('signal_perms')" self.symbols["setrlimit"] = "add_process('setrlimit')" self.symbols["rlimitinh"] = "add_process('rlimitinh')" self.symbols["dyntransition"] = "add_process('dyntransition')" self.symbols["setcurrent"] = "add_process('setcurrent')" self.symbols["execmem"] = "add_process('execmem')" self.symbols["execstack"] = "add_process('execstack')" self.symbols["execheap"] = "add_process('execheap')" self.symbols["setkeycreate"] = "add_process('setkeycreate')" self.symbols["setsockcreate"] = "add_process('setsockcreate')" self.symbols["chown"] = "add_capability('chown')" self.symbols["dac_override"] = "add_capability('dac_override')" self.symbols["dac_read_search"] = "add_capability('dac_read_search')" self.symbols["fowner"] = "add_capability('fowner')" self.symbols["fsetid"] = "add_capability('fsetid')" self.symbols["setgid"] = "add_capability('setgid')" self.symbols["setegid"] = "add_capability('setgid')" self.symbols["setresgid"] = "add_capability('setgid')" self.symbols["setregid"] = "add_capability('setgid')" self.symbols["setresuid"] = "add_capability('setuid')" self.symbols["setuid"] = "add_capability('setuid')" self.symbols["seteuid"] = "add_capability('setuid')" self.symbols["setreuid"] = "add_capability('setuid')" self.symbols["setresuid"] = "add_capability('setuid')" self.symbols["setpcap"] = "add_capability('setpcap')" self.symbols["linux_immutable"] = "add_capability('linux_immutable')" self.symbols["net_bind_service"] = "add_capability('net_bind_service')" self.symbols["net_broadcast"] = "add_capability('net_broadcast')" self.symbols["net_admin"] = "add_capability('net_admin')" self.symbols["net_raw"] = "add_capability('net_raw')" self.symbols["ipc_lock"] = "add_capability('ipc_lock')" self.symbols["ipc_owner"] = "add_capability('ipc_owner')" self.symbols["sys_module"] = "add_capability('sys_module')" self.symbols["sys_rawio"] = "add_capability('sys_rawio')" self.symbols["chroot"] = "add_capability('sys_chroot')" self.symbols["sys_chroot"] = "add_capability('sys_chroot')" self.symbols["sys_ptrace"] = "add_capability('sys_ptrace')" self.symbols["sys_pacct"] = "add_capability('sys_pacct')" self.symbols["mount"] = "add_capability('sys_admin')" self.symbols["unshare"] = "add_capability('sys_admin')" self.symbols["sys_admin"] = "add_capability('sys_admin')" self.symbols["sys_boot"] = "add_capability('sys_boot')" self.symbols["sys_nice"] = "add_capability('sys_nice')" self.symbols["sys_resource"] = "add_capability('sys_resource')" self.symbols["sys_time"] = "add_capability('sys_time')" self.symbols["sys_tty_config"] = "add_capability('sys_tty_config')" self.symbols["mknod"] = "add_capability('mknod')" self.symbols["lease"] = "add_capability('lease')" self.symbols["audit_write"] = "add_capability('audit_write')" self.symbols["audit_control"] = "add_capability('audit_control')" self.symbols["setfcap"] = "add_capability('setfcap')" self.DEFAULT_DIRS = {} self.DEFAULT_DIRS["/etc"] = ["etc_rw", [], etc_rw] self.DEFAULT_DIRS["/tmp"] = ["tmp", [], tmp] self.DEFAULT_DIRS["rw"] = ["rw", [], rw] self.DEFAULT_DIRS["/usr/lib/systemd/system"] = ["unit_file", [], unit_file] self.DEFAULT_DIRS["/lib/systemd/system"] = ["unit_file", [], unit_file] self.DEFAULT_DIRS["/etc/systemd/system"] = ["unit_file", [], unit_file] self.DEFAULT_DIRS["/var/cache"] = ["var_cache", [], var_cache] self.DEFAULT_DIRS["/var/lib"] = ["var_lib", [], var_lib] self.DEFAULT_DIRS["/var/log"] = ["var_log", [], var_log] self.DEFAULT_DIRS["/var/run"] = ["var_run", [], var_run] self.DEFAULT_DIRS["/var/spool"] = ["var_spool", [], var_spool] self.DEFAULT_EXT = {} self.DEFAULT_EXT["_tmp_t"] = tmp self.DEFAULT_EXT["_unit_file_t"] = unit_file self.DEFAULT_EXT["_var_cache_t"] = var_cache self.DEFAULT_EXT["_var_lib_t"] = var_lib self.DEFAULT_EXT["_var_log_t"] = var_log self.DEFAULT_EXT["_var_run_t"] = var_run self.DEFAULT_EXT["_var_spool_t"] = var_spool self.DEFAULT_EXT["_port_t"] = network self.DEFAULT_KEYS = ["/etc", "/var/cache", "/var/log", "/tmp", "rw", "/var/lib", "/var/run", "/var/spool", "/etc/systemd/system", "/usr/lib/systemd/system", "/lib/systemd/system"] self.DEFAULT_TYPES = ( (self.generate_daemon_types, self.generate_daemon_rules), (self.generate_dbusd_types, self.generate_dbusd_rules), (self.generate_inetd_types, self.generate_inetd_rules), (self.generate_cgi_types, self.generate_cgi_rules), (self.generate_sandbox_types, self.generate_sandbox_rules), (self.generate_userapp_types, self.generate_userapp_rules), (self.generate_existing_user_types, self.generate_existing_user_rules), (self.generate_min_login_user_types, self.generate_login_user_rules), (self.generate_x_login_user_types, self.generate_x_login_user_rules), (self.generate_login_user_types, self.generate_login_user_rules), (self.generate_admin_user_types, self.generate_login_user_rules), (self.generate_root_user_types, self.generate_root_user_rules), (self.generate_new_types, self.generate_new_rules)) if not re.match(r"^[a-zA-Z0-9-_]+$", name): raise ValueError(_("Name must be alphanumeric with no spaces. Consider using option \"-n MODULENAME\"")) if type == CGI: self.name = "httpd_%s_script" % name else: self.name = name self.file_name = name self.capabilities = [] self.processes = [] self.type = type self.initscript = "" self.program = None self.in_tcp = [False, False, False, []] self.in_udp = [False, False, False, []] self.out_tcp = [False, False, False, []] self.out_udp = [False, False, False, []] self.use_resolve = False self.use_tmp = False self.use_uid = False self.use_syslog = False self.use_kerberos = False self.manage_krb5_rcache = False self.use_pam = False self.use_dbus = False self.use_audit = False self.use_etc = self.type not in [EUSER, NEWTYPE] self.use_localization = self.type not in [EUSER, NEWTYPE] self.use_fd = self.type not in [EUSER, NEWTYPE] self.use_terminal = False self.use_mail = False self.booleans = {} self.files = {} self.dirs = {} self.found_tcp_ports = [] self.found_udp_ports = [] self.need_tcp_type = False self.need_udp_type = False self.admin_domains = [] self.existing_domains = [] self.transition_domains = [] self.transition_users = [] self.roles = [] def __isnetset(self, l): return l[ALL] or l[RESERVED] or l[UNRESERVED] or len(l[PORTS]) > 0 def set_admin_domains(self, admin_domains): self.admin_domains = admin_domains def set_existing_domains(self, existing_domains): self.existing_domains = existing_domains def set_admin_roles(self, roles): self.roles = roles def set_transition_domains(self, transition_domains): self.transition_domains = transition_domains def set_transition_users(self, transition_users): self.transition_users = transition_users def use_in_udp(self): return self.__isnetset(self.in_udp) def use_out_udp(self): return self.__isnetset(self.out_udp) def use_udp(self): return self.use_in_udp() or self.use_out_udp() def use_in_tcp(self): return self.__isnetset(self.in_tcp) def use_out_tcp(self): return self.__isnetset(self.out_tcp) def use_tcp(self): return self.use_in_tcp() or self.use_out_tcp() def use_network(self): return self.use_tcp() or self.use_udp() def find_port(self, port, protocol="tcp"): for begin, end, p in self.ports.keys(): if port >= begin and port <= end and protocol == p: return self.ports[begin, end, protocol] return None def set_program(self, program): if self.type not in APPLICATIONS: raise ValueError(_("User Role types can not be assigned executables.")) self.program = program def set_init_script(self, initscript): if self.type != DAEMON: raise ValueError(_("Only Daemon apps can use an init script.")) self.initscript = initscript def set_in_tcp(self, all, reserved, unreserved, ports): self.in_tcp = [all, reserved, unreserved, verify_ports(ports)] def set_in_udp(self, all, reserved, unreserved, ports): self.in_udp = [all, reserved, unreserved, verify_ports(ports)] def set_out_tcp(self, all, ports): self.out_tcp = [all, False, False, verify_ports(ports)] def set_out_udp(self, all, ports): self.out_udp = [all, False, False, verify_ports(ports)] def set_use_resolve(self, val): if type(val) is not bool: raise ValueError(_("use_resolve must be a boolean value ")) self.use_resolve = val def set_use_syslog(self, val): if type(val) is not bool: raise ValueError(_("use_syslog must be a boolean value ")) self.use_syslog = val def set_use_kerberos(self, val): if type(val) is not bool: raise ValueError(_("use_kerberos must be a boolean value ")) self.use_kerberos = val def set_manage_krb5_rcache(self, val): if type(val) is not bool: raise ValueError(_("manage_krb5_rcache must be a boolean value ")) self.manage_krb5_rcache = val def set_use_pam(self, val): self.use_pam = (val is True) def set_use_dbus(self, val): self.use_dbus = (val is True) def set_use_audit(self, val): self.use_audit = (val is True) def set_use_etc(self, val): self.use_etc = (val is True) def set_use_localization(self, val): self.use_localization = (val is True) def set_use_fd(self, val): self.use_fd = (val is True) def set_use_terminal(self, val): self.use_terminal = (val is True) def set_use_mail(self, val): self.use_mail = (val is True) def set_use_tmp(self, val): if self.type in USERS: raise ValueError(_("USER Types automatically get a tmp type")) if val: self.DEFAULT_DIRS["/tmp"][1].append("/tmp") else: self.DEFAULT_DIRS["/tmp"][1] = [] def set_use_uid(self, val): self.use_uid = (val is True) def generate_uid_rules(self): if self.use_uid: return re.sub("TEMPLATETYPE", self.name, executable.te_uid_rules) else: return "" def generate_syslog_rules(self): if self.use_syslog: return re.sub("TEMPLATETYPE", self.name, executable.te_syslog_rules) else: return "" def generate_resolve_rules(self): if self.use_resolve: return re.sub("TEMPLATETYPE", self.name, executable.te_resolve_rules) else: return "" def generate_kerberos_rules(self): if self.use_kerberos: return re.sub("TEMPLATETYPE", self.name, executable.te_kerberos_rules) else: return "" def generate_manage_krb5_rcache_rules(self): if self.manage_krb5_rcache: return re.sub("TEMPLATETYPE", self.name, executable.te_manage_krb5_rcache_rules) else: return "" def generate_pam_rules(self): newte = "" if self.use_pam: newte = re.sub("TEMPLATETYPE", self.name, executable.te_pam_rules) return newte def generate_audit_rules(self): newte = "" if self.use_audit: newte = re.sub("TEMPLATETYPE", self.name, executable.te_audit_rules) return newte def generate_etc_rules(self): newte = "" if self.use_etc: newte = re.sub("TEMPLATETYPE", self.name, executable.te_etc_rules) return newte def generate_fd_rules(self): newte = "" if self.use_fd: newte = re.sub("TEMPLATETYPE", self.name, executable.te_fd_rules) return newte def generate_localization_rules(self): newte = "" if self.use_localization: newte = re.sub("TEMPLATETYPE", self.name, executable.te_localization_rules) return newte def generate_dbus_rules(self): newte = "" if self.type != DBUS and self.use_dbus: newte = re.sub("TEMPLATETYPE", self.name, executable.te_dbus_rules) return newte def generate_mail_rules(self): newte = "" if self.use_mail: newte = re.sub("TEMPLATETYPE", self.name, executable.te_mail_rules) return newte def generate_network_action(self, protocol, action, port_name): line = "" method = "corenet_%s_%s_%s" % (protocol, action, port_name) if method in sepolicy.get_methods(): line = "%s(%s_t)\n" % (method, self.name) else: line = """ gen_require(` type %s_t; ') allow %s_t %s_t:%s_socket name_%s; """ % (port_name, self.name, port_name, protocol, action) return line def generate_network_types(self): for i in self.in_tcp[PORTS]: rec = self.find_port(int(i), "tcp") if rec is None: self.need_tcp_type = True else: port_name = rec[0][:-2] line = self.generate_network_action("tcp", "bind", port_name) # line = "corenet_tcp_bind_%s(%s_t)\n" % (port_name, self.name) if line not in self.found_tcp_ports: self.found_tcp_ports.append(line) for i in self.out_tcp[PORTS]: rec = self.find_port(int(i), "tcp") if rec is None: self.need_tcp_type = True else: port_name = rec[0][:-2] line = self.generate_network_action("tcp", "connect", port_name) # line = "corenet_tcp_connect_%s(%s_t)\n" % (port_name, self.name) if line not in self.found_tcp_ports: self.found_tcp_ports.append(line) for i in self.in_udp[PORTS]: rec = self.find_port(int(i), "udp") if rec is None: self.need_udp_type = True else: port_name = rec[0][:-2] line = self.generate_network_action("udp", "bind", port_name) # line = "corenet_udp_bind_%s(%s_t)\n" % (port_name, self.name) if line not in self.found_udp_ports: self.found_udp_ports.append(line) if self.need_udp_type is True or self.need_tcp_type is True: return re.sub("TEMPLATETYPE", self.name, network.te_types) return "" def __find_path(self, file): for d in self.DEFAULT_DIRS: if file.find(d) == 0: self.DEFAULT_DIRS[d][1].append(file) return self.DEFAULT_DIRS[d] self.DEFAULT_DIRS["rw"][1].append(file) return self.DEFAULT_DIRS["rw"] def add_capability(self, capability): if capability not in self.capabilities: self.capabilities.append(capability) def set_types(self, types): self.types = types def add_process(self, process): if process not in self.processes: self.processes.append(process) def add_boolean(self, name, description): self.booleans[name] = description def add_file(self, file): self.files[file] = self.__find_path(file) def add_dir(self, file): self.dirs[file] = self.__find_path(file) def generate_capabilities(self): newte = "" self.capabilities.sort() if len(self.capabilities) > 0: newte = "allow %s_t self:capability { %s };\n" % (self.name, " ".join(self.capabilities)) return newte def generate_process(self): newte = "" self.processes.sort() if len(self.processes) > 0: newte = "allow %s_t self:process { %s };\n" % (self.name, " ".join(self.processes)) return newte def generate_network_rules(self): newte = "" if self.use_network(): newte = "\n" newte += re.sub("TEMPLATETYPE", self.name, network.te_network) if self.use_tcp(): newte += "\n" newte += re.sub("TEMPLATETYPE", self.name, network.te_tcp) if self.use_in_tcp(): newte += re.sub("TEMPLATETYPE", self.name, network.te_in_tcp) if self.need_tcp_type and len(self.in_tcp[PORTS]) > 0: newte += re.sub("TEMPLATETYPE", self.name, network.te_in_need_port_tcp) if self.need_tcp_type and len(self.out_tcp[PORTS]) > 0: newte += re.sub("TEMPLATETYPE", self.name, network.te_out_need_port_tcp) if self.in_tcp[ALL]: newte += re.sub("TEMPLATETYPE", self.name, network.te_in_all_ports_tcp) if self.in_tcp[RESERVED]: newte += re.sub("TEMPLATETYPE", self.name, network.te_in_reserved_ports_tcp) if self.in_tcp[UNRESERVED]: newte += re.sub("TEMPLATETYPE", self.name, network.te_in_unreserved_ports_tcp) if self.out_tcp[ALL]: newte += re.sub("TEMPLATETYPE", self.name, network.te_out_all_ports_tcp) if self.out_tcp[RESERVED]: newte += re.sub("TEMPLATETYPE", self.name, network.te_out_reserved_ports_tcp) if self.out_tcp[UNRESERVED]: newte += re.sub("TEMPLATETYPE", self.name, network.te_out_unreserved_ports_tcp) for i in self.found_tcp_ports: newte += i if self.use_udp(): newte += "\n" newte += re.sub("TEMPLATETYPE", self.name, network.te_udp) if self.need_udp_type: newte += re.sub("TEMPLATETYPE", self.name, network.te_in_need_port_udp) if self.use_in_udp(): newte += re.sub("TEMPLATETYPE", self.name, network.te_in_udp) if self.in_udp[ALL]: newte += re.sub("TEMPLATETYPE", self.name, network.te_in_all_ports_udp) if self.in_udp[RESERVED]: newte += re.sub("TEMPLATETYPE", self.name, network.te_in_reserved_ports_udp) if self.in_udp[UNRESERVED]: newte += re.sub("TEMPLATETYPE", self.name, network.te_in_unreserved_ports_udp) for i in self.found_udp_ports: newte += i return newte def generate_transition_rules(self): newte = "" for app in self.transition_domains: tmp = re.sub("TEMPLATETYPE", self.name, user.te_transition_rules) newte += re.sub("APPLICATION", app, tmp) if self.type == USER: for u in self.transition_users: temp = re.sub("TEMPLATETYPE", self.name, executable.te_run_rules) newte += re.sub("USER", u.split("_u")[0], temp) return newte def generate_admin_rules(self): newte = "" if self.type == EUSER: for d in self.existing_domains: name = d.split("_t")[0] role = name + "_r" for app in self.admin_domains: tmp = re.sub("TEMPLATETYPE", name, user.te_admin_domain_rules) if role not in self.all_roles: tmp = re.sub(role, "system_r", tmp) newte += re.sub("APPLICATION", app, tmp) return newte if self.type == RUSER: newte += re.sub("TEMPLATETYPE", self.name, user.te_admin_rules) for app in self.admin_domains: tmp = re.sub("TEMPLATETYPE", self.name, user.te_admin_domain_rules) newte += re.sub("APPLICATION", app, tmp) for u in self.transition_users: role = u.split("_u")[0] if (role + "_r") in self.all_roles: tmp = re.sub("TEMPLATETYPE", self.name, user.te_admin_trans_rules) newte += re.sub("USER", role, tmp) return newte def generate_dbus_if(self): newif = "" if self.use_dbus: newif = re.sub("TEMPLATETYPE", self.name, executable.if_dbus_rules) return newif def generate_sandbox_if(self): newif = "" if self.type != SANDBOX: return newif newif = re.sub("TEMPLATETYPE", self.name, executable.if_sandbox_rules) return newif def generate_admin_if(self): newif = "" newtypes = "" if self.initscript != "": newtypes += re.sub("TEMPLATETYPE", self.name, executable.if_initscript_admin_types) newif += re.sub("TEMPLATETYPE", self.name, executable.if_initscript_admin) for d in self.DEFAULT_KEYS: if len(self.DEFAULT_DIRS[d][1]) > 0: newtypes += re.sub("TEMPLATETYPE", self.name, self.DEFAULT_DIRS[d][2].if_admin_types) newif += re.sub("TEMPLATETYPE", self.name, self.DEFAULT_DIRS[d][2].if_admin_rules) if newif != "": ret = re.sub("TEMPLATETYPE", self.name, executable.if_begin_admin) ret += newtypes ret += re.sub("TEMPLATETYPE", self.name, executable.if_middle_admin) ret += newif ret += re.sub("TEMPLATETYPE", self.name, executable.if_end_admin) return ret return "" def generate_cgi_types(self): return re.sub("TEMPLATETYPE", self.file_name, executable.te_cgi_types) def generate_sandbox_types(self): return re.sub("TEMPLATETYPE", self.file_name, executable.te_sandbox_types) def generate_userapp_types(self): return re.sub("TEMPLATETYPE", self.name, executable.te_userapp_types) def generate_inetd_types(self): return re.sub("TEMPLATETYPE", self.name, executable.te_inetd_types) def generate_dbusd_types(self): return re.sub("TEMPLATETYPE", self.name, executable.te_dbusd_types) def generate_min_login_user_types(self): return re.sub("TEMPLATETYPE", self.name, user.te_min_login_user_types) def generate_login_user_types(self): return re.sub("TEMPLATETYPE", self.name, user.te_login_user_types) def generate_admin_user_types(self): return re.sub("TEMPLATETYPE", self.name, user.te_admin_user_types) def generate_existing_user_types(self): if len(self.existing_domains) == 0: raise ValueError(_("'%s' policy modules require existing domains") % poltype[self.type]) newte = re.sub("TEMPLATETYPE", self.name, user.te_existing_user_types) newte += """gen_require(`""" for d in self.existing_domains: newte += """ type %s;""" % d role = d.split("_t")[0] + "_r" if role in self.all_roles: newte += """ role %s;""" % role newte += """ ') """ return newte def generate_x_login_user_types(self): return re.sub("TEMPLATETYPE", self.name, user.te_x_login_user_types) def generate_root_user_types(self): return re.sub("TEMPLATETYPE", self.name, user.te_root_user_types) def generate_new_types(self): newte = "" if len(self.types) == 0: raise ValueError(_("Type field required")) for t in self.types: for i in self.DEFAULT_EXT: if t.endswith(i): print(t, t[:-len(i)]) newte += re.sub("TEMPLATETYPE", t[:-len(i)], self.DEFAULT_EXT[i].te_types) break if NEWTYPE and newte == "": default_ext = [] for i in self.DEFAULT_EXT: default_ext.append(i) raise ValueError(_("You need to define a new type which ends with: \n %s") % "\n ".join(default_ext)) return newte def generate_new_rules(self): return "" def generate_daemon_types(self): newte = re.sub("TEMPLATETYPE", self.name, executable.te_daemon_types) if self.initscript != "": newte += re.sub("TEMPLATETYPE", self.name, executable.te_initscript_types) return newte def generate_tmp_types(self): if self.use_tmp: return re.sub("TEMPLATETYPE", self.name, tmp.te_types) else: return "" def generate_booleans(self): newte = "" for b in self.booleans: tmp = re.sub("BOOLEAN", b, boolean.te_boolean) newte += re.sub("DESCRIPTION", self.booleans[b], tmp) return newte def generate_boolean_rules(self): newte = "" for b in self.booleans: newte += re.sub("BOOLEAN", b, boolean.te_rules) return newte def generate_sandbox_te(self): return re.sub("TEMPLATETYPE", self.name, executable.te_sandbox_types) def generate_cgi_te(self): return re.sub("TEMPLATETYPE", self.name, executable.te_cgi_types) def generate_daemon_rules(self): newif = re.sub("TEMPLATETYPE", self.name, executable.te_daemon_rules) return newif def generate_new_type_if(self): newif = "" for t in self.types: for i in self.DEFAULT_EXT: if t.endswith(i): reqtype = t[:-len(i)] + "_t" newif += re.sub("TEMPLATETYPE", t[:-len(i)], self.DEFAULT_EXT[i].if_rules) break return newif def generate_login_user_rules(self): return re.sub("TEMPLATETYPE", self.name, user.te_login_user_rules) def generate_existing_user_rules(self): nerules = re.sub("TEMPLATETYPE", self.name, user.te_existing_user_rules) return nerules def generate_x_login_user_rules(self): return re.sub("TEMPLATETYPE", self.name, user.te_x_login_user_rules) def generate_root_user_rules(self): newte = re.sub("TEMPLATETYPE", self.name, user.te_root_user_rules) return newte def generate_userapp_rules(self): return re.sub("TEMPLATETYPE", self.name, executable.te_userapp_rules) def generate_inetd_rules(self): return re.sub("TEMPLATETYPE", self.name, executable.te_inetd_rules) def generate_dbusd_rules(self): return re.sub("TEMPLATETYPE", self.name, executable.te_dbusd_rules) def generate_tmp_rules(self): if self.use_tmp: return re.sub("TEMPLATETYPE", self.name, tmp.te_rules) else: return "" def generate_cgi_rules(self): newte = "" newte += re.sub("TEMPLATETYPE", self.name, executable.te_cgi_rules) return newte def generate_sandbox_rules(self): newte = "" newte += re.sub("TEMPLATETYPE", self.name, executable.te_sandbox_rules) return newte def generate_user_if(self): newif = "" if self.use_terminal or self.type == USER: newif = re.sub("TEMPLATETYPE", self.name, executable.if_user_program_rules) if self.type in (TUSER, XUSER, AUSER, LUSER): newif += re.sub("TEMPLATETYPE", self.name, executable.if_role_change_rules) return newif def generate_if(self): newif = "" newif += re.sub("TEMPLATETYPE", self.name, executable.if_heading_rules) if self.program: newif += re.sub("TEMPLATETYPE", self.name, executable.if_program_rules) if self.initscript != "": newif += re.sub("TEMPLATETYPE", self.name, executable.if_initscript_rules) for d in self.DEFAULT_KEYS: if len(self.DEFAULT_DIRS[d][1]) > 0: newif += re.sub("TEMPLATETYPE", self.name, self.DEFAULT_DIRS[d][2].if_rules) for i in self.DEFAULT_DIRS[d][1]: if os.path.exists(i) and stat.S_ISSOCK(os.stat(i)[stat.ST_MODE]): newif += re.sub("TEMPLATETYPE", self.name, self.DEFAULT_DIRS[d][2].if_stream_rules) break newif += self.generate_user_if() newif += self.generate_dbus_if() newif += self.generate_admin_if() newif += self.generate_sandbox_if() newif += self.generate_new_type_if() newif += self.generate_new_rules() return newif def generate_default_types(self): return self.DEFAULT_TYPES[self.type][0]() def generate_default_rules(self): if self.DEFAULT_TYPES[self.type][1]: return self.DEFAULT_TYPES[self.type][1]() return "" def generate_roles_rules(self): newte = "" if self.type in (TUSER, XUSER, AUSER, LUSER): roles = "" if len(self.roles) > 0: newte += re.sub("TEMPLATETYPE", self.name, user.te_sudo_rules) newte += re.sub("TEMPLATETYPE", self.name, user.te_newrole_rules) for role in self.roles: tmp = re.sub("TEMPLATETYPE", self.name, user.te_roles_rules) newte += re.sub("ROLE", role, tmp) return newte def generate_te(self): newte = self.generate_default_types() for d in self.DEFAULT_KEYS: if len(self.DEFAULT_DIRS[d][1]) > 0: # CGI scripts already have a rw_t if self.type != CGI or d != "rw": newte += re.sub("TEMPLATETYPE", self.name, self.DEFAULT_DIRS[d][2].te_types) if self.type != EUSER: newte += """ ######################################## # # %s local policy # """ % self.name newte += self.generate_capabilities() newte += self.generate_process() newte += self.generate_network_types() newte += self.generate_tmp_types() newte += self.generate_booleans() newte += self.generate_default_rules() newte += self.generate_boolean_rules() for d in self.DEFAULT_KEYS: if len(self.DEFAULT_DIRS[d][1]) > 0: if self.type == EUSER: newte_tmp = "" for domain in self.existing_domains: newte_tmp += re.sub("TEMPLATETYPE_t", domain[:-2] + "_t", self.DEFAULT_DIRS[d][2].te_rules) newte += re.sub("TEMPLATETYPE_rw_t", self.name + "_rw_t", newte_tmp) else: newte += re.sub("TEMPLATETYPE", self.name, self.DEFAULT_DIRS[d][2].te_rules) for i in self.DEFAULT_DIRS[d][1]: if os.path.exists(i) and stat.S_ISSOCK(os.stat(i)[stat.ST_MODE]): if self.type == EUSER: for domain in self.existing_domains: newte += re.sub("TEMPLATETYPE", domain[:-2], self.DEFAULT_DIRS[d][2].te_stream_rules) else: newte += re.sub("TEMPLATETYPE", self.name, self.DEFAULT_DIRS[d][2].te_stream_rules) break newte += self.generate_tmp_rules() newte += self.generate_network_rules() newte += self.generate_fd_rules() newte += self.generate_etc_rules() newte += self.generate_pam_rules() newte += self.generate_uid_rules() newte += self.generate_audit_rules() newte += self.generate_syslog_rules() newte += self.generate_localization_rules() newte += self.generate_resolve_rules() newte += self.generate_roles_rules() newte += self.generate_mail_rules() newte += self.generate_transition_rules() newte += self.generate_admin_rules() newte += self.generate_dbus_rules() newte += self.generate_kerberos_rules() newte += self.generate_manage_krb5_rcache_rules() return newte def generate_fc(self): newfc = "" fclist = [] for i in self.files.keys(): if os.path.exists(i) and stat.S_ISSOCK(os.stat(i)[stat.ST_MODE]): t1 = re.sub("TEMPLATETYPE", self.name, self.files[i][2].fc_sock_file) else: t1 = re.sub("TEMPLATETYPE", self.name, self.files[i][2].fc_file) t2 = re.sub("FILENAME", i, t1) fclist.append(re.sub("FILETYPE", self.files[i][0], t2)) for i in self.dirs.keys(): t1 = re.sub("TEMPLATETYPE", self.name, self.dirs[i][2].fc_dir) t2 = re.sub("FILENAME", i, t1) fclist.append(re.sub("FILETYPE", self.dirs[i][0], t2)) if self.type in USERS + [SANDBOX]: if len(fclist) == 0: return executable.fc_user if self.type not in USERS + [SANDBOX, EUSER, NEWTYPE] and not self.program: raise ValueError(_("You must enter the executable path for your confined process")) if self.program: t1 = re.sub("EXECUTABLE", self.program, executable.fc_program) fclist.append(re.sub("TEMPLATETYPE", self.name, t1)) if self.initscript != "": t1 = re.sub("EXECUTABLE", self.initscript, executable.fc_initscript) fclist.append(re.sub("TEMPLATETYPE", self.name, t1)) fclist.sort() newfc = "\n".join(fclist) return newfc def generate_user_sh(self): newsh = "" if self.type not in (TUSER, XUSER, AUSER, LUSER, RUSER): return newsh roles = "" for role in self.roles: roles += " %s_r" % role if roles != "": roles += " system_r" tmp = re.sub("TEMPLATETYPE", self.name, script.users) newsh += re.sub("ROLES", roles, tmp) if self.type == RUSER or self.type == AUSER: for u in self.transition_users: tmp = re.sub("TEMPLATETYPE", self.name, script.admin_trans) newsh += re.sub("USER", u, tmp) if self.type == LUSER: newsh += re.sub("TEMPLATETYPE", self.name, script.min_login_user_default_context) else: newsh += re.sub("TEMPLATETYPE", self.name, script.x_login_user_default_context) return newsh def generate_sh(self): temp = re.sub("TEMPLATETYPE", self.file_name, script.compile) temp = re.sub("DOMAINTYPE", self.name, temp) if self.type == EUSER: newsh = re.sub("TEMPLATEFILE", "%s" % self.file_name, temp) else: newsh = re.sub("TEMPLATEFILE", self.file_name, temp) newsh += re.sub("DOMAINTYPE", self.name, script.manpage) if self.program: newsh += re.sub("FILENAME", self.program, script.restorecon) if self.initscript != "": newsh += re.sub("FILENAME", self.initscript, script.restorecon) for i in self.files.keys(): newsh += re.sub("FILENAME", i, script.restorecon) for i in self.dirs.keys(): newsh += re.sub("FILENAME", i, script.restorecon) for i in self.in_tcp[PORTS] + self.out_tcp[PORTS]: if self.find_port(i, "tcp") is None: t1 = re.sub("PORTNUM", "%d" % i, script.tcp_ports) newsh += re.sub("TEMPLATETYPE", self.name, t1) for i in self.in_udp[PORTS]: if self.find_port(i, "udp") is None: t1 = re.sub("PORTNUM", "%d" % i, script.udp_ports) newsh += re.sub("TEMPLATETYPE", self.name, t1) newsh += self.generate_user_sh() newsh += re.sub("TEMPLATEFILE", self.file_name, script.rpm) return newsh def generate_spec(self): newspec = "" selinux_policynvr = get_rpm_nvr_list("selinux-policy") if selinux_policynvr is None: selinux_policyver = "0.0.0" else: selinux_policyver = selinux_policynvr[1] newspec += spec.header_comment_section if self.type in APPLICATIONS: newspec += spec.define_relabel_files_begin if self.program: newspec += re.sub("FILENAME", self.program, spec.define_relabel_files_end) if self.initscript != "": newspec += re.sub("FILENAME", self.initscript, spec.define_relabel_files_end) for i in self.files.keys(): newspec += re.sub("FILENAME", i, spec.define_relabel_files_end) for i in self.dirs.keys(): newspec += re.sub("FILENAME", i, spec.define_relabel_files_end) newspec += re.sub("VERSION", selinux_policyver, spec.base_section) newspec = re.sub("MODULENAME", self.file_name, newspec) newspec = re.sub("DOMAINNAME", self.name, newspec) if len(self.rpms) > 0: newspec += "Requires(post): %s\n" % ", ".join(self.rpms) newspec += re.sub("MODULENAME", self.file_name, spec.mid_section) newspec = re.sub("DOMAINNAME", self.name, newspec) newspec = re.sub("TODAYSDATE", time.strftime("%a %b %e %Y"), newspec) if self.type not in APPLICATIONS: newspec = re.sub("%relabel_files", "", newspec) # Remove man pages from EUSER spec file if self.type == EUSER: newspec = re.sub(".%s_selinux.8." % self.name, "", newspec) # Remove user context file from non users spec file if self.type not in (TUSER, XUSER, AUSER, LUSER, RUSER): newspec = re.sub(".%s_u." % self.name, "", newspec) return newspec def write_spec(self, out_dir): specfile = "%s/%s_selinux.spec" % (out_dir, self.file_name) fd = open(specfile, "w") fd.write(self.generate_spec()) fd.close() return specfile def write_te(self, out_dir): tefile = "%s/%s.te" % (out_dir, self.file_name) fd = open(tefile, "w") fd.write(self.generate_te()) fd.close() return tefile def write_sh(self, out_dir): shfile = "%s/%s.sh" % (out_dir, self.file_name) fd = open(shfile, "w") fd.write(self.generate_sh()) fd.close() os.chmod(shfile, 0o750) return shfile def write_if(self, out_dir): iffile = "%s/%s.if" % (out_dir, self.file_name) fd = open(iffile, "w") fd.write(self.generate_if()) fd.close() return iffile def write_fc(self, out_dir): fcfile = "%s/%s.fc" % (out_dir, self.file_name) fd = open(fcfile, "w") fd.write(self.generate_fc()) fd.close() return fcfile def __extract_rpms(self): import dnf with dnf.Base() as base: base.read_all_repos() base.fill_sack(load_system_repo=True) query = base.sack.query() pq = query.available() pq = pq.filter(file=self.program) for pkg in pq: self.rpms.append(pkg.name) for fname in pkg.files: for b in self.DEFAULT_DIRS: if b == "/etc": continue if fname.startswith(b): if os.path.isfile(fname): self.add_file(fname) else: self.add_dir(fname) sq = query.available() sq = sq.filter(provides=pkg.source_name) for bpkg in sq: for fname in bpkg.files: for b in self.DEFAULT_DIRS: if b == "/etc": continue if fname.startswith(b): if os.path.isfile(fname): self.add_file(fname) else: self.add_dir(fname) def gen_writeable(self): try: self.__extract_rpms() except ImportError: pass if os.path.isfile("/var/run/%s.pid" % self.name): self.add_file("/var/run/%s.pid" % self.name) if os.path.isdir("/var/run/%s" % self.name): self.add_dir("/var/run/%s" % self.name) if os.path.isdir("/var/log/%s" % self.name): self.add_dir("/var/log/%s" % self.name) if os.path.isfile("/var/log/%s.log" % self.name): self.add_file("/var/log/%s.log" % self.name) if os.path.isdir("/var/lib/%s" % self.name): self.add_dir("/var/lib/%s" % self.name) if os.path.isfile("/etc/rc.d/init.d/%s" % self.name): self.set_init_script(r"/etc/rc\.d/init\.d/%s" % self.name) # we don't want to have subdir in the .fc policy file # if we already specify labeling for parent dir temp_basepath = [] for p in self.DEFAULT_DIRS.keys(): temp_dirs = [] try: temp_basepath = self.DEFAULT_DIRS[p][1][0] + "/" except IndexError: continue for i in self.DEFAULT_DIRS[p][1]: if i.startswith(temp_basepath): temp_dirs.append(i) else: continue if len(temp_dirs) != 0: for i in temp_dirs: if i in self.dirs.keys(): del self.dirs[i] elif i in self.files.keys(): del self.files[i] else: continue self.DEFAULT_DIRS[p][1] = list(set(self.DEFAULT_DIRS[p][1]) - set(temp_dirs)) def gen_symbols(self): if self.type not in APPLICATIONS: return if not os.path.exists(self.program): sys.stderr.write(""" ************************************* Warning %s does not exist ************************************* """ % self.program) return fd = os.popen("nm -D %s \| grep U" % self.program) for s in fd.read().split(): for b in self.symbols: if s.startswith(b): exec("self.%s" % self.symbols[b]) fd.close() def generate(self, out_dir=os.getcwd()): out = _("Created the following files:\n") out += "%s # %s\n" % (self.write_te(out_dir), _("Type Enforcement file")) out += "%s # %s\n" % (self.write_if(out_dir), _("Interface file")) out += "%s # %s\n" % (self.write_fc(out_dir), _("File Contexts file")) if self.type != NEWTYPE: out += "%s # %s\n" % (self.write_spec(out_dir), _("Spec file")) out += "%s # %s\n" % (self.write_sh(out_dir), _("Setup Script")) return out	50.38 KB	2025-09-21 09:25
📁 help	-	2026-03-21 15:45
📄 interface.py # Copyright (C) 2012 Red Hat # see file 'COPYING' for use and warranty information # # policygentool is a tool for the initial generation of SELinux policy # # 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 re import sys import sepolicy ADMIN_TRANSITION_INTERFACE = "_admin$" USER_TRANSITION_INTERFACE = "_role$" __all__ = ['get_all_interfaces', 'get_interfaces_from_xml', 'get_admin', 'get_user', 'get_interface_dict', 'get_interface_format_text', 'get_interface_compile_format_text', 'get_xml_file', 'interface_compile_test'] ## ## I18N ## PROGNAME = "selinux-python" try: import gettext kwargs = {} if sys.version_info < (3,): kwargs['unicode'] = True t = gettext.translation(PROGNAME, localedir="/usr/share/locale", *kwargs, fallback=True) _ = t.gettext except: try: import builtins builtins.__dict__['_'] = str except ImportError: import __builtin__ __builtin__.__dict__['_'] = unicode def get_interfaces_from_xml(path): """ Get all interfaces from given xml file""" interfaces_list = [] idict = get_interface_dict(path) for k in idict.keys(): interfaces_list.append(k) return interfaces_list def get_all_interfaces(path=""): from sepolicy import get_methods all_interfaces = [] if not path: all_interfaces = get_methods() else: xml_path = get_xml_file(path) all_interfaces = get_interfaces_from_xml(xml_path) return all_interfaces def get_admin(path=""): """ Get all domains with an admin interface from installed policy.""" """ If xml_path is specified, func returns an admin interface from specified xml file""" admin_list = [] if path: try: xml_path = get_xml_file(path) idict = get_interface_dict(xml_path) for k in idict.keys(): if k.endswith("_admin"): admin_list.append(k) except IOError as e: sys.stderr.write("%s: %s\n" % (e.__class__.__name__, str(e))) sys.exit(1) else: for i in sepolicy.get_methods(): if i.endswith("_admin"): admin_list.append(i.split("_admin")[0]) return admin_list def get_user(path=""): """ Get all domains with SELinux user role interface""" """ If xml_path is specified, func returns an user role interface from specified xml file""" trans_list = [] if path: try: xml_path = get_xml_file(path) idict = get_interface_dict(xml_path) for k in idict.keys(): if k.endswith("_role"): if (("%s_exec_t" % k[:-5]) in sepolicy.get_all_types()): trans_list.append(k) except IOError as e: sys.stderr.write("%s: %s\n" % (e.__class__.__name__, str(e))) sys.exit(1) else: for i in sepolicy.get_methods(): m = re.findall("(.)%s" % USER_TRANSITION_INTERFACE, i) if len(m) > 0: if "%s_exec_t" % m[0] in sepolicy.get_all_types(): trans_list.append(m[0]) return trans_list interface_dict = None def get_interface_dict(path="/usr/share/selinux/devel/policy.xml"): global interface_dict import os import xml.etree.ElementTree if interface_dict: return interface_dict interface_dict = {} param_list = [] xml_path = """<?xml version="1.0" encoding="ISO-8859-1" standalone="no"?> <policy> <layer name="admin"> """ xml_path += path xml_path += """ </layer> </policy> """ try: if os.path.isfile(path): tree = xml.etree.ElementTree.parse(path) else: tree = xml.etree.ElementTree.fromstring(xml_path) for l in tree.findall("layer"): for m in l.findall("module"): for i in m.iter('interface'): for e in i.findall("param"): param_list.append(e.get('name')) interface_dict[(i.get("name"))] = [param_list, (i.find('summary').text), "interface"] param_list = [] for i in m.iter('template'): for e in i.findall("param"): param_list.append(e.get('name')) interface_dict[(i.get("name"))] = [param_list, (i.find('summary').text), "template"] param_list = [] except IOError: pass return interface_dict def get_interface_format_text(interface, path="/usr/share/selinux/devel/policy.xml"): idict = get_interface_dict(path) interface_text = "%s(%s) %s" % (interface, ", ".join(idict[interface][0]), " ".join(idict[interface][1].split("\n"))) return interface_text def get_interface_compile_format_text(interfaces_dict, interface): from .templates import test_module param_tmp = [] for i in interfaces_dict[interface][0]: param_tmp.append(test_module.dict_values[i]) interface_text = "%s(%s)\n" % (interface, ", ".join(param_tmp)) return interface_text def generate_compile_te(interface, idict, name="compiletest"): from .templates import test_module te = "" te += re.sub("TEMPLATETYPE", name, test_module.te_test_module) te += get_interface_compile_format_text(idict, interface) return te def get_xml_file(if_file): """ Returns xml format of interfaces for given .if policy file""" import os try: from commands import getstatusoutput except ImportError: from subprocess import getstatusoutput basedir = os.path.dirname(if_file) + "/" filename = os.path.basename(if_file).split(".")[0] rc, output = getstatusoutput("/usr/bin/python3 /usr/share/selinux/devel/include/support/segenxml.py -w -m %s" % (basedir + filename)) if rc != 0: sys.stderr.write("\n Could not process selected interface file.\n") sys.stderr.write("\n%s" % output) sys.exit(1) else: return output def interface_compile_test(interface, path="/usr/share/selinux/devel/policy.xml"): exclude_interfaces = ["userdom", "kernel", "corenet", "files", "dev"] exclude_interface_type = ["template"] try: from commands import getstatusoutput except ImportError: from subprocess import getstatusoutput import os policy_files = {'pp': "compiletest.pp", 'te': "compiletest.te", 'fc': "compiletest.fc", 'if': "compiletest.if"} idict = get_interface_dict(path) if not (interface.split("_")[0] in exclude_interfaces or idict[interface][2] in exclude_interface_type): print(_("Compiling %s interface") % interface) try: fd = open(policy_files['te'], "w") fd.write(generate_compile_te(interface, idict)) fd.close() rc, output = getstatusoutput("make -f /usr/share/selinux/devel/Makefile %s" % policy_files['pp']) if rc != 0: sys.stderr.write(output) sys.stderr.write(_("\nCompile test for %s failed.\n") % interface) except EnvironmentError as e: sys.stderr.write(_("\nCompile test for %s has not run. %s\n") % (interface, e)) for v in policy_files.values(): if os.path.exists(v): os.remove(v) else: sys.stderr.write(_("\nCompiling of %s interface is not supported.") % interface)	8 KB	2023-12-13 14:46
📄 manpage.py # Copyright (C) 2012-2013 Red Hat # AUTHOR: Dan Walsh <dwalsh@redhat.com> # AUTHOR: Miroslav Grepl <mgrepl@redhat.com> # 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 # # __all__ = ['ManPage', 'HTMLManPages', 'gen_domains'] import string import selinux import sepolicy import os import time typealias_types = { "antivirus_t":("amavis_t", "clamd_t", "clamscan_t", "freshclam_t"), "cluster_t":("rgmanager_t", "corosync_t", "aisexec_t", "pacemaker_t"), "svirt_t":("qemu_t"), "httpd_t":("phpfpm_t"), } equiv_dict = {"smbd": ["samba"], "httpd": ["apache"], "virtd": ["virt", "libvirt"], "named": ["bind"], "fsdaemon": ["smartmon"], "mdadm": ["raid"]} equiv_dirs = ["/var"] man_date = time.strftime("%y-%m-%d", time.gmtime( int(os.environ.get('SOURCE_DATE_EPOCH', time.time())))) modules_dict = None def gen_modules_dict(path="/usr/share/selinux/devel/policy.xml"): global modules_dict if modules_dict: return modules_dict import xml.etree.ElementTree modules_dict = {} try: tree = xml.etree.ElementTree.fromstring(sepolicy.policy_xml(path)) for l in tree.findall("layer"): for m in l.findall("module"): name = m.get("name") if name == "user" or name == "unconfined": continue if name == "unprivuser": name = "user" if name == "unconfineduser": name = "unconfined" for b in m.findall("summary"): modules_dict[name] = b.text except IOError: pass return modules_dict users = None users_range = None def get_all_users_info(): global users global users_range if users and users_range: return users, users_range users = [] users_range = {} allusers = [] allusers_info = sepolicy.info(sepolicy.USER) for d in allusers_info: allusers.append(d['name']) if 'range' in d: users_range[d['name'].split("_")[0]] = d['range'] for u in allusers: if u not in ["system_u", "root", "unconfined_u"]: users.append(u.replace("_u", "")) users.sort() return users, users_range all_entrypoints = None def get_entrypoints(): global all_entrypoints if not all_entrypoints: all_entrypoints = next(sepolicy.info(sepolicy.ATTRIBUTE, "entry_type"))["types"] return all_entrypoints domains = None def gen_domains(): global domains if domains: return domains domains = [] for d in sepolicy.get_all_domains(): found = False domain = d[:-2] # if domain + "_exec_t" not in get_entrypoints(): # continue if domain in domains: continue domains.append(domain) for role in sepolicy.get_all_roles(): if role[:-2] in domains or role == "system_r": continue domains.append(role[:-2]) domains.sort() return domains exec_types = None def _gen_exec_types(): global exec_types if exec_types is None: exec_types = next(sepolicy.info(sepolicy.ATTRIBUTE, "exec_type"))["types"] return exec_types entry_types = None def _gen_entry_types(): global entry_types if entry_types is None: entry_types = next(sepolicy.info(sepolicy.ATTRIBUTE, "entry_type"))["types"] return entry_types types = None def _gen_types(): global types if types: return types all_types = sepolicy.info(sepolicy.TYPE) types = {} for rec in all_types: try: types[rec["name"]] = rec["attributes"] except: types[rec["name"]] = [] return types def prettyprint(f, trim): return " ".join(f[:-len(trim)].split("_")) def get_alphabet_manpages(manpage_list): alphabet_manpages = dict.fromkeys(string.ascii_letters, []) for i in string.ascii_letters: temp = [] for j in manpage_list: if j.split("/")[-1][0] == i: temp.append(j.split("/")[-1]) alphabet_manpages[i] = sorted(temp) return alphabet_manpages def convert_manpage_to_html(html_manpage, manpage): try: from commands import getstatusoutput except ImportError: from subprocess import getstatusoutput rc, output = getstatusoutput("/usr/bin/groff -man -Thtml %s 2>/dev/null" % manpage) if rc == 0: print(html_manpage, "has been created") fd = open(html_manpage, 'w') fd.write(output) fd.close() class HTMLManPages: """ Generate a HTML Manpages on an given SELinux domains """ def __init__(self, manpage_roles, manpage_domains, path, os_version): self.manpage_roles = get_alphabet_manpages(manpage_roles) self.manpage_domains = get_alphabet_manpages(manpage_domains) self.os_version = os_version self.old_path = path + "/" self.new_path = self.old_path self.__gen_html_manpages() def __gen_html_manpages(self): self._write_html_manpage() self._gen_index() self._gen_css() def _write_html_manpage(self): if not os.path.isdir(self.new_path): os.mkdir(self.new_path) for domain in self.manpage_domains.values(): if len(domain): for d in domain: convert_manpage_to_html((self.new_path + d.rsplit("_selinux", 1)[0] + ".html"), self.old_path + d) for role in self.manpage_roles.values(): if len(role): for r in role: convert_manpage_to_html((self.new_path + r.rsplit("_selinux", 1)[0] + ".html"), self.old_path + r) def _gen_index(self): html = self.new_path + "index.html" fd = open(html, 'w') fd.write(""" <html> <head> <link rel=stylesheet type="text/css" href="style.css" title="style"> <title>SELinux man pages</title> </head> <body> <h1>SELinux man pages for %s</h1> <hr> <table><tr> <td valign="middle"> <h3>SELinux roles</h3> """ % self.os_version) for letter in self.manpage_roles: if len(self.manpage_roles[letter]): fd.write(""" <a href=#%s_role>%s</a>""" % (letter, letter)) fd.write(""" </td> </tr></table> <pre> """) rolename_body = "" for letter in self.manpage_roles: if len(self.manpage_roles[letter]): rolename_body += "<p>" for r in self.manpage_roles[letter]: rolename = r.rsplit("_selinux", 1)[0] rolename_body += "<a name=%s_role></a><a href=%s.html>%s_selinux(8)</a> - Security Enhanced Linux Policy for the %s SELinux user\n" % (letter, rolename, rolename, rolename) fd.write("""%s </pre> <hr> <table><tr> <td valign="middle"> <h3>SELinux domains</h3>""" % rolename_body) for letter in self.manpage_domains: if len(self.manpage_domains[letter]): fd.write(""" <a href=#%s_domain>%s</a> """ % (letter, letter)) fd.write(""" </td> </tr></table> <pre> """) domainname_body = "" for letter in self.manpage_domains: if len(self.manpage_domains[letter]): domainname_body += "<p>" for r in self.manpage_domains[letter]: domainname = r.rsplit("_selinux", 1)[0] domainname_body += "<a name=%s_domain></a><a href=%s.html>%s_selinux(8)</a> - Security Enhanced Linux Policy for the %s SELinux processes\n" % (letter, domainname, domainname, domainname) fd.write("""%s </pre> </body> </html> """ % domainname_body) fd.close() print("%s has been created" % html) def _gen_css(self): style_css = self.old_path + "style.css" fd = open(style_css, 'w') fd.write(""" html, body { background-color: #fcfcfc; font-family: arial, sans-serif; font-size: 110%; color: #333; } h1, h2, h3, h4, h5, h5 { color: #2d7c0b; font-family: arial, sans-serif; margin-top: 25px; } a { color: #336699; text-decoration: none; } a:visited { color: #4488bb; } a:hover, a:focus, a:active { color: #07488A; text-decoration: none; } a.func { color: red; text-decoration: none; } a.file { color: red; text-decoration: none; } pre.code { background-color: #f4f0f4; // font-family: monospace, courier; font-size: 110%; margin-left: 0px; margin-right: 60px; padding-top: 5px; padding-bottom: 5px; padding-left: 8px; padding-right: 8px; border: 1px solid #AADDAA; } .url { font-family: serif; font-style: italic; color: #440064; } """) fd.close() print("%s has been created" % style_css) class ManPage: """ Generate a Manpage on an SELinux domain in the specified path """ modules_dict = None enabled_str = ["Disabled", "Enabled"] manpage_domains = [] manpage_roles = [] def __init__(self, domainname, path="/tmp", root="/", source_files=False, html=False): self.html = html self.source_files = source_files self.root = root self.portrecs = sepolicy.gen_port_dict()[0] self.domains = gen_domains() self.all_domains = sepolicy.get_all_domains() self.all_attributes = sepolicy.get_all_attributes() self.all_bools = sepolicy.get_all_bools() self.all_port_types = sepolicy.get_all_port_types() self.all_roles = sepolicy.get_all_roles() self.all_users = get_all_users_info()[0] self.all_users_range = get_all_users_info()[1] self.all_file_types = sepolicy.get_all_file_types() self.role_allows = sepolicy.get_all_role_allows() self.types = _gen_types() self.exec_types = _gen_exec_types() self.entry_types = _gen_entry_types() if self.source_files: self.fcpath = self.root + "file_contexts" else: self.fcpath = self.root + selinux.selinux_file_context_path() self.fcdict = sepolicy.get_fcdict(self.fcpath) os.makedirs(path, exist_ok=True) self.path = path if self.source_files: self.xmlpath = self.root + "policy.xml" else: self.xmlpath = self.root + "/usr/share/selinux/devel/policy.xml" self.booleans_dict = sepolicy.gen_bool_dict(self.xmlpath) self.domainname, self.short_name = sepolicy.gen_short_name(domainname) self.type = self.domainname + "_t" self._gen_bools() self.man_page_path = "%s/%s_selinux.8" % (path, self.domainname) self.fd = open(self.man_page_path, 'w') if self.domainname + "_r" in self.all_roles: self.__gen_user_man_page() if self.html: self.manpage_roles.append(self.man_page_path) else: if self.html: self.manpage_domains.append(self.man_page_path) self.__gen_man_page() self.fd.close() for k in equiv_dict.keys(): if k == self.domainname: for alias in equiv_dict[k]: self.__gen_man_page_link(alias) def _gen_bools(self): self.bools = [] self.domainbools = [] types = [self.type] if self.domainname in equiv_dict: for t in equiv_dict[self.domainname]: if t + "_t" in self.all_domains: types.append(t + "_t") for t in types: domainbools, bools = sepolicy.get_bools(t) self.bools += bools self.domainbools += domainbools self.bools.sort() self.domainbools.sort() def get_man_page_path(self): return self.man_page_path def __gen_user_man_page(self): self.role = self.domainname + "_r" if not self.modules_dict: self.modules_dict = gen_modules_dict(self.xmlpath) try: self.desc = self.modules_dict[self.domainname] except: self.desc = "%s user role" % self.domainname if self.domainname in self.all_users: self.attributes = next(sepolicy.info(sepolicy.TYPE, (self.type)))["attributes"] self._user_header() self._user_attribute() self._can_sudo() self._xwindows_login() # until a new policy build with login_userdomain attribute #self.terminal_login() self._network() self._booleans() self._home_exec() self._transitions() else: self._role_header() self._booleans() self._port_types() self._mcs_types() self._writes() self._footer() def __gen_man_page_link(self, alias): path = "%s/%s_selinux.8" % (self.path, alias) self.fd = open("%s/%s_selinux.8" % (self.path, alias), 'w') self.fd.write(".so man8/%s_selinux.8" % self.domainname) self.fd.close() print(path) def __gen_man_page(self): self.anon_list = [] self.attributes = {} self.ptypes = [] self._get_ptypes() for domain_type in self.ptypes: try: if typealias_types[domain_type]: fd = self.fd man_page_path = self.man_page_path for t in typealias_types[domain_type]: self._typealias_gen_man(t) self.fd = fd self.man_page_path = man_page_path except KeyError: continue self.attributes[domain_type] = next(sepolicy.info(sepolicy.TYPE, ("%s") % domain_type))["attributes"] self._header() self._entrypoints() self._process_types() self._mcs_types() self._booleans() self._nsswitch_domain() self._port_types() self._writes() self._file_context() self._public_content() self._footer() def _get_ptypes(self): for f in self.all_domains: if f.startswith(self.short_name) or f.startswith(self.domainname): self.ptypes.append(f) def _typealias_gen_man(self, t): self.man_page_path = "%s/%s_selinux.8" % (self.path, t[:-2]) self.ports = [] self.booltext = "" self.fd = open(self.man_page_path, 'w') self._typealias(t[:-2]) self._footer() self.fd.close() def _typealias(self,typealias): self.fd.write('.TH "%(typealias)s_selinux" "8" "%(date)s" "%(typealias)s" "SELinux Policy %(typealias)s"' % {'typealias':typealias, 'date': man_date}) self.fd.write(r""" .SH "NAME" %(typealias)s_selinux \- Security Enhanced Linux Policy for the %(typealias)s processes .SH "DESCRIPTION" %(typealias)s_t SELinux domain type is now associated with %(domainname)s domain type (%(domainname)s_t). """ % {'typealias':typealias, 'domainname':self.domainname}) self.fd.write(r""" Please see .B %(domainname)s_selinux man page for more details. """ % {'domainname':self.domainname}) def _header(self): self.fd.write('.TH "%(domainname)s_selinux" "8" "%(date)s" "%(domainname)s" "SELinux Policy %(domainname)s"' % {'domainname': self.domainname, 'date': man_date}) self.fd.write(r""" .SH "NAME" %(domainname)s_selinux \- Security Enhanced Linux Policy for the %(domainname)s processes .SH "DESCRIPTION" Security-Enhanced Linux secures the %(domainname)s processes via flexible mandatory access control. The %(domainname)s processes execute with the %(domainname)s_t SELinux type. You can check if you have these processes running by executing the \fBps\fP command with the \fB\-Z\fP qualifier. For example: .B ps -eZ \| grep %(domainname)s_t """ % {'domainname': self.domainname}) def _format_boolean_desc(self, b): desc = self.booleans_dict[b][2][0].lower() + self.booleans_dict[b][2][1:] if desc[-1] == ".": desc = desc[:-1] return desc def _gen_bool_text(self): booltext = "" for b, enabled in self.domainbools + self.bools: if b.endswith("anon_write") and b not in self.anon_list: self.anon_list.append(b) else: if b not in self.booleans_dict: continue booltext += """ .PP If you want to %s, you must turn on the %s boolean. %s by default. .EX .B setsebool -P %s 1 .EE """ % (self._format_boolean_desc(b), b, self.enabled_str[enabled], b) return booltext def _booleans(self): self.booltext = self._gen_bool_text() if self.booltext != "": self.fd.write(""" .SH BOOLEANS SELinux policy is customizable based on least access required. %s policy is extremely flexible and has several booleans that allow you to manipulate the policy and run %s with the tightest access possible. """ % (self.domainname, self.domainname)) self.fd.write(self.booltext) def _nsswitch_domain(self): nsswitch_types = [] nsswitch_booleans = ['authlogin_nsswitch_use_ldap', 'kerberos_enabled'] nsswitchbooltext = "" for k in self.attributes.keys(): if "nsswitch_domain" in self.attributes[k]: nsswitch_types.append(k) if len(nsswitch_types): self.fd.write(""" .SH NSSWITCH DOMAIN """) for b in nsswitch_booleans: nsswitchbooltext += """ .PP If you want to %s for the %s, you must turn on the %s boolean. .EX .B setsebool -P %s 1 .EE """ % (self._format_boolean_desc(b), (", ".join(nsswitch_types)), b, b) self.fd.write(nsswitchbooltext) def _process_types(self): if len(self.ptypes) == 0: return self.fd.write(r""" .SH PROCESS TYPES SELinux defines process types (domains) for each process running on the system .PP You can see the context of a process using the \fB\-Z\fP option to \fBps\bP .PP Policy governs the access confined processes have to files. SELinux %(domainname)s policy is very flexible allowing users to setup their %(domainname)s processes in as secure a method as possible. .PP The following process types are defined for %(domainname)s: """ % {'domainname': self.domainname}) self.fd.write(""" .EX .B %s .EE""" % ", ".join(self.ptypes)) self.fd.write(""" .PP Note: .B semanage permissive -a %(domainname)s_t can be used to make the process type %(domainname)s_t permissive. SELinux does not deny access to permissive process types, but the AVC (SELinux denials) messages are still generated. """ % {'domainname': self.domainname}) def _port_types(self): self.ports = [] for f in self.all_port_types: if f.startswith(self.short_name) or f.startswith(self.domainname): self.ports.append(f) if len(self.ports) == 0: return self.fd.write(""" .SH PORT TYPES SELinux defines port types to represent TCP and UDP ports. .PP You can see the types associated with a port by using the following command: .B semanage port -l .PP Policy governs the access confined processes have to these ports. SELinux %(domainname)s policy is very flexible allowing users to setup their %(domainname)s processes in as secure a method as possible. .PP The following port types are defined for %(domainname)s:""" % {'domainname': self.domainname}) for p in self.ports: self.fd.write(""" .EX .TP 5 .B %s .TP 10 .EE """ % p) once = True for prot in ("tcp", "udp"): if (p, prot) in self.portrecs: if once: self.fd.write(""" Default Defined Ports:""") once = False self.fd.write(r""" %s %s .EE""" % (prot, ",".join(self.portrecs[(p, prot)]))) def _file_context(self): flist = [] flist_non_exec = [] mpaths = [] for f in self.all_file_types: if f.startswith(self.domainname): flist.append(f) if f not in self.exec_types or f not in self.entry_types: flist_non_exec.append(f) if f in self.fcdict: mpaths = mpaths + self.fcdict[f]["regex"] if len(mpaths) == 0: return mpaths.sort() mdirs = {} for mp in mpaths: found = False for md in mdirs: if mp.startswith(md): mdirs[md].append(mp) found = True break if not found: for e in equiv_dirs: if mp.startswith(e) and mp.endswith('(/.)?'): mdirs[mp[:-6]] = [] break equiv = [] for m in mdirs: if len(mdirs[m]) > 0: equiv.append(m) self.fd.write(r""" .SH FILE CONTEXTS SELinux requires files to have an extended attribute to define the file type. .PP You can see the context of a file using the \fB\-Z\fP option to \fBls\bP .PP Policy governs the access confined processes have to these files. SELinux %(domainname)s policy is very flexible allowing users to setup their %(domainname)s processes in as secure a method as possible. .PP """ % {'domainname': self.domainname}) if len(equiv) > 0: self.fd.write(r""" .PP .B EQUIVALENCE DIRECTORIES """) for e in equiv: self.fd.write(r""" .PP %(domainname)s policy stores data with multiple different file context types under the %(equiv)s directory. If you would like to store the data in a different directory you can use the semanage command to create an equivalence mapping. If you wanted to store this data under the /srv directory you would execute the following command: .PP .B semanage fcontext -a -e %(equiv)s /srv/%(alt)s .br .B restorecon -R -v /srv/%(alt)s .PP """ % {'domainname': self.domainname, 'equiv': e, 'alt': e.split('/')[-1]}) if flist_non_exec: self.fd.write(r""" .PP .B STANDARD FILE CONTEXT SELinux defines the file context types for the %(domainname)s, if you wanted to store files with these types in a different paths, you need to execute the semanage command to specify alternate labeling and then use restorecon to put the labels on disk. .B semanage fcontext -a -t %(type)s '/srv/my%(domainname)s_content(/.)?' .br .B restorecon -R -v /srv/my%(domainname)s_content Note: SELinux often uses regular expressions to specify labels that match multiple files. """ % {'domainname': self.domainname, "type": flist_non_exec[-1]}) self.fd.write(r""" .I The following file types are defined for %(domainname)s: """ % {'domainname': self.domainname}) flist.sort() for f in flist: self.fd.write(""" .EX .PP .B %s .EE - %s """ % (f, sepolicy.get_description(f))) if f in self.fcdict: plural = "" if len(self.fcdict[f]["regex"]) > 1: plural = "s" self.fd.write(""" .br .TP 5 Path%s: %s""" % (plural, self.fcdict[f]["regex"][0])) for x in self.fcdict[f]["regex"][1:]: self.fd.write(", %s" % x) self.fd.write(""" .PP Note: File context can be temporarily modified with the chcon command. If you want to permanently change the file context you need to use the .B semanage fcontext command. This will modify the SELinux labeling database. You will need to use .B restorecon to apply the labels. """) def _see_also(self): ret = "" for d in self.domains: if d == self.domainname: continue if d.startswith(self.short_name): ret += ", %s_selinux(8)" % d if d.startswith(self.domainname + "_"): ret += ", %s_selinux(8)" % d self.fd.write(ret) def _public_content(self): if len(self.anon_list) > 0: self.fd.write(""" .SH SHARING FILES If you want to share files with multiple domains (Apache, FTP, rsync, Samba), you can set a file context of public_content_t and public_content_rw_t. These context allow any of the above domains to read the content. If you want a particular domain to write to the public_content_rw_t domain, you must set the appropriate boolean. .TP Allow %(domainname)s servers to read the /var/%(domainname)s directory by adding the public_content_t file type to the directory and by restoring the file type. .PP .B semanage fcontext -a -t public_content_t "/var/%(domainname)s(/.)?" .br .B restorecon -F -R -v /var/%(domainname)s .pp .TP Allow %(domainname)s servers to read and write /var/%(domainname)s/incoming by adding the public_content_rw_t type to the directory and by restoring the file type. You also need to turn on the %(domainname)s_anon_write boolean. .PP .B semanage fcontext -a -t public_content_rw_t "/var/%(domainname)s/incoming(/.)?" .br .B restorecon -F -R -v /var/%(domainname)s/incoming .br .B setsebool -P %(domainname)s_anon_write 1 """ % {'domainname': self.domainname}) for b in self.anon_list: desc = self.booleans_dict[b][2][0].lower() + self.booleans_dict[b][2][1:] self.fd.write(""" .PP If you want to %s, you must turn on the %s boolean. .EX .B setsebool -P %s 1 .EE """ % (desc, b, b)) def _footer(self): self.fd.write(""" .SH "COMMANDS" .B semanage fcontext can also be used to manipulate default file context mappings. .PP .B semanage permissive can also be used to manipulate whether or not a process type is permissive. .PP .B semanage module can also be used to enable/disable/install/remove policy modules. """) if len(self.ports) > 0: self.fd.write(""" .B semanage port can also be used to manipulate the port definitions """) if self.booltext != "": self.fd.write(""" .B semanage boolean can also be used to manipulate the booleans """) self.fd.write(""" .PP .B system-config-selinux is a GUI tool available to customize SELinux policy settings. .SH AUTHOR This manual page was auto-generated using .B "sepolicy manpage". .SH "SEE ALSO" selinux(8), %s(8), semanage(8), restorecon(8), chcon(1), sepolicy(8)""" % (self.domainname)) if self.booltext != "": self.fd.write(", setsebool(8)") self._see_also() def _valid_write(self, check, attributes): if check in [self.type, "domain"]: return False if check.endswith("_t"): for a in attributes: if a in self.types[check]: return False return True def _entrypoints(self): entrypoints = [x['target'] for x in filter(lambda y: y['source'] == self.type and y['class'] == 'file' and 'entrypoint' in y['permlist'], sepolicy.get_all_allow_rules() )] if len(entrypoints) == 0: return self.fd.write(""" .SH "ENTRYPOINTS" """) if len(entrypoints) > 1: entrypoints_str = "\\fB%s\\fP file types" % ", ".join(entrypoints) else: entrypoints_str = "\\fB%s\\fP file type" % entrypoints[0] self.fd.write(""" The %s_t SELinux type can be entered via the %s. The default entrypoint paths for the %s_t domain are the following: """ % (self.domainname, entrypoints_str, self.domainname)) if "bin_t" in entrypoints: entrypoints.remove("bin_t") self.fd.write(""" All executables with the default executable label, usually stored in /usr/bin and /usr/sbin.""") paths = [] for entrypoint in entrypoints: if entrypoint in self.fcdict: paths += self.fcdict[entrypoint]["regex"] self.fd.write(""" %s""" % ", ".join(paths)) def _mcs_types(self): try: mcs_constrained_type = next(sepolicy.info(sepolicy.ATTRIBUTE, "mcs_constrained_type")) except StopIteration: return if self.type not in mcs_constrained_type['types']: return self.fd.write (""" .SH "MCS Constrained" The SELinux process type %(type)s_t is an MCS (Multi Category Security) constrained type. Sometimes this separation is referred to as sVirt. These types are usually used for securing multi-tenant environments, such as virtualization, containers or separation of users. The tools used to launch MCS types, pick out a different MCS label for each process group. For example one process might be launched with %(type)s_t:s0:c1,c2, and another process launched with %(type)s_t:s0:c3,c4. The SELinux kernel only allows these processes can only write to content with a matching MCS label, or a MCS Label of s0. A process running with the MCS level of s0:c1,c2 is not allowed to write to content with the MCS label of s0:c3,c4 """ % {'type': self.domainname}) def _writes(self): # add assigned attributes src_list = [self.type] try: src_list += list(filter(lambda x: x['name'] == self.type, sepolicy.get_all_types_info()))[0]['attributes'] except: pass permlist = list(filter(lambda x: x['source'] in src_list and set(['open', 'write']).issubset(x['permlist']) and x['class'] == 'file', sepolicy.get_all_allow_rules())) if permlist is None or len(permlist) == 0: return all_writes = [] attributes = ["proc_type", "sysctl_type"] for i in permlist: if self._valid_write(i['target'], attributes): if i['target'] not in all_writes: all_writes.append(i['target']) if len(all_writes) == 0: return self.fd.write(""" .SH "MANAGED FILES" """) self.fd.write(""" The SELinux process type %s_t can manage files labeled with the following file types. The paths listed are the default paths for these file types. Note the processes UID still need to have DAC permissions. """ % self.domainname) all_writes.sort() if "file_type" in all_writes: all_writes = ["file_type"] for f in all_writes: self.fd.write(""" .br .B %s """ % f) if f in self.fcdict: for path in self.fcdict[f]["regex"]: self.fd.write("""\t%s .br """ % path) def _get_users_range(self): if self.domainname in self.all_users_range: return self.all_users_range[self.domainname] return "s0" def _user_header(self): self.fd.write('.TH "%(type)s_selinux" "8" "%(type)s" "mgrepl@redhat.com" "%(type)s SELinux Policy documentation"' % {'type': self.domainname}) self.fd.write(r""" .SH "NAME" %(user)s_u \- \fB%(desc)s\fP - Security Enhanced Linux Policy .SH DESCRIPTION \fB%(user)s_u\fP is an SELinux User defined in the SELinux policy. SELinux users have default roles, \fB%(user)s_r\fP. The default role has a default type, \fB%(user)s_t\fP, associated with it. The SELinux user will usually login to a system with a context that looks like: .B %(user)s_u:%(user)s_r:%(user)s_t:%(range)s Linux users are automatically assigned an SELinux users at login. Login programs use the SELinux User to assign initial context to the user's shell. SELinux policy uses the context to control the user's access. By default all users are assigned to the SELinux user via the \fB__default__\fP flag On Targeted policy systems the \fB__default__\fP user is assigned to the \fBunconfined_u\fP SELinux user. You can list all Linux User to SELinux user mapping using: .B semanage login -l If you wanted to change the default user mapping to use the %(user)s_u user, you would execute: .B semanage login -m -s %(user)s_u __default__ """ % {'desc': self.desc, 'user': self.domainname, 'range': self._get_users_range()}) if "login_userdomain" in self.attributes and "login_userdomain" in self.all_attributes: self.fd.write(""" If you want to map the one Linux user (joe) to the SELinux user %(user)s, you would execute: .B $ semanage login -a -s %(user)s_u joe """ % {'user': self.domainname}) def _can_sudo(self): sudotype = "%s_sudo_t" % self.domainname self.fd.write(""" .SH SUDO """) if sudotype in self.types: role = self.domainname + "_r" self.fd.write(""" The SELinux user %(user)s can execute sudo. You can set up sudo to allow %(user)s to transition to an administrative domain: Add one or more of the following record to sudoers using visudo. """ % {'user': self.domainname}) for adminrole in self.role_allows[role]: self.fd.write(""" USERNAME ALL=(ALL) ROLE=%(admin)s_r TYPE=%(admin)s_t COMMAND .br sudo will run COMMAND as %(user)s_u:%(admin)s_r:%(admin)s_t:LEVEL """ % {'admin': adminrole[:-2], 'user': self.domainname}) self.fd.write(""" You might also need to add one or more of these new roles to your SELinux user record. List the SELinux roles your SELinux user can reach by executing: .B $ semanage user -l \|grep selinux_name Modify the roles list and add %(user)s_r to this list. .B $ semanage user -m -R '%(roles)s' %(user)s_u For more details you can see semanage man page. """ % {'user': self.domainname, "roles": " ".join([role] + self.role_allows[role])}) else: self.fd.write(""" The SELinux type %s_t is not allowed to execute sudo. """ % self.domainname) def _user_attribute(self): self.fd.write(""" .SH USER DESCRIPTION """) if "unconfined_usertype" in self.attributes: self.fd.write(""" The SELinux user %s_u is an unconfined user. It means that a mapped Linux user to this SELinux user is supposed to be allow all actions. """ % self.domainname) if "unpriv_userdomain" in self.attributes: self.fd.write(""" The SELinux user %s_u is defined in policy as a unprivileged user. SELinux prevents unprivileged users from doing administration tasks without transitioning to a different role. """ % self.domainname) if "admindomain" in self.attributes: self.fd.write(""" The SELinux user %s_u is an admin user. It means that a mapped Linux user to this SELinux user is intended for administrative actions. Usually this is assigned to a root Linux user. """ % self.domainname) def _xwindows_login(self): if "x_domain" in self.all_attributes: self.fd.write(""" .SH X WINDOWS LOGIN """) if "x_domain" in self.attributes: self.fd.write(""" The SELinux user %s_u is able to X Windows login. """ % self.domainname) else: self.fd.write(""" The SELinux user %s_u is not able to X Windows login. """ % self.domainname) def _terminal_login(self): if "login_userdomain" in self.all_attributes: self.fd.write(""" .SH TERMINAL LOGIN """) if "login_userdomain" in self.attributes: self.fd.write(""" The SELinux user %s_u is able to terminal login. """ % self.domainname) else: self.fd.write(""" The SELinux user %s_u is not able to terminal login. """ % self.domainname) def _network(self): from sepolicy import network self.fd.write(""" .SH NETWORK """) for net in ("tcp", "udp"): portdict = network.get_network_connect(self.type, net, "name_bind") if len(portdict) > 0: self.fd.write(""" .TP The SELinux user %s_u is able to listen on the following %s ports. """ % (self.domainname, net)) for p in portdict: for t, ports in portdict[p]: self.fd.write(""" .B %s """ % ",".join(ports)) portdict = network.get_network_connect(self.type, "tcp", "name_connect") if len(portdict) > 0: self.fd.write(""" .TP The SELinux user %s_u is able to connect to the following tcp ports. """ % (self.domainname)) for p in portdict: for t, ports in portdict[p]: self.fd.write(""" .B %s """ % ",".join(ports)) def _home_exec(self): permlist = list(filter(lambda x: x['source'] == self.type and x['target'] == 'user_home_type' and x['class'] == 'file' and set(['ioctl', 'read', 'getattr', 'execute', 'execute_no_trans', 'open']).issubset(set(x['permlist'])), sepolicy.get_all_allow_rules())) self.fd.write(""" .SH HOME_EXEC """) if permlist is not None: self.fd.write(""" The SELinux user %s_u is able execute home content files. """ % self.domainname) else: self.fd.write(""" The SELinux user %s_u is not able execute home content files. """ % self.domainname) def _transitions(self): self.fd.write(r""" .SH TRANSITIONS Three things can happen when %(type)s attempts to execute a program. \fB1.\fP SELinux Policy can deny %(type)s from executing the program. .TP \fB2.\fP SELinux Policy can allow %(type)s to execute the program in the current user type. Execute the following to see the types that the SELinux user %(type)s can execute without transitioning: .B sesearch -A -s %(type)s -c file -p execute_no_trans .TP \fB3.\fP SELinux can allow %(type)s to execute the program and transition to a new type. Execute the following to see the types that the SELinux user %(type)s can execute and transition: .B $ sesearch -A -s %(type)s -c process -p transition """ % {'type': self.type}) def _role_header(self): self.fd.write('.TH "%(user)s_selinux" "8" "%(user)s" "mgrepl@redhat.com" "%(user)s SELinux Policy documentation"' % {'user': self.domainname}) self.fd.write(r""" .SH "NAME" %(user)s_r \- \fB%(desc)s\fP - Security Enhanced Linux Policy .SH DESCRIPTION SELinux supports Roles Based Access Control (RBAC), some Linux roles are login roles, while other roles need to be transition into. .I Note: Examples in this man page will use the .B staff_u SELinux user. Non login roles are usually used for administrative tasks. For example, tasks that require root privileges. Roles control which types a user can run processes with. Roles often have default types assigned to them. The default type for the %(user)s_r role is %(user)s_t. The .B newrole program to transition directly to this role. .B newrole -r %(user)s_r -t %(user)s_t .B sudo is the preferred method to do transition from one role to another. You setup sudo to transition to %(user)s_r by adding a similar line to the /etc/sudoers file. USERNAME ALL=(ALL) ROLE=%(user)s_r TYPE=%(user)s_t COMMAND .br sudo will run COMMAND as staff_u:%(user)s_r:%(user)s_t:LEVEL When using a non login role, you need to setup SELinux so that your SELinux user can reach %(user)s_r role. Execute the following to see all of the assigned SELinux roles: .B semanage user -l You need to add %(user)s_r to the staff_u user. You could setup the staff_u user to be able to use the %(user)s_r role with a command like: .B $ semanage user -m -R 'staff_r system_r %(user)s_r' staff_u """ % {'desc': self.desc, 'user': self.domainname}) troles = [] for i in self.role_allows: if self.domainname + "_r" in self.role_allows[i]: troles.append(i) if len(troles) > 0: plural = "" if len(troles) > 1: plural = "s" self.fd.write(""" SELinux policy also controls which roles can transition to a different role. You can list these rules using the following command. .B search --role_allow SELinux policy allows the %s role%s can transition to the %s_r role. """ % (", ".join(troles), plural, self.domainname))	39.92 KB	2025-09-21 09:25
📄 network.py # Copyright (C) 2012 Red Hat # see file 'COPYING' for use and warranty information # # setrans is a tool for analyzing process transitions in SELinux policy # # 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 sepolicy def get_types(src, tclass, perm, check_bools=False): allows = sepolicy.search([sepolicy.ALLOW], {sepolicy.SOURCE: src, sepolicy.CLASS: tclass, sepolicy.PERMS: perm}) nlist = [] if allows: for i in map(lambda y: y[sepolicy.TARGET], filter(lambda x: set(perm).issubset(x[sepolicy.PERMS]) and (not check_bools or x["enabled"]), allows)): if i not in nlist: nlist.append(i) return nlist def get_network_connect(src, protocol, perm, check_bools=False): portrecs, portrecsbynum = sepolicy.gen_port_dict() d = {} tlist = get_types(src, "%s_socket" % protocol, [perm], check_bools) if len(tlist) > 0: d[(src, protocol, perm)] = [] for i in tlist: if i == "ephemeral_port_type": if "unreserved_port_type" in tlist: continue i = "ephemeral_port_t" if i == "unreserved_port_t": if "unreserved_port_type" in tlist: continue if "port_t" in tlist: continue if i == "port_t": d[(src, protocol, perm)].append((i, ["all ports without defined types"])) if i == "port_type": d[(src, protocol, perm)].append((i, ["all ports"])) elif i == "unreserved_port_type": d[(src, protocol, perm)].append((i, ["all ports >= 1024"])) elif i == "reserved_port_type": d[(src, protocol, perm)].append((i, ["all ports < 1024"])) elif i == "rpc_port_type": d[(src, protocol, perm)].append((i, ["all ports >= 512 and < 1024"])) else: try: d[(src, protocol, perm)].append((i, portrecs[(i, protocol)])) except KeyError: pass return d	2.74 KB	2023-12-13 14:46
📄 sedbus.py import sys import dbus import dbus.service import dbus.mainloop.glib class SELinuxDBus (object): def __init__(self): self.bus = dbus.SystemBus() self.dbus_object = self.bus.get_object("org.selinux", "/org/selinux/object") def semanage(self, buf): ret = self.dbus_object.semanage(buf, dbus_interface="org.selinux") return ret def restorecon(self, path): ret = self.dbus_object.restorecon(path, dbus_interface="org.selinux") return ret def setenforce(self, value): ret = self.dbus_object.setenforce(value, dbus_interface="org.selinux") return ret def customized(self): ret = self.dbus_object.customized(dbus_interface="org.selinux") return ret def semodule_list(self): ret = self.dbus_object.semodule_list(dbus_interface="org.selinux") return ret def relabel_on_boot(self, value): ret = self.dbus_object.relabel_on_boot(value, dbus_interface="org.selinux") return ret def change_default_mode(self, value): ret = self.dbus_object.change_default_mode(value, dbus_interface="org.selinux") return ret def change_default_policy(self, value): ret = self.dbus_object.change_default_policy(value, dbus_interface="org.selinux") return ret if __name__ == "__main__": try: dbus_proxy = SELinuxDBus() resp = dbus_proxy.setenforce(int(sys.argv[1])) print(resp) except dbus.DBusException as e: print(e)	1.48 KB	2023-12-13 14:46
📁 templates	-	2026-03-21 15:45
📄 transition.py # Copyright (C) 2011 Red Hat # see file 'COPYING' for use and warranty information # # setrans is a tool for analyzing process transitions in SELinux policy # # 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 sepolicy __all__ = ['setrans'] def _entrypoint(src): trans = sepolicy.search([sepolicy.ALLOW], {sepolicy.SOURCE: src}) return map(lambda y: y[sepolicy.TARGET], filter(lambda x: "entrypoint" in x[sepolicy.PERMS], trans)) def _get_trans(src): src_list = [src] + list(filter(lambda x: x['name'] == src, sepolicy.get_all_types_info()))[0]['attributes'] trans_list = list(filter(lambda x: x['source'] in src_list and x['class'] == 'process', sepolicy.get_all_transitions())) return trans_list class setrans: def __init__(self, source, dest=None): self.seen = [] self.sdict = {} self.source = source self.dest = dest self._process(self.source) def _process(self, source): if source in self.sdict: return self.sdict[source] self.sdict[source] = {} trans = _get_trans(source) if not trans: return self.sdict[source]["name"] = source if not self.dest: self.sdict[source]["map"] = trans else: self.sdict[source]["map"] = list(map(lambda y: y, filter(lambda x: x["transtype"] == self.dest, trans))) self.sdict[source]["child"] = list(map(lambda y: y["transtype"], filter(lambda x: x["transtype"] not in [self.dest, source], trans))) for s in self.sdict[source]["child"]: self._process(s) def out(self, name, header=""): buf = "" if name in self.seen: return buf self.seen.append(name) if "map" in self.sdict[name]: for t in self.sdict[name]["map"]: cond = sepolicy.get_conditionals(t["source"], t["transtype"], "process", ["transition"]) if cond: buf += "%s%s @ %s --> %s %s\n" % (header, t["source"], t["target"], t["transtype"], sepolicy.get_conditionals_format_text(cond)) else: buf += "%s%s @ %s --> %s\n" % (header, t["source"], t["target"], t["transtype"]) if "child" in self.sdict[name]: for x in self.sdict[name]["child"]: buf += self.out(x, "%s%s ... " % (header, name)) return buf def output(self): self.seen = [] print(self.out(self.source))	3.15 KB	2023-12-13 14:46

Terminal