<h2>Edit File: assertte.py</h2><form method="post"><textarea name="content" rows="20" cols="80"># Copyright 2020, Microsoft Corporation
# Copyright 2020, Chris PeBenito &lt;pebenito@ieee.org&gt;
#
# SPDX-License-Identifier: LGPL-2.1-only
#

import logging
from typing import List, Union

from ..exception import InvalidCheckValue
from ..policyrep import AnyTERule
from ..terulequery import TERuleQuery
from .checkermodule import CheckerModule
from .descriptors import ConfigDescriptor, ConfigSetDescriptor, ConfigPermissionSetDescriptor

SOURCE_OPT = &quot;source&quot;
TARGET_OPT = &quot;target&quot;
CLASS_OPT = &quot;tclass&quot;
PERMS_OPT = &quot;perms&quot;
EXEMPT_SRC_OPT = &quot;exempt_source&quot;
EXEMPT_TGT_OPT = &quot;exempt_target&quot;
EXPECT_SRC_OPT = &quot;expect_source&quot;
EXPECT_TGT_OPT = &quot;expect_target&quot;


class AssertTE(CheckerModule):

    &quot;&quot;&quot;Checker module for asserting a type enforcement allow rule exists (or not).&quot;&quot;&quot;

    check_type = &quot;assert_te&quot;
    check_config = frozenset((SOURCE_OPT, TARGET_OPT, CLASS_OPT, PERMS_OPT, EXEMPT_SRC_OPT,
                              EXEMPT_TGT_OPT, EXPECT_SRC_OPT, EXPECT_TGT_OPT))

    source = ConfigDescriptor(&quot;lookup_type_or_attr&quot;)
    target = ConfigDescriptor(&quot;lookup_type_or_attr&quot;)
    tclass = ConfigSetDescriptor(&quot;lookup_class&quot;, strict=True, expand=False)
    perms = ConfigPermissionSetDescriptor()

    exempt_source = ConfigSetDescriptor(&quot;lookup_type_or_attr&quot;, strict=False, expand=True)
    exempt_target = ConfigSetDescriptor(&quot;lookup_type_or_attr&quot;, strict=False, expand=True)
    expect_source = ConfigSetDescriptor(&quot;lookup_type_or_attr&quot;, strict=True, expand=True)
    expect_target = ConfigSetDescriptor(&quot;lookup_type_or_attr&quot;, strict=True, expand=True)

    def __init__(self, policy, checkname, config) -&gt; None:
        super().__init__(policy, checkname, config)
        self.log = logging.getLogger(__name__)

        self.source = config.get(SOURCE_OPT)
        self.target = config.get(TARGET_OPT)
        self.tclass = config.get(CLASS_OPT)
        self.perms = config.get(PERMS_OPT)

        self.exempt_source = config.get(EXEMPT_SRC_OPT)
        self.exempt_target = config.get(EXEMPT_TGT_OPT)
        self.expect_source = config.get(EXPECT_SRC_OPT)
        self.expect_target = config.get(EXPECT_TGT_OPT)

        if not any((self.source, self.target, self.tclass, self.perms)):
            raise InvalidCheckValue(
                &quot;At least one of source, target, tclass, or perms options must be set.&quot;)

        source_exempt_expect_overlap = self.exempt_source &amp; self.expect_source
        if source_exempt_expect_overlap:
            self.log.info(&quot;Overlap in expect_source and exempt_source: {}&quot;.
                          format(&quot;, &quot;.join(i.name for i in source_exempt_expect_overlap)))

        target_exempt_expect_overlap = self.exempt_target &amp; self.expect_target
        if target_exempt_expect_overlap:
            self.log.info(&quot;Overlap in expect_target and exempt_target: {}&quot;.
                          format(&quot;, &quot;.join(i.name for i in target_exempt_expect_overlap)))

    def run(self) -&gt; List:
        assert any((self.source, self.target, self.tclass, self.perms)), \
            &quot;AssertTe no options set, this is a bug.&quot;

        self.log.info(&quot;Checking TE allow rule assertion.&quot;)

        query = TERuleQuery(self.policy,
                            source=self.source,
                            target=self.target,
                            tclass=self.tclass,
                            perms=self.perms,
                            ruletype=(&quot;allow&quot;,))

        unseen_sources = set(self.expect_source)
        unseen_targets = set(self.expect_target)
        failures: List[Union[AnyTERule, str]] = []
        for rule in sorted(query.results()):
            srcs = set(rule.source.expand())
            tgts = set(rule.target.expand())

            unseen_sources -= srcs
            unseen_targets -= tgts
            if (srcs - self.expect_source - self.exempt_source) and \
                    (tgts - self.expect_target - self.exempt_target):

                self.log_fail(str(rule))
                failures.append(rule)
            else:
                self.log_ok(str(rule))

        for item in unseen_sources:
            failure = &quot;Expected rule with source \&quot;{}\&quot; not found.&quot;.format(item)
            self.log_fail(failure)
            failures.append(failure)

        for item in unseen_targets:
            failure = &quot;Expected rule with target \&quot;{}\&quot; not found.&quot;.format(item)
            self.log_fail(failure)
            failures.append(failure)

        self.log.debug(&quot;{} failure(s)&quot;.format(failures))
        return failures
</textarea><br><input type="hidden" name="edit_file" value="assertte.py"><input type="submit" value="Save"></form>