regression-tests.dnsdist/test_Console.py

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