Fix: Protect against wrong IPv4 format from 3rd party service.

 On some servers, get_IP() returned chinese chars instead if a proper IPv4. This protects against invalid IP formats.

src/aleph/services/utils.py

@@ -1,18 +1,48 @@
+import logging
+import re
 import socket
 
 import aiohttp
 
+logger = logging.getLogger(__name__)
 
-async def get_IP():
-    ip = None
+IP4_SERVICE_URL = "https://v4.ident.me/"
+IP4_SOCKET_ENDPOINT = "8.8.8.8"
+
+
+def is_valid_ip4(ip: str) -> bool:
+    return bool(re.match(r"\d+\.\d+\.\d+\.\d+", ip))
+
+
+async def get_ip4_from_service() -> str:
+    """Get the public IPv4 of this system by calling a third-party service"""
+    async with aiohttp.ClientSession() as session:
+        async with session.get(IP4_SERVICE_URL) as resp:
+            resp.raise_for_status()
+            ip = await resp.text()
+
+            if is_valid_ip4(ip):
+                return ip
+            else:
+                raise ValueError(f"Response does not match IPv4 format: {ip}")
+
+
+def get_ip4_from_socket() -> str:
+    """Get the public IPv4 of this system by inspecting a socket connection.
+    Warning: This returns a local IP address when running behind a NAT, e.g. on Docker.
+    """
+    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
     try:
-        async with aiohttp.ClientSession() as session:
-            async with session.get("https://v4.ident.me/") as resp:
-                ip = await resp.text()
-    except:
-        s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
-        s.connect(("8.8.8.8", 80))
-        ip = s.getsockname()[0]
+        s.connect((IP4_SOCKET_ENDPOINT, 80))
+        return s.getsockname()[0]
+    finally:
         s.close()
 
-    return ip
+
+async def get_IP() -> str:
+    """Get the public IPv4 of this system."""
+    try:
+        return await get_ip4_from_service()
+    except Exception as error:
+        logging.exception("Error when fetching IPv4 from service")
+        return get_ip4_from_socket()

tests/services/test_utils.py

@@ -0,0 +1,33 @@
+import pytest
+
+from aleph.services.utils import (
+    is_valid_ip4,
+    get_ip4_from_service,
+    get_ip4_from_socket,
+    get_IP,
+)
+
+
+def test_is_valid_ip4():
+    assert is_valid_ip4("1.2.3.4")
+    assert is_valid_ip4("123.456.789.123")
+    assert not is_valid_ip4("")
+    assert not is_valid_ip4("Hello !")
+    assert not is_valid_ip4("a.b.c.d")
+
+
+@pytest.mark.asyncio
+async def test_get_ip4_from_service():
+    ip4 = await get_ip4_from_service()
+    assert is_valid_ip4(ip4)
+
+
+def test_get_ip4_from_socket():
+    ip4 = get_ip4_from_socket()
+    assert is_valid_ip4(ip4)
+
+
+@pytest.mark.asyncio
+async def test_get_IP():
+    ip4 = await get_IP()
+    assert is_valid_ip4(ip4)