Adding upstream version 3.2.1.upstream/3.2.1upstream

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

1 files changed, 335 insertions, 0 deletions

diff --git a/tests/deckard/rplint.py b/tests/deckard/rplint.py
new file mode 100755
index 0000000..c0aa6e9
--- /dev/null
+++ b/tests/deckard/rplint.py
@@ -0,0 +1,335 @@
+#!/usr/bin/env python3
+
+import itertools
+import os
+import sys
+from typing import Any, Callable, Iterable, Iterator, Optional, List, Union, Set  # noqa
+
+import pydnstest.augwrap
+import pydnstest.matchpart
+import pydnstest.scenario
+
+Element = Union["Entry", "Step", pydnstest.scenario.Range]
+
+RCODES = {"NOERROR", "FORMERR", "SERVFAIL", "NXDOMAIN", "NOTIMP", "REFUSED", "YXDOMAIN", "YXRRSET",
+          "NXRRSET", "NOTAUTH", "NOTZONE", "BADVERS", "BADSIG", "BADKEY", "BADTIME", "BADMODE",
+          "BADNAME", "BADALG", "BADTRUNC", "BADCOOKIE"}
+FLAGS = {"QR", "AA", "TC", "RD", "RA", "AD", "CD"}
+SECTIONS = {"question", "answer", "authority", "additional"}
+
+
+class RplintError(ValueError):
+    def __init__(self, fails):
+        msg = ""
+        for fail in fails:
+            msg += str(fail) + "\n"
+        super().__init__(msg)
+
+
+def get_line_number(file: str, char_number: int) -> int:
+    pos = 0
+    for number, line in enumerate(open(file)):
+        pos += len(line)
+        if pos >= char_number:
+            return number + 2
+    return 0
+
+
+def is_empty(iterable: Iterator[Any]) -> bool:
+    try:
+        next(iterable)
+    except StopIteration:
+        return True
+    return False
+
+
+class Entry:
+    def __init__(self, node: pydnstest.augwrap.AugeasNode) -> None:
+        self.match = {m.value for m in node.match("/match")}
+        self.adjust = {a.value for a in node.match("/adjust")}
+        self.authority = list(node.match("/section/authority/record"))
+        self.reply = {r.value for r in node.match("/reply")}
+        self.records = list(node.match("/section/*/record"))
+        self.node = node
+
+
+class Step:
+    def __init__(self, node: pydnstest.augwrap.AugeasNode) -> None:
+        self.node = node
+        self.type = node["/type"].value
+        try:
+            self.entry = Entry(node["/entry"])  # type: Optional[Entry]
+        except KeyError:
+            self.entry = None
+
+
+class RplintFail:
+    def __init__(self, test: "RplintTest",
+                 element: Optional[Element] = None,
+                 etc: str = "") -> None:
+        self.path = test.path
+        self.element = element  # type: Optional[Element]
+        self.line = get_line_number(self.path, element.node.char if element is not None else 0)
+        self.etc = etc
+        self.check = None  # type: Optional[Callable[[RplintTest], List[RplintFail]]]
+
+    def __str__(self):
+        if self.etc:
+            return "{}:{} {}: {} ({})".format(os.path.basename(self.path), self.line,
+                                              self.check.__name__, self.check.__doc__, self.etc)
+        return "{}:{} {}: {}".format(os.path.basename(self.path), self.line, self.check.__name__,
+                                     self.check.__doc__)
+
+
+class RplintTest:
+    def __init__(self, path: str) -> None:
+        aug = pydnstest.augwrap.AugeasWrapper(confpath=os.path.realpath(path),
+                                              lens='Deckard',
+                                              loadpath=os.path.join(os.path.dirname(__file__),
+                                                                    'pydnstest'))
+        self.node = aug.tree
+        self.name = os.path.basename(path)
+        self.path = path
+
+        _, self.config = pydnstest.scenario.parse_file(os.path.realpath(path))
+        self.range_entries = [Entry(node) for node in self.node.match("/scenario/range/entry")]
+        self.steps = [Step(node) for node in self.node.match("/scenario/step")]
+        self.step_entries = [step.entry for step in self.steps if step.entry is not None]
+        self.entries = self.range_entries + self.step_entries
+
+        self.ranges = [pydnstest.scenario.Range(n) for n in self.node.match("/scenario/range")]
+
+        self.fails = None  # type: Optional[List[RplintFail]]
+        self.checks = [
+            entry_more_than_one_rcode,
+            entry_no_qname_qtype_copy_query,
+            # Commented out for now until we implement selective turning off of checks
+            # entry_ns_in_authority,
+            range_overlapping_ips,
+            range_shadowing_match_rules,
+            step_check_answer_no_match,
+            step_query_match,
+            step_section_unchecked,
+            step_unchecked_match,
+            step_unchecked_rcode,
+            scenario_ad_or_rrsig_no_ta,
+            scenario_timestamp,
+            config_trust_anchor_trailing_period_missing,
+            step_duplicate_id,
+        ]
+
+    def run_checks(self) -> bool:
+        """returns True iff all tests passed"""
+        self.fails = []
+        for check in self.checks:
+            fails = check(self)
+            for fail in fails:
+                fail.check = check
+            self.fails += fails
+
+        if self.fails == []:
+            return True
+        return False
+
+    def print_fails(self) -> None:
+        if self.fails is None:
+            raise RuntimeError("Maybe you should run some test first…")
+        for fail in self.fails:
+            print(fail)
+
+
+def config_trust_anchor_trailing_period_missing(test: RplintTest) -> List[RplintFail]:
+    """Trust-anchor option in configuration contains domain without trailing period"""
+    for conf in test.config:
+        if conf[0] == "trust-anchor":
+            if conf[1].split()[0][-1] != ".":
+                return [RplintFail(test, etc=conf[1])]
+    return []
+
+
+def scenario_timestamp(test: RplintTest) -> List[RplintFail]:
+    """RRSSIG record present in test but no val-override-date or val-override-timestamp in config"""
+    rrsigs = []
+    for entry in test.entries:
+        for record in entry.records:
+            if record["/type"].value == "RRSIG":
+                rrsigs.append(RplintFail(test, entry))
+    if rrsigs:
+        for k in test.config:
+            if k[0] == "val-override-date" or k[0] == "val-override-timestamp":
+                return []
+    return rrsigs
+
+
+def entry_no_qname_qtype_copy_query(test: RplintTest) -> List[RplintFail]:
+    """ENTRY without qname and qtype in MATCH and without copy_query in ADJUST"""
+    fails = []
+    for entry in test.range_entries:
+        if "question" not in entry.match and ("qname" not in entry.match or
+                                              "qtype" not in entry.match):
+            if "copy_query" not in entry.adjust:
+                fails.append(RplintFail(test, entry))
+    return fails
+
+
+def entry_ns_in_authority(test: RplintTest) -> List[RplintFail]:
+    """ENTRY has authority section with NS records, consider using MATCH subdomain"""
+    fails = []
+    for entry in test.range_entries:
+        if entry.authority and "subdomain" not in entry.match:
+            for record in entry.authority:
+                if record["/type"].value == "NS":
+                    fails.append(RplintFail(test, entry))
+    return fails
+
+
+def entry_more_than_one_rcode(test: RplintTest) -> List[RplintFail]:
+    """ENTRY has more than one rcode in MATCH"""
+    fails = []
+    for entry in test.entries:
+        if len(RCODES & entry.reply) > 1:
+            fails.append(RplintFail(test, entry))
+    return fails
+
+
+def scenario_ad_or_rrsig_no_ta(test: RplintTest) -> List[RplintFail]:
+    """AD or RRSIG present in test but no trust-anchor present in config"""
+    dnssec = []
+    for entry in test.entries:
+        if "AD" in entry.reply or "AD" in entry.match:
+            dnssec.append(RplintFail(test, entry))
+        else:
+            for record in entry.records:
+                if record["/type"].value == "RRSIG":
+                    dnssec.append(RplintFail(test, entry))
+
+    if dnssec:
+        for k in test.config:
+            if k[0] == "trust-anchor":
+                return []
+    return dnssec
+
+
+def step_query_match(test: RplintTest) -> List[RplintFail]:
+    """STEP QUERY has a MATCH rule"""
+    return [RplintFail(test, step) for step in test.steps if step.type == "QUERY" and
+            step.entry and step.entry.match]
+
+
+def step_check_answer_no_match(test: RplintTest) -> List[RplintFail]:
+    """ENTRY in STEP CHECK_ANSWER has no MATCH rule"""
+    return [RplintFail(test, step) for step in test.steps if step.type == "CHECK_ANSWER" and
+            step.entry and not step.entry.match]
+
+
+def step_unchecked_rcode(test: RplintTest) -> List[RplintFail]:
+    """ENTRY specifies rcode but STEP MATCH does not check for it."""
+    fails = []
+    for step in test.steps:
+        if step.type == "CHECK_ANSWER" and step.entry and "all" not in step.entry.match:
+            if step.entry.reply & RCODES and "rcode" not in step.entry.match:
+                fails.append(RplintFail(test, step.entry))
+    return fails
+
+
+def step_unchecked_match(test: RplintTest) -> List[RplintFail]:
+    """ENTRY specifies flags but MATCH does not check for them"""
+    fails = []
+    for step in test.steps:
+        if step.type == "CHECK_ANSWER":
+            entry = step.entry
+            if entry and "all" not in entry.match and entry.reply - RCODES and \
+               "flags" not in entry.match:
+                fails.append(RplintFail(test, entry, str(entry.reply - RCODES)))
+    return fails
+
+
+def step_section_unchecked(test: RplintTest) -> List[RplintFail]:
+    """ENTRY has non-empty sections but MATCH does not check for all of them"""
+    fails = []
+    for step in test.steps:
+        if step.type == "CHECK_ANSWER" and step.entry and "all" not in step.entry.match:
+            for section in SECTIONS:
+                if not is_empty(step.node.match("/entry/section/" + section + "/*")):
+                    if section not in step.entry.match:
+                        fails.append(RplintFail(test, step.entry, section))
+    return fails
+
+
+def range_overlapping_ips(test: RplintTest) -> List[RplintFail]:
+    """RANGE has common IPs with some previous overlapping RANGE"""
+    fails = []
+    for r1, r2 in itertools.combinations(test.ranges, 2):
+        # If the ranges overlap
+        if min(r1.b, r2.b) >= max(r1.a, r2.a):
+            if r1.addresses & r2.addresses:
+                info = "previous range on line %d" % get_line_number(test.path, r1.node.char)
+                fails.append(RplintFail(test, r2, info))
+    return fails
+
+
+def range_shadowing_match_rules(test: RplintTest) -> List[RplintFail]:
+    """ENTRY has no effect since one of previous entries has the same or broader match rules"""
+    fails = []
+    for r in test.ranges:
+        for e1, e2 in itertools.combinations(r.stored, 2):
+            try:
+                e1.match(e2.message)
+            except ValueError:
+                pass
+            else:
+                info = "previous entry on line %d" % get_line_number(test.path, e1.node.char)
+                if e1.match_fields > e2.match_fields:
+                    continue
+                if "subdomain" not in e1.match_fields and "subdomain" in e2.match_fields:
+                    continue
+                fails.append(RplintFail(test, e2, info))
+    return fails
+
+
+def step_duplicate_id(test: RplintTest) -> List[RplintFail]:
+    """STEP has the same ID as one of previous ones"""
+    fails = []
+    step_numbers = set()  # type: Set[int]
+    for step in test.steps:
+        if step.node.value in step_numbers:
+            fails.append(RplintFail(test, step))
+        else:
+            step_numbers.add(step.node.value)
+    return fails
+
+
+# TODO: This will make sense after we fix how we handle defaults in deckard.aug and scenario.py
+# We might just not use defaults altogether as testbound does
+# if "copy_id" not in adjust:
+#    entry_error(test, entry, "copy_id should be in ADJUST")
+
+def test_run_rplint(rpl_path: str) -> None:
+    t = RplintTest(rpl_path)
+    passed = t.run_checks()
+    if not passed:
+        raise RplintError(t.fails)
+
+
+def main():
+    try:
+        test_path = sys.argv[1]
+    except IndexError:
+        print("usage: %s <path to rpl file>" % sys.argv[0])
+        sys.exit(2)
+    if not os.path.isfile(test_path):
+        print("rplint.py works on single file only.")
+        print("Use rplint.sh with --scenarios=<directory with rpls> to run on rpls.")
+        sys.exit(2)
+    print("Linting %s" % test_path)
+    t = RplintTest(test_path)
+    passed = t.run_checks()
+    t.print_fails()
+
+    if passed:
+        sys.exit(0)
+    sys.exit(1)
+
+
+if __name__ == '__main__':
+    main()