[nat64] add functional tests (#8161)

The test includes:
- Counters (protocol & error (4to6 no mapping, other errors are not
  expected))
- ICMP ping Connectivity
- UDP connectivity

tests/scripts/thread-cert/border_router/nat64/test_single_border_router.py

@@ -30,6 +30,13 @@ import unittest
 
 import config
 import thread_cert
+import ipv6
+
+import ipaddress
+
+# For NAT64 connectivity tests
+import socket
+import select
 
 # Test description:
 #   This test verifies the advertisement of local NAT64 prefix in Thread network
@@ -79,6 +86,28 @@ class Nat64SingleBorderRouter(thread_cert.TestCase):
         },
     }
 
+    def get_host_ip(self):
+        sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+        sock.setblocking(False)
+        # Note: The address is not important, we use this function to get the host ip address.
+        sock.connect(('8.8.8.8', 54321))
+        host_ip, _ = sock.getsockname()
+        sock.close()
+        return host_ip
+
+    def receive_from(self, sock, timeout_seconds):
+        ready = select.select([sock], [], [], timeout_seconds)
+        if ready[0]:
+            return sock.recv(1024)
+        else:
+            raise AssertionError("No data recevied")
+
+    def listen_udp(self, addr, port):
+        sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+        sock.setblocking(False)
+        sock.bind((addr, port))
+        return sock
+
     def test(self):
         br = self.nodes[BR]
         router = self.nodes[ROUTER]
@@ -157,12 +186,57 @@ class Nat64SingleBorderRouter(thread_cert.TestCase):
         br.enable_br()
         self.simulator.go(config.BORDER_ROUTER_STARTUP_DELAY)
 
-        # Same NAT64 prefix is advertised to Network Data.
-        self.assertEqual(len(br.get_netdata_nat64_prefix()), 1)
-        self.assertEqual(nat64_prefix, br.get_netdata_nat64_prefix()[0])
+        #
+        # Case 5. NAT64 connectivity and counters.
+        #
+        host_ip = self.get_host_ip()
+        self.assertTrue(router.ping(ipaddr=host_ip))
+
+        mappings = br.get_nat64_mappings()
+        self.assertEqual(mappings[0]['counters']['ICMP']['4to6']['packets'], 1)
+        self.assertEqual(mappings[0]['counters']['ICMP']['6to4']['packets'], 1)
+        self.assertEqual(mappings[0]['counters']['total']['4to6']['packets'], 1)
+        self.assertEqual(mappings[0]['counters']['total']['6to4']['packets'], 1)
+
+        counters = br.get_nat64_counters()
+        self.assertEqual(counters['protocol']['ICMP']['4to6']['packets'], 1)
+        self.assertEqual(counters['protocol']['ICMP']['6to4']['packets'], 1)
+
+        sock = self.listen_udp('0.0.0.0', 54321)
+        router.udp_start('::', 54321)
+        # We can use IPv4 addresses for commands like UDP send.
+        # The address will be converted to an IPv6 address by CLI.
+        router.udp_send(10, host_ip, 54321)
+        self.assertTrue(len(self.receive_from(sock, timeout_seconds=1)) == 10)
+
+        sock.close()
+
+        counters = br.get_nat64_counters()
+        self.assertEqual(counters['protocol']['UDP']['6to4']['packets'], 1)
+        mappings = br.get_nat64_mappings()
+        self.assertEqual(mappings[0]['counters']['UDP']['6to4']['packets'], 1)
+
+        # We should be able to get a IPv4 mapped IPv6 address.
+        # 203.0.113.1, RFC5737 TEST-NET-3, should be unreachable.
+        mapped_ip6_address = str(
+            ipv6.synthesize_ip6_address(ipaddress.IPv6Network(nat64_prefix), ipaddress.IPv4Address('203.0.113.1')))
+        self.assertFalse(router.ping(ipaddr=mapped_ip6_address))
+
+        mappings = br.get_nat64_mappings()
+        self.assertEqual(mappings[0]['counters']['ICMP']['4to6']['packets'], 1)
+        self.assertEqual(mappings[0]['counters']['ICMP']['6to4']['packets'], 2)
+        self.assertEqual(mappings[0]['counters']['total']['4to6']['packets'], 1)
+        self.assertEqual(mappings[0]['counters']['total']['6to4']['packets'], 3)
+
+        counters = br.get_nat64_counters()
+        self.assertEqual(counters['protocol']['ICMP']['4to6']['packets'], 1)
+        self.assertEqual(counters['protocol']['ICMP']['6to4']['packets'], 2)
 
         #
-        # Case 5. Disable and re-enable ethernet on the border router.
+        # Case 6. Disable and re-enable ethernet on the border router.
+        # Note: disable_ether will remove default route but enable_ether won't add it back,
+        # NAT64 connectivity tests will fail after this.
+        # TODO: Add a default IPv4 route after enable_ether.
         #
         br.disable_ether()
         self.simulator.go(5)
@@ -176,6 +250,9 @@ class Nat64SingleBorderRouter(thread_cert.TestCase):
         # Same NAT64 prefix is advertised to Network Data.
         self.assertEqual(len(br.get_netdata_nat64_prefix()), 1)
         self.assertEqual(nat64_prefix, br.get_netdata_nat64_prefix()[0])
+        # Same NAT64 prefix is advertised to Network Data.
+        self.assertEqual(len(br.get_netdata_nat64_prefix()), 1)
+        self.assertEqual(nat64_prefix, br.get_netdata_nat64_prefix()[0])
 
 
 if __name__ == '__main__':

tests/scripts/thread-cert/ipv6.py

@@ -29,10 +29,10 @@
 
 import abc
 import io
+import ipaddress
 import struct
 
 from binascii import hexlify
-from ipaddress import ip_address
 
 import common
 
@@ -93,6 +93,25 @@ def calculate_checksum(data):
         return checksum
 
 
+def synthesize_ip6_address(ip6_network: ipaddress.IPv6Network,
+                           ip4_address: ipaddress.IPv4Address) -> ipaddress.IPv6Address:
+    """ Synthesize an IPv6 address from a prefix for NAT64 and an IPv4 address.
+
+    Only supports /96 network for now.
+
+    Args:
+        ip6_network: The network for NAT64.
+        ip4_address: The IPv4 address.
+
+    Returns:
+        ipaddress.IPv6Address: The synthesized IPv6 address.
+    """
+    if ip6_network.prefixlen != 96:
+        # We are only using /96 networks in openthread
+        raise NotImplementedError("synthesize_ip6_address only supports /96 networks")
+    return ipaddress.IPv6Address(int(ip6_network.network_address) | int(ip4_address))
+
+
 class PacketFactory(object):
     """ Interface for classes that produce objects from data. """
 
@@ -209,7 +228,7 @@ class IPv6PseudoHeader(ConvertibleToBytes):
         if isinstance(value, bytearray):
             value = bytes(value)
 
-        return ip_address(value)
+        return ipaddress.ip_address(value)
 
     @property
     def source_address(self):
@@ -267,7 +286,7 @@ class IPv6Header(ConvertibleToBytes, BuildableFromBytes):
         if isinstance(value, bytearray):
             value = bytes(value)
 
-        return ip_address(value)
+        return ipaddress.ip_address(value)
 
     @property
     def source_address(self):

tests/scripts/thread-cert/node.py

@@ -1999,6 +1999,97 @@ class NodeImpl:
         self.send_command(cmd)
         return self._expect_command_output()[0].split(' ')[0]
 
+    def get_nat64_mappings(self):
+        cmd = 'nat64 mappings'
+        self.send_command(cmd)
+        result = self._expect_command_output()
+        session = None
+        session_counters = None
+        sessions = []
+
+        for line in result:
+            m = re.match(
+                r'\|\s+([a-f0-9]+)\s+\|\s+(.+)\s+\|\s+(.+)\s+\|\s+(\d+)s\s+\|\s+(\d+)\s+\|\s+(\d+)\s+\|\s+(\d+)\s+\|\s+(\d+)\s+\|',
+                line)
+            if m:
+                groups = m.groups()
+                if session:
+                    session['counters'] = session_counters
+                    sessions.append(session)
+                session = {
+                    'id': groups[0],
+                    'ip6': groups[1],
+                    'ip4': groups[2],
+                    'expiry': int(groups[3]),
+                }
+                session_counters = {}
+                session_counters['total'] = {
+                    '4to6': {
+                        'packets': int(groups[4]),
+                        'bytes': int(groups[5]),
+                    },
+                    '6to4': {
+                        'packets': int(groups[6]),
+                        'bytes': int(groups[7]),
+                    },
+                }
+                continue
+            if not session:
+                continue
+            m = re.match(r'\|\s+\|\s+(.+)\s+\|\s+(\d+)\s+\|\s+(\d+)\s+\|\s+(\d+)\s+\|\s+(\d+)\s+\|', line)
+            if m:
+                groups = m.groups()
+                session_counters[groups[0]] = {
+                    '4to6': {
+                        'packets': int(groups[1]),
+                        'bytes': int(groups[2]),
+                    },
+                    '6to4': {
+                        'packets': int(groups[3]),
+                        'bytes': int(groups[4]),
+                    },
+                }
+        if session:
+            session['counters'] = session_counters
+            sessions.append(session)
+        return sessions
+
+    def get_nat64_counters(self):
+        cmd = 'nat64 counters'
+        self.send_command(cmd)
+        result = self._expect_command_output()
+
+        protocol_counters = {}
+        error_counters = {}
+        for line in result:
+            m = re.match(r'\|\s+(.+)\s+\|\s+(\d+)\s+\|\s+(\d+)\s+\|\s+(\d+)\s+\|\s+(\d+)\s+\|', line)
+            if m:
+                groups = m.groups()
+                protocol_counters[groups[0]] = {
+                    '4to6': {
+                        'packets': int(groups[1]),
+                        'bytes': int(groups[2]),
+                    },
+                    '6to4': {
+                        'packets': int(groups[3]),
+                        'bytes': int(groups[4]),
+                    },
+                }
+                continue
+            m = re.match(r'\|\s+(.+)\s+\|\s+(\d+)\s+\|\s+(\d+)\s+\|', line)
+            if m:
+                groups = m.groups()
+                error_counters[groups[0]] = {
+                    '4to6': {
+                        'packets': int(groups[1]),
+                    },
+                    '6to4': {
+                        'packets': int(groups[2]),
+                    },
+                }
+                continue
+        return {'protocol': protocol_counters, 'errors': error_counters}
+
     def get_netdata_nat64_prefix(self):
         prefixes = []
         routes = self.get_routes()