<commit_before>#!/usr/bin/env python3

from contextlib import suppress
import glob
import itertools
import os
import sys

import dns.name

import pydnstest.augwrap
import pydnstest.matchpart
import pydnstest.scenario

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):
    pass


def get_line_number(file, char_number):
    pos = 0
    for number, line in enumerate(open(file)):
        pos += len(line)
        if pos >= char_number:
            return number + 2


def is_empty(iterable):
    try:
        next(iterable)
    except StopIteration:
        return True
    return False


class Entry:
    def __init__(self, node):
        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):
        self.node = node
        self.type = node["/type"].value
        try:
            self.entry = Entry(node["/entry"])
        except KeyError:
            self.entry = None


class RplintTest:
    def __init__(self, path):
        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.results = None
        self.checks = [
                    #    entry_more_than_one_rcode,
                    #    entry_no_qname_qtype_copy_query,
                    #    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):
        """returns True iff all tests passed"""
        self.results = ""
        failed = False
        for check in self.checks:
            fails = check(self)
            for fail in fails:
                pos = get_line_number(self.path, fail)
                self.results += " ".join(["line", str(pos), check.__name__, check.__doc__, "\n"])

        if self.results == "":
            return True
        return False

    def print_results(self):
        print(self.results)


def config_trust_anchor_trailing_period_missing(test):
    """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 [0]
    return []


def scenario_timestamp(test):
    """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(record.char)
    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):
    """ENTRY without qname and qtype in MATCH and without copy_query in ADJUST"""
    fails = []
    for entry in test.range_entries:
        if ("qname" not in entry.match or "qtype" not in entry.match) \
	   and "question" not in entry.match:
            if "copy_query" not in entry.adjust:
                fails.append(entry.node.char)
    return fails


def entry_ns_in_authority(test):
    """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(entry.node.char)
    return fails


def entry_more_than_one_rcode(test):
    """ENTRY has more than one rcode in MATCH"""
    fails = []
    for entry in test.entries:
        if len(RCODES & entry.reply) > 1:
            fails.append(entry.node.char)
    return fails


def scenario_ad_or_rrsig_no_ta(test):
    """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(entry.node.char)
        else:
            for record in entry.records:
                if record["/type"].value == "RRSIG":
                    dnssec.append(entry.node.char)

    if dnssec:
        for k in test.config:
            if k[0] == "trust-anchor":
                return []
    return dnssec


def step_query_match(test):
    """STEP QUERY has a MATCH rule"""
    return [step.node.char for step in test.steps if step.type == "QUERY" and step.entry.match]


def step_check_answer_no_match(test):
    """ENTRY in STEP CHECK_ANSWER has no MATCH rule"""
    return [step.entry.node.char for step in test.steps if step.type == "CHECK_ANSWER"
            and not step.entry.match]


def step_unchecked_rcode(test):
    """ENTRY specifies rcode but STEP MATCH does not check for it."""
    fails = []<commit_msg>rplint: clean-up<commit_after>#!/usr/bin/env python3

from contextlib import suppress
import glob
import itertools
import os
import sys

import dns.name

import pydnstest.augwrap
import pydnstest.matchpart
import pydnstest.scenario

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):
    pass


def get_line_number(file, char_number):
    pos = 0
    for number, line in enumerate(open(file)):
        pos += len(line)
        if pos >= char_number:
            return number + 2


def is_empty(iterable):
    try:
        next(iterable)
    except StopIteration:
        return True
    return False


class Entry:
    def __init__(self, node):
        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):
        self.node = node
        self.type = node["/type"].value
        try:
            self.entry = Entry(node["/entry"])
        except KeyError:
            self.entry = None


class RplintTest:
    def __init__(self, path):
        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.results = None
        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):
        """returns True iff all tests passed"""
        self.results = ""
        for check in self.checks:
            fails = check(self)
            for fail in fails:
                pos = get_line_number(self.path, fail)
                self.results += " ".join(["line", str(pos), check.__name__, check.__doc__, "\n"])

        if self.results == "":
            return True
        return False

    def print_results(self):
        print(self.results)


def config_trust_anchor_trailing_period_missing(test):
    """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 [0]
    return []


def scenario_timestamp(test):
    """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(record.char)
    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):
    """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(entry.node.char)
    return fails


def entry_ns_in_authority(test):
    """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(entry.node.char)
    return fails


def entry_more_than_one_rcode(test):
    """ENTRY has more than one rcode in MATCH"""
    fails = []
    for entry in test.entries:
        if len(RCODES & entry.reply) > 1:
            fails.append(entry.node.char)
    return fails


def scenario_ad_or_rrsig_no_ta(test):
    """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(entry.node.char)
        else:
            for record in entry.records:
                if record["/type"].value == "RRSIG":
                    dnssec.append(entry.node.char)

    if dnssec:
        for k in test.config:
            if k[0] == "trust-anchor":
                return []
    return dnssec


def step_query_match(test):
    """STEP QUERY has a MATCH rule"""
    return [step.node.char for step in test.steps if step.type == "QUERY" and step.entry.match]


def step_check_answer_no_match(test):
    """ENTRY in STEP CHECK_ANSWER has no MATCH rule"""
    return [step.entry.node.char for step in test.steps if step.type == "CHECK_ANSWER" and
            not step.entry.match]


def step_unchecked_rcode(test):
    """ENTRY specifies rcode but STEP MATCH does not check for it."""
    fails = []
{'fraction_non_alphanumeric': 0.11025317395500779, 'fraction_numerical': 0.001781869478060732, 'mean_word_length': 2.0073677160080377, 'pattern_counts': {'":': 16, '<': 3, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 14, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2292920', 'n_tokens_mistral': 4070, 'n_tokens_neox': 3938, 'n_words': 1023}