New: OsOperations::is_port_available(ip: str, number: int)

.github/workflows/ci.yml

@@ -117,7 +117,7 @@ jobs:
     - name: Build local image ${{ matrix.alpine }}
       run: docker build --build-arg PYTHON_VERSION="${{ matrix.python }}" -t "${{ env.RUN_CFG__DOCKER_IMAGE_NAME }}" -f Dockerfile--${{ matrix.platform }}.tmpl .
     - name: Run
-      run: docker run -t -v ${{ github.workspace }}/${{ env.RUN_CFG__LOGS_DIR }}:/home/test/testgres/logs "${{ env.RUN_CFG__DOCKER_IMAGE_NAME }}"
+      run: docker run -t --cap-add=NET_ADMIN -v ${{ github.workspace }}/${{ env.RUN_CFG__LOGS_DIR }}:/home/test/testgres/logs "${{ env.RUN_CFG__DOCKER_IMAGE_NAME }}"
     - name: Upload Logs
       uses: actions/upload-artifact@v7
       if: always() # IT IS IMPORTANT!

Dockerfile--astralinux_1_7.tmpl

@@ -12,6 +12,9 @@ RUN apt install -y openssh-server
 
 RUN apt install -y git
 
+# this installs "ip" utility
+RUN apt install -y iproute2
+
 # RUN apt install -y mc
 # RUN apt install -y nano
 

Dockerfile--ubuntu_24_04.tmpl

@@ -12,6 +12,9 @@ RUN apt install -y netcat-traditional
 
 RUN apt install -y git
 
+# this installs "ip" utility
+RUN apt install -y iproute2
+
 # RUN apt install -y mc
 
 # --------------------------------------------- base2_with_python-3

run_tests3.sh

@@ -2,6 +2,10 @@
 
 set -eux
 
+for i in {2..11}; do
+    sudo ip addr add 127.0.0.$i/32 dev lo
+done
+
 # prepare python environment
 VENV_PATH="/tmp/testgres_venv"
 rm -rf $VENV_PATH

src/local_ops.py

@@ -16,6 +16,7 @@
 import threading
 import copy
 import signal as os_signal
+import ipaddress
 
 from .exceptions import ExecUtilException
 from .exceptions import InvalidOperationException
@@ -599,6 +600,52 @@ def is_port_free(self, number: int) -> bool:
             except OSError:
                 return False
 
+    def is_port_available(self, ip: str, number: int) -> bool:
+        assert type(ip) is str
+        assert ip != ""
+        assert type(number) is int
+        assert number >= 0
+        assert number <= 65535  # OK?
+
+        try:
+            addr = ipaddress.ip_address(ip)
+            if addr.version == 4:
+                return __class__._is_port_available_vX(
+                    ip,
+                    number,
+                    socket.AF_INET,
+                )
+            if addr.version == 6:
+                return self._is_port_available_vX(
+                    ip,
+                    number,
+                    socket.AF_INET6,
+                )
+        except ValueError:
+            raise RuntimeError("Unknown format of IP: {!r}".format(ip))
+
+        raise RuntimeError("Unsupported IP version: {!r}".format(ip))
+
+    @staticmethod
+    def _is_port_available_vX(
+        ip: str,
+        number: int,
+        address_family: socket.AddressFamily,
+    ) -> bool:
+        assert type(ip) is str
+        assert ip != ""
+        assert type(number) is int
+        assert number >= 0
+        assert number <= 65535  # OK?
+
+        with socket.socket(address_family, socket.SOCK_STREAM) as s:
+            s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+            try:
+                s.bind((ip, number))
+                return True
+            except OSError:
+                return False
+
     def get_tempdir(self) -> str:
         r = tempfile.gettempdir()
         assert r is not None

src/os_ops.py

@@ -148,5 +148,13 @@ def is_port_free(self, number: int):
         assert type(number) is int
         raise NotImplementedError()
 
+    def is_port_available(self, ip: str, number: int) -> bool:
+        assert type(ip) is str
+        assert ip != ""
+        assert type(number) is int
+        assert number >= 0
+        assert number <= 65535  # OK?
+        raise NotImplementedError()
+
     def get_tempdir(self) -> str:
         raise NotImplementedError()

src/remote_ops.py

@@ -11,6 +11,8 @@
 import copy
 import re
 import signal as os_signal
+import struct
+import ipaddress
 
 from .exceptions import ExecUtilException
 from .exceptions import InvalidOperationException
@@ -734,6 +736,112 @@ def is_port_free(self, number: int) -> bool:
             out=output
         )
 
+    def is_port_available(self, ip: str, number: int) -> bool:
+        assert type(ip) is str
+        assert ip != ""
+        assert type(number) is int
+        assert number >= 0
+        assert number <= 65535  # OK?
+
+        try:
+            addr = ipaddress.ip_address(ip)
+            if addr.version == 4:
+                return self._is_port_available_v4(
+                    addr,
+                    number,
+                )
+            if addr.version == 6:
+                return self._is_port_available_v6(
+                    addr,
+                    number,
+                )
+        except ValueError:
+            raise RuntimeError("Unknown format of IP: {!r}".format(ip))
+
+        raise RuntimeError("Unsupported IP version: {!r}".format(ip))
+
+    # --------------------------------------------------------------------
+    def _is_port_available_v4(self, addr: ipaddress.IPv4Address, number: int) -> bool:
+        assert type(addr) is ipaddress.IPv4Address
+        assert type(number) is int
+        assert number >= 0
+        assert number <= 65535  # OK?
+
+        ip_packed = addr.packed
+
+        # 1. The IP address really depends on the architecture (we only flip it)
+        ip_le = format(struct.unpack("I", ip_packed)[0], "08X")
+        ip_be = format(struct.unpack("!I", ip_packed)[0], "08X")
+
+        # 2. The port is ALWAYS output in Big Endian (just convert the number to HEX)
+        port_hex = format(number, "04X")
+
+        # 3. Build a command
+        # Byte 0: 0x7F
+        # Byte 1: 'E'
+        # Byte 2: 'L'
+        # Byte 3: 'F'
+        # Byte 4: Bit class (01 — 32-bit, 02 — 64-bit)
+        # Byte 5: Byte order (01 — Little Endian, 02 — Big Endian)
+        grep_cmd_s = (
+            'if od -An -t x1 -N 1 -j 5 /bin/bash | grep -q "01"; then '
+            '  if grep -q -E "^\\s*[0-9]+:\\s*' + ip_le + ':' + port_hex + '\\s+" /proc/net/tcp; then echo "BUSY"; else echo "FREE"; fi; '
+            'else '
+            '  if grep -q -E "^\\s*[0-9]+:\\s*' + ip_be + ':' + port_hex + '\\s+" /proc/net/tcp; then echo "BUSY"; else echo "FREE"; fi; '
+            'fi'
+        )
+
+        return self._run_grep(grep_cmd_s)
+
+    # --------------------------------------------------------------------
+    def _is_port_available_v6(self, addr: ipaddress.IPv6Address, number: int) -> bool:
+        assert type(addr) is ipaddress.IPv6Address
+        assert type(number) is int
+        assert number >= 0
+        assert number <= 65535  # OK?
+
+        ip_bytes = addr.packed
+        words = struct.unpack("!IIII", ip_bytes)
+
+        # 1. The IP address really depends on the architecture (we only flip it)
+        ip_le = "".join(format(struct.unpack("<I", struct.pack(">I", w))[0], "08X") for w in words)
+        ip_be = "".join(format(w, "08X") for w in words)
+
+        # 2. The port is ALWAYS output in Big Endian (just convert the number to HEX)
+        port_hex = format(number, "04X")
+
+        # 3. Bash script checks architecture: if 5th byte of /bin/bash is 1, then it is Little Endian
+        grep_cmd_s = (
+            'if od -An -t x1 -N 1 -j 5 /bin/bash | grep -q "01"; then '
+            '  if grep -q -E "^\\s*[0-9]+:\\s*' + ip_le + ':' + port_hex + '\\s+" /proc/net/tcp6; then echo "BUSY"; else echo "FREE"; fi; '
+            'else '
+            '  if grep -q -E "^\\s*[0-9]+:\\s*' + ip_be + ':' + port_hex + '\\s+" /proc/net/tcp6; then echo "BUSY"; else echo "FREE"; fi; '
+            'fi'
+        )
+
+        return self._run_grep(grep_cmd_s)
+
+    # --------------------------------------------------------------------
+    def _run_grep(self, grep_cmd_s: str) -> bool:
+        assert type(grep_cmd_s) is str
+
+        cmd = ["/bin/bash", "-c", grep_cmd_s]
+
+        output = self.exec_command(
+            cmd=cmd,
+            encoding=get_default_encoding(),
+        )
+
+        if output == "BUSY\n":
+            return False
+
+        if output == "FREE\n":
+            return True
+
+        errMsg = "grep returned unexpected output: {!r}".format(output)
+        raise RuntimeError(errMsg)
+
+    # --------------------------------------------------------------------
     def get_tempdir(self) -> str:
         command = ["mktemp", "-u", "-d"]