diff options
Diffstat (limited to 'lib/python2.7/site-packages/sepolgen/policygen.py')
| -rw-r--r-- | lib/python2.7/site-packages/sepolgen/policygen.py | 400 |
1 files changed, 400 insertions, 0 deletions
diff --git a/lib/python2.7/site-packages/sepolgen/policygen.py b/lib/python2.7/site-packages/sepolgen/policygen.py new file mode 100644 index 0000000..34c8401 --- /dev/null +++ b/lib/python2.7/site-packages/sepolgen/policygen.py @@ -0,0 +1,400 @@ +# Authors: Karl MacMillan <kmacmillan@mentalrootkit.com> +# +# Copyright (C) 2006 Red Hat +# see file 'COPYING' for use and warranty information +# +# This program is free software; you can redistribute it and/or +# modify it under the terms of the GNU General Public License as +# published by the Free Software Foundation; version 2 only +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program; if not, write to the Free Software +# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA +# + +""" +classes and algorithms for the generation of SELinux policy. +""" + +import itertools +import textwrap + +import selinux.audit2why as audit2why +try: + from setools import * +except: + pass + +from . import refpolicy +from . import objectmodel +from . import access +from . import interfaces +from . import matching +from . import util +# Constants for the level of explanation from the generation +# routines +NO_EXPLANATION = 0 +SHORT_EXPLANATION = 1 +LONG_EXPLANATION = 2 + +class PolicyGenerator: + """Generate a reference policy module from access vectors. + + PolicyGenerator generates a new reference policy module + or updates an existing module based on requested access + in the form of access vectors. + + It generates allow rules and optionally module require + statements and reference policy interfaces. By default + only allow rules are generated. The methods .set_gen_refpol + and .set_gen_requires turns on interface generation and + requires generation respectively. + + PolicyGenerator can also optionally add comments explaining + why a particular access was allowed based on the audit + messages that generated the access. The access vectors + passed in must have the .audit_msgs field set correctly + and .explain set to SHORT|LONG_EXPLANATION to enable this + feature. + + The module created by PolicyGenerator can be passed to + output.ModuleWriter to output a text representation. + """ + def __init__(self, module=None): + """Initialize a PolicyGenerator with an optional + existing module. + + If the module paramater is not None then access + will be added to the passed in module. Otherwise + a new reference policy module will be created. + """ + self.ifgen = None + self.explain = NO_EXPLANATION + self.gen_requires = False + if module: + self.moduel = module + else: + self.module = refpolicy.Module() + + self.dontaudit = False + + self.domains = None + def set_gen_refpol(self, if_set=None, perm_maps=None): + """Set whether reference policy interfaces are generated. + + To turn on interface generation pass in an interface set + to use for interface generation. To turn off interface + generation pass in None. + + If interface generation is enabled requires generation + will also be enabled. + """ + if if_set: + self.ifgen = InterfaceGenerator(if_set, perm_maps) + self.gen_requires = True + else: + self.ifgen = None + self.__set_module_style() + + + def set_gen_requires(self, status=True): + """Set whether module requires are generated. + + Passing in true will turn on requires generation and + False will disable generation. If requires generation is + disabled interface generation will also be disabled and + can only be re-enabled via .set_gen_refpol. + """ + self.gen_requires = status + + def set_gen_explain(self, explain=SHORT_EXPLANATION): + """Set whether access is explained. + """ + self.explain = explain + + def set_gen_dontaudit(self, dontaudit): + self.dontaudit = dontaudit + + def __set_module_style(self): + if self.ifgen: + refpolicy = True + else: + refpolicy = False + for mod in self.module.module_declarations(): + mod.refpolicy = refpolicy + + def set_module_name(self, name, version="1.0"): + """Set the name of the module and optionally the version. + """ + # find an existing module declaration + m = None + for mod in self.module.module_declarations(): + m = mod + if not m: + m = refpolicy.ModuleDeclaration() + self.module.children.insert(0, m) + m.name = name + m.version = version + if self.ifgen: + m.refpolicy = True + else: + m.refpolicy = False + + def get_module(self): + # Generate the requires + if self.gen_requires: + gen_requires(self.module) + + """Return the generated module""" + return self.module + + def __add_allow_rules(self, avs): + for av in avs: + rule = refpolicy.AVRule(av) + if self.dontaudit: + rule.rule_type = rule.DONTAUDIT + rule.comment = "" + if self.explain: + rule.comment = str(refpolicy.Comment(explain_access(av, verbosity=self.explain))) + if av.type == audit2why.ALLOW: + rule.comment += "\n#!!!! This avc is allowed in the current policy" + if av.type == audit2why.DONTAUDIT: + rule.comment += "\n#!!!! This avc has a dontaudit rule in the current policy" + + if av.type == audit2why.BOOLEAN: + if len(av.data) > 1: + rule.comment += "\n#!!!! This avc can be allowed using one of the these booleans:\n# %s" % ", ".join([x[0] for x in av.data]) + else: + rule.comment += "\n#!!!! This avc can be allowed using the boolean '%s'" % av.data[0][0] + + if av.type == audit2why.CONSTRAINT: + rule.comment += "\n#!!!! This avc is a constraint violation. You would need to modify the attributes of either the source or target types to allow this access." + rule.comment += "\n#Constraint rule: " + rule.comment += "\n#\t" + av.data[0] + for reason in av.data[1:]: + rule.comment += "\n#\tPossible cause is the source %s and target %s are different." % reason + + try: + if ( av.type == audit2why.TERULE and + "write" in av.perms and + ( "dir" in av.obj_class or "open" in av.perms )): + if not self.domains: + self.domains = seinfo(ATTRIBUTE, name="domain")[0]["types"] + types=[] + + for i in [x[TCONTEXT] for x in sesearch([ALLOW], {SCONTEXT: av.src_type, CLASS: av.obj_class, PERMS: av.perms})]: + if i not in self.domains: + types.append(i) + if len(types) == 1: + rule.comment += "\n#!!!! The source type '%s' can write to a '%s' of the following type:\n# %s\n" % ( av.src_type, av.obj_class, ", ".join(types)) + elif len(types) >= 1: + rule.comment += "\n#!!!! The source type '%s' can write to a '%s' of the following types:\n# %s\n" % ( av.src_type, av.obj_class, ", ".join(types)) + except: + pass + self.module.children.append(rule) + + + def add_access(self, av_set): + """Add the access from the access vector set to this + module. + """ + # Use the interface generator to split the access + # into raw allow rules and interfaces. After this + # a will contain a list of access that should be + # used as raw allow rules and the interfaces will + # be added to the module. + if self.ifgen: + raw_allow, ifcalls = self.ifgen.gen(av_set, self.explain) + self.module.children.extend(ifcalls) + else: + raw_allow = av_set + + # Generate the raw allow rules from the filtered list + self.__add_allow_rules(raw_allow) + + def add_role_types(self, role_type_set): + for role_type in role_type_set: + self.module.children.append(role_type) + +def explain_access(av, ml=None, verbosity=SHORT_EXPLANATION): + """Explain why a policy statement was generated. + + Return a string containing a text explanation of + why a policy statement was generated. The string is + commented and wrapped and can be directly inserted + into a policy. + + Params: + av - access vector representing the access. Should + have .audit_msgs set appropriately. + verbosity - the amount of explanation provided. Should + be set to NO_EXPLANATION, SHORT_EXPLANATION, or + LONG_EXPLANATION. + Returns: + list of strings - strings explaining the access or an empty + string if verbosity=NO_EXPLANATION or there is not sufficient + information to provide an explanation. + """ + s = [] + + def explain_interfaces(): + if not ml: + return + s.append(" Interface options:") + for match in ml.all(): + ifcall = call_interface(match.interface, ml.av) + s.append(' %s # [%d]' % (ifcall.to_string(), match.dist)) + + + # Format the raw audit data to explain why the + # access was requested - either long or short. + if verbosity == LONG_EXPLANATION: + for msg in av.audit_msgs: + s.append(' %s' % msg.header) + s.append(' scontext="%s" tcontext="%s"' % + (str(msg.scontext), str(msg.tcontext))) + s.append(' class="%s" perms="%s"' % + (msg.tclass, refpolicy.list_to_space_str(msg.accesses))) + s.append(' comm="%s" exe="%s" path="%s"' % (msg.comm, msg.exe, msg.path)) + s.extend(textwrap.wrap('message="' + msg.message + '"', 80, initial_indent=" ", + subsequent_indent=" ")) + explain_interfaces() + elif verbosity: + s.append(' src="%s" tgt="%s" class="%s", perms="%s"' % + (av.src_type, av.tgt_type, av.obj_class, av.perms.to_space_str())) + # For the short display we are only going to use the additional information + # from the first audit message. For the vast majority of cases this info + # will always be the same anyway. + if len(av.audit_msgs) > 0: + msg = av.audit_msgs[0] + s.append(' comm="%s" exe="%s" path="%s"' % (msg.comm, msg.exe, msg.path)) + explain_interfaces() + return s + +def call_interface(interface, av): + params = [] + args = [] + + params.extend(interface.params.values()) + params.sort(key=lambda param: param.num, reverse=True) + + ifcall = refpolicy.InterfaceCall() + ifcall.ifname = interface.name + + for i in range(len(params)): + if params[i].type == refpolicy.SRC_TYPE: + ifcall.args.append(av.src_type) + elif params[i].type == refpolicy.TGT_TYPE: + ifcall.args.append(av.tgt_type) + elif params[i].type == refpolicy.OBJ_CLASS: + ifcall.args.append(av.obj_class) + else: + print(params[i].type) + assert(0) + + assert(len(ifcall.args) > 0) + + return ifcall + +class InterfaceGenerator: + def __init__(self, ifs, perm_maps=None): + self.ifs = ifs + self.hack_check_ifs(ifs) + self.matcher = matching.AccessMatcher(perm_maps) + self.calls = [] + + def hack_check_ifs(self, ifs): + # FIXME: Disable interfaces we can't call - this is a hack. + # Because we don't handle roles, multiple paramaters, etc., + # etc., we must make certain we can actually use a returned + # interface. + for x in ifs.interfaces.values(): + params = [] + params.extend(x.params.values()) + params.sort(key=lambda param: param.num, reverse=True) + for i in range(len(params)): + # Check that the paramater position matches + # the number (e.g., $1 is the first arg). This + # will fail if the parser missed something. + if (i + 1) != params[i].num: + x.enabled = False + break + # Check that we can handle the param type (currently excludes + # roles. + if params[i].type not in [refpolicy.SRC_TYPE, refpolicy.TGT_TYPE, + refpolicy.OBJ_CLASS]: + x.enabled = False + break + + def gen(self, avs, verbosity): + raw_av = self.match(avs) + ifcalls = [] + for ml in self.calls: + ifcall = call_interface(ml.best().interface, ml.av) + if verbosity: + ifcall.comment = refpolicy.Comment(explain_access(ml.av, ml, verbosity)) + ifcalls.append((ifcall, ml)) + + d = [] + for ifcall, ifs in ifcalls: + found = False + for o_ifcall in d: + if o_ifcall.matches(ifcall): + if o_ifcall.comment and ifcall.comment: + o_ifcall.comment.merge(ifcall.comment) + found = True + if not found: + d.append(ifcall) + + return (raw_av, d) + + + def match(self, avs): + raw_av = [] + for av in avs: + ans = matching.MatchList() + self.matcher.search_ifs(self.ifs, av, ans) + if len(ans): + self.calls.append(ans) + else: + raw_av.append(av) + + return raw_av + + +def gen_requires(module): + """Add require statements to the module. + """ + def collect_requires(node): + r = refpolicy.Require() + for avrule in node.avrules(): + r.types.update(avrule.src_types) + r.types.update(avrule.tgt_types) + for obj in avrule.obj_classes: + r.add_obj_class(obj, avrule.perms) + + for ifcall in node.interface_calls(): + for arg in ifcall.args: + # FIXME - handle non-type arguments when we + # can actually figure those out. + r.types.add(arg) + + for role_type in node.role_types(): + r.roles.add(role_type.role) + r.types.update(role_type.types) + + r.types.discard("self") + + node.children.insert(0, r) + + # FUTURE - this is untested on modules with any sort of + # nesting + for node in module.nodes(): + collect_requires(node) + + |