fix(gateway): require http(s) URLs before urlopen

urllib.request.urlopen also opens file: and ftp: URLs, and the gateway
base URL and pairing targets are not guaranteed to be http(s). Add
require_http_url() to gateway_embed/urls.py and call it before each
urlopen. Health probes turn a rejected scheme into the existing soft
error; pairing raises ValueError.

The urlopen calls themselves stay flagged by the dynamic-urllib rule,
so each carries a one-line nosemgrep for that rule that points at the
check.
This commit is contained in:
2026-11-06 19:29:21 +05:30
parent 73d33a7228
commit a9a1900ee4
3 changed files with 18 additions and 3 deletions
+4 -1
View File
@@ -7,6 +7,8 @@ import urllib.error
import urllib.request
from dataclasses import dataclass
from .urls import require_http_url
logger = logging.getLogger(__name__)
DEFAULT_BASE_URL = "http://127.0.0.1:8765"
@@ -23,8 +25,9 @@ class HealthSnapshot:
def _get_status(url: str, *, timeout: float) -> tuple[int | None, str]:
request = urllib.request.Request(url, method="GET")
try:
request = urllib.request.Request(require_http_url(url), method="GET")
# nosemgrep: python.lang.security.audit.dynamic-urllib-use-detected.dynamic-urllib-use-detected -- scheme checked by require_http_url
with urllib.request.urlopen(request, timeout=timeout) as response:
return int(response.status), ""
except urllib.error.HTTPError as exc:
+3 -2
View File
@@ -10,7 +10,7 @@ from dataclasses import dataclass
from typing import Any, Optional
from ..common_types import Readable
from .urls import is_unusable_phone_url, reject_loopback_url
from .urls import is_unusable_phone_url, reject_loopback_url, require_http_url
logger = logging.getLogger(__name__)
@@ -61,7 +61,8 @@ def _authorized_get(
headers = {}
if token:
headers["Authorization"] = f"Bearer {token}"
request = urllib.request.Request(url, headers=headers, method="GET")
request = urllib.request.Request(require_http_url(url), headers=headers, method="GET")
# nosemgrep: python.lang.security.audit.dynamic-urllib-use-detected.dynamic-urllib-use-detected -- scheme checked by require_http_url
with urllib.request.urlopen(request, timeout=timeout) as response:
return _read_capped(response, limit)
+11
View File
@@ -182,3 +182,14 @@ def reject_loopback_url(url: str | None) -> str | None:
if is_unusable_phone_url(url):
return None
return str(url).strip()
def require_http_url(url: str) -> str:
"""Return *url* when it is an http(s) URL, else raise ``ValueError``.
``urllib.request.urlopen`` also opens ``file:`` and ``ftp:`` URLs, so every
gateway request passes its URL through here before it is opened.
"""
if urlparse(url).scheme not in ("http", "https"):
raise ValueError(f"gateway URL must use http or https: {url!r}")
return url