Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
35 changes: 35 additions & 0 deletions src/infrastructure/external/ai_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@
AI 客户端封装
提供统一的 AI 调用接口
"""
import ipaddress
import os
import json
import base64
Expand Down Expand Up @@ -33,6 +34,38 @@
)


def _sanitize_no_proxy_env() -> None:
"""Strip CIDR prefix lengths from IPv6 entries in NO_PROXY / no_proxy.

httpx <= 0.28.1 wraps NO_PROXY IPv6 entries in brackets *including* the
CIDR mask (e.g. ``[::1/128]``), which the URL parser rejects as an invalid
port. Stripping the ``/prefix`` part is safe because httpx doesn't
support CIDR range matching anyway — it only does exact-host comparison.

See https://github.com/encode/httpx/pull/3741
"""
for key in ("NO_PROXY", "no_proxy"):
value = os.environ.get(key)
if not value:
continue
parts = [h.strip() for h in value.split(",")]
cleaned: list[str] = []
changed = False
for part in parts:
if "/" in part:
host, _, prefix = part.partition("/")
try:
ipaddress.IPv6Address(host)
cleaned.append(host)
changed = True
continue
except ValueError:
pass
cleaned.append(part)
Comment on lines +55 to +64
Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

Optimization: Adding a check for : before attempting to parse the host as an IPv6 address can avoid the overhead of the try/except block for IPv4 CIDRs (e.g., 127.0.0.0/8) and other non-IPv6 entries in NO_PROXY. Additionally, the prefix variable is unused and can be replaced with _ to follow Python conventions.

Suggested change
if "/" in part:
host, _, prefix = part.partition("/")
try:
ipaddress.IPv6Address(host)
cleaned.append(host)
changed = True
continue
except ValueError:
pass
cleaned.append(part)
if "/" in part and ":" in part:
host, _, _ = part.partition("/")
try:
ipaddress.IPv6Address(host)
cleaned.append(host)
changed = True
continue
except ValueError:
pass
cleaned.append(part)

if changed:
os.environ[key] = ",".join(cleaned)


class AIClient:
"""AI 客户端封装"""

Expand Down Expand Up @@ -61,6 +94,8 @@ def _initialize_client(self) -> Optional[AsyncOpenAI]:
os.environ['HTTP_PROXY'] = self.settings.proxy_url
os.environ['HTTPS_PROXY'] = self.settings.proxy_url

_sanitize_no_proxy_env()

return AsyncOpenAI(
api_key=self.settings.api_key,
base_url=self.settings.base_url
Expand Down
38 changes: 37 additions & 1 deletion tests/unit/test_ai_client.py
Original file line number Diff line number Diff line change
@@ -1,9 +1,10 @@
import asyncio
import os
from types import SimpleNamespace

import pytest

from src.infrastructure.external.ai_client import AIClient
from src.infrastructure.external.ai_client import AIClient, _sanitize_no_proxy_env
from src.services.ai_request_compat import build_responses_input


Expand Down Expand Up @@ -244,3 +245,38 @@ def test_parse_response_uses_first_json_object_when_response_contains_multiple_o
```""")

assert result == {"ok": True, "reason": "first"}


# -- _sanitize_no_proxy_env tests --


def test_sanitize_no_proxy_strips_ipv6_cidr(monkeypatch):
monkeypatch.setenv("NO_PROXY", "localhost,127.0.0.0/8,::1/128")
_sanitize_no_proxy_env()
assert os.environ["NO_PROXY"] == "localhost,127.0.0.0/8,::1"


def test_sanitize_no_proxy_strips_lowercase_variant(monkeypatch):
monkeypatch.setenv("no_proxy", "localhost,::1/128,fe80::1/64")
_sanitize_no_proxy_env()
assert os.environ["no_proxy"] == "localhost,::1,fe80::1"


def test_sanitize_no_proxy_preserves_ipv4_cidr(monkeypatch):
monkeypatch.setenv("NO_PROXY", "10.0.0.0/8,192.168.0.0/16")
_sanitize_no_proxy_env()
assert os.environ["NO_PROXY"] == "10.0.0.0/8,192.168.0.0/16"


def test_sanitize_no_proxy_noop_without_env(monkeypatch):
monkeypatch.delenv("NO_PROXY", raising=False)
monkeypatch.delenv("no_proxy", raising=False)
_sanitize_no_proxy_env()


def test_sanitize_no_proxy_handles_both_keys(monkeypatch):
monkeypatch.setenv("NO_PROXY", "::1/128")
monkeypatch.setenv("no_proxy", "fe80::1/10")
_sanitize_no_proxy_env()
assert os.environ["NO_PROXY"] == "::1"
assert os.environ["no_proxy"] == "fe80::1"