return result.decode('UTF-8')
@classmethod
- def sendConsoleCommand(cls, command, timeout=5.0):
+ def sendConsoleCommand(cls, command, timeout=5.0, IPv6=False):
ourNonce = libnacl.utils.rand_nonce()
theirNonce = None
- sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+ sock = socket.socket(socket.AF_INET if not IPv6 else socket.AF_INET6, socket.SOCK_STREAM)
sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
if timeout:
sock.settimeout(timeout)
- sock.connect(("127.0.0.1", cls._consolePort))
+ sock.connect(("::1", cls._consolePort, 0, 0) if IPv6 else ("127.0.0.1", cls._consolePort))
sock.send(ourNonce)
theirNonce = sock.recv(len(ourNonce))
if len(theirNonce) != len(ourNonce):
#!/usr/bin/env python
import base64
import dns
+import os
import socket
import time
from dnsdisttests import DNSDistTest
version = self.sendConsoleCommand('showVersion()')
self.assertTrue(version.startswith('dnsdist '))
+class TestConsoleAllowedV6(DNSDistTest):
+
+ _consoleKey = DNSDistTest.generateConsoleKey()
+ _consoleKeyB64 = base64.b64encode(_consoleKey).decode('ascii')
+
+ _config_params = ['_consoleKeyB64', '_consolePort', '_testServerPort']
+ _config_template = """
+ setKey("%s")
+ controlSocket("[::1]:%s")
+ newServer{address="127.0.0.1:%d"}
+ """
+
+ def testConsoleAllowed(self):
+ """
+ Console: Allowed IPv6
+ """
+ if 'SKIP_IPV6_TESTS' in os.environ:
+ raise unittest.SkipTest('IPv6 tests are disabled')
+ version = self.sendConsoleCommand('showVersion()', IPv6=True)
+ self.assertTrue(version.startswith('dnsdist '))
+
class TestConsoleNotAllowed(DNSDistTest):
_consoleKey = DNSDistTest.generateConsoleKey()