<h2>Edit File: dta.py</h2><form method="post"><textarea name="content" rows="20" cols="80"># Copyright 2014-2015, Tresys Technology, LLC
#
# SPDX-License-Identifier: LGPL-2.1-only
#
# pylint: disable=unsubscriptable-object

import itertools
import logging
from collections import defaultdict
from contextlib import suppress
from typing import DefaultDict, Iterable, List, NamedTuple, Optional, Union

try:
    import networkx as nx
    from networkx.exception import NetworkXError, NetworkXNoPath, NodeNotFound
except ImportError:
    logging.getLogger(__name__).debug(&quot;NetworkX failed to import.&quot;)

from .descriptors import EdgeAttrDict, EdgeAttrList
from .policyrep import AnyTERule, SELinuxPolicy, TERuletype, Type

__all__ = [&#039;DomainTransitionAnalysis&#039;, &#039;DomainTransition&#039;, &#039;DomainEntrypoint&#039;, &#039;DTAPath&#039;]


class DomainEntrypoint(NamedTuple):

    &quot;&quot;&quot;Entrypoint list entry named tuple output format.&quot;&quot;&quot;

    name: Type
    entrypoint: List[AnyTERule]
    execute: List[AnyTERule]
    type_transition: List[AnyTERule]


class DomainTransition(NamedTuple):

    &quot;&quot;&quot;Transition step output named tuple format.&quot;&quot;&quot;

    source: Type
    target: Type
    transition: List[AnyTERule]
    entrypoints: List[DomainEntrypoint]
    setexec: List[AnyTERule]
    dyntransition: List[AnyTERule]
    setcurrent: List[AnyTERule]


#
# Typing
#
DTAPath = Iterable[DomainTransition]
RuleHash = DefaultDict[Type, List[AnyTERule]]


class DomainTransitionAnalysis:

    &quot;&quot;&quot;Domain transition analysis.&quot;&quot;&quot;

    _exclude: List[Type]
    _reverse: bool

    def __init__(self, policy: SELinuxPolicy, reverse: bool = False,
                 exclude: Optional[Iterable[Union[Type, str]]] = None) -&gt; None:
        &quot;&quot;&quot;
        Parameter:
        policy   The policy to analyze.

        Keyword Parameters:
        reverse  True means reverse the direction of the analysis (find parent domains).
        exclude  An iterable of types to exclude from the analysis.
        &quot;&quot;&quot;
        self.log = logging.getLogger(__name__)

        self.policy = policy
        self.exclude = exclude  # type: ignore # https://github.com/python/mypy/issues/220
        self.reverse = reverse
        self.rebuildgraph = True
        self.rebuildsubgraph = True

        try:
            self.G = nx.DiGraph()
            self.subG = self.G.copy()
        except NameError:
            self.log.critical(&quot;NetworkX is not available.  This is &quot;
                              &quot;requried for Domain Transition Analysis.&quot;)
            self.log.critical(&quot;This is typically in the python3-networkx package.&quot;)
            raise

    @property
    def reverse(self) -&gt; bool:
        return self._reverse

    @reverse.setter
    def reverse(self, direction) -&gt; None:
        self._reverse = bool(direction)
        self.rebuildsubgraph = True

    @property
    def exclude(self) -&gt; List[Type]:
        return self._exclude

    @exclude.setter
    def exclude(self, types: Optional[Iterable[Union[Type, str]]]) -&gt; None:
        if types:
            self._exclude = [self.policy.lookup_type(t) for t in types]
        else:
            self._exclude = []

        self.rebuildsubgraph = True

    def shortest_path(self, source: Union[Type, str], target: Union[Type, str]) \
            -&gt; Iterable[DTAPath]:
        &quot;&quot;&quot;
        Generator which yields one shortest domain transition path
        between the source and target types (there may be more).

        Parameters:
        source  The source type.
        target  The target type.

        Yield: generator(steps)

        steps   A generator that returns the tuple of
                source, target, and rules for each
                domain transition.
        &quot;&quot;&quot;
        s: Type = self.policy.lookup_type(source)
        t: Type = self.policy.lookup_type(target)

        if self.rebuildsubgraph:
            self._build_subgraph()

        self.log.info(&quot;Generating one domain transition path from {0} to {1}...&quot;.format(s, t))

        with suppress(NetworkXNoPath, NodeNotFound):
            # NodeNotFound: the type is valid but not in graph, e.g. excluded
            # NetworkXNoPath: no paths or the target type is
            # not in the graph
            # pylint: disable=unexpected-keyword-arg, no-value-for-parameter
            yield self.__generate_steps(nx.shortest_path(self.subG, source=s, target=t))

    def all_paths(self, source: Union[Type, str], target: Union[Type, str], maxlen: int = 2) \
            -&gt; Iterable[DTAPath]:
        &quot;&quot;&quot;
        Generator which yields all domain transition paths between
        the source and target up to the specified maximum path
        length.

        Parameters:
        source   The source type.
        target   The target type.
        maxlen   Maximum length of paths.

        Yield: generator(steps)

        steps    A generator that returns the tuple of
                 source, target, and rules for each
                 domain transition.
        &quot;&quot;&quot;
        if maxlen &lt; 1:
            raise ValueError(&quot;Maximum path length must be positive.&quot;)

        s: Type = self.policy.lookup_type(source)
        t: Type = self.policy.lookup_type(target)

        if self.rebuildsubgraph:
            self._build_subgraph()

        self.log.info(&quot;Generating all domain transition paths from {0} to {1}, max length {2}...&quot;.
                      format(s, t, maxlen))

        with suppress(NetworkXNoPath, NodeNotFound):
            # NodeNotFound: the type is valid but not in graph, e.g. excluded
            # NetworkXNoPath: no paths or the target type is
            # not in the graph
            for path in nx.all_simple_paths(self.subG, s, t, maxlen):
                yield self.__generate_steps(path)

    def all_shortest_paths(self, source: Union[Type, str], target: Union[Type, str]) \
            -&gt; Iterable[DTAPath]:
        &quot;&quot;&quot;
        Generator which yields all shortest domain transition paths
        between the source and target types.

        Parameters:
        source   The source type.
        target   The target type.

        Yield: generator(steps)

        steps    A generator that returns the tuple of
                 source, target, and rules for each
                 domain transition.
        &quot;&quot;&quot;
        s: Type = self.policy.lookup_type(source)
        t: Type = self.policy.lookup_type(target)

        if self.rebuildsubgraph:
            self._build_subgraph()

        self.log.info(&quot;Generating all shortest domain transition paths from {0} to {1}...&quot;.
                      format(s, t))

        with suppress(NetworkXNoPath, NodeNotFound):
            # NodeNotFound: the type is valid but not in graph, e.g. excluded
            # NetworkXNoPath: no paths or the target type is
            # not in the graph
            for path in nx.all_shortest_paths(self.subG, s, t):
                yield self.__generate_steps(path)

    def transitions(self, type_: Union[Type, str]) -&gt; DTAPath:
        &quot;&quot;&quot;
        Generator which yields all domain transitions out of a
        specified source type.

        Parameters:
        type_   The starting type.

        Yield: generator(steps)

        steps   A generator that returns the tuple of
                source, target, and rules for each
                domain transition.
        &quot;&quot;&quot;
        s: Type = self.policy.lookup_type(type_)

        if self.rebuildsubgraph:
            self._build_subgraph()

        self.log.info(&quot;Generating all domain transitions {1} {0}&quot;.
                      format(s, &quot;in to&quot; if self.reverse else &quot;out from&quot;))

        with suppress(NetworkXError):
            # NetworkXError: the type is valid but not in graph, e.g. excluded
            for source, target in self.subG.out_edges(s):
                edge = Edge(self.subG, source, target)

                if self.reverse:
                    real_source, real_target = target, source
                else:
                    real_source, real_target = source, target

                yield DomainTransition(real_source,
                                       real_target,
                                       edge.transition,
                                       self.__generate_entrypoints(edge),
                                       edge.setexec,
                                       edge.dyntransition,
                                       edge.setcurrent)

    def get_stats(self) -&gt; str:  # pragma: no cover
        &quot;&quot;&quot;
        Get the domain transition graph statistics.

        Return: str
        &quot;&quot;&quot;
        if self.rebuildgraph:
            self._build_graph()

        return f&quot;Graph nodes: {nx.number_of_nodes(self.G)}\n&quot; \
               f&quot;Graph edges: {nx.number_of_edges(self.G)}&quot;

    #
    # Internal functions follow
    #
    @staticmethod
    def __generate_entrypoints(edge: &#039;Edge&#039;) -&gt; List[DomainEntrypoint]:
        &quot;&quot;&quot;
        Creates a list of entrypoint, execute, and
        type_transition rules for each entrypoint.

        Parameter:
        data     The dictionary of entrypoints.

        Return: list of tuple(type, entry, exec, trans)

        type     The entrypoint type.
        entry    The list of entrypoint rules.
        exec     The list of execute rules.
        trans    The list of type_transition rules.
        &quot;&quot;&quot;
        return [DomainEntrypoint(e, edge.entrypoint[e], edge.execute[e], edge.type_transition[e])
                for e in edge.entrypoint]

    def __generate_steps(self, path: List[Type]) -&gt; DTAPath:
        &quot;&quot;&quot;
        Generator which yields the source, target, and associated rules
        for each domain transition.

        Parameter:
        path     A list of graph node names representing an information flow path.

        Yield: tuple(source, target, transition, entrypoints,
                     setexec, dyntransition, setcurrent)

        source          The source type for this step of the domain transition.
        target          The target type for this step of the domain transition.
        transition      The list of transition rules.
        entrypoints     Generator which yields entrypoint-related rules.
        setexec         The list of setexec rules.
        dyntranstion    The list of dynamic transition rules.
        setcurrent      The list of setcurrent rules.
        &quot;&quot;&quot;

        for s in range(1, len(path)):
            source: Type = path[s - 1]
            target: Type = path[s]
            edge = Edge(self.subG, source, target)

            # Yield the actual source and target.
            # The above perspective is reversed
            # if the graph has been reversed.
            if self.reverse:
                real_source, real_target = target, source
            else:
                real_source, real_target = source, target

            yield DomainTransition(real_source, real_target,
                                   edge.transition,
                                   self.__generate_entrypoints(edge),
                                   edge.setexec,
                                   edge.dyntransition,
                                   edge.setcurrent)

    #
    # Graph building functions
    #

    # Domain transition requirements:
    #
    # Standard transitions a-&gt;b:
    # allow a b:process transition;
    # allow a b_exec:file execute;
    # allow b b_exec:file entrypoint;
    #
    # and at least one of:
    # allow a self:process setexec;
    # type_transition a b_exec:process b;
    #
    # Dynamic transition x-&gt;y:
    # allow x y:process dyntransition;
    # allow x self:process setcurrent;
    #
    # Algorithm summary:
    # 1. iterate over all rules
    #   1. skip non allow/type_transition rules
    #   2. if process transition or dyntransition, create edge,
    #      initialize rule lists, add the (dyn)transition rule
    #   3. if process setexec or setcurrent, add to appropriate dict
    #      keyed on the subject
    #   4. if file exec, entrypoint, or type_transition:process,
    #      add to appropriate dict keyed on subject,object.
    # 2. Iterate over all graph edges:
    #   1. if there is a transition rule (else add to invalid
    #      transition list):
    #       1. use set intersection to find matching exec
    #          and entrypoint rules. If none, add to invalid
    #          transition list.
    #       2. for each valid entrypoint, add rules to the
    #          edge&#039;s lists if there is either a
    #          type_transition for it or the source process
    #          has setexec permissions.
    #       3. If there are neither type_transitions nor
    #          setexec permissions, add to the invalid
    #          transition list
    #   2. if there is a dyntransition rule (else add to invalid
    #      dyntrans list):
    #       1. If the source has a setcurrent rule, add it
    #          to the edge&#039;s list, else add to invalid
    #          dyntransition list.
    # 3. Iterate over all graph edges:
    #   1. if the edge has an invalid trans and dyntrans, delete
    #      the edge.
    #   2. if the edge has an invalid trans, clear the related
    #      lists on the edge.
    #   3. if the edge has an invalid dyntrans, clear the related
    #      lists on the edge.
    #
    def _build_graph(self) -&gt; None:
        self.G.clear()
        self.G.name = &quot;Domain transition graph for {0}.&quot;.format(self.policy)

        self.log.info(&quot;Building domain transition graph from {0}...&quot;.format(self.policy))

        # hash tables keyed on domain type
        setexec: RuleHash = defaultdict(list)
        setcurrent: RuleHash = defaultdict(list)

        # hash tables keyed on (domain, entrypoint file type)
        # the parameter for defaultdict has to be callable
        # hence the lambda for the nested defaultdict
        execute: DefaultDict[Type, RuleHash] = defaultdict(lambda: defaultdict(list))
        entrypoint: DefaultDict[Type, RuleHash] = defaultdict(lambda: defaultdict(list))

        # hash table keyed on (domain, entrypoint, target domain)
        type_trans: DefaultDict[Type, DefaultDict[Type, RuleHash]] = \
            defaultdict(lambda: defaultdict(lambda: defaultdict(list)))

        for rule in self.policy.terules():
            if rule.ruletype == TERuletype.allow:
                if rule.tclass not in [&quot;process&quot;, &quot;file&quot;]:
                    continue

                if rule.tclass == &quot;process&quot;:
                    if &quot;transition&quot; in rule.perms:
                        for s, t in itertools.product(rule.source.expand(), rule.target.expand()):
                            # only add edges if they actually
                            # transition to a new type
                            if s != t:
                                edge = Edge(self.G, s, t, create=True)
                                edge.transition.append(rule)

                    if &quot;dyntransition&quot; in rule.perms:
                        for s, t in itertools.product(rule.source.expand(), rule.target.expand()):
                            # only add edges if they actually
                            # transition to a new type
                            if s != t:
                                e = Edge(self.G, s, t, create=True)
                                e.dyntransition.append(rule)

                    if &quot;setexec&quot; in rule.perms:
                        for s in rule.source.expand():
                            setexec[s].append(rule)

                    if &quot;setcurrent&quot; in rule.perms:
                        for s in rule.source.expand():
                            setcurrent[s].append(rule)

                else:
                    if &quot;execute&quot; in rule.perms:
                        for s, t in itertools.product(
                                rule.source.expand(),
                                rule.target.expand()):
                            execute[s][t].append(rule)

                    if &quot;entrypoint&quot; in rule.perms:
                        for s, t in itertools.product(rule.source.expand(), rule.target.expand()):
                            entrypoint[s][t].append(rule)

            elif rule.ruletype == TERuletype.type_transition:
                if rule.tclass != &quot;process&quot;:
                    continue

                d = rule.default
                for s, t in itertools.product(rule.source.expand(), rule.target.expand()):
                    type_trans[s][t][d].append(rule)

        invalid_edge: List[Edge] = []
        clear_transition: List[Edge] = []
        clear_dyntransition: List[Edge] = []

        for s, t in self.G.edges():
            edge = Edge(self.G, s, t)
            invalid_trans = False
            invalid_dyntrans = False

            if edge.transition:
                # get matching domain exec w/entrypoint type
                entry = set(entrypoint[t].keys())
                exe = set(execute[s].keys())
                match = entry.intersection(exe)

                if not match:
                    # there are no valid entrypoints
                    invalid_trans = True
                else:
                    # TODO try to improve the
                    # efficiency in this loop
                    for m in match:
                        # pylint: disable=unsupported-assignment-operation
                        if s in setexec or type_trans[s][m]:
                            # add key for each entrypoint
                            edge.entrypoint[m] += entrypoint[t][m]
                            edge.execute[m] += execute[s][m]

                        if type_trans[s][m][t]:
                            edge.type_transition[m] += type_trans[s][m][t]

                    if s in setexec:
                        edge.setexec.extend(setexec[s])

                    if not edge.setexec and not edge.type_transition:
                        invalid_trans = True
            else:
                invalid_trans = True

            if edge.dyntransition:
                if s in setcurrent:
                    edge.setcurrent.extend(setcurrent[s])
                else:
                    invalid_dyntrans = True
            else:
                invalid_dyntrans = True

            # cannot change the edges while iterating over them,
            # so keep appropriate lists
            if invalid_trans and invalid_dyntrans:
                invalid_edge.append(edge)
            elif invalid_trans:
                clear_transition.append(edge)
            elif invalid_dyntrans:
                clear_dyntransition.append(edge)

        # Remove invalid transitions
        self.G.remove_edges_from(invalid_edge)
        for edge in clear_transition:
            # if only the regular transition is invalid,
            # clear the relevant lists
            del edge.transition
            del edge.execute
            del edge.entrypoint
            del edge.type_transition
            del edge.setexec
        for edge in clear_dyntransition:
            # if only the dynamic transition is invalid,
            # clear the relevant lists
            del edge.dyntransition
            del edge.setcurrent

        self.rebuildgraph = False
        self.rebuildsubgraph = True
        self.log.info(&quot;Completed building domain transition graph.&quot;)
        self.log.debug(&quot;Graph stats: nodes: {0}, edges: {1}.&quot;.format(
            nx.number_of_nodes(self.G),
            nx.number_of_edges(self.G)))

    def __remove_excluded_entrypoints(self) -&gt; None:
        invalid_edges: List[Edge] = []
        for source, target in self.subG.edges():
            edge = Edge(self.subG, source, target)
            entrypoints = set(edge.entrypoint)
            entrypoints.intersection_update(self.exclude)

            if not entrypoints:
                # short circuit if there are no
                # excluded entrypoint types on
                # this edge.
                continue

            for e in entrypoints:
                # clear the entrypoint data
                # pylint: disable=unsupported-delete-operation
                del edge.entrypoint[e]
                del edge.execute[e]

                with suppress(KeyError):  # setexec
                    del edge.type_transition[e]

            # cannot delete the edges while iterating over them
            if not edge.entrypoint and not edge.dyntransition:
                invalid_edges.append(edge)

        self.subG.remove_edges_from(invalid_edges)

    def _build_subgraph(self) -&gt; None:
        if self.rebuildgraph:
            self._build_graph()

        self.log.info(&quot;Building domain transition subgraph.&quot;)
        self.log.debug(&quot;Excluding {0}&quot;.format(self.exclude))
        self.log.debug(&quot;Reverse {0}&quot;.format(self.reverse))

        # reverse graph for reverse DTA
        if self.reverse:
            self.subG = self.G.reverse(copy=True)
        else:
            self.subG = self.G.copy()

        if self.exclude:
            # delete excluded domains from subgraph
            self.subG.remove_nodes_from(self.exclude)

            # delete excluded entrypoints from subgraph
            self.__remove_excluded_entrypoints()

        self.rebuildsubgraph = False
        self.log.info(&quot;Completed building domain transition subgraph.&quot;)
        self.log.debug(&quot;Subgraph stats: nodes: {0}, edges: {1}.&quot;.format(
            nx.number_of_nodes(self.subG),
            nx.number_of_edges(self.subG)))


class Edge:

    &quot;&quot;&quot;
    A graph edge.  Also used for returning domain transition steps.

    Parameters:
    graph       The NetworkX graph.
    source      The source type of the edge.
    target      The target tyep of the edge.

    Keyword Parameters:
    create      (T/F) create the edge if it does not exist.
                The default is False.
    &quot;&quot;&quot;

    transition = EdgeAttrList(&#039;transition&#039;)
    setexec = EdgeAttrList(&#039;setexec&#039;)
    dyntransition = EdgeAttrList(&#039;dyntransition&#039;)
    setcurrent = EdgeAttrList(&#039;setcurrent&#039;)
    entrypoint = EdgeAttrDict(&#039;entrypoint&#039;)
    execute = EdgeAttrDict(&#039;execute&#039;)
    type_transition = EdgeAttrDict(&#039;type_transition&#039;)

    def __init__(self, graph, source: Type, target: Type, create: bool = False) -&gt; None:
        self.G = graph
        self.source: Type = source
        self.target: Type = target

        if not self.G.has_edge(source, target):
            if not create:
                raise ValueError(&quot;Edge does not exist in graph&quot;)
            else:
                self.G.add_edge(source, target)
                self.transition = None
                self.entrypoint = None
                self.execute = None
                self.type_transition = None
                self.setexec = None
                self.dyntransition = None
                self.setcurrent = None

    def __getitem__(self, key):
        # This is implemented so this object can be used in NetworkX
        # functions that operate on (source, target) tuples
        if isinstance(key, slice):
            return [self._index_to_item(i) for i in range(* key.indices(2))]
        else:
            return self._index_to_item(key)

    def _index_to_item(self, index: int) -&gt; Type:
        &quot;&quot;&quot;Return source or target based on index.&quot;&quot;&quot;
        if index == 0:
            return self.source
        elif index == 1:
            return self.target
        else:
            raise IndexError(&quot;Invalid index (edges only have 2 items): {0}&quot;.format(index))
</textarea><br><input type="hidden" name="edit_file" value="dta.py"><input type="submit" value="Save"></form>