fix(tls): replace raw urllib with requests for external HTTPS calls (frozen-build cert failures) (#903)

Co-authored-by: Jonathan Singer <jonathansinger@Mac-4051.lan>
Co-authored-by: Ahmed Allam <ahmed39652003@gmail.com>

strix/config/codex.py

@@ -18,12 +18,12 @@ import logging
 import secrets
 import threading
 import time
-import urllib.error
 import urllib.parse
-import urllib.request
 from pathlib import Path
 from typing import TYPE_CHECKING, Any
 
+import requests
+
 
 if TYPE_CHECKING:
     from collections.abc import Iterator
@@ -221,26 +221,19 @@ def _first(query: dict[str, list[str]], key: str) -> str | None:
 
 
 def _post_form(payload: dict[str, str]) -> dict[str, Any]:
-    body = urllib.parse.urlencode(payload).encode("ascii")
-    request = urllib.request.Request(  # noqa: S310 - fixed https OAuth endpoint
-        TOKEN_URL,
-        data=body,
-        headers={
-            "Content-Type": "application/x-www-form-urlencoded",
-            "Accept": "application/json",
-        },
-        method="POST",
-    )
     try:
-        with urllib.request.urlopen(  # noqa: S310  # nosec B310 - fixed https endpoint
-            request, timeout=_TOKEN_TIMEOUT
-        ) as response:
-            data = json.loads(response.read() or b"{}")
-    except urllib.error.HTTPError as exc:
-        detail = exc.read().decode("utf-8", "replace")[:300]
-        raise CodexAuthError("token_http_error", f"HTTP {exc.code}: {detail}") from exc
-    except (urllib.error.URLError, TimeoutError, OSError) as exc:
+        response = requests.post(
+            TOKEN_URL,
+            data=payload,
+            headers={"Accept": "application/json"},
+            timeout=_TOKEN_TIMEOUT,
+        )
+    except requests.RequestException as exc:
         raise CodexAuthError("unavailable", str(exc)) from exc
+    if response.status_code >= 400:
+        detail = response.text[:300]
+        raise CodexAuthError("token_http_error", f"HTTP {response.status_code}: {detail}")
+    data = json.loads(response.content or b"{}")
     if not isinstance(data, dict):
         raise CodexAuthError("bad_response", "token endpoint returned non-object")
     return data

strix/interface/utils.py

@@ -11,11 +11,10 @@ import tempfile
 from dataclasses import dataclass, field
 from pathlib import Path
 from typing import Any
-from urllib.error import HTTPError, URLError
 from urllib.parse import urlparse
-from urllib.request import Request, urlopen
 
 import docker
+import requests
 from docker.errors import DockerException, ImageNotFound
 from rich.console import Console
 from rich.panel import Panel
@@ -1088,13 +1087,12 @@ def resolve_diff_scope_context(
 def _is_http_git_repo(url: str) -> bool:
     check_url = f"{url.rstrip('/')}/info/refs?service=git-upload-pack"
     try:
-        req = Request(check_url, headers={"User-Agent": "git/strix"})  # noqa: S310
-        with urlopen(req, timeout=10) as resp:  # noqa: S310  # nosec B310
-            return "x-git-upload-pack-advertisement" in resp.headers.get("Content-Type", "")
-    except HTTPError as e:
-        return e.code == 401
-    except (URLError, OSError, ValueError):
+        resp = requests.get(check_url, headers={"User-Agent": "git/strix"}, timeout=10)
+    except (requests.RequestException, ValueError):
         return False
+    if resp.status_code >= 400:
+        return resp.status_code == 401
+    return "x-git-upload-pack-advertisement" in resp.headers.get("Content-Type", "")
 
 
 def infer_target_type(target: str) -> tuple[str, dict[str, str]]:  # noqa: PLR0911

strix/interface/viewer/auth.py

@@ -15,12 +15,12 @@ import base64
 import contextlib
 import json
 import logging
-import urllib.error
-import urllib.request
 from datetime import UTC, datetime
 from pathlib import Path
 from typing import Any
 
+import requests
+
 from strix.config.loader import load_settings
 
 
@@ -147,21 +147,17 @@ def _post_json(path: str, payload: dict[str, Any], *, timeout: int) -> tuple[int
     map, not raised.
     """
     url = f"{_app_url()}{path}"
-    body = json.dumps(payload).encode("utf-8")
-    request = urllib.request.Request(  # noqa: S310 - fixed https relay URL
-        url,
-        data=body,
-        headers={"Content-Type": "application/json", "Accept": "application/json"},
-        method="POST",
-    )
     try:
-        with urllib.request.urlopen(request, timeout=timeout) as response:  # noqa: S310  # nosec B310
-            return response.status, _parse_body(response.read())
-    except urllib.error.HTTPError as exc:
-        return exc.code, _parse_body(exc.read())
-    except (urllib.error.URLError, TimeoutError, OSError) as exc:
+        response = requests.post(
+            url,
+            json=payload,
+            headers={"Accept": "application/json"},
+            timeout=timeout,
+        )
+    except requests.RequestException as exc:
         logger.warning("relay request to %s failed: %s", path, exc)
         raise RelayError("unavailable") from exc
+    return response.status_code, _parse_body(response.content)
 
 
 def _parse_body(raw: bytes) -> dict[str, Any]:

strix/telemetry/posthog.py

@@ -1,9 +1,9 @@
-import json
 import logging
-import urllib.request
 from datetime import datetime
 from typing import TYPE_CHECKING, Any
 
+import requests
+
 from strix.config import load_settings
 from strix.telemetry._common import (
     SESSION_ID,
@@ -37,13 +37,7 @@ def _send(event: str, properties: dict[str, Any]) -> bool:
             "distinct_id": SESSION_ID,
             "properties": properties,
         }
-        req = urllib.request.Request(  # noqa: S310
-            f"{_POSTHOG_HOST}/capture/",
-            data=json.dumps(payload).encode(),
-            headers={"Content-Type": "application/json"},
-        )
-        with urllib.request.urlopen(req, timeout=10):  # noqa: S310  # nosec B310
-            pass
+        requests.post(f"{_POSTHOG_HOST}/capture/", json=payload, timeout=10)
     except Exception:  # noqa: BLE001
         logger.debug("posthog send failed for event %s", event, exc_info=True)
         return False

strix/telemetry/scarf.py

@@ -2,10 +2,11 @@ from __future__ import annotations
 
 import logging
 import urllib.parse
-import urllib.request
 from datetime import datetime
 from typing import TYPE_CHECKING, Any
 
+import requests
+
 from strix.config import load_settings
 from strix.telemetry._common import (
     SESSION_ID,
@@ -42,9 +43,7 @@ def _send(event: str, properties: dict[str, Any]) -> bool:
         url = f"{_SCARF_ENDPOINT}{path}"
         if query:
             url = f"{url}?{query}"
-        req = urllib.request.Request(url, method="POST")  # noqa: S310
-        with urllib.request.urlopen(req, timeout=10):  # noqa: S310  # nosec B310
-            pass
+        requests.post(url, timeout=10)
     except Exception:  # noqa: BLE001
         logger.debug("scarf send failed for event %s", event, exc_info=True)
         return False

tests/test_codex_auth.py

@@ -7,8 +7,10 @@ import hashlib
 import json
 import time
 from typing import TYPE_CHECKING, Any
+from unittest import mock
 
 import pytest
+import requests
 
 from strix.config import codex
 
@@ -52,6 +54,18 @@ def test_authorize_url_carries_pkce_and_client() -> None:
     assert "state=st8" in url
 
 
+def test_post_form_returns_parsed_body() -> None:
+    resp = mock.MagicMock()
+    resp.status_code = 200
+    resp.content = b'{"access_token": "tok"}'
+
+    with mock.patch.object(requests, "post", return_value=resp) as post:
+        data = codex._post_form({"grant_type": "refresh_token"})
+
+    assert data == {"access_token": "tok"}
+    assert post.call_args.kwargs["timeout"] == codex._TOKEN_TIMEOUT
+
+
 @pytest.mark.parametrize(
     ("value", "expected"),
     [