Security hardening (maziggy/bambuddy-security #21)

.basedpyright/baseline.json

@@ -2745,30 +2745,6 @@
             }
         ],
         "./backend/app/services/external_camera.py": [
-            {
-                "code": "reportOptionalMemberAccess",
-                "range": {
-                    "startColumn": 10,
-                    "endColumn": 21,
-                    "lineCount": 1
-                }
-            },
-            {
-                "code": "reportOptionalMemberAccess",
-                "range": {
-                    "startColumn": 10,
-                    "endColumn": 24,
-                    "lineCount": 1
-                }
-            },
-            {
-                "code": "reportOptionalMemberAccess",
-                "range": {
-                    "startColumn": 10,
-                    "endColumn": 23,
-                    "lineCount": 1
-                }
-            },
             {
                 "code": "reportOptionalSubscript",
                 "range": {

backend/app/api/routes/_oidc_helpers.py

@@ -16,8 +16,9 @@ from urllib.parse import urlparse
 from backend.app.api.routes._url_safety import (
     CLOUD_METADATA_HOSTNAMES,
     CLOUD_METADATA_IPS,
-    NUMERIC_IP_RE,
+    is_numeric_encoded_ip,
     unwrap_ipv4_mapped,
+    url_hostname,
 )
 
 
@@ -54,17 +55,22 @@ def assert_safe_public_https_url(url: str) -> None:
     if parsed.scheme.lower() != "https":
         raise ValueError("icon URL must use https://")
 
-    hostname = (parsed.hostname or "").lower()
+    hostname = url_hostname(parsed)
 
     # "https:///path" parses to an empty hostname; without this it reaches the
     # ip_address() ValueError branch and is accepted as a symbolic hostname.
     if not hostname:
         raise ValueError("icon URL must include a hostname")
 
+    # HTTP clients map other dot characters and fullwidth forms to ASCII
+    # before they connect: a host the checks below cannot read is refused.
+    if not hostname.isascii():
+        raise ValueError("icon URL must use an ASCII host name (punycode, xn--..., for international names)")
+
     if hostname in CLOUD_METADATA_HOSTNAMES:
         raise ValueError("icon URL must not point to a cloud metadata endpoint")
 
-    if NUMERIC_IP_RE.match(hostname):
+    if is_numeric_encoded_ip(hostname):
         raise ValueError("icon URL must not use numeric-encoded IP addresses")
 
     try:

backend/app/api/routes/_url_safety.py

@@ -25,7 +25,7 @@ from __future__ import annotations
 
 import ipaddress
 import re
-from urllib.parse import urlparse
+from urllib.parse import ParseResult, urlparse
 
 # Cloud-provider metadata endpoints — the classic SSRF credential-exfil
 # targets. Both guards reject these unconditionally.
@@ -53,17 +53,41 @@ CLOUD_METADATA_HOSTNAMES = frozenset(
 )
 
 
-# libc and browsers parse numeric-encoded IP forms (decimal ``2130706433``
-# for 127.0.0.1, hex ``0x7f000001``) but Python's ``ipaddress.ip_address``
-# raises ValueError on these, so they slip past the IP-class checks if
-# not caught first. Used by both guards to reject up-front.
-NUMERIC_IP_RE = re.compile(r"^(0x[0-9a-f]+|[0-9]+)$", re.I)
+# libc and browsers parse numeric-encoded IP forms (decimal ``2130706433`` for
+# 127.0.0.1, hex ``0x7f000001``, dotted hex or octal, short forms such as
+# ``127.1``) but Python's ``ipaddress`` raises ValueError on them, so they slip
+# past the IP-class checks if not caught first. Used by both guards to reject
+# them up front, see ``is_numeric_encoded_ip``. One label of such an address
+# is decimal, octal or hex.
+_NUMERIC_LABEL_RE = re.compile(r"^(0x[0-9a-f]*|[0-9]+)$", re.I)
+
+
+def url_hostname(parsed: ParseResult) -> str:
+    """The URL's hostname as resolvers see it: lower case, no trailing dot."""
+    return (parsed.hostname or "").lower().rstrip(".")
+
+
+def is_numeric_encoded_ip(hostname: str) -> bool:
+    """Whether *hostname* is an IPv4 address in a notation other than plain
+    dotted decimal: one number, dotted hex or octal, a leading zero, or fewer
+    than four parts. Resolvers read all of them as an address; the checks on
+    addresses only see standard notation.
+    """
+    labels = hostname.split(".")
+    if not all(_NUMERIC_LABEL_RE.match(label) for label in labels):
+        return False
+    try:
+        ipaddress.IPv4Address(hostname)
+    except ValueError:
+        return True
+    return False
 
 
 def unwrap_ipv4_mapped(
     addr: ipaddress.IPv4Address | ipaddress.IPv6Address,
 ) -> ipaddress.IPv4Address | ipaddress.IPv6Address:
-    """Return the underlying IPv4 for an IPv4-mapped IPv6 address, else return *addr*.
+    """Return the underlying IPv4 for an IPv4-mapped IPv6 address, *addr* without
+    its zone id, else *addr*.
 
     ``::ffff:127.0.0.1`` and similar mapped forms must be unwrapped before
     the per-class checks (``is_private``, ``is_loopback``, …) — otherwise
@@ -72,6 +96,10 @@ def unwrap_ipv4_mapped(
     """
     if isinstance(addr, ipaddress.IPv6Address) and addr.ipv4_mapped is not None:
         return addr.ipv4_mapped
+    # A zone id is ignored on connect to an address that is not link-local,
+    # but it makes the address compare unequal to the same one without it.
+    if isinstance(addr, ipaddress.IPv6Address) and addr.scope_id is not None:
+        return ipaddress.IPv6Address(str(addr).split("%", 1)[0])
     return addr
 
 
@@ -112,7 +140,7 @@ def assert_safe_lan_service_url(url: str, *, label: str) -> None:
     if parsed.scheme.lower() not in ("http", "https"):
         raise ValueError(f"{label} must use http or https")
 
-    hostname = (parsed.hostname or "").lower()
+    hostname = url_hostname(parsed)
 
     # "http:///path" parses to an empty hostname. Never a valid destination,
     # and without this it falls through the ip_address() ValueError branch
@@ -120,10 +148,16 @@ def assert_safe_lan_service_url(url: str, *, label: str) -> None:
     if not hostname:
         raise ValueError(f"{label} must include a hostname")
 
+    # HTTP clients map other dot characters, fullwidth digits and similar
+    # forms to plain ASCII before they connect, so a host the checks below
+    # cannot read is refused rather than let through as a name.
+    if not hostname.isascii():
+        raise ValueError(f"{label} must use an ASCII host name (punycode, xn--..., for international names)")
+
     if hostname in CLOUD_METADATA_HOSTNAMES:
         raise ValueError(f"{label} must not point to a cloud metadata endpoint")
 
-    if NUMERIC_IP_RE.match(hostname):
+    if is_numeric_encoded_ip(hostname):
         raise ValueError(f"{label} must not use numeric-encoded IP addresses; use standard dotted-decimal notation")
 
     try:

backend/app/services/external_camera.py

@@ -54,7 +54,16 @@ def _blocked_host_reason(hostname: str) -> str | None:
     accept them, so refusing to understand them here would only mean not seeing
     where the request is actually going.
     """
-    host = hostname.lower()
+    from backend.app.api.routes._url_safety import CLOUD_METADATA_HOSTNAMES, CLOUD_METADATA_IPS, unwrap_ipv4_mapped
+
+    # A trailing dot names the same host to every resolver.
+    host = hostname.lower().rstrip(".")
+
+    # The HTTP clients map other dot characters, fullwidth digits and similar
+    # forms to plain ASCII before they connect: a host not written in ASCII
+    # cannot be judged here, so it is refused.
+    if not host.isascii():
+        return "a host name not written in ASCII (use punycode, xn--...)"
 
     ip: ipaddress.IPv4Address | ipaddress.IPv6Address | None = None
     try:
@@ -73,15 +82,16 @@ def _blocked_host_reason(hostname: str) -> str | None:
         # intermittent local DNS.
         if host == "localhost" or host.endswith(".localhost"):
             return "localhost"
-        if host in ("metadata.google.internal", "metadata.google"):
+        if host in CLOUD_METADATA_HOSTNAMES or host == "metadata.google":
             return "a cloud metadata service"
         return None
 
-    # ::ffff:127.0.0.1 is loopback wearing an IPv6 spelling.
-    mapped = getattr(ip, "ipv4_mapped", None)
-    if mapped is not None:
-        ip = mapped
+    # ::ffff:127.0.0.1 is loopback wearing an IPv6 spelling, and a zone id
+    # only hides the address it is attached to.
+    ip = unwrap_ipv4_mapped(ip)
 
+    if ip in CLOUD_METADATA_IPS:
+        return "a cloud metadata service"
     if ip.is_loopback:
         return "loopback"
     if ip.is_unspecified:

backend/tests/unit/test_external_camera_ssrf.py

@@ -264,3 +264,32 @@ class TestUsbDevicePaths:
             path_cls.return_value.__str__.return_value = "/dev/video7"
             assert _safe_usb_device_path("/dev/video7") == "/dev/video7"
         path_cls.assert_called_once_with("/dev/video7")
+
+
+class TestTheSameHostsSpelledOtherwise:
+    """A trailing dot is the same host to every resolver, and the metadata
+    endpoints outside the link-local range are still metadata endpoints."""
+
+    @pytest.mark.parametrize(
+        "host",
+        [
+            "127.0.0.1.",
+            "169.254.169.254.",
+            "localhost.",
+            "metadata.google.internal.",
+            "metadata.goog",
+            "100.100.100.200",
+            "[fd00:ec2::254]",
+            "169\u3002254\u3002169\u3002254",
+            "\uff11\uff12\uff17.\uff10.\uff10.\uff11",
+            "\u24dcetadata.google.internal",
+            "[fd00:ec2::254%251]",
+            "[fd00:ec2::254%1]",
+        ],
+    )
+    def test_is_refused(self, host):
+        assert _sanitize_camera_url(f"http://{host}/snap.jpg", HTTP_SCHEMES) is None
+
+    @pytest.mark.parametrize("host", ["192.168.1.50.", "camera.lan.", "100.64.0.9", "xn--kamera-kche-dlb.lan"])
+    def test_a_lan_camera_still_passes(self, host):
+        assert _sanitize_camera_url(f"http://{host}/snap.jpg", HTTP_SCHEMES) is not None

backend/tests/unit/test_outbound_url_ssrf_guards.py

@@ -62,6 +62,32 @@ UNIVERSALLY_BLOCKED = [
     "http://metadata.google.internal/",
     "http://METADATA.GOOGLE.INTERNAL/computeMetadata/v1/",
     "http://metadata.goog/",
+    # Classic inet_aton notations other than a single number: libc and HTTP
+    # clients resolve them to the same addresses.
+    "http://0xa9.0xfe.0xa9.0xfe/",
+    "http://0xA9.254.169.254/",
+    "http://169.254.43518/",
+    "http://169.16689662/",
+    "http://0251.0376.0251.0376/",
+    "http://0169.254.169.254/",
+    "http://0x7f.1/",
+    # A trailing dot is the same host.
+    "http://169.254.169.254./",
+    "http://metadata.google.internal./",
+    "http://0x7f000001./",
+    # Other dot characters, fullwidth digits and enclosed letters: HTTP clients
+    # map them to the plain forms before connecting.
+    "http://169\u3002254\u3002169\u3002254/",
+    "http://169\uff0e254\uff0e169\uff0e254/",
+    "http://169\uff61254\uff61169\uff61254/",
+    "http://\uff11\uff16\uff19.\uff12\uff15\uff14.\uff11\uff16\uff19.\uff12\uff15\uff14/",
+    "http://\u24dcetadata.google.internal/",
+    "http://metadata\u3002google\u3002internal/",
+    "http://0xa9\u30020xfe\u30020xa9\u30020xfe/",
+    # A zone id on an address that is not link-local is ignored on connect.
+    "http://[fd00:ec2::254%251]/",
+    "http://[fd00:ec2::254%25eth0]/",
+    "http://[fd00:ec2::254%1]/",
 ]
 
 # The normal self-hosted topology — the LAN tier must permit all of these.
@@ -73,6 +99,12 @@ LAN_ALLOWED = [
     "http://172.16.4.9:8080",
     "https://ntfy.example.com/",
     "http://spoolman.lan:7912",
+    "http://3dprinter.lan:8080",
+    "http://node-01:8080",
+    "http://192.168.1.50.:8123",
+    "http://[fd12:3456::1]:8123",
+    "http://xn--drucker-kche-dlb.lan:8080",
+    "http://[fe80::1%25eth0]:8123",
 ]
 
 
@@ -452,6 +484,7 @@ def test_ha_guard_keeps_ipv6_literals_bracketed():
         "169.254.169.254",
         "100.100.100.200",
         "fd00:ec2::254",
+        "fd00:ec2::254%1",
         "0.0.0.0",  # nosec B104 — rejection fixture, not a bind address: the assertion below is that the guard refuses it
         "239.255.255.250",
     ],
@@ -674,3 +707,23 @@ def test_classification_lists_do_not_drift_from_the_routes():
     actual = _request_body_url_fields()
     stale = (GUARDED_BODY_URLS | NOT_A_FETCH_TARGET | KNOWN_UNGUARDED_NEEDS_SCHEME_AWARE_GUARD) - actual
     assert not stale, f"Classification entries no longer match any route: {sorted(stale)}"
+
+
+@pytest.mark.parametrize(
+    "host",
+    [
+        "0xa9.0xfe.0xa9.0xfe",
+        "169.254.43518",
+        "0251.0376.0251.0376",
+        "0169.254.169.254",
+        "0x7f.1",
+        "169.254.169.254.",
+        "metadata.google.internal.",
+        "127\u30020\u30020\u30021",
+        "169\uff0e254\uff0e169\uff0e254",
+    ],
+)
+def test_public_tier_rejects_other_ipv4_notations_over_https(host: str):
+    """Over https, so the scheme check cannot be what refuses them."""
+    with pytest.raises(ValueError):
+        assert_safe_public_https_url(f"https://{host}/icon.png")

backend/tests/unit/test_url_safety.py

@@ -11,7 +11,7 @@ import pytest
 
 from backend.app.api.routes._url_safety import (
     CLOUD_METADATA_IPS,
-    NUMERIC_IP_RE,
+    is_numeric_encoded_ip,
     unwrap_ipv4_mapped,
 )
 
@@ -37,10 +37,14 @@ def test_cloud_metadata_set_is_frozen():
         "0xFFFFFFFF",  # uppercase hex
         "0",
         "4294967295",  # max uint32
+        "0xa9.0xfe.0xa9.0xfe",  # dotted hex
+        "169.254.43518",  # short form
+        "0251.0376.0251.0376",  # dotted octal
+        "0169.254.169.254",  # a leading zero
     ],
 )
-def test_numeric_ip_re_matches_encoded_forms(candidate):
-    assert NUMERIC_IP_RE.match(candidate) is not None
+def test_numeric_encoded_forms_are_recognised(candidate):
+    assert is_numeric_encoded_ip(candidate) is True
 
 
 @pytest.mark.parametrize(
@@ -51,10 +55,12 @@ def test_numeric_ip_re_matches_encoded_forms(candidate):
         "spoolman.lan",
         "::1",
         "localhost",
+        "3dprinter.lan",
+        "node-01",
     ],
 )
-def test_numeric_ip_re_rejects_normal_forms(candidate):
-    assert NUMERIC_IP_RE.match(candidate) is None
+def test_normal_forms_are_not_numeric_encoded(candidate):
+    assert is_numeric_encoded_ip(candidate) is False
 
 
 def test_unwrap_ipv4_mapped_unwraps_mapped_address():
@@ -72,3 +78,11 @@ def test_unwrap_ipv4_mapped_passes_through_pure_ipv4():
 def test_unwrap_ipv4_mapped_passes_through_pure_ipv6():
     addr = ipaddress.ip_address("2001:db8::1")
     assert unwrap_ipv4_mapped(addr) is addr
+
+
+def test_unwrap_drops_a_zone_id():
+    """Ignored on connect to an address that is not link-local: the address
+    reached is the one without it."""
+    result = unwrap_ipv4_mapped(ipaddress.ip_address("fd00:ec2::254%251"))
+    assert result == ipaddress.ip_address("fd00:ec2::254")
+    assert result in CLOUD_METADATA_IPS