regression-tests.dnsdist/test_Console.py

   1 #!/usr/bin/env python
   2 import base64
   3 import dns
   4 import os
   5 import socket
   6 import time
   7 from dnsdisttests import DNSDistTest
   8
   9 class TestConsoleAllowed(DNSDistTest):
  10
  11     _consoleKey = DNSDistTest.generateConsoleKey()
  12     _consoleKeyB64 = base64.b64encode(_consoleKey).decode('ascii')
  13
  14     _config_params = ['_consoleKeyB64', '_consolePort', '_testServerPort']
  15     _config_template = """
  16     setKey("%s")
  17     controlSocket("127.0.0.1:%s")
  18     newServer{address="127.0.0.1:%d"}
  19     """
  20
  21     def testConsoleAllowed(self):
  22         """
  23         Console: Allowed
  24         """
  25         version = self.sendConsoleCommand('showVersion()')
  26         self.assertTrue(version.startswith('dnsdist '))
  27
  28 class TestConsoleAllowedV6(DNSDistTest):
  29
  30     _consoleKey = DNSDistTest.generateConsoleKey()
  31     _consoleKeyB64 = base64.b64encode(_consoleKey).decode('ascii')
  32
  33     _config_params = ['_consoleKeyB64', '_consolePort', '_testServerPort']
  34     _config_template = """
  35     setKey("%s")
  36     controlSocket("[::1]:%s")
  37     newServer{address="127.0.0.1:%d"}
  38     """
  39
  40     def testConsoleAllowed(self):
  41         """
  42         Console: Allowed IPv6
  43         """
  44         if 'SKIP_IPV6_TESTS' in os.environ:
  45             raise unittest.SkipTest('IPv6 tests are disabled')
  46         version = self.sendConsoleCommand('showVersion()', IPv6=True)
  47         self.assertTrue(version.startswith('dnsdist '))
  48
  49 class TestConsoleNotAllowed(DNSDistTest):
  50
  51     _consoleKey = DNSDistTest.generateConsoleKey()
  52     _consoleKeyB64 = base64.b64encode(_consoleKey).decode('ascii')
  53
  54     _config_params = ['_consoleKeyB64', '_consolePort', '_testServerPort']
  55     _config_template = """
  56     setKey("%s")
  57     controlSocket("127.0.0.1:%s")
  58     setConsoleACL({'192.0.2.1'})
  59     newServer{address="127.0.0.1:%d"}
  60     """
  61
  62     def testConsoleAllowed(self):
  63         """
  64         Console: Not allowed by the ACL
  65         """
  66         self.assertRaises(socket.error, self.sendConsoleCommand, 'showVersion()')
  67
  68 class TestConsoleNoKey(DNSDistTest):
  69
  70     _consoleKey = DNSDistTest.generateConsoleKey()
  71     _consoleKeyB64 = base64.b64encode(_consoleKey).decode('ascii')
  72
  73     _config_params = ['_consolePort', '_testServerPort']
  74     _config_template = """
  75     controlSocket("127.0.0.1:%s")
  76     newServer{address="127.0.0.1:%d"}
  77     """
  78
  79     def testConsoleAllowed(self):
  80         """
  81         Console: No key, the connection should not be allowed
  82         """
  83         self.assertRaises(socket.error, self.sendConsoleCommand, 'showVersion()')
  84
  85 class TestConsoleConcurrentConnections(DNSDistTest):
  86
  87     _consoleKey = DNSDistTest.generateConsoleKey()
  88     _consoleKeyB64 = base64.b64encode(_consoleKey).decode('ascii')
  89     _maxConns = 2
  90
  91     _config_params = ['_consoleKeyB64', '_consolePort', '_testServerPort', '_maxConns']
  92     _config_template = """
  93     setKey("%s")
  94     controlSocket("127.0.0.1:%s")
  95     newServer{address="127.0.0.1:%d"}
  96     setConsoleMaximumConcurrentConnections(%d)
  97     """
  98
  99     def testConsoleConnectionsLimit(self):
 100         """
 101         Console: Check the maximum number of connections
 102         """
 103         conns = []
 104         # open the maximum number of connections
 105         for _ in range(self._maxConns):
 106             conn = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
 107             conn.connect(("127.0.0.1", self._consolePort))
 108             conns.append(conn)
 109
 110         # we now hold all the slots, let's try to establish a new connection
 111         self.assertRaises(socket.error, self.sendConsoleCommand, 'showVersion()')
 112
 113         # free one slot
 114         conns[0].close()
 115         conns[0] = None
 116         time.sleep(1)
 117
 118         # this should work
 119         version = self.sendConsoleCommand('showVersion()')
 120         self.assertTrue(version.startswith('dnsdist '))