cgnat: T9473: reject external addresses used in more than one range

Ports are allocated per external range, starting over for each one. An
address listed in two ranges of one pool, or in the pools of two rules
with overlapping port ranges, was therefore handed out twice: two
subscribers got the same external address and port block. The capacity
check also counted such an address twice.

Such overlaps are now rejected. Pools of different rules may still share
an address when their port ranges are distinct.

1 files changed, 71 insertions, 0 deletions

diff --git a/src/conf_mode/nat_cgnat.py b/src/conf_mode/nat_cgnat.py
index ef1b4259a..5d2f0143c 100755
--- a/src/conf_mode/nat_cgnat.py
+++ b/src/conf_mode/nat_cgnat.py
@@ -145,6 +145,51 @@ class IPOperations:
         # Return the summarized IP networks list
         return list(ipaddress.summarize_address_range(range_start, range_end))
 
+    def get_bounds(self) -> tuple[int, int]:
+        """Return the first and last address of a prefix or range as integers
+
+        Example:
+            % ip = IPOperations('192.0.2.0/30')
+            % ip.get_bounds()
+            (3221225984, 3221225987)
+        """
+        if self.ip_network:
+            return (
+                int(self.ip_network.network_address),
+                int(self.ip_network.broadcast_address),
+            )
+        start_ip, end_ip = self.ip_prefix.split('-')
+        return int(ipaddress.ip_address(start_ip)), int(ipaddress.ip_address(end_ip))
+
+
+def _find_range_overlap(ranges: list[tuple[str, str]]) -> tuple | None:
+    """Return the first two overlapping (owner, range) pairs or None"""
+    bounds = sorted(
+        (IPOperations(ip_range).get_bounds(), owner, ip_range)
+        for owner, ip_range in ranges
+    )
+    # compare against the range reaching furthest so far, not just the
+    # previous one - it may contain several of the following ranges
+    widest = None
+    widest_end = -1
+    for (start, end), owner, ip_range in bounds:
+        if start <= widest_end:
+            return widest, (owner, ip_range)
+        if end > widest_end:
+            widest = (owner, ip_range)
+            widest_end = end
+    return None
+
+
+def _verify_range_overlap(ranges: list[tuple[str, str]], reason: str = '') -> None:
+    overlap = _find_range_overlap(ranges)
+    if overlap:
+        (owner_a, range_a), (owner_b, range_b) = overlap
+        raise ConfigError(
+            f'Range "{range_a}" of "{owner_a}" overlaps with range '
+            f'"{range_b}" of "{owner_b}"{reason}!'
+        )
+
 
 def _delete_conntrack_entries(source_prefixes: list[ipaddress.IPv4Network]) -> None:
     """Delete all conntrack entries for the list of prefixes"""
@@ -357,6 +402,32 @@ def verify(config):
                 f'specified parameters'
             )
 
+    # Ports are allocated per range, an address in two ranges of a pool is
+    # handed out twice. Pools may share an address on distinct port ranges.
+    external_ranges = {
+        pool_name: [
+            (f'external pool {pool_name}', ip_range)
+            for ip_range in config['pool']['external'][pool_name]['range']
+        ]
+        for pool_name in used_external_pools
+    }
+    for ranges in external_ranges.values():
+        _verify_range_overlap(ranges)
+
+    port_ranges = {}
+    for pool_name in used_external_pools:
+        port_range = config['pool']['external'][pool_name]['external_port_range']
+        port_ranges[pool_name] = tuple(map(int, port_range.split('-')))
+    pool_names = list(external_ranges)
+    for index, pool_a in enumerate(pool_names):
+        for pool_b in pool_names[index + 1:]:
+            (start_a, end_a), (start_b, end_b) = port_ranges[pool_a], port_ranges[pool_b]
+            if start_a <= end_b and start_b <= end_a:
+                _verify_range_overlap(
+                    external_ranges[pool_a] + external_ranges[pool_b],
+                    ' and their port ranges overlap',
+                )
+
 
 def generate(config):
     if 'deleted' in config: