test/test_dns.py

   1 #!/usr/bin/env python3
   2
   3 import unittest
   4
   5 from framework import VppTestCase
   6 from asfframework import VppTestRunner
   7 from ipaddress import *
   8
   9 from scapy.layers.inet import IP, UDP
  10 from scapy.layers.l2 import Ether
  11 from scapy.layers.dns import DNS, DNSQR
  12
  13
  14 class TestDns(VppTestCase):
  15     """Dns Test Cases"""
  16
  17     @classmethod
  18     def setUpClass(cls):
  19         super(TestDns, cls).setUpClass()
  20
  21     @classmethod
  22     def tearDownClass(cls):
  23         super(TestDns, cls).tearDownClass()
  24
  25     def setUp(self):
  26         super(TestDns, self).setUp()
  27
  28         self.create_pg_interfaces(range(1))
  29
  30         for i in self.pg_interfaces:
  31             i.admin_up()
  32             i.config_ip4()
  33             i.resolve_arp()
  34
  35     def tearDown(self):
  36         super(TestDns, self).tearDown()
  37
  38     def create_stream(self, src_if):
  39         """Create input packet stream for defined interface.
  40
  41         :param VppInterface src_if: Interface to create packet stream for.
  42         """
  43         good_request = (
  44             Ether(dst=src_if.local_mac, src=src_if.remote_mac)
  45             / IP(src=src_if.remote_ip4)
  46             / UDP(sport=1234, dport=53)
  47             / DNS(rd=1, qd=DNSQR(qname="bozo.clown.org"))
  48         )
  49
  50         bad_request = (
  51             Ether(dst=src_if.local_mac, src=src_if.remote_mac)
  52             / IP(src=src_if.remote_ip4)
  53             / UDP(sport=1234, dport=53)
  54             / DNS(rd=1, qd=DNSQR(qname="no.clown.org"))
  55         )
  56         pkts = [good_request, bad_request]
  57         return pkts
  58
  59     def verify_capture(self, dst_if, capture):
  60         """Verify captured input packet stream for defined interface.
  61
  62         :param VppInterface dst_if: Interface to verify captured packet stream
  63             for.
  64         :param list capture: Captured packet stream.
  65         """
  66         self.logger.info("Verifying capture on interface %s" % dst_if.name)
  67         for packet in capture:
  68             dns = packet[DNS]
  69             self.assertEqual(dns.an[0].rdata, "1.2.3.4")
  70
  71     def test_dns_unittest(self):
  72         """DNS Name Resolver Basic Functional Test"""
  73
  74         # Set up an upstream name resolver. We won't actually go there
  75         self.vapi.dns_name_server_add_del(
  76             is_ip6=0, is_add=1, server_address=IPv4Address("8.8.8.8").packed
  77         )
  78
  79         # Enable name resolution
  80         self.vapi.dns_enable_disable(enable=1)
  81
  82         # Manually add a static dns cache entry
  83         self.logger.info(self.vapi.cli("dns cache add bozo.clown.org 1.2.3.4"))
  84
  85         # Test the binary API
  86         rv = self.vapi.dns_resolve_name(name=b"bozo.clown.org")
  87         self.assertEqual(rv.ip4_address, IPv4Address("1.2.3.4").packed)
  88
  89         # Configure 127.0.0.1/8 on the pg interface
  90         self.vapi.sw_interface_add_del_address(
  91             sw_if_index=self.pg0.sw_if_index, prefix="127.0.0.1/8"
  92         )
  93
  94         # Send a couple of DNS request packets, one for bozo.clown.org
  95         # and one for no.clown.org which won't resolve
  96
  97         pkts = self.create_stream(self.pg0)
  98         self.pg0.add_stream(pkts)
  99         self.pg_enable_capture(self.pg_interfaces)
 100
 101         self.pg_start()
 102         pkts = self.pg0.get_capture(1)
 103         self.verify_capture(self.pg0, pkts)
 104
 105         # Make sure that the cache contents are correct
 106         str = self.vapi.cli("show dns cache verbose")
 107         self.assertIn("1.2.3.4", str)
 108         self.assertIn("[P] no.clown.org:", str)
 109
 110
 111 if __name__ == "__main__":
 112     unittest.main(testRunner=VppTestRunner)