1
0
mirror of https://github.com/samba-team/samba.git synced 2024-12-23 17:34:34 +03:00
samba-mirror/python/samba/tests/dns_base.py
Rob van der Linde 6ac4833678 python: tests: update all super calls to python 3 style in tests
Signed-off-by: Rob van der Linde <rob@catalyst.net.nz>
Reviewed-by: Douglas Bagnall <douglas.bagnall@catalyst.net.nz>
Reviewed-by: Andrew Bartlett <abartlet@samba.org>

[abartlet@samba.org Some python2 style super() calls remain due
 to being an actual, even if reasonable, behaviour change]
2023-11-30 01:05:32 +00:00

438 lines
14 KiB
Python

# Unix SMB/CIFS implementation.
# Copyright (C) Kai Blin <kai@samba.org> 2011
# Copyright (C) Ralph Boehme <slow@samba.org> 2016
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
from samba.tests import TestCaseInTempDir
from samba.dcerpc import dns, dnsp
from samba import gensec, tests
from samba import credentials
import struct
import samba.ndr as ndr
import random
import socket
import uuid
import time
class DNSTest(TestCaseInTempDir):
def setUp(self):
super().setUp()
self.timeout = None
def errstr(self, errcode):
"Return a readable error code"
string_codes = [
"OK",
"FORMERR",
"SERVFAIL",
"NXDOMAIN",
"NOTIMP",
"REFUSED",
"YXDOMAIN",
"YXRRSET",
"NXRRSET",
"NOTAUTH",
"NOTZONE",
"0x0B",
"0x0C",
"0x0D",
"0x0E",
"0x0F",
"BADSIG",
"BADKEY"
]
return string_codes[errcode]
def assert_rcode_equals(self, rcode, expected):
"Helper function to check return code"
self.assertEqual(rcode, expected, "Expected RCODE %s, got %s" %
(self.errstr(expected), self.errstr(rcode)))
def assert_dns_rcode_equals(self, packet, rcode):
"Helper function to check return code"
p_errcode = packet.operation & dns.DNS_RCODE
self.assertEqual(p_errcode, rcode, "Expected RCODE %s, got %s" %
(self.errstr(rcode), self.errstr(p_errcode)))
def assert_dns_opcode_equals(self, packet, opcode):
"Helper function to check opcode"
p_opcode = packet.operation & dns.DNS_OPCODE
self.assertEqual(p_opcode, opcode, "Expected OPCODE %s, got %s" %
(opcode, p_opcode))
def make_name_packet(self, opcode, qid=None):
"Helper creating a dns.name_packet"
p = dns.name_packet()
if qid is None:
p.id = random.randint(0x0, 0xff00)
p.operation = opcode
p.questions = []
p.additional = []
return p
def finish_name_packet(self, packet, questions):
"Helper to finalize a dns.name_packet"
packet.qdcount = len(questions)
packet.questions = questions
def make_name_question(self, name, qtype, qclass):
"Helper creating a dns.name_question"
q = dns.name_question()
q.name = name
q.question_type = qtype
q.question_class = qclass
return q
def make_txt_record(self, records):
rdata_txt = dns.txt_record()
s_list = dnsp.string_list()
s_list.count = len(records)
s_list.str = records
rdata_txt.txt = s_list
return rdata_txt
def get_dns_domain(self):
"Helper to get dns domain"
return self.creds.get_realm().lower()
def dns_transaction_udp(self, packet, host,
dump=False, timeout=None):
"send a DNS query and read the reply"
s = None
if timeout is None:
timeout = self.timeout
try:
send_packet = ndr.ndr_pack(packet)
if dump:
print(self.hexdump(send_packet))
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, 0)
s.settimeout(timeout)
s.connect((host, 53))
s.sendall(send_packet, 0)
recv_packet = s.recv(2048, 0)
if dump:
print(self.hexdump(recv_packet))
response = ndr.ndr_unpack(dns.name_packet, recv_packet)
return (response, recv_packet)
finally:
if s is not None:
s.close()
def dns_transaction_tcp(self, packet, host,
dump=False, timeout=None):
"send a DNS query and read the reply, also return the raw packet"
s = None
if timeout is None:
timeout = self.timeout
try:
send_packet = ndr.ndr_pack(packet)
if dump:
print(self.hexdump(send_packet))
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0)
s.settimeout(timeout)
s.connect((host, 53))
tcp_packet = struct.pack('!H', len(send_packet))
tcp_packet += send_packet
s.sendall(tcp_packet)
recv_packet = s.recv(0xffff + 2, 0)
if dump:
print(self.hexdump(recv_packet))
response = ndr.ndr_unpack(dns.name_packet, recv_packet[2:])
finally:
if s is not None:
s.close()
# unpacking and packing again should produce same bytestream
my_packet = ndr.ndr_pack(response)
self.assertEqual(my_packet, recv_packet[2:])
return (response, recv_packet[2:])
def make_txt_update(self, prefix, txt_array, zone=None, ttl=900):
p = self.make_name_packet(dns.DNS_OPCODE_UPDATE)
updates = []
name = zone or self.get_dns_domain()
u = self.make_name_question(name, dns.DNS_QTYPE_SOA, dns.DNS_QCLASS_IN)
updates.append(u)
self.finish_name_packet(p, updates)
updates = []
r = dns.res_rec()
r.name = "%s.%s" % (prefix, name)
r.rr_type = dns.DNS_QTYPE_TXT
r.rr_class = dns.DNS_QCLASS_IN
r.ttl = ttl
r.length = 0xffff
rdata = self.make_txt_record(txt_array)
r.rdata = rdata
updates.append(r)
p.nscount = len(updates)
p.nsrecs = updates
return p
def check_query_txt(self, prefix, txt_array, zone=None):
name = "%s.%s" % (prefix, zone or self.get_dns_domain())
p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
questions = []
q = self.make_name_question(name, dns.DNS_QTYPE_TXT, dns.DNS_QCLASS_IN)
questions.append(q)
self.finish_name_packet(p, questions)
(response, response_packet) =\
self.dns_transaction_udp(p, host=self.server_ip)
self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
self.assertEqual(response.ancount, 1)
self.assertEqual(response.answers[0].rdata.txt.str, txt_array)
class DNSTKeyTest(DNSTest):
def setUp(self):
super().setUp()
self.settings = {}
self.settings["lp_ctx"] = self.lp_ctx = tests.env_loadparm()
self.settings["target_hostname"] = self.server
self.creds = credentials.Credentials()
self.creds.guess(self.lp_ctx)
self.creds.set_username(tests.env_get_var_value('USERNAME'))
self.creds.set_password(tests.env_get_var_value('PASSWORD'))
self.creds.set_kerberos_state(credentials.MUST_USE_KERBEROS)
self.newrecname = "tkeytsig.%s" % self.get_dns_domain()
def tkey_trans(self, creds=None):
"Do a TKEY transaction and establish a gensec context"
if creds is None:
creds = self.creds
self.key_name = "%s.%s" % (uuid.uuid4(), self.get_dns_domain())
p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
q = self.make_name_question(self.key_name,
dns.DNS_QTYPE_TKEY,
dns.DNS_QCLASS_IN)
questions = []
questions.append(q)
self.finish_name_packet(p, questions)
r = dns.res_rec()
r.name = self.key_name
r.rr_type = dns.DNS_QTYPE_TKEY
r.rr_class = dns.DNS_QCLASS_IN
r.ttl = 0
r.length = 0xffff
rdata = dns.tkey_record()
rdata.algorithm = "gss-tsig"
rdata.inception = int(time.time())
rdata.expiration = int(time.time()) + 60 * 60
rdata.mode = dns.DNS_TKEY_MODE_GSSAPI
rdata.error = 0
rdata.other_size = 0
self.g = gensec.Security.start_client(self.settings)
self.g.set_credentials(creds)
self.g.set_target_service("dns")
self.g.set_target_hostname(self.server)
self.g.want_feature(gensec.FEATURE_SIGN)
self.g.start_mech_by_name("spnego")
finished = False
client_to_server = b""
(finished, server_to_client) = self.g.update(client_to_server)
self.assertFalse(finished)
data = [x if isinstance(x, int) else ord(x) for x in list(server_to_client)]
rdata.key_data = data
rdata.key_size = len(data)
r.rdata = rdata
additional = [r]
p.arcount = 1
p.additional = additional
(response, response_packet) =\
self.dns_transaction_tcp(p, self.server_ip)
self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
tkey_record = response.answers[0].rdata
server_to_client = bytes(tkey_record.key_data)
(finished, client_to_server) = self.g.update(server_to_client)
self.assertTrue(finished)
self.verify_packet(response, response_packet)
def verify_packet(self, response, response_packet, request_mac=b""):
self.assertEqual(response.additional[0].rr_type, dns.DNS_QTYPE_TSIG)
tsig_record = response.additional[0].rdata
mac = bytes(tsig_record.mac)
# Cut off tsig record from dns response packet for MAC verification
# and reset additional record count.
key_name_len = len(self.key_name) + 2
tsig_record_len = len(ndr.ndr_pack(tsig_record)) + key_name_len + 10
# convert str/bytes to a list (of string char or int)
# so it can be modified
response_packet_list = [x if isinstance(x, int) else ord(x) for x in response_packet]
del response_packet_list[-tsig_record_len:]
response_packet_list[11] = 0
# convert modified list (of string char or int) to str/bytes
response_packet_wo_tsig = bytes(response_packet_list)
fake_tsig = dns.fake_tsig_rec()
fake_tsig.name = self.key_name
fake_tsig.rr_class = dns.DNS_QCLASS_ANY
fake_tsig.ttl = 0
fake_tsig.time_prefix = tsig_record.time_prefix
fake_tsig.time = tsig_record.time
fake_tsig.algorithm_name = tsig_record.algorithm_name
fake_tsig.fudge = tsig_record.fudge
fake_tsig.error = 0
fake_tsig.other_size = 0
fake_tsig_packet = ndr.ndr_pack(fake_tsig)
data = request_mac + response_packet_wo_tsig + fake_tsig_packet
self.g.check_packet(data, data, mac)
def sign_packet(self, packet, key_name):
"Sign a packet, calculate a MAC and add TSIG record"
packet_data = ndr.ndr_pack(packet)
fake_tsig = dns.fake_tsig_rec()
fake_tsig.name = key_name
fake_tsig.rr_class = dns.DNS_QCLASS_ANY
fake_tsig.ttl = 0
fake_tsig.time_prefix = 0
fake_tsig.time = int(time.time())
fake_tsig.algorithm_name = "gss-tsig"
fake_tsig.fudge = 300
fake_tsig.error = 0
fake_tsig.other_size = 0
fake_tsig_packet = ndr.ndr_pack(fake_tsig)
data = packet_data + fake_tsig_packet
mac = self.g.sign_packet(data, data)
mac_list = [x if isinstance(x, int) else ord(x) for x in list(mac)]
rdata = dns.tsig_record()
rdata.algorithm_name = "gss-tsig"
rdata.time_prefix = 0
rdata.time = fake_tsig.time
rdata.fudge = 300
rdata.original_id = packet.id
rdata.error = 0
rdata.other_size = 0
rdata.mac = mac_list
rdata.mac_size = len(mac_list)
r = dns.res_rec()
r.name = key_name
r.rr_type = dns.DNS_QTYPE_TSIG
r.rr_class = dns.DNS_QCLASS_ANY
r.ttl = 0
r.length = 0xffff
r.rdata = rdata
additional = [r]
packet.additional = additional
packet.arcount = 1
return mac
def bad_sign_packet(self, packet, key_name):
"""Add bad signature for a packet by bitflipping
the final byte in the MAC"""
mac_list = [x if isinstance(x, int) else ord(x) for x in list("badmac")]
rdata = dns.tsig_record()
rdata.algorithm_name = "gss-tsig"
rdata.time_prefix = 0
rdata.time = int(time.time())
rdata.fudge = 300
rdata.original_id = packet.id
rdata.error = 0
rdata.other_size = 0
rdata.mac = mac_list
rdata.mac_size = len(mac_list)
r = dns.res_rec()
r.name = key_name
r.rr_type = dns.DNS_QTYPE_TSIG
r.rr_class = dns.DNS_QCLASS_ANY
r.ttl = 0
r.length = 0xffff
r.rdata = rdata
additional = [r]
packet.additional = additional
packet.arcount = 1
def search_record(self, name):
p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
questions = []
q = self.make_name_question(name, dns.DNS_QTYPE_TXT, dns.DNS_QCLASS_IN)
questions.append(q)
self.finish_name_packet(p, questions)
(response, response_packet) =\
self.dns_transaction_udp(p, self.server_ip)
return response.operation & 0x000F
def make_update_request(self, delete=False):
"Create a DNS update request"
rr_class = dns.DNS_QCLASS_IN
ttl = 900
if delete:
rr_class = dns.DNS_QCLASS_NONE
ttl = 0
p = self.make_name_packet(dns.DNS_OPCODE_UPDATE)
q = self.make_name_question(self.get_dns_domain(),
dns.DNS_QTYPE_SOA,
dns.DNS_QCLASS_IN)
questions = []
questions.append(q)
self.finish_name_packet(p, questions)
updates = []
r = dns.res_rec()
r.name = self.newrecname
r.rr_type = dns.DNS_QTYPE_TXT
r.rr_class = rr_class
r.ttl = ttl
r.length = 0xffff
rdata = self.make_txt_record(['"This is a test"'])
r.rdata = rdata
updates.append(r)
p.nscount = len(updates)
p.nsrecs = updates
return p