mirror of
https://github.com/SELinuxProject/setools
synced 2025-03-18 08:35:19 +00:00
infoflow: adds boolean selection to seinfoflow
Adds the option to the infoflow analysis to filter conditional policy based on the default or user specified boolean values. Signed-off-by: Daniel Riechers <daniel.riechers@rockwellcollins.com> Signed-off-by: David Graziano <david.graziano@rockwellcollins.com>
This commit is contained in:
Riechers, Daniel J
parent
36c6a4a721
commit
b662b07cc0
24
seinfoflow
24
seinfoflow
@ -53,6 +53,9 @@ opts.add_argument("-w", "--min_weight", default=3, type=int,
|
||||
help="Minimum permission weight. Default is 3.")
|
||||
opts.add_argument("-l", "--limit_flows", default=0, type=int,
|
||||
help="Limit to the specified number of flows. Default is unlimited.")
|
||||
opts.add_argument("-b", "--booleans", default=None,
|
||||
help="Specify the boolean values to use."
|
||||
" Options are default, or \"foo:true bar:false ...\"")
|
||||
opts.add_argument("exclude", nargs="*",
|
||||
help="List of excluded types in the analysis.")
|
||||
|
||||
@ -75,11 +78,28 @@ elif args.verbose:
|
||||
else:
|
||||
logging.basicConfig(level=logging.WARNING, format='%(message)s')
|
||||
|
||||
booleans = None
|
||||
if args.booleans == 'default':
|
||||
booleans = {}
|
||||
elif args.booleans is not None:
|
||||
booleans = {}
|
||||
for boolean in args.booleans.split(','):
|
||||
try:
|
||||
key, value = boolean.split(':')
|
||||
if value.lower() == 'true':
|
||||
booleans[key] = True
|
||||
elif value.lower() == 'false':
|
||||
booleans[key] = False
|
||||
else:
|
||||
parser.error("Conditional value must be true or false.")
|
||||
except ValueError:
|
||||
parser.error("Expected boolean format foo:true,bar:false")
|
||||
|
||||
try:
|
||||
p = setools.SELinuxPolicy(args.policy)
|
||||
m = setools.PermissionMap(args.map)
|
||||
g = setools.InfoFlowAnalysis(p, m, min_weight=args.min_weight, exclude=args.exclude)
|
||||
|
||||
g = setools.InfoFlowAnalysis(p, m, min_weight=args.min_weight, exclude=args.exclude,
|
||||
booleans=booleans)
|
||||
if args.shortest_path or args.all_paths:
|
||||
if args.shortest_path:
|
||||
paths = g.all_shortest_paths(args.source, args.target)
|
||||
|
||||
@ -24,6 +24,7 @@ import networkx as nx
|
||||
from networkx.exception import NetworkXError, NetworkXNoPath, NodeNotFound
|
||||
|
||||
from .descriptors import EdgeAttrIntMax, EdgeAttrList
|
||||
from .exception import RuleNotConditional
|
||||
from .policyrep import TERuletype
|
||||
|
||||
__all__ = ['InfoFlowAnalysis']
|
||||
@ -33,7 +34,7 @@ class InfoFlowAnalysis:
|
||||
|
||||
"""Information flow analysis."""
|
||||
|
||||
def __init__(self, policy, perm_map, min_weight=1, exclude=None):
|
||||
def __init__(self, policy, perm_map, min_weight=1, exclude=None, booleans=None):
|
||||
"""
|
||||
Parameters:
|
||||
policy The policy to analyze.
|
||||
@ -42,6 +43,10 @@ class InfoFlowAnalysis:
|
||||
(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.
|
||||
"""
|
||||
self.log = logging.getLogger(__name__)
|
||||
|
||||
@ -50,6 +55,7 @@ class InfoFlowAnalysis:
|
||||
self.min_weight = min_weight
|
||||
self.perm_map = perm_map
|
||||
self.exclude = exclude
|
||||
self.booleans = booleans
|
||||
self.rebuildgraph = True
|
||||
self.rebuildsubgraph = True
|
||||
|
||||
@ -327,6 +333,8 @@ class InfoFlowAnalysis:
|
||||
self.log.info("Building information flow subgraph...")
|
||||
self.log.debug("Excluding {0!r}".format(self.exclude))
|
||||
self.log.debug("Min weight {0}".format(self.min_weight))
|
||||
self.log.debug("Exclude disabled conditional policy: {0}".format(
|
||||
self.booleans is not None))
|
||||
|
||||
# delete excluded types from subgraph
|
||||
nodes = [n for n in self.G.nodes() if n not in self.exclude]
|
||||
@ -344,6 +352,36 @@ class InfoFlowAnalysis:
|
||||
|
||||
self.subG.remove_edges_from(delete_list)
|
||||
|
||||
if self.booleans is not None:
|
||||
delete_list = []
|
||||
for s, t in self.subG.edges():
|
||||
edge = Edge(self.subG, s, t)
|
||||
rule_list = []
|
||||
for rule in iter(edge.rules):
|
||||
try:
|
||||
if rule.conditional:
|
||||
bool_enabled = rule.conditional.evaluate(**self.booleans)
|
||||
# if conditional is true then delete the false rules
|
||||
if bool_enabled:
|
||||
for false_rule in rule.conditional.false_rules():
|
||||
if false_rule in iter(edge.rules):
|
||||
rule_list.append(false_rule)
|
||||
# if conditional is false then delete the true rules
|
||||
else:
|
||||
for true_rule in rule.conditional.true_rules():
|
||||
if true_rule in iter(edge.rules):
|
||||
rule_list.append(true_rule)
|
||||
except RuleNotConditional as e:
|
||||
pass
|
||||
deleted_rules = []
|
||||
for rule in rule_list:
|
||||
if rule not in deleted_rules:
|
||||
edge.rules.remove(rule)
|
||||
deleted_rules.append(rule)
|
||||
if len(edge.rules) == 0:
|
||||
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("Subgraph stats: nodes: {0}, edges: {1}.".format(
|
||||
|
||||
52
tests/conditionalinfoflow.conf
Normal file
52
tests/conditionalinfoflow.conf
Normal file
@ -0,0 +1,52 @@
|
||||
class infoflow
|
||||
|
||||
sid kernel
|
||||
|
||||
class infoflow
|
||||
{
|
||||
hi_w
|
||||
hi_r
|
||||
med_r
|
||||
med_w
|
||||
}
|
||||
|
||||
type system;
|
||||
role system;
|
||||
role system types system;
|
||||
|
||||
#################################################
|
||||
|
||||
type src;
|
||||
type tgt;
|
||||
type flow_true;
|
||||
type flow_false;
|
||||
|
||||
type src_remain;
|
||||
type tgt_remain;
|
||||
type flow_remain;
|
||||
|
||||
bool condition false;
|
||||
|
||||
allow src_remain flow_remain:infoflow hi_w;
|
||||
allow tgt_remain flow_remain:infoflow hi_r;
|
||||
|
||||
if (condition) {
|
||||
allow src flow_true:infoflow hi_w;
|
||||
allow tgt flow_true:infoflow hi_r;
|
||||
allow tgt flow_true:infoflow hi_r;
|
||||
|
||||
allow src_remain flow_remain:infoflow med_w;
|
||||
allow tgt_remain flow_remain:infoflow med_r;
|
||||
}
|
||||
else {
|
||||
allow src flow_false:infoflow hi_w;
|
||||
allow tgt flow_false:infoflow hi_r;
|
||||
}
|
||||
|
||||
#################################################
|
||||
|
||||
#users
|
||||
user system roles system;
|
||||
|
||||
#isids
|
||||
sid kernel system:system:system
|
||||
144
tests/conditionalinfoflow.py
Normal file
144
tests/conditionalinfoflow.py
Normal file
@ -0,0 +1,144 @@
|
||||
# Copyright 2014-2015, Tresys Technology, LLC
|
||||
#
|
||||
# This file is part of SETools.
|
||||
#
|
||||
# SETools 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.
|
||||
#
|
||||
# SETools 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 SETools. If not, see <http://www.gnu.org/licenses/>.
|
||||
#
|
||||
import os
|
||||
import unittest
|
||||
|
||||
from setools import InfoFlowAnalysis
|
||||
from setools import TERuletype as TERT
|
||||
from setools.exception import InvalidType
|
||||
from setools.permmap import PermissionMap
|
||||
from setools.policyrep import Type
|
||||
|
||||
from . import mixins
|
||||
from .policyrep.util import compile_policy
|
||||
|
||||
|
||||
# Note: the testing for having correct rules on every edge is only
|
||||
# performed once on the full graph, since it is assumed that NetworkX's
|
||||
# Digraph.subgraph() function correctly copies the edge attributes into
|
||||
# the subgraph.
|
||||
|
||||
|
||||
class ConditionalInfoFlowAnalysisTest(unittest.TestCase):
|
||||
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
cls.p = compile_policy("tests/conditionalinfoflow.conf", mls=False)
|
||||
cls.m = PermissionMap("tests/perm_map")
|
||||
cls.a = InfoFlowAnalysis(cls.p, cls.m)
|
||||
|
||||
@classmethod
|
||||
def tearDownClass(cls):
|
||||
os.unlink(cls.p.path)
|
||||
|
||||
def test_001_keep_conditional_rules(self):
|
||||
"""Keep all conditional rules."""
|
||||
self.a.booleans = None
|
||||
self.a._rebuildgraph = True
|
||||
self.a._build_subgraph()
|
||||
|
||||
source = self.p.lookup_type("src")
|
||||
target = self.p.lookup_type("tgt")
|
||||
flow_true = self.p.lookup_type("flow_true")
|
||||
flow_false = self.p.lookup_type("flow_false")
|
||||
|
||||
r = self.a.G.edges[source, flow_true]["rules"]
|
||||
self.assertEqual(len(r), 1)
|
||||
r = self.a.G.edges[flow_true, target]["rules"]
|
||||
self.assertEqual(len(r), 1)
|
||||
r = self.a.G.edges[source, flow_false]["rules"]
|
||||
self.assertEqual(len(r), 1)
|
||||
r = self.a.G.edges[flow_false, target]["rules"]
|
||||
self.assertEqual(len(r), 1)
|
||||
|
||||
def test_002_default_conditional_rules(self):
|
||||
"""Keep only default conditional rules."""
|
||||
self.a.booleans = {}
|
||||
self.a._rebuildgraph = True
|
||||
self.a._build_subgraph()
|
||||
|
||||
source = self.p.lookup_type("src")
|
||||
target = self.p.lookup_type("tgt")
|
||||
flow_true = self.p.lookup_type("flow_true")
|
||||
flow_false = self.p.lookup_type("flow_false")
|
||||
|
||||
r = self.a.G.edges[source, flow_true]["rules"]
|
||||
self.assertEqual(len(r), 0)
|
||||
r = self.a.G.edges[flow_true, target]["rules"]
|
||||
self.assertEqual(len(r), 0)
|
||||
r = self.a.G.edges[source, flow_false]["rules"]
|
||||
self.assertEqual(len(r), 1)
|
||||
r = self.a.G.edges[flow_false, target]["rules"]
|
||||
self.assertEqual(len(r), 1)
|
||||
|
||||
def test_003_user_conditional_true(self):
|
||||
"""Keep only conditional rules selected by user specified booleans (True Case.)"""
|
||||
self.a.booleans = {"condition": True}
|
||||
self.a.rebuildgraph = True
|
||||
self.a._build_subgraph()
|
||||
|
||||
source = self.p.lookup_type("src")
|
||||
target = self.p.lookup_type("tgt")
|
||||
flow_true = self.p.lookup_type("flow_true")
|
||||
flow_false = self.p.lookup_type("flow_false")
|
||||
|
||||
r = self.a.G.edges[source, flow_true]["rules"]
|
||||
self.assertEqual(len(r), 1)
|
||||
r = self.a.G.edges[flow_true, target]["rules"]
|
||||
self.assertEqual(len(r), 1)
|
||||
r = self.a.G.edges[source, flow_false]["rules"]
|
||||
self.assertEqual(len(r), 0)
|
||||
r = self.a.G.edges[flow_false, target]["rules"]
|
||||
self.assertEqual(len(r), 0)
|
||||
|
||||
def test_004_user_conditional_false(self):
|
||||
"""Keep only conditional rules selected by user specified booleans (False Case.)"""
|
||||
self.a.booleans = {"condition": False}
|
||||
self.a.rebuildgraph = True
|
||||
self.a._build_subgraph()
|
||||
|
||||
source = self.p.lookup_type("src")
|
||||
target = self.p.lookup_type("tgt")
|
||||
flow_true = self.p.lookup_type("flow_true")
|
||||
flow_false = self.p.lookup_type("flow_false")
|
||||
|
||||
r = self.a.G.edges[source, flow_true]["rules"]
|
||||
self.assertEqual(len(r), 0)
|
||||
r = self.a.G.edges[flow_true, target]["rules"]
|
||||
self.assertEqual(len(r), 0)
|
||||
r = self.a.G.edges[source, flow_false]["rules"]
|
||||
self.assertEqual(len(r), 1)
|
||||
r = self.a.G.edges[flow_false, target]["rules"]
|
||||
self.assertEqual(len(r), 1)
|
||||
|
||||
def test_005_remaining_edges(self):
|
||||
"""Keep edges when rules are deleted, but there are still remaining rules on the edge."""
|
||||
self.a.booleans = {}
|
||||
self.a.rebuildgraph = True
|
||||
self.a._build_subgraph()
|
||||
|
||||
source = self.p.lookup_type("src_remain")
|
||||
target = self.p.lookup_type("tgt_remain")
|
||||
flow = self.p.lookup_type("flow_remain")
|
||||
|
||||
r = self.a.G.edges[source, flow]["rules"]
|
||||
self.assertEqual(len(r), 1)
|
||||
self.assertEqual(str(r[0]), 'allow src_remain flow_remain:infoflow hi_w;')
|
||||
r = self.a.G.edges[flow, target]["rules"]
|
||||
self.assertEqual(len(r), 1)
|
||||
self.assertEqual(str(r[0]), 'allow tgt_remain flow_remain:infoflow hi_r;')
|
||||
Loading…
Reference in New Issue
Block a user