"""Pruebas sin red: formato de los mensajes, autoritativo, resolver y reglas de cache.""" import os import struct import unittest import dns_auth import dns_resolver import dns_wire as w import mini_cache from netlab import HERE A, NS, MX, TXT = w.TYPES["A"], w.TYPES["NS"], w.TYPES["MX"], w.TYPES["TXT"] def zone(name): return dns_auth.Zone(*dns_auth.load_zone(os.path.join(HERE, "red", "zones", name))) def ask(server_zone, name, rtype=A, limit=None): query = w.encode(w.Message(7, 0, question=[(name, rtype)])) return w.decode(dns_auth.respond(server_zone, query, limit)[0]) class Format(unittest.TestCase): def test_query_bytes_follow_rfc1035(self): raw = w.encode(w.Message(0x1234, w.FLAG_RD, question=[("www.uni.test.", A)])) self.assertEqual(raw[:12], bytes.fromhex("1234 0100 0001 0000 0000 0000")) self.assertEqual(raw[12:], b"\x03www\x03uni\x04test\x00\x00\x01\x00\x01") def test_round_trip(self): records = [w.Record("a.test.", A, 60, "10.0.0.1"), w.Record("a.test.", w.TYPES["AAAA"], 60, "2001:db8::1"), w.Record("a.test.", w.TYPES["CNAME"], 60, "b.test."), w.Record("a.test.", MX, 60, (10, "m.a.test.")), w.Record("a.test.", TXT, 60, "hola " * 100)] back = w.decode(w.encode(w.Message(1, w.FLAG_QR, answer=records))) self.assertEqual([record.data for record in back.answer], [record.data for record in records]) def test_compression_uses_a_pointer_to_the_repeated_suffix(self): message = w.Message(1, 0, question=[("www.wayra.test.", A)], answer=[w.Record("www.wayra.test.", A, 60, "10.0.0.1")]) raw = w.encode(message) self.assertEqual(raw[12 + 16 + 4:12 + 16 + 6], b"\xc0\x0c") self.assertEqual(w.decode(raw).answer[0].name, "www.wayra.test.") def test_rejects_pointer_loop(self): raw = struct.pack("!6H", 1, 0, 1, 0, 0, 0) + b"\xc0\x0c" + b"\0\1\0\1" with self.assertRaises(w.WireError): w.decode(raw) def test_rejects_cut_messages(self): raw = w.encode(w.Message(1, w.FLAG_QR, answer=[w.Record("a.test.", A, 60, "10.0.0.1")])) for size in range(len(raw)): with self.assertRaises(w.WireError): w.decode(raw[:size]) class Authoritative(unittest.TestCase): def test_root_refers_to_tld_with_glue_and_without_authority(self): reply = ask(zone("root.zone"), "www.wayra.test.") self.assertFalse(reply.flags & w.FLAG_AA) self.assertEqual(reply.answer, []) self.assertEqual(reply.authority[0].data, "ns1.nic.test.") self.assertEqual(reply.additional[0].data, "10.212.4.10") def test_final_answer_is_authoritative(self): reply = ask(zone("wayra.zone"), "www.wayra.test.") self.assertTrue(reply.flags & w.FLAG_AA) self.assertEqual(reply.answer[0].data, "10.212.6.10") def test_names_ignore_case(self): self.assertEqual(ask(zone("wayra.zone"), "WwW.WaYrA.TeSt.").answer[0].data, "10.212.6.10") def test_cname_is_followed_inside_the_zone(self): reply = ask(zone("wayra.zone"), "blog.wayra.test.") self.assertEqual([record.rtype for record in reply.answer], [w.TYPES["CNAME"], A]) def test_nxdomain_and_nodata_differ(self): wayra = zone("wayra.zone") self.assertEqual(ask(wayra, "nada.wayra.test.").rcode, w.NXDOMAIN) nodata = ask(wayra, "www.wayra.test.", MX) self.assertEqual(nodata.rcode, w.NOERROR) self.assertEqual(nodata.answer, []) self.assertEqual(nodata.authority[0].rtype, w.TYPES["SOA"]) def test_refuses_names_outside_the_zone(self): self.assertEqual(ask(zone("wayra.zone"), "www.otra.test.").rcode, w.REFUSED) def test_large_answer_is_truncated_only_on_udp(self): wayra = zone("wayra.zone") short = ask(wayra, "grande.wayra.test.", TXT, dns_auth.UDP_LIMIT) self.assertTrue(short.flags & w.FLAG_TC) self.assertEqual(short.answer, []) whole = ask(wayra, "grande.wayra.test.", TXT) self.assertFalse(whole.flags & w.FLAG_TC) self.assertGreater(len(whole.answer[0].data), 512) class ReplyChecks(unittest.TestCase): QUESTION = ("www.wayra.test.", A) def reply(self, msg_id=99, flags=w.FLAG_QR, question=None): return w.Message(msg_id, flags, question=[question or self.QUESTION]) def accepts(self, reply, server="10.0.0.1"): return dns_resolver.acceptable_reply(reply, 99, self.QUESTION, (server, 53), "10.0.0.1") def test_matching_reply_is_accepted(self): self.assertTrue(self.accepts(self.reply())) def test_wrong_id_is_rejected(self): self.assertFalse(self.accepts(self.reply(msg_id=98))) def test_wrong_question_is_rejected(self): self.assertFalse(self.accepts(self.reply(question=("banco.test.", A)))) self.assertFalse(self.accepts(self.reply(question=("www.wayra.test.", MX)))) def test_wrong_server_is_rejected(self): self.assertFalse(self.accepts(self.reply(), server="10.0.0.2")) def test_a_query_is_not_a_reply(self): self.assertFalse(self.accepts(self.reply(flags=0))) def test_glue_outside_the_delegated_zone_is_ignored(self): reply = w.Message(1, w.FLAG_QR, additional=[ w.Record("ns1.wayra.test.", A, 60, "10.0.0.1"), w.Record("www.banco.test.", A, 60, "10.6.6.6")]) self.assertEqual([r.name for r in dns_resolver.usable_glue(reply, "wayra.test.")], ["ns1.wayra.test."]) def test_bailiwick_does_not_match_a_mere_text_suffix(self): self.assertFalse(w.in_bailiwick("xwayra.test.", "wayra.test.")) self.assertTrue(w.in_bailiwick("a.wayra.test.", "wayra.test.")) class CacheTtl(unittest.TestCase): def setUp(self): self.now = 1000.0 self.cache = dns_resolver.Cache(clock=lambda: self.now) self.record = w.Record("a.test.", A, 10, "10.0.0.1") def test_ttl_counts_down_and_expires(self): self.cache.put("a.test.", A, [self.record]) self.assertEqual(self.cache.get("a.test.", A)[0][0].ttl, 10) self.now += 4 self.assertEqual(self.cache.get("a.test.", A)[0][0].ttl, 6) self.now += 6 self.assertIsNone(self.cache.get("a.test.", A)) def test_ttl_zero_is_not_stored(self): self.cache.put("a.test.", A, [w.Record("a.test.", A, 0, "10.0.0.1")]) self.assertIsNone(self.cache.get("a.test.", A)) def test_negative_answers_are_remembered(self): self.cache.put("x.test.", A, [], ttl=5, rcode=w.NXDOMAIN) self.assertEqual(self.cache.get("x.test.", A), ([], w.NXDOMAIN)) class HttpCacheRules(unittest.TestCase): def test_lifetime(self): self.assertEqual(mini_cache.freshness_lifetime("max-age=60"), 60) self.assertEqual(mini_cache.freshness_lifetime("public, max-age=5"), 5) self.assertEqual(mini_cache.freshness_lifetime("max-age=60, no-cache"), 0) self.assertEqual(mini_cache.freshness_lifetime("max-age=hola"), 0) self.assertEqual(mini_cache.freshness_lifetime(None), 0) def test_what_can_be_stored(self): self.assertTrue(mini_cache.storable(200, "max-age=60", None)) self.assertTrue(mini_cache.storable(200, "no-cache", '"x"')) self.assertFalse(mini_cache.storable(200, "no-store, max-age=60", '"x"')) self.assertFalse(mini_cache.storable(200, "private, max-age=60", None)) self.assertFalse(mini_cache.storable(404, "max-age=60", None)) self.assertFalse(mini_cache.storable(200, None, None)) if __name__ == "__main__": unittest.main()