fix: support CLI CIDR proxy bypasses and explicit CA bundles

docs/cli.md

@@ -377,3 +377,23 @@ extore manage logout --all
 商品可以通过 CLI 移入回收站和恢复：`extore admin product delete --product PRODUCT_ID --yes`、`extore admin product restore --product PRODUCT_ID`。商品授权下使用 `extore manage product delete/restore`，需要明确批准的 `product.delete` 权限；已有授权不会自动扩权。列表的 `--view active|deleted|all|history` 默认显示未删除商品，`history` 还保留永久移出回收站的商品用于履约查询。删除只停售，原卡密、任务、领取链接和附件保留，聚合队列仍包括旧任务。完整示例见 [商品回收站](cli-owner.md#商品回收站)。
 
 清空回收站：`extore admin trash empty --yes`，平台管理员必须另加 `--shop SHOP_ID`；商品授权只可执行 `extore manage trash empty --product PRODUCT_ID --yes`。单商品使用 `product purge --product PRODUCT_ID --yes`。这些操作永久移出商品、不可恢复，需要独立的 `product.purge` 权限，保留原卡密、任务与交付。CLI 提交明确 ID 快照，空回收站不写入，超过 500 个商品拒绝隐式批量。详见 [永久移出回收站](cli-owner.md#永久移出回收站)。
+
+### 私有 CA 与 CIDR 绕过规则
+
+`--proxy-env` 的 `NO_PROXY` 支持 IPv4/IPv6 CIDR，例如 `10.0.0.0/8,2001:db8::/32`。
+CIDR 仅匹配 URL 中的 IP 字面量，不解析域名，避免将代理侧 DNS 泄露到本地。
+域名目标继续使用现有域名绕过规则；非法 CIDR 会在打开私密 profile 前报错。
+
+需要企业或运行环境 CA 时，显式传入 `--ca-bundle /path/to/ca.pem`，或设置
+`EXTORE_CA_BUNDLE`。此 PEM 文件中的 CA 仅添加到当前 CLI 的默认可信 CA，
+适用于目标 HTTPS 和 HTTPS 代理连接，不改变系统信任库，不保存到 profile。
+证书链和主机名校验保持开启；不提供跳过 TLS 校验的开关。
+`SSL_CERT_FILE`、`SSL_CERT_DIR` 不会自动生效。若环境提供了受信任的
+`SSL_CERT_FILE`，可明确选择它：
+
+```sh
+EXTORE_CA_BUNDLE="$SSL_CERT_FILE" extore manage --proxy-env queues
+```
+
+CA 文件无法读取或不是有效 PEM 时，CLI 在打开 profile 前返回
+`invalid_ca_bundle`，不回显文件路径或文件内容。

extore/http_proxy.py

@@ -4,9 +4,11 @@
 import ipaddress
 import logging
 import os
+import ssl
 from dataclasses import dataclass, field
 from urllib.parse import urlsplit
 
+import certifi
 import httpx
 
 
@@ -66,6 +68,11 @@ def add_proxy_arguments(parser):
         default=argparse.SUPPRESS,
         help="opt in to HTTP_PROXY, HTTPS_PROXY, ALL_PROXY and NO_PROXY",
     )
+    parser.add_argument(
+        "--ca-bundle",
+        default=argparse.SUPPRESS,
+        help="additional trusted PEM CAs for this CLI only (or EXTORE_CA_BUNDLE)",
+    )
     children = next(
         (
             action
@@ -87,6 +94,8 @@ class ProxySettings:
     mode: str = "direct"
     url: str | None = field(default=None, repr=False)
     routes: tuple = field(default=(), repr=False)
+    bypass_networks: tuple = field(default=(), repr=False)
+    tls_context: ssl.SSLContext | None = field(default=None, repr=False, compare=False)
 
 
 class _SocksAuthenticationFilter(logging.Filter):
@@ -111,20 +120,62 @@ def _env(name, environ):
     return environ.get(name.lower(), environ.get(name, ""))
 
 
+def _ca_context(args, environ):
+    path = getattr(args, "ca_bundle", None)
+    if path is None:
+        path = environ.get("EXTORE_CA_BUNDLE", "")
+    if not path:
+        return None
+    try:
+        # Ambient SSL_CERT_FILE/DIR remain disabled. Add explicit trust locally.
+        context = ssl.create_default_context(cafile=certifi.where())
+        context.load_verify_locations(cafile=path)
+        return context
+    except (OSError, ValueError, TypeError):
+        from .manage_client import ManageError
+
+        raise ManageError(
+            "Invalid CLI CA bundle configuration", code="invalid_ca_bundle"
+        ) from None
+
+
+class _CIDRClient(httpx.Client):
+    """CIDR rules match literal IP hosts only; never perform local proxy DNS."""
+
+    def __init__(self, *, bypass_networks, **kwargs):
+        self._bypass_networks = bypass_networks
+        super().__init__(**kwargs)
+
+    def _transport_for_url(self, url):
+        try:
+            address = ipaddress.ip_address(url.host)
+        except ValueError:
+            pass
+        else:
+            if any(address in network for network in self._bypass_networks):
+                return self._transport
+        return super()._transport_for_url(url)
+
+
 def resolve_proxy(args=None, *, environ=None):
     environ = os.environ if environ is None else environ
+    tls_context = _ca_context(args, environ)
     explicit = getattr(args, "proxy", None)
     use_env = getattr(args, "proxy_env", False)
     if explicit is not None and use_env:
         _invalid()
     if explicit is not None:
-        return ProxySettings("explicit", validate_proxy(explicit))
+        return ProxySettings(
+            "explicit", validate_proxy(explicit), tls_context=tls_context
+        )
     if not use_env:
         dedicated = environ.get("EXTORE_PROXY", "")
         return (
-            ProxySettings("explicit", validate_proxy(dedicated))
+            ProxySettings(
+                "explicit", validate_proxy(dedicated), tls_context=tls_context
+            )
             if dedicated
-            else ProxySettings()
+            else ProxySettings(tls_context=tls_context)
         )
     routes = {}
     for scheme in ("http", "https", "all"):
@@ -137,6 +188,7 @@ def resolve_proxy(args=None, *, environ=None):
     if len(bypass) > 16384:
         _invalid()
     bypass_all = False
+    networks = []
     for hostname in (part.strip() for part in bypass.split(",")):
         if not hostname:
             continue
@@ -159,6 +211,9 @@ def resolve_proxy(args=None, *, environ=None):
                 ):
                     _invalid()
                 pattern = hostname.rstrip("/")
+            elif "/" in hostname:
+                networks.append(ipaddress.ip_network(hostname, strict=False))
+                continue
             else:
                 if any(char in hostname for char in "@/?#"):
                     _invalid()
@@ -184,36 +239,48 @@ def resolve_proxy(args=None, *, environ=None):
             _invalid()
         routes[pattern] = None
     return ProxySettings(
-        "environment", routes=() if bypass_all else tuple(routes.items())
+        "environment",
+        routes=() if bypass_all else tuple(routes.items()),
+        bypass_networks=() if bypass_all else tuple(networks),
+        tls_context=tls_context,
     )
 
 
 def make_client(settings=None, *, timeout=30, transport=None):
     settings = resolve_proxy() if settings is None else settings
     _protect_proxy_logging(settings)
     opened = []
+    verify = settings.tls_context or True
+
+    def proxy_url(value):
+        if settings.tls_context is not None and urlsplit(value).scheme == "https":
+            return httpx.Proxy(value, ssl_context=settings.tls_context)
+        return value
+
     try:
         kwargs = {
             "timeout": timeout,
             "transport": transport,
-            "verify": True,
+            "verify": verify,
             "follow_redirects": False,
             "trust_env": False,
         }
         if settings.url is not None:
-            kwargs["proxy"] = settings.url
+            kwargs["proxy"] = proxy_url(settings.url)
         if settings.routes:
             mounts = {}
             for pattern, value in settings.routes:
                 if value is None:
                     mounts[pattern] = None
                 else:
                     mount = httpx.HTTPTransport(
-                        proxy=value, verify=True, trust_env=False
+                        proxy=proxy_url(value), verify=verify, trust_env=False
                     )
                     opened.append(mount)
                     mounts[pattern] = mount
             kwargs["mounts"] = mounts
+        if settings.bypass_networks:
+            return _CIDRClient(bypass_networks=settings.bypass_networks, **kwargs)
         return httpx.Client(**kwargs)
     except (ValueError, ImportError, httpx.InvalidURL):
         for opened_transport in opened:

pyproject.toml

@@ -19,6 +19,7 @@ classifiers = [
 ]
 dependencies = [
     "argon2-cffi>=25.1.0",
+    "certifi>=2026.7.22",
     "cryptography>=50.0.2",
     "fastapi>=0.142.2",
     "httpx[socks]>=0.28.1",

tests/test_cli_proxy.py

@@ -3,7 +3,13 @@
 
 import pytest
 
-from extore import customer_cli, manage_client, owner_client
+from extore import (
+    commerce_client,
+    commerce_commands,
+    customer_cli,
+    manage_client,
+    owner_client,
+)
 from extore.http_proxy import (
     ProxySettings,
     make_client,
@@ -18,6 +24,7 @@ def parser():
     commands = result.add_subparsers(dest="command", required=True)
     for module in (manage_client, owner_client, customer_cli):
         module.add_parser(commands)
+    commerce_commands.add_parser(commands)
     return result
 
 
@@ -152,6 +159,9 @@ def close(self):
         manage_client.ManageClient(data, proxy_settings=settings),
         owner_client.OwnerClient(data, proxy_settings=settings),
         customer_cli.CustomerClient(data, proxy_settings=settings),
+        commerce_client.CommerceClient(
+            "https://store.example", "test-client", proxy_settings=settings
+        ),
     ]
     for configuration in configurations:
         assert configuration["proxy"] == settings.url
@@ -215,3 +225,134 @@ def test_socks_debug_trace_does_not_log_auth_and_keeps_ordinary_logs(
     assert "synthetic-secret" not in caplog.text
     assert "setup_socks5_connection.complete" in caplog.text
     assert "connect_tcp.started" in caplog.text
+
+
+@pytest.mark.parametrize("role", ["manage", "admin", "customer"])
+@pytest.mark.parametrize("before", [True, False])
+def test_ca_bundle_flag_positions(role, before):
+    command = [role]
+    if before:
+        command += ["--ca-bundle", "explicit.pem"]
+    command += ["login"] if role != "customer" else ["products"]
+    if role != "manage":
+        command += ["--origin", "https://store.example"]
+    if not before:
+        command += ["--ca-bundle", "explicit.pem"]
+    assert parser().parse_args(command).ca_bundle == "explicit.pem"
+
+
+@pytest.mark.parametrize(
+    "network,matching,nonmatching",
+    [
+        ("10.0.0.0/8", "10.12.34.56", "11.12.34.56"),
+        ("192.168.1.42/24", "192.168.1.7", "192.168.2.7"),
+        ("2001:db8::/32", "2001:db8::1234", "2001:db9::1234"),
+    ],
+)
+def test_cidr_routes_match_only_literal_ips(network, matching, nonmatching):
+    import httpx
+
+    settings = resolve_proxy(
+        argparse.Namespace(proxy_env=True),
+        environ={"ALL_PROXY": "http://proxy.example:3128", "NO_PROXY": network},
+    )
+    with make_client(settings) as client:
+
+        def route(host):
+            host = f"[{host}]" if ":" in host else host
+            return client._transport_for_url(httpx.URL(f"https://{host}/"))
+
+        assert route(matching) is client._transport
+        assert route(nonmatching) is not client._transport
+        assert route("service.example") is not client._transport
+
+
+@pytest.mark.parametrize(
+    "value", ["10.0.0.0/99", "host.example/8", "user:secret@host/8", "2001:db8::/999"]
+)
+def test_invalid_cidr_redacted(value):
+    with pytest.raises(ManageError) as caught:
+        resolve_proxy(argparse.Namespace(proxy_env=True), environ={"NO_PROXY": value})
+    assert caught.value.code == "invalid_proxy"
+    assert value not in str(caught.value)
+    assert "secret" not in str(caught.value)
+
+
+@pytest.mark.parametrize("module", [manage_client, owner_client, customer_cli])
+def test_invalid_ca_before_profile_and_redacted(module, tmp_path, monkeypatch):
+    ca = tmp_path / "private-path-secret.pem"
+    ca.write_text("invalid certificate")
+    monkeypatch.setenv("EXTORE_CA_BUNDLE", str(ca))
+    profile = tmp_path / "not-created" / "profile.json"
+    with pytest.raises(ManageError) as caught:
+        module.execute(argparse.Namespace(profile=profile))
+    assert caught.value.code == "invalid_ca_bundle"
+    assert str(ca) not in str(caught.value)
+    assert "secret" not in str(caught.value)
+    assert not profile.parent.exists()
+
+
+def test_explicit_ca_context_preserves_verification_and_ignores_ambient(monkeypatch):
+    import ssl
+
+    import certifi
+
+    monkeypatch.setenv("SSL_CERT_FILE", "/untrusted-ambient.pem")
+    settings = resolve_proxy(
+        argparse.Namespace(ca_bundle=certifi.where()),
+        environ={"EXTORE_CA_BUNDLE": "/ignored.pem"},
+    )
+    assert settings.tls_context.verify_mode == ssl.CERT_REQUIRED
+    assert settings.tls_context.check_hostname is True
+    assert certifi.where() not in repr(settings)
+    with make_client(settings) as client:
+        assert client.follow_redirects is False
+
+
+@pytest.mark.parametrize("scheme", ["https", "HTTPS"])
+def test_ca_bundle_applies_to_https_proxy_and_target(monkeypatch, scheme):
+    import certifi
+
+    captured = {}
+    transports = []
+
+    class CapturedTransport:
+        def __init__(self, **kwargs):
+            transports.append(kwargs)
+
+    monkeypatch.setattr("extore.http_proxy.httpx.HTTPTransport", CapturedTransport)
+    monkeypatch.setattr(
+        "extore.http_proxy.httpx.Client", lambda **kwargs: captured.update(kwargs)
+    )
+    settings = resolve_proxy(
+        argparse.Namespace(proxy_env=True),
+        environ={
+            "EXTORE_CA_BUNDLE": certifi.where(),
+            "HTTPS_PROXY": f"{scheme}://user:synthetic-secret@proxy.example:443",
+        },
+    )
+    make_client(settings)
+    assert captured["verify"] is settings.tls_context
+    assert transports[0]["verify"] is settings.tls_context
+    assert transports[0]["proxy"].ssl_context is settings.tls_context
+    assert captured["trust_env"] is False
+    assert "synthetic-secret" not in repr(settings)
+
+
+@pytest.mark.parametrize("before", [True, False])
+def test_commerce_ca_bundle_flag_positions(before):
+    command = ["commerce"]
+    if before:
+        command += ["--ca-bundle", "explicit.pem"]
+    command += [
+        "metadata",
+        "--origin",
+        "https://store.example",
+        "--client-id",
+        "test-client",
+        "--output",
+        "output.json",
+    ]
+    if not before:
+        command += ["--ca-bundle", "explicit.pem"]
+    assert parser().parse_args(command).ca_bundle == "explicit.pem"