← Back
Editing: infoflow.py
# Copyright 2014-2015, Tresys Technology, LLC # # SPDX-License-Identifier: LGPL-2.1-only # import enum import itertools import logging from collections.abc import Iterable, Mapping from contextlib import suppress from dataclasses import dataclass, InitVar import typing try: import networkx as nx from networkx.exception import NetworkXError, NetworkXNoPath, NodeNotFound except ImportError as iex: logging.getLogger(__name__).debug(f"{iex.name} failed to import.") from . import exception, mixins, permmap, policyrep, query from .descriptors import CriteriaDescriptor, EdgeAttrIntMax, EdgeAttrList InfoFlowPath = Iterable['InfoFlowStep'] __all__: typing.Final[tuple[str, ...]] = ("InfoFlowAnalysis", "InfoFlowStep", "InfoFlowPath") class InfoFlowAnalysis(query.DirectedGraphAnalysis): """ Information flow analysis. Parameters: policy The policy to analyze. perm_map The permission map or path to the permission map file. Keyword Parameters source The source type of the analysis. target The target type of the analysis. mode The analysis mode (see InfoFlowAnalysisMode) min_weight The minimum permission weight to include in the analysis. (default is 1) exclude The types excluded from the information flow analysis. (default is none) booleans If None, all rules will be added to the analysis (default). otherwise it should be set to a dict with keys corresponding to boolean names and values of True/False. Any unspecified booleans will use the policy's default values. """ class Mode(enum.Enum): """Information flow analysis modes""" ShortestPaths = "All shortest paths" AllPaths = "All paths up to" # N steps FlowsOut = "Flows out of the source type." FlowsIn = "Flows into the target type." DIRECT_MODES: typing.Final[tuple[Mode, ...]] = (Mode.FlowsIn, Mode.FlowsOut) TRANSITIVE_MODES: typing.Final[tuple[Mode, ...]] = (Mode.ShortestPaths, Mode.AllPaths) source = CriteriaDescriptor[policyrep.Type](lookup_function="lookup_type") target = CriteriaDescriptor[policyrep.Type](lookup_function="lookup_type") mode = Mode.ShortestPaths booleans: Mapping[str, bool] | None def __init__(self, policy: policyrep.SELinuxPolicy, perm_map: permmap.PermissionMap, /, *, min_weight: int = 1, source: policyrep.Type | str | None = None, target: policyrep.Type | str | None = None, mode: Mode = Mode.ShortestPaths, depth_limit: int | None = 1, exclude: Iterable[policyrep.Type | str] | None = None, booleans: Mapping[str, bool] | None = None) -> None: super().__init__(policy, perm_map=perm_map, min_weight=min_weight, source=source, target=target, mode=mode, depth_limit=depth_limit, exclude=exclude, booleans=booleans) self._min_weight: int self._perm_map: permmap.PermissionMap self._depth_limit: int | None self.rebuildgraph = True self.rebuildsubgraph = True try: self.G = nx.DiGraph() self.subG = self.G.copy() except NameError: self.log.critical("NetworkX is not available. This is " "required for Information Flow Analysis.") self.log.critical("This is typically in the python3-networkx package.") raise @property def depth_limit(self) -> int | None: return self._depth_limit @depth_limit.setter def depth_limit(self, value: int | None) -> None: if value is not None and value < 1: raise ValueError("Information flow max depth must be positive.") self._depth_limit = value # no subgraph rebuild needed. @property def min_weight(self) -> int: return self._min_weight @min_weight.setter def min_weight(self, weight: int) -> None: if not 1 <= weight <= 10: raise ValueError( "Min information flow weight must be an integer 1-10.") self._min_weight = weight self.rebuildsubgraph = True @property def perm_map(self) -> permmap.PermissionMap: return self._perm_map @perm_map.setter def perm_map(self, perm_map: permmap.PermissionMap) -> None: self._perm_map = perm_map self.rebuildgraph = True self.rebuildsubgraph = True @property def exclude(self) -> list[policyrep.Type]: return self._exclude @exclude.setter def exclude(self, types: Iterable[policyrep.Type | str] | None) -> None: if types: self._exclude: list[policyrep.Type] = [self.policy.lookup_type(t) for t in types] else: self._exclude = [] self.rebuildsubgraph = True def results(self) -> Iterable[InfoFlowPath] | Iterable["InfoFlowStep"]: if self.rebuildsubgraph: self._build_subgraph() self.log.info(f"Generating information flow results from {self.policy}") self.log.debug(f"{self.source=}") self.log.debug(f"{self.target=}") self.log.debug(f"{self.mode=}, {self.depth_limit=}") with suppress(NetworkXNoPath, NodeNotFound, NetworkXError): match self.mode: case InfoFlowAnalysis.Mode.ShortestPaths: if not all((self.source, self.target)): raise ValueError("Source and target types must be specified.") self.log.info("Generating all shortest information flow paths from " f"{self.source} to {self.target}...") for path in nx.all_shortest_paths(self.subG, self.source, self.target): yield (InfoFlowStep(self.subG, source, target) for source, target in nx.utils.misc.pairwise(path)) case InfoFlowAnalysis.Mode.AllPaths: if not all((self.source, self.target)): raise ValueError("Source and target types must be specified.") self.log.info("Generating all information flow paths from " f"{self.source} to {self.target}, " f"max length {self.depth_limit}...") for path in nx.all_simple_paths(self.subG, self.source, self.target, cutoff=self.depth_limit): yield (InfoFlowStep(self.subG, source, target) for source, target in nx.utils.misc.pairwise(path)) case InfoFlowAnalysis.Mode.FlowsOut: if not self.source: raise ValueError("Source type must be specified.") self.log.info(f"Generating all information flows out of {self.source}, " f"max depth {self.depth_limit}") for source, target in nx.bfs_edges(self.subG, self.source, depth_limit=self.depth_limit): yield InfoFlowStep(self.subG, source, target) case InfoFlowAnalysis.Mode.FlowsIn: if not self.target: raise ValueError("Target type must be specified.") self.log.info(f"Generating all information flows into {self.target} ", f"max depth {self.depth_limit}") # swap source and target since bfs_edges is reversed. for target, source in nx.bfs_edges(self.subG, self.target, reverse=True, depth_limit=self.depth_limit): yield InfoFlowStep(self.subG, source, target) case _: raise ValueError(f"Unknown analysis mode: {self.mode}") def graphical_results(self) -> "nx.DiGraph": """ Return the results of the analysis as a NetworkX directed graph. Caller has the responsibility of converting the graph to a visualization. For example, to convert to a pygraphviz graph: pgv = nx.nx_agraph.to_agraph(g.graphical_results()) pgv.layout(prog="dot") """ if self.rebuildsubgraph: self._build_subgraph() self.log.info(f"Generating graphical information flow results from {self.policy}") self.log.debug(f"{self.source=}") self.log.debug(f"{self.target=}") self.log.debug(f"{self.mode=}, {self.depth_limit=}") try: match self.mode: case InfoFlowAnalysis.Mode.ShortestPaths: if not all((self.source, self.target)): raise ValueError("Source and target types must be specified.") self.log.info("Generating all shortest information flow paths from " f"{self.source} to {self.target}...") paths = nx.all_shortest_paths(self.subG, self.source, self.target) edges = [pair for path in paths for pair in nx.utils.misc.pairwise(path)] out = nx.DiGraph() out.add_edges_from(edges) return out case InfoFlowAnalysis.Mode.AllPaths: if not all((self.source, self.target)): raise ValueError("Source and target types must be specified.") self.log.info("Generating all information flow paths from " f"{self.source} to {self.target}, " f"max length {self.depth_limit}...") paths = nx.all_simple_paths(self.subG, self.source, self.target, cutoff=self.depth_limit) edges = [pair for path in paths for pair in nx.utils.misc.pairwise(path)] out = nx.DiGraph() out.add_edges_from(edges) return out case InfoFlowAnalysis.Mode.FlowsOut: if not self.source: raise ValueError("Source type must be specified.") self.log.info(f"Generating all information flows out of {self.source}, " f"max depth {self.depth_limit}") return nx.bfs_tree(self.subG, self.source, depth_limit=self.depth_limit) case InfoFlowAnalysis.Mode.FlowsIn: if not self.target: raise ValueError("Target type must be specified.") self.log.info(f"Generating all information flows into {self.target} ", f"max depth {self.depth_limit}") out = nx.bfs_tree(self.subG, self.target, reverse=True, depth_limit=self.depth_limit) # output is reversed, un-reverse it return nx.reverse(out, copy=False) case _: raise ValueError(f"Unknown analysis mode: {self.mode}") except Exception as ex: raise exception.AnalysisException( f"Unable to generate graphical results: {ex}") from ex def get_stats(self) -> str: # pragma: no cover """ Get the information flow graph statistics. Return: str """ if self.rebuildgraph: self._build_graph() return f"{nx.number_of_nodes(self.G)=}\n" \ f"{nx.number_of_edges(self.G)=}\n" \ f"{len(self.G)=}\n" # # Internal functions follow # def _generate_steps(self, path: list[policyrep.Type]) -> InfoFlowPath: """ Generator which returns the source, target, and associated rules for each information flow step. Parameter: path A list of graph node names representing an information flow path. Yield: tuple(source, target, rules) source The source type for this step of the information flow. target The target type for this step of the information flow. rules The list of rules creating this information flow step. """ for source, target in nx.utils.misc.pairwise(path): yield InfoFlowStep(self.subG, source, target) # # # Graph building functions # # # 1. _build_graph determines the flow in each direction for each TE # rule and then expands the rule. All information flows are # included in this main graph: memory is traded off for efficiency # as the main graph should only need to be rebuilt if permission # weights change. # 2. _build_subgraph derives a subgraph which removes all excluded # types (nodes) and edges (information flows) which are below the # minimum weight. This subgraph is rebuilt only if the main graph # is rebuilt or the minimum weight or excluded types change. def _build_graph(self) -> None: self.G.clear() self.G.name = f"Information flow graph for {self.policy}." self.perm_map.map_policy(self.policy) self.log.info(f"Building information flow graph from {self.policy}...") self.log.debug(f"{self.perm_map=}") for rule in self.policy.terules(): if rule.ruletype != policyrep.TERuletype.allow: continue weight = self.perm_map.rule_weight(typing.cast(policyrep.AVRule, rule)) for s, t in itertools.product(rule.source.expand(), rule.target.expand()): # only add flows if they actually flow # in or out of the source type type if s != t: if weight.write: edge = InfoFlowStep(self.G, s, t, create=True) edge.rules.append(rule) edge.weight = weight.write if weight.read: edge = InfoFlowStep(self.G, t, s, create=True) edge.rules.append(rule) edge.weight = weight.read self.rebuildgraph = False self.rebuildsubgraph = True self.log.info("Completed building information flow graph.") self.log.debug(f"Graph stats: nodes: {nx.number_of_nodes(self.G)}, " f"edges: {nx.number_of_edges(self.G)}.") def _build_subgraph(self) -> None: if self.rebuildgraph: self._build_graph() self.log.info("Building information flow subgraph...") self.log.debug(f"{self.min_weight=}") self.log.debug(f"{self.exclude=}") self.log.debug(f"{self.booleans=}") # delete excluded types from subgraph nodes = [n for n in self.G.nodes() if n not in self.exclude] self.subG = self.G.subgraph(nodes).copy() # delete edges below minimum weight. # no need if weight is 1, since that # does not exclude any edges. if self.min_weight > 1: delete_list = [] for s, t in self.subG.edges(): edge = InfoFlowStep(self.subG, s, t) if edge.weight < self.min_weight: delete_list.append(edge) self.subG.remove_edges_from(delete_list) if self.booleans is not None: delete_list = [] for s, t in self.subG.edges(): edge = InfoFlowStep(self.subG, s, t) # collect disabled rules rule_list = [] # pylint: disable=not-an-iterable for rule in edge.rules: if not rule.enabled(**self.booleans): rule_list.append(rule) deleted_rules: list[policyrep.AVRule] = [] for rule in rule_list: if rule not in deleted_rules: edge.rules.remove(rule) deleted_rules.append(rule) if not edge.rules: delete_list.append(edge) self.subG.remove_edges_from(delete_list) self.rebuildsubgraph = False self.log.info("Completed building information flow subgraph.") self.log.debug(f"Subgraph stats: nodes: {nx.number_of_nodes(self.subG)}, " f"edges: {nx.number_of_edges(self.subG)}.") @dataclass class InfoFlowStep(mixins.NetworkXGraphEdge): """ A graph edge. Also used for returning information flow steps. Parameters: graph The NetworkX graph. source The source type of the edge. target The target type of the edge. Keyword Parameters: create (T/F) create the edge if it does not exist. The default is False. """ G: "nx.DiGraph" source: policyrep.Type target: policyrep.Type create: InitVar[bool] = False rules = EdgeAttrList() # use capacity to store the info flow weight so # we can use network flow algorithms naturally. # The weight for each edge is 1 since each info # flow step is no more costly than another # (see below add_edge() call) weight = EdgeAttrIntMax('capacity') def __post_init__(self, create) -> None: if not self.G.has_edge(self.source, self.target): if create: self.G.add_edge(self.source, self.target, weight=1) self.rules = None self.weight = None else: raise ValueError("InfoFlowStep does not exist in graph") def __format__(self, spec: str) -> str: if spec == "full": rules = "\n".join(f" {r}" for r in sorted(self.rules)) return f"{self.source} -> {self.target}\n{rules}" elif not spec: return f"{self.source} -> {self.target}" else: return super().__format__(spec) def __str__(self): return self.__format__("full")
Save File
Cancel