Commit 23f4b153 authored by Štěpán Balážik's avatar Štěpán Balážik
Browse files

first version

Includes somewhat working parser and an object model of Scenario and
parent e6d35f30
# mockdns[^1] – toolchain for generating, parsing, running and evaluating mock DNS scenarios
`mockdns` will basically be a love child of Deckard and DNS Maze with some more flexibility and expressibily.
See [deckard#69]( for initial design discussion.
_This is very much work in progress._
## Instalation
* install packages from `requirements.txt`
* put the parent directory of this repository on your `PYTHONPATH`
Nothing fancy yet.
## Writing Deckard-like scenarios
See `examples/iter_minim_a.yaml` for a transliteration of `iter_minim_rpl.rpl` from Deckard.
## Generating scenarios programmaticaly
Instanciate `Scenario` class from `scenario/` and run it with (for now non-existent) runner.
## Overall structure
* `scenario`: base class describing one test run
* `parser`: parser using `strictyaml` schemas which spits out a `Scenario` object to be passed to runner
* `parser/actors`: subclasses of `Actor` from used in parsing scenarios (see `parser/` for the base classes and [this comment]( for the initial idea)
* `runner`: this will run `Scenario` objects and generated a binary (test passed/failed) and a structured (PCAP? C-DNS?) output for further analysis (empty for now)
* `contrib`: code from elsewhere (notably `matchpart` from Deckard for now – until I refactor it for better use)
## What's completely missing (i.e. not even an empty directory has been created)
* tools for generating (DNS Sketch) scenarios and transforming them from the old format
* analysis tools
* proper documentation
* tests! (but thanks to the modularity of `Actors` test writing should be pretty convinient)
# Roadmap/TODO:
_Order is particular but not strict._
* find a better name for `Range` from `scenario/`
* implement a runner
* this includes some kind of configuration parsing (probably the one from Deckard first until the new Knot Resolver config is available)
* implement a [zone matcher](
* this is very much needed for the Maze scenarios to work well
* also it will be a big step towards convenience of scenario writing
* refactor and better integrate `contrib/`
* implement a convertor from RPL
* measure performance of the parser
* reimplement/integrate DNS Maze and DNS Sketch feature sets
* add proper installation
* integrate to the new CI for Knot Resolver
[^1]: Name is very much WIP as well. There is some [Go package]( of this name but both [PyPI]( and [Read the docs]( names are free for now. :)
\ No newline at end of file
"""matchpart is used to compare two DNS messages using a single criterion"""
from typing import Any, Hashable, Sequence, Tuple, Union # noqa
import dns.edns
import dns.rcode
import dns.set
MismatchValue = Union[str, Sequence[Any]]
class DataMismatch(Exception):
def __init__(self, exp_val, got_val):
self.exp_val = exp_val
self.got_val = got_val
def format_value(value: MismatchValue) -> str:
if isinstance(value, list):
return " ".join([str(val) for val in value])
return str(value)
def __str__(self) -> str:
return 'expected "{}" got "{}"'.format(
self.format_value(self.exp_val), self.format_value(self.got_val)
def __eq__(self, other):
return (
isinstance(other, DataMismatch)
and self.exp_val == other.exp_val
and self.got_val == other.got_val
def __ne__(self, other):
return not self.__eq__(other)
def key(self) -> Tuple[Hashable, Hashable]:
def make_hashable(value):
if isinstance(value, (list, dns.set.Set)):
value = (make_hashable(item) for item in value)
value = tuple(value)
return value
return (make_hashable(self.exp_val), make_hashable(self.got_val))
def __hash__(self) -> int:
return hash(self.key)
def compare_val(exp, got):
"""Compare arbitraty objects, throw exception if different. """
if exp != got:
raise DataMismatch(exp, got)
return True
def compare_rrs(expected, got):
""" Compare lists of RR sets, throw exception if different. """
for rr in expected:
if rr not in got:
raise DataMismatch(expected, got)
for rr in got:
if rr not in expected:
raise DataMismatch(expected, got)
if len(expected) != len(got):
raise DataMismatch(expected, got)
return True
def compare_rrs_types(exp_val, got_val, skip_rrsigs):
"""sets of RR types in both sections must match"""
def rr_ordering_key(rrset):
if rrset.covers:
return rrset.covers, 1 # RRSIGs go to the end of RRtype list
return rrset.rdtype, 0
def key_to_text(rrtype, rrsig):
if not rrsig:
return dns.rdatatype.to_text(rrtype)
return "RRSIG(%s)" % dns.rdatatype.to_text(rrtype)
if skip_rrsigs:
exp_val = (rrset for rrset in exp_val if rrset.rdtype != dns.rdatatype.RRSIG)
got_val = (rrset for rrset in got_val if rrset.rdtype != dns.rdatatype.RRSIG)
exp_types = frozenset(rr_ordering_key(rrset) for rrset in exp_val)
got_types = frozenset(rr_ordering_key(rrset) for rrset in got_val)
if exp_types != got_types:
exp_types = tuple(key_to_text(*i) for i in sorted(exp_types))
got_types = tuple(key_to_text(*i) for i in sorted(got_types))
raise DataMismatch(exp_types, got_types)
def check_question(question):
if len(question) > 2:
raise NotImplementedError("More than one record in QUESTION SECTION.")
def match_opcode(exp, got):
return compare_val(exp.opcode(), got.opcode())
def match_qtype(exp, got):
if not exp.question and not got.question:
return True
if not exp.question:
raise DataMismatch("<empty question>", got.question[0].rdtype)
if not got.question:
raise DataMismatch(exp.question[0].rdtype, "<empty question>")
return compare_val(exp.question[0].rdtype, got.question[0].rdtype)
def match_qname(exp, got):
if not exp.question and not got.question:
return True
if not exp.question:
raise DataMismatch("<empty question>", got.question[0].name)
if not got.question:
raise DataMismatch(exp.question[0].name, "<empty question>")
return compare_val(exp.question[0].name, got.question[0].name)
def match_qcase(exp, got):
if not exp.question and not got.question:
return True
if not exp.question:
raise DataMismatch("<empty question>", got.question[0].name.labels)
if not got.question:
raise DataMismatch(exp.question[0].name.labels, "<empty question>")
return compare_val(exp.question[0].name.labels, got.question[0].name.labels)
def match_subdomain(exp, got):
if not exp.question:
return True
if got.question:
qname = got.question[0].name
qname =
if exp.question[0].name.is_superdomain(qname):
return True
raise DataMismatch(exp, got)
def match_flags(exp, got):
return compare_val(dns.flags.to_text(exp.flags), dns.flags.to_text(got.flags))
def match_rcode(exp, got):
return compare_val(dns.rcode.to_text(exp.rcode()), dns.rcode.to_text(got.rcode()))
def match_answer(exp, got):
return compare_rrs(exp.answer, got.answer)
def match_answertypes(exp, got):
return compare_rrs_types(exp.answer, got.answer, skip_rrsigs=True)
def match_answerrrsigs(exp, got):
return compare_rrs_types(exp.answer, got.answer, skip_rrsigs=False)
def match_authority(exp, got):
return compare_rrs(exp.authority, got.authority)
def match_additional(exp, got):
return compare_rrs(exp.additional, got.additional)
def match_edns(exp, got):
if got.edns != exp.edns:
raise DataMismatch(exp.edns, got.edns)
if got.payload != exp.payload:
raise DataMismatch(exp.payload, got.payload)
def match_nsid(exp, got):
nsid_opt = None
for opt in exp.options:
if opt.otype == dns.edns.NSID:
nsid_opt = opt
# Find matching NSID
for opt in got.options:
if opt.otype == dns.edns.NSID:
if not nsid_opt:
raise DataMismatch(None,
if opt == nsid_opt:
return True
raise DataMismatch(,
if nsid_opt:
raise DataMismatch(, None)
return True
"opcode": match_opcode,
"qtype": match_qtype,
"qname": match_qname,
"qcase": match_qcase,
"subdomain": match_subdomain,
"flags": match_flags,
"rcode": match_rcode,
"answer": match_answer,
"answertypes": match_answertypes,
"answerrrsigs": match_answerrrsigs,
"authority": match_authority,
"additional": match_additional,
"edns": match_edns,
"nsid": match_nsid,
def match_part(exp, got, code):
return MATCH[code](exp, got)
except KeyError as ex:
raise NotImplementedError('unknown match request "%s"' % code) from ex
# Reimplementation of iter_minim_a.rpl from Deckard showcasing some of the features of the new parser
config: "No configuration handling yet. :("
# Parsing:
# --------
# In an interactive python console, run
# from mockdns.parser import ParsedScenario
# scenario = ParsedScenario("examples/iter_minim_a.yaml")
# and explore the resulting object.
# defaults for Matchpart:
# -----------------------
# criteria: []
# adjust: []
# message:
# id: <random>
# opcode: QUERY
# rcode: NOERROR
# flags: QR
# eflags: [] # ENS flags
# question: []
# answer: []
# authority: []
# additional: []
# edns:
# version: 0
# payload: 4096
- first_step: 0 # default: start of scenario
last_step: 100 # default: end of scenario
- # can be both list and one value
- type: Matchpart
criteria: opcode, qtype, qname # can be both list and comma-separated
- copy_id # can be both list and comma-separated
question: . IN NS # can be both list and one value
answer: . 3600 IN NS K.ROOT-SERVERS.NET. # ttl is required for now
additional: K.ROOT-SERVERS.NET. 3600 IN A
- type: Matchpart
criteria: opcode, qtype, qname
adjust: copy_id
question: com. IN NS
authority: com. 3600 IN NS
additional: 3600 IN A
- addresses:
- type: Matchpart
- opcode
- qtype
- qname
adjust: copy_id
question: IN NS
authority: 3600 IN NS
additional: 3600 IN A
- addresses:
- type: Matchpart
criteria: opcode, qtype, qname
adjust: copy_id
question: IN A
answer: 3600 IN A
authority: 3600 IN NS
additional: 3600 IN A
- step_id: 1
type: Static
question: IN A
type: Checkpart
criteria: opcode, qtype, qname, flags, rcode, answer
flags: QR, RD, RA
question: IN A
answer: 3600 IN A
ignore_missing_imports = True
from mockdns.parser.parser import ParsedScenario
"""Define Abstract Base Classes for Actors"""
from abc import ABC, abstractmethod
from typing import Any, Dict, Iterable, Iterator, Mapping, Optional
from dns.message import Message
import strictyaml
from mockdns.scenario import DataMismatch
class Actor(ABC):
def __init__(self, yaml: strictyaml.YAML):
self.options = yaml["options"].data
self.line_number = yaml.start_line
self.yaml = yaml
def schema() -> strictyaml.Map:
def set_up(self) -> None:
def __repr__(self) -> str:
return f"{self.__class__.__name__}({self.options}, line: {self.line_number})"
class Generator(Actor):
def generate(self) -> Iterable[Message]:
def __iter__(self) -> Iterator[Message]:
return iter(self.generate())
class Matcher(Actor):
def match(self, query: Message) -> Optional[Message]:
def __call__(self, query: Message) -> Optional[Message]:
return self.match(query)
class CheckerMismatch(DataMismatch):
def __init__(self, actor: Actor):
super().__init__(f"mismatch on line {actor.line_number}")
class Checker(Actor):
def check(self, query: Message, answer: Message) -> None:
def __call__(self, query: Message, answer: Message) -> None:
return self.check(query, answer)
import glob
import os.path
__all__ = [
for file_name in glob.glob(os.path.join(__path__[0], "*.py")) # type: ignore # mypy issue #1422
if os.path.basename(file_name) not in ("", "")
from copy import deepcopy
from typing import Iterable, Optional
import dns.message
from dns.opcode import _by_text as OPCODES
from dns.rcode import _by_text as RCODES
from dns.flags import _by_text as FLAGS
from dns.flags import _edns_by_text as EFLAGS
import strictyaml
from mockdns.parser.actor_abc import Checker, CheckerMismatch, Generator, Matcher, Actor
from mockdns.parser.schema import DNSQuestion, DNSRecord, ListOrCommaSeparated, OneOrListOf
from mockdns.contrib.matchpart import matchpart
from mockdns.scenario import MatcherNoResponse
class _Matchpart(Actor):
ADJUST_OPTIONS = ["copy_id", "copy_query", "do_not_answer"]
def schema():
return strictyaml.Map(
"criteria": ListOrCommaSeparated(strictyaml.Enum(matchpart.MATCH.keys())),
strictyaml.Optional("adjust", default=[]): ListOrCommaSeparated(strictyaml.Enum(_Matchpart.ADJUST_OPTIONS)),
"message": strictyaml.Map(
strictyaml.Optional("id"): strictyaml.Int(),
strictyaml.Optional("opcode", default="QUERY"): strictyaml.Enum(
strictyaml.Optional("rcode", default="NOERROR"): strictyaml.Enum(
strictyaml.Optional("flags", default=["QR"]): ListOrCommaSeparated(
strictyaml.Optional("eflags", default=[]): ListOrCommaSeparated(
strictyaml.Optional("question", default=[]): OneOrListOf(DNSQuestion()),
strictyaml.Optional("answer", default=[]): OneOrListOf(DNSRecord()),
strictyaml.Optional("authority", default=[]): OneOrListOf(DNSRecord()),
strictyaml.Optional("additional", default=[]): OneOrListOf(DNSRecord()),
strictyaml.Optional("edns", default={"version": 0, "payload": 4096}): strictyaml.Map(
strictyaml.Optional("version", default=0): strictyaml.Int(),
strictyaml.Optional("payload", default=4096): strictyaml.Int(),
def set_up(self):
# Build the message from options.
self._message = dns.message.Message()
parsed = self.options["message"]
self._message.flags = sum([dns.flags.from_text(f) for f in set(parsed["flags"])])
for section in ["question", "answer", "authority", "additional"]:
if isinstance(parsed[section], dns.rrset.RRset):
getattr(self._message, section).append(parsed[section])
for record in parsed[section]:
getattr(self._message, section).append(record)
eflags = sum([dns.flags.edns_from_text(f) for f in set(parsed["eflags"])])
self._message.use_edns(edns=parsed["edns"]["version"], ednsflags=eflags, payload=parsed["edns"]["payload"])
def _adjust_message(self, source: dns.message.Message, destination: dns.message.Message) -> None:
if "do_not_answer" in self.options["adjust"]:
# FIXME: It's kind of weird to have this as adjust field but it was like this
# in the old Deckard, so let's leave it that way for now
raise MatcherNoResponse
if "copy_query" in self.options["adjust"]:
destination.question = source.question
elif "qname" in self.options["criteria"]:
# To ensure that case of qname is copied if we are matching it
destination.question = source.question
if "copy_id" in self.options["adjust"]: =
def match(self, query: dns.message.Message) -> Optional[dns.message.Message]:
for criterion in self.options["criteria"]:
matchpart.match_part(query, self._message, criterion)
except matchpart.DataMismatch:
return None
answer = deepcopy(self._message)
self._adjust_message(query, answer)
return answer
def check(self, query, answer) -> None:
message = deepcopy(self._message)
self._adjust_message(query, message)
for criterion in self.options["criteria"]:
matchpart.match_part(query, self._message, criterion)
except matchpart.DataMismatch as ex:
raise CheckerMismatch(self) from ex
# Matcher and Checker both implement __call__ so in order to reuse code,
# we subclass _Matchpart like this:
class Matchpart(_Matchpart, Matcher):
class Checkpart(_Matchpart, Checker):
adjust: copy_id, copy_query