regression-tests.dnsdist/test_TLS.py

   1 #!/usr/bin/env python
   2 import dns
   3 from dnsdisttests import DNSDistTest
   4
   5 class TestTLS(DNSDistTest):
   6
   7     _serverKey = 'server.key'
   8     _serverCert = 'server.chain'
   9     _serverName = 'tls.tests.dnsdist.org'
  10     _caCert = 'ca.pem'
  11     _tlsServerPort = 8453
  12     _config_template = """
  13     newServer{address="127.0.0.1:%s"}
  14     addTLSLocal("127.0.0.1:%s", "%s", "%s")
  15     """
  16     _config_params = ['_testServerPort', '_tlsServerPort', '_serverCert', '_serverKey']
  17
  18     def testTLSSimple(self):
  19         """
  20         TLS: Single query
  21         """
  22         name = 'single.tls.tests.powerdns.com.'
  23         query = dns.message.make_query(name, 'A', 'IN', use_edns=False)
  24         response = dns.message.make_response(query)
  25         rrset = dns.rrset.from_text(name,
  26                                     3600,
  27                                     dns.rdataclass.IN,
  28                                     dns.rdatatype.A,
  29                                     '127.0.0.1')
  30         response.answer.append(rrset)
  31
  32         conn = self.openTLSConnection(self._tlsServerPort, self._serverName, self._caCert)
  33
  34         self.sendTCPQueryOverConnection(conn, query, response=response)
  35         (receivedQuery, receivedResponse) = self.recvTCPResponseOverConnection(conn, useQueue=True)
  36         self.assertTrue(receivedQuery)
  37         self.assertTrue(receivedResponse)
  38         receivedQuery.id = query.id
  39         self.assertEquals(query, receivedQuery)
  40         self.assertEquals(response, receivedResponse)
  41
  42     def testTLKA(self):
  43         """
  44         TLS: Several queries over the same connection
  45         """
  46         name = 'ka.tls.tests.powerdns.com.'
  47         query = dns.message.make_query(name, 'A', 'IN', use_edns=False)
  48         response = dns.message.make_response(query)
  49         rrset = dns.rrset.from_text(name,
  50                                     3600,
  51                                     dns.rdataclass.IN,
  52                                     dns.rdatatype.A,
  53                                     '127.0.0.1')
  54         response.answer.append(rrset)
  55
  56         conn = self.openTLSConnection(self._tlsServerPort, self._serverName, self._caCert)
  57
  58         for idx in range(5):
  59             self.sendTCPQueryOverConnection(conn, query, response=response)
  60             (receivedQuery, receivedResponse) = self.recvTCPResponseOverConnection(conn, useQueue=True)
  61             self.assertTrue(receivedQuery)
  62             self.assertTrue(receivedResponse)
  63             receivedQuery.id = query.id
  64             self.assertEquals(query, receivedQuery)
  65             self.assertEquals(response, receivedResponse)
  66
  67     def testTLSPipelining(self):
  68         """
  69         TLS: Several queries over the same connection without waiting for the responses
  70         """
  71         name = 'pipelining.tls.tests.powerdns.com.'
  72         query = dns.message.make_query(name, 'A', 'IN', use_edns=False)
  73         response = dns.message.make_response(query)
  74         rrset = dns.rrset.from_text(name,
  75                                     3600,
  76                                     dns.rdataclass.IN,
  77                                     dns.rdatatype.A,
  78                                     '127.0.0.1')
  79         response.answer.append(rrset)
  80
  81         conn = self.openTLSConnection(self._tlsServerPort, self._serverName, self._caCert)
  82
  83         for idx in range(100):
  84             self.sendTCPQueryOverConnection(conn, query, response=response)
  85
  86         for idx in range(100):
  87             (receivedQuery, receivedResponse) = self.recvTCPResponseOverConnection(conn, useQueue=True)
  88             self.assertTrue(receivedQuery)
  89             self.assertTrue(receivedResponse)
  90             receivedQuery.id = query.id
  91             self.assertEquals(query, receivedQuery)
  92             self.assertEquals(response, receivedResponse)