Verified Commit 6e0ba029 authored by Petr Špaček's avatar Petr Špaček
Browse files

support explicit specification of OPCODE using REPLY

Default OPCODE is QUERY so existing tests should not require changes.

refs: #11
parent c1991d34
Pipeline #26377 passed with stage
in 1 minute and 6 seconds
......@@ -133,6 +133,7 @@ Particular use of data in an ``ENTRY`` depends on context and is different
for ``STEP`` types and ``RANGE`` blocks, see details below.
In any case, entry starts with ``ENTRY_BEGIN`` and ends with ``ENTRY_END`` keywords and share ``REPLY`` and ``SECTION`` definitions.
Some fields in DNS messages have default values which can be overriden by explicit specification.
Format of query messages (for ``STEP QUERY``)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
......@@ -142,9 +143,9 @@ Format of query messages (for ``STEP QUERY``)
STEP <n> QUERY
ENTRY_BEGIN
REPLY <flags> ; REPLY is a bad keyword name, these flags will be sent out!
SECTION QUESTION ; it is possible to replace QUESTION section or omit it
<name> <class> <type> ; to simulate weird queries
REPLY <OPCODE flags> ; REPLY is a bad keyword name, OPCODE and flags will be sent out!
SECTION QUESTION ; it is possible to replace QUESTION section or omit it
<name> <class> <type> ; to simulate weird queries
ENTRY_END
The message will be assigned a random message ID, converted into DNS wire format, and sent to the binary under test.
......@@ -160,7 +161,7 @@ Format of expected messages (for ``STEP CHECK_ANSWER``)
ENTRY_BEGIN
MATCH <match element list> ; MATCH elements define what message fields will be compared
REPLY <RCODE flags> ; REPLY field here defines expected RCODE as well as flags!
REPLY <OPCODE RCODE flags> ; REPLY field here defines expected OPCODE, RCODE as well as flags!
SECTION QUESTION
<name> <class> <type> ; to simulate weird queries
SECTION <type2>
......@@ -179,7 +180,7 @@ Entries in ``RANGE`` blocks are used to answer queries *from binaries under test
ENTRY_BEGIN
MATCH <match element list> ; all MATCH elements must match before using this answer template
ADJUST <adjust element list> ; ADJUST fields will be modified before answering
REPLY <RCODE flags> ; RCODE and flags to be set in the outgoing answer
REPLY <OPCODE RCODE flags> ; OPCODE, RCODE, and flags to be set in the outgoing answer
SECTION <type1>
<RR sets>
SECTION <type2>
......@@ -202,7 +203,10 @@ Entries present in Deckard scenario define values *expected* in DNS messages. Th
============ =========================================================================================
element DNS message fields and additional rules
============ =========================================================================================
opcode message ``OPCODE`` = standard query (value 0)
opcode ``OPCODE`` as `defined in IANA registry <https://www.iana.org/assignments/dns-parameters/dns-parameters.xhtml#dns-parameters-5>`_
- *expected* message ``OPCODE`` is defined by ``REPLY`` keyword
qtype RR type in question section [qmatch]_
qname name in question section (case insensitive) [qmatch]_
qcase name in question section (case sensitive) [qmatch]_
......@@ -302,6 +306,19 @@ Example:
SECTION ADDITIONAL
ns.example.com. IN A 1.2.3.4
Default values for DNS messages
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
========== ===========================================================================================
feature default value
========== ===========================================================================================
ADJUST copy_id
EDNS version 0 with buffer size 4096 B
MATCH opcode, qtype, qname
REPLY QUERY, NOERROR
========== ===========================================================================================
Entry with RAW data
^^^^^^^^^^^^^^^^^^^
An entry might have special section named ``RAW``. This section is used only for sending raw,
......
......@@ -33,7 +33,7 @@ let hex_re = /[0-9a-fA-F]+/
let match_option = "opcode" | "qtype" | "qcase" | "qname" | "subdomain" | "flags" | "rcode" | "question" | "answer" | "authority" | "additional" | "all" | "TCP" | "ttl"
let adjust_option = "copy_id" | "copy_query"
let reply_option = "QR" | "TC" | "AA" | "AD" | "RD" | "RA" | "CD" | "DO" | "NOERROR" | "FORMERR" | "SERVFAIL" | "NXDOMAIN" | "NOTIMP" | "REFUSED" | "YXDOMAIN" | "YXRRSET" | "NXRRSET" | "NOTAUTH" | "NOTZONE" | "BADVERS"
let reply_option = "QR" | "TC" | "AA" | "AD" | "RD" | "RA" | "CD" | "DO" | "NOERROR" | "FORMERR" | "SERVFAIL" | "NXDOMAIN" | "NOTIMP" | "REFUSED" | "YXDOMAIN" | "YXRRSET" | "NXRRSET" | "NOTAUTH" | "NOTZONE" | "BADVERS" | "QUERY" | "IQUERY" | "STATUS" | "NOTIFY" | "UPDATE"
let step_option = "REPLY" | "QUERY" | "CHECK_ANSWER" | "CHECK_OUT_QUERY" | /TIME_PASSES[ \t]+ELAPSE/
let mandatory = [del_str "MANDATORY" . label "mandatory" . value "true" . comment_or_eol]
......
......@@ -202,19 +202,7 @@ class Entry:
self.match_fields = ['opcode', 'qtype', 'qname']
# FLAGS
self.fields = [f.value for f in node.match("/reply")]
flags = []
rcode = dns.rcode.from_text(self.default_rc)
for code in self.fields:
if code == 'DO':
self.message.want_dnssec(True)
continue
try:
rcode = dns.rcode.from_text(code)
except dns.rcode.UnknownRcode:
flags.append(code)
self.message.flags = dns.flags.from_text(' '.join(flags))
self.message.set_rcode(rcode)
self.process_reply_line(node)
# ADJUST
self.adjust_fields = [m.value for m in node.match("/adjust")]
......@@ -303,6 +291,67 @@ class Entry:
txt += 'ENTRY_END\n'
return txt
def process_reply_line(self, node):
"""Extracts flags, rcode and opcode from given node and adjust dns message accordingly"""
self.fields = [f.value for f in node.match("/reply")]
if 'DO' in self.fields:
self.message.want_dnssec(True)
opcode = self.get_opcode(fields=self.fields)
rcode = self.get_rcode(fields=self.fields)
self.message.flags = self.get_flags(fields=self.fields)
if rcode is not None:
self.message.set_rcode(rcode)
if opcode is not None:
self.message.set_opcode(opcode)
@classmethod
def get_flags(cls, fields):
"""From `fields` extracts and returns flags"""
flags = []
for code in fields:
try:
dns.flags.from_text(code) # throws KeyError on failure
flags.append(code)
except KeyError:
pass
return dns.flags.from_text(' '.join(flags))
@classmethod
def get_rcode(cls, fields):
"""
From `fields` extracts and returns rcode.
Throws `ValueError` if there are more then one rcodes
"""
rcodes = []
for code in fields:
try:
rcodes.append(dns.rcode.from_text(code))
except dns.rcode.UnknownRcode:
pass
if len(rcodes) > 1:
raise ValueError("Parse failed, too many rcode values.", rcodes)
if len(rcodes) == 0:
return None
return rcodes[0]
@classmethod
def get_opcode(cls, fields):
"""
From `fields` extracts and returns opcode.
Throws `ValueError` if there are more then one opcodes
"""
opcodes = []
for code in fields:
try:
opcodes.append(dns.opcode.from_text(code))
except dns.opcode.UnknownOpcode:
pass
if len(opcodes) > 1:
raise ValueError("Parse failed, too many opcode values.")
if len(opcodes) == 0:
return None
return opcodes[0]
def match_part(self, code, msg):
""" Compare scripted reply to given message using single criteria. """
if code not in self.match_fields and 'all' not in self.match_fields:
......
""" This is unittest file for scenario.py """
import pytest
from pydnstest.scenario import Entry
RCODE_FLAGS = ['NOERROR', 'FORMERR', 'SERVFAIL', 'NXDOMAIN', 'NOTIMP', 'REFUSED', 'YXDOMAIN',
'YXRRSET', 'NXRRSET', 'NOTAUTH', 'NOTZONE', 'BADVERS']
OPCODE_FLAGS = ['QUERY', 'IQUERY', 'STATUS', 'NOTIFY', 'UPDATE']
FLAGS = ['QR', 'TC', 'AA', 'AD', 'RD', 'RA', 'CD']
def test_entry__get_flags():
"""Checks if all rcodes and opcodes are filtered out"""
expected_flags = Entry.get_flags(FLAGS)
for flag in RCODE_FLAGS + OPCODE_FLAGS:
rcode_flags = Entry.get_flags(FLAGS + [flag])
assert rcode_flags == expected_flags, \
'Entry._get_flags does not filter out "{flag}"'.format(flag=flag)
def test_entry__get_rcode():
"""
Checks if the error is raised for multiple rcodes
checks if None is returned for no rcode
checks if flags and opcode are filtered out
"""
with pytest.raises(ValueError):
Entry.get_rcode(RCODE_FLAGS[:2])
assert Entry.get_rcode(FLAGS) is None
assert Entry.get_rcode([]) is None
for rcode in RCODE_FLAGS:
given_rcode = Entry.get_rcode(FLAGS + OPCODE_FLAGS + [rcode])
assert given_rcode is not None, 'Entry.get_rcode does not recognize {rcode}'.format(
rcode=rcode)
def test_entry__get_opcode():
"""
Checks if the error is raised for multiple opcodes
checks if None is returned for no opcode
checks if flags and opcode are filtered out
"""
with pytest.raises(ValueError):
Entry.get_opcode(OPCODE_FLAGS[:2])
assert Entry.get_opcode(FLAGS) is None
assert Entry.get_opcode([]) is None
for opcode in OPCODE_FLAGS:
given_rcode = Entry.get_opcode(FLAGS + RCODE_FLAGS + [opcode])
assert given_rcode is not None, 'Entry.get_opcode does not recognize {opcode}'.format(
opcode=opcode)
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment