From cf57bfdeaf637de22bd3fe3e97334562e990cb23 Mon Sep 17 00:00:00 2001 From: hsq Date: Tue, 15 Sep 2026 21:39:37 +0800 Subject: [PATCH] feat: support standard outbound proxy environment --- .env.dev.example | 5 + .env.example | 4 + README.md | 7 + README.zh-CN.md | 4 + config/runtime.exs | 3 + .../docs/getting-started/configuration.mdx | 18 ++ .../alerts/delivery/webhook_delivery.ex | 2 +- .../catalog/openai_pricing_importer.ex | 2 +- lib/codex_pooler/catalog/sync/discovery.ex | 2 +- .../gateway/operational_settings.ex | 22 +- .../gateway/payloads/transport_envelope.ex | 21 +- .../gateway/transports/file_bridge.ex | 8 +- .../gateway/transports/upstream_dispatch.ex | 4 +- .../connection_upgrade.ex | 54 +++- lib/codex_pooler/platform/outbound_http.ex | 257 +++++++++++++++++- lib/codex_pooler/status/feed_client.ex | 6 +- lib/codex_pooler/upstreams/auth/codex_auth.ex | 19 +- .../saved_reset_usage_enrichment.ex | 2 +- .../upstreams/reconciliation/usage_probe.ex | 2 +- .../upstreams/saved_resets/redemption.ex | 8 +- scripts/self-host/generate-env.sh | 3 + .../payloads/transport_envelope_test.exs | 2 +- .../platform/outbound_http_test.exs | 123 ++++++++- test/codex_pooler/runtime_config_test.exs | 39 +++ 24 files changed, 558 insertions(+), 59 deletions(-) diff --git a/.env.dev.example b/.env.dev.example index f855483b1..56f7573f3 100644 --- a/.env.dev.example +++ b/.env.dev.example @@ -2,6 +2,11 @@ # App-specific variables keep self-host `.env` release credentials from leaking # into direct `mix` and `mix test` runs. +# Optional standard proxy environment variables for outbound traffic. +# http_proxy=http://proxy.example.com:3128 +# https_proxy=http://proxy.example.com:3128 +# no_proxy=localhost,127.0.0.1,.example.internal + CODEX_POOLER_DEV_POSTGRES_HOST=localhost CODEX_POOLER_DEV_POSTGRES_PORT=5433 CODEX_POOLER_DEV_POSTGRES_DB=codex_pooler_dev diff --git a/.env.example b/.env.example index 537bf0744..b0c3a28c6 100644 --- a/.env.example +++ b/.env.example @@ -4,6 +4,10 @@ CODEX_POOLER_IMAGE=ghcr.io/icoretech/codex-pooler CODEX_POOLER_IMAGE_TAG=latest CODEX_POOLER_HTTP_PORT=4000 +# Optional standard proxy environment variables for outbound traffic. +# http_proxy=http://proxy.example.com:3128 +# https_proxy=http://proxy.example.com:3128 +# no_proxy=localhost,127.0.0.1,.example.internal PHX_HOST=localhost OBAN_MODE=all diff --git a/README.md b/README.md index 7b3ea8621..6a3331033 100644 --- a/README.md +++ b/README.md @@ -2165,6 +2165,13 @@ the database: - `CODEX_POOLER_UPSTREAM_SECRET_KEY` and `CODEX_POOLER_UPSTREAM_SECRET_KEY_VERSION`, upstream secret encryption root and version; the key must be 32 raw bytes or base64-encoded 32 bytes +- `http_proxy` and `https_proxy`, optional `http://` proxy URLs selected by the + outbound target scheme, including WebSocket traffic; optional URL credentials + use HTTP Basic proxy authentication +- `no_proxy`, an optional comma-separated bypass list supporting exact hosts, + domain suffixes, IP addresses, CIDR ranges, optional ports, and `*`; lowercase variables + take precedence over the uppercase `HTTP_PROXY`, `HTTPS_PROXY`, and `NO_PROXY` + variants Operational controls such as file limits, ingress trust, gateway diagnostics, route-class admission, circuit thresholds, metrics auth, operator email, model diff --git a/README.zh-CN.md b/README.zh-CN.md index 8f64f7a8c..a626e0027 100644 --- a/README.zh-CN.md +++ b/README.zh-CN.md @@ -2029,6 +2029,10 @@ user-agent 历史。 - `CODEX_POOLER_UPSTREAM_SECRET_KEY` 和 `CODEX_POOLER_UPSTREAM_SECRET_KEY_VERSION`，上游密钥加密根和版本；key 必须是 32 raw bytes 或 base64-encoded 32 bytes +- `http_proxy` 和 `https_proxy`，按出站目标协议选择的可选 `http://` 代理 URL， + 同时适用于 WebSocket；URL 中可选的凭据使用 HTTP Basic 代理认证 +- `no_proxy`，可选的逗号分隔绕过列表，支持精确主机、域名后缀、IP 地址、CIDR、 + 可选端口和 `*`；小写变量优先于大写的 `HTTP_PROXY`、`HTTPS_PROXY` 和 `NO_PROXY` 文件限制、入口信任、网关诊断、路由类别准入、熔断阈值、指标认证、运营者邮箱、 模型元数据、上游超时、OpenAI 价格 catalog URL 和 SMTP 投递等运营控制项位于 diff --git a/config/runtime.exs b/config/runtime.exs index d5d2d7374..56abe8724 100644 --- a/config/runtime.exs +++ b/config/runtime.exs @@ -14,6 +14,9 @@ config :codex_pooler, CodexPooler.Gateway.Transports.Websocket.NativeCompactionTrace, mode: native_compaction_trace_mode +config :codex_pooler, CodexPooler.Platform.OutboundHTTP, + proxy_config: CodexPooler.Platform.OutboundHTTP.proxy_config_from_env!() + if System.get_env("PHX_SERVER") in ~w(true 1) do config :codex_pooler, CodexPoolerWeb.Endpoint, server: true end diff --git a/docs-site/src/content/docs/getting-started/configuration.mdx b/docs-site/src/content/docs/getting-started/configuration.mdx index b871d95cb..a32dbd9a7 100644 --- a/docs-site/src/content/docs/getting-started/configuration.mdx +++ b/docs-site/src/content/docs/getting-started/configuration.mdx @@ -31,6 +31,12 @@ as `https://localhost`; the port mapping is the local URL to open. Set release environment variables for values the app needs before it can read database settings. +Set `http_proxy` and `https_proxy` to `http://` proxy URLs to route outbound +HTTP, HTTPS, and WebSocket traffic by target scheme. Credentials embedded in a +proxy URL are sent as HTTP Basic proxy authentication. Use the comma-separated +`no_proxy` list for exact hosts, domain suffixes, IP addresses, CIDR ranges, +optional ports, or `*`. Lowercase variables take precedence over their uppercase variants. + @@ -51,6 +57,18 @@ Set release environment variables for values the app needs before it can read da + + + + + + + + + + + + diff --git a/lib/codex_pooler/alerts/delivery/webhook_delivery.ex b/lib/codex_pooler/alerts/delivery/webhook_delivery.ex index fb5808039..f188a75ac 100644 --- a/lib/codex_pooler/alerts/delivery/webhook_delivery.ex +++ b/lib/codex_pooler/alerts/delivery/webhook_delivery.ex @@ -181,7 +181,7 @@ defmodule CodexPooler.Alerts.Delivery.WebhookDelivery do decode_body: false, receive_timeout: @receive_timeout_ms, retry: false, - finch: OutboundHTTP.pool_options() + finch: OutboundHTTP.pool_options_for_url(url) ) rescue exception in [ diff --git a/lib/codex_pooler/catalog/openai_pricing_importer.ex b/lib/codex_pooler/catalog/openai_pricing_importer.ex index 3940648e1..db232667d 100644 --- a/lib/codex_pooler/catalog/openai_pricing_importer.ex +++ b/lib/codex_pooler/catalog/openai_pricing_importer.ex @@ -64,7 +64,7 @@ defmodule CodexPooler.Catalog.OpenAIPricingImporter do decode_body: false, receive_timeout: :timer.seconds(30), retry: false, - finch: OutboundHTTP.pool_options() + finch: OutboundHTTP.pool_options_for_url(url) ) do {:ok, %{status: status, body: body}} when status in 200..299 and is_binary(body) -> {:ok, body} diff --git a/lib/codex_pooler/catalog/sync/discovery.ex b/lib/codex_pooler/catalog/sync/discovery.ex index a6128dc1e..d77367933 100644 --- a/lib/codex_pooler/catalog/sync/discovery.ex +++ b/lib/codex_pooler/catalog/sync/discovery.ex @@ -71,7 +71,7 @@ defmodule CodexPooler.Catalog.Sync.Discovery do case Req.get(url, retry: false, receive_timeout: 30_000, - finch: OutboundHTTP.pool_options(), + finch: OutboundHTTP.pool_options_for_url(url), headers: CloudflareCookies.request_headers(url, model_catalog_headers(identity, token)) ) diff --git a/lib/codex_pooler/gateway/operational_settings.ex b/lib/codex_pooler/gateway/operational_settings.ex index 88b2bff86..fc9fc7ab3 100644 --- a/lib/codex_pooler/gateway/operational_settings.ex +++ b/lib/codex_pooler/gateway/operational_settings.ex @@ -202,14 +202,32 @@ defmodule CodexPooler.Gateway.OperationalSettings do Finch HTTP/1 pool options for gateway Req requests, built by `OutboundHTTP.pool_options/1` from this snapshot's idle bound: dispatch through `TransportEnvelope.req_timeout_options/1` and the file bridge upload - PUT. Callers outside the gateway use `OutboundHTTP.pool_options/0`, which - reads the same Instance Setting. + PUT. Callers outside the gateway use `OutboundHTTP.pool_options_for_url/2`, + which reads the same Instance Setting. The URL-aware helpers include the + proxy selected for the target and apply `no_proxy`. """ @spec upstream_http_pool_options() :: OutboundHTTP.pool_options() def upstream_http_pool_options do OutboundHTTP.pool_options(current().upstream_conn_max_idle_time_ms) end + @spec upstream_http_pool_options(keyword()) :: OutboundHTTP.pool_options() + def upstream_http_pool_options(conn_opts) when is_list(conn_opts) do + OutboundHTTP.pool_options_with_conn_opts( + current().upstream_conn_max_idle_time_ms, + conn_opts + ) + end + + @spec upstream_http_pool_options(String.t(), keyword()) :: OutboundHTTP.pool_options() + def upstream_http_pool_options(url, conn_opts) when is_binary(url) and is_list(conn_opts) do + OutboundHTTP.pool_options_for_url( + url, + current().upstream_conn_max_idle_time_ms, + conn_opts + ) + end + @spec firewall_enabled?(t()) :: boolean() def firewall_enabled?(%__MODULE__{firewall_allowlist: allowlist}), do: allowlist != [] diff --git a/lib/codex_pooler/gateway/payloads/transport_envelope.ex b/lib/codex_pooler/gateway/payloads/transport_envelope.ex index 59335ec57..80a6ba41f 100644 --- a/lib/codex_pooler/gateway/payloads/transport_envelope.ex +++ b/lib/codex_pooler/gateway/payloads/transport_envelope.ex @@ -74,15 +74,22 @@ defmodule CodexPooler.Gateway.Payloads.TransportEnvelope do # `pool_max_idle_time` stays unset: stopping an idle per-origin pool can race # a request that has just looked it up, and stale connections are already # dropped at checkout. - @spec req_timeout_options(TimeoutConfig.t() | timeout_settings()) :: keyword() - def req_timeout_options(timeouts) do + @spec req_timeout_options(TimeoutConfig.t() | timeout_settings(), String.t() | nil) :: keyword() + def req_timeout_options(timeouts, url \\ nil) do + pool_options = + if is_binary(url) do + OperationalSettings.upstream_http_pool_options(url, + transport_opts: [timeout: timeouts.connect_timeout_ms] + ) + else + OperationalSettings.upstream_http_pool_options( + transport_opts: [timeout: timeouts.connect_timeout_ms] + ) + end + [ receive_timeout: timeouts.receive_timeout_ms, - finch: - [ - pool_timeout: timeouts.pool_timeout_ms, - conn_opts: [transport_opts: [timeout: timeouts.connect_timeout_ms]] - ] ++ OperationalSettings.upstream_http_pool_options() + finch: [pool_timeout: timeouts.pool_timeout_ms] ++ pool_options ] end diff --git a/lib/codex_pooler/gateway/transports/file_bridge.ex b/lib/codex_pooler/gateway/transports/file_bridge.ex index 7e8392565..70718d775 100644 --- a/lib/codex_pooler/gateway/transports/file_bridge.ex +++ b/lib/codex_pooler/gateway/transports/file_bridge.ex @@ -68,7 +68,7 @@ defmodule CodexPooler.Gateway.Transports.FileBridge do with {:ok, body, byte_size} <- readable_file_stream(path) do upload_url |> upload_request() - |> Req.put(upload_req_options(body, content_type, byte_size)) + |> Req.put(upload_req_options(upload_url, body, content_type, byte_size)) |> normalize_upload_response(opts) end rescue @@ -214,7 +214,7 @@ defmodule CodexPooler.Gateway.Transports.FileBridge do retry: false, headers: headers(identity, token, forwarded_headers(opts)) ] - |> Keyword.merge(TransportEnvelope.req_timeout_options(timeouts)) + |> Keyword.merge(TransportEnvelope.req_timeout_options(timeouts, url)) url |> Req.post(request_options) @@ -267,7 +267,7 @@ defmodule CodexPooler.Gateway.Transports.FileBridge do path == root or String.starts_with?(path, root <> "/") end - defp upload_req_options(body, content_type, byte_size) do + defp upload_req_options(upload_url, body, content_type, byte_size) do configured_upload_req_options() |> Keyword.merge( body: body, @@ -278,7 +278,7 @@ defmodule CodexPooler.Gateway.Transports.FileBridge do ], redirect: false, retry: false, - finch: OperationalSettings.upstream_http_pool_options() + finch: OperationalSettings.upstream_http_pool_options(upload_url, []) ) end diff --git a/lib/codex_pooler/gateway/transports/upstream_dispatch.ex b/lib/codex_pooler/gateway/transports/upstream_dispatch.ex index af7d1d36e..fce9a5a2a 100644 --- a/lib/codex_pooler/gateway/transports/upstream_dispatch.ex +++ b/lib/codex_pooler/gateway/transports/upstream_dispatch.ex @@ -401,7 +401,7 @@ defmodule CodexPooler.Gateway.Transports.UpstreamDispatch do ]) ) ] - |> Keyword.merge(TransportEnvelope.req_timeout_options(timeouts)) + |> Keyword.merge(TransportEnvelope.req_timeout_options(timeouts, url)) result = Req.post(url, request_options) CloudflareCookies.store_from_result(url, result) @@ -462,7 +462,7 @@ defmodule CodexPooler.Gateway.Transports.UpstreamDispatch do retry: false, headers: upstream_header_list ] - |> Keyword.merge(TransportEnvelope.req_timeout_options(timeouts)) + |> Keyword.merge(TransportEnvelope.req_timeout_options(timeouts, url)) request_options = if streaming_request?(payload, opts) do diff --git a/lib/codex_pooler/gateway/transports/websocket/upstream_websocket_session/connection_upgrade.ex b/lib/codex_pooler/gateway/transports/websocket/upstream_websocket_session/connection_upgrade.ex index d9c4c9d9a..620ce1916 100644 --- a/lib/codex_pooler/gateway/transports/websocket/upstream_websocket_session/connection_upgrade.ex +++ b/lib/codex_pooler/gateway/transports/websocket/upstream_websocket_session/connection_upgrade.ex @@ -1,6 +1,8 @@ defmodule CodexPooler.Gateway.Transports.Websocket.UpstreamWebsocketSession.ConnectionUpgrade do @moduledoc false + alias CodexPooler.Platform.OutboundHTTP + @type request_caller :: {pid(), reference()} | nil @type upgrade_response :: %{status: non_neg_integer() | nil, headers: Mint.Types.headers()} @connect_ready_tag :upstream_websocket_connect_ready @@ -37,28 +39,47 @@ defmodule CodexPooler.Gateway.Transports.Websocket.UpstreamWebsocketSession.Conn end defp connect_websocket(%{connect_scheme: :http} = target, timeouts, request_caller) do - connect_in_task(target, timeouts, request_caller) + proxy_options = OutboundHTTP.proxy_options_for_url(target.uri) + connect_in_task(target, timeouts, request_caller, proxy_options) end defp connect_websocket(%{connect_scheme: :https} = target, timeouts, request_caller) do - raw_target = %{target | connect_scheme: :http} + proxy_options = OutboundHTTP.proxy_options_for_url(target.uri) + + if proxy_options == [] do + raw_target = %{target | connect_scheme: :http} - with {:ok, raw_conn} <- connect_in_task(raw_target, timeouts, request_caller) do - upgrade_tls_connection(raw_conn, target, timeouts, request_caller) + with {:ok, raw_conn} <- connect_in_task(raw_target, timeouts, request_caller, []) do + upgrade_tls_connection(raw_conn, target, timeouts, request_caller) + end + else + # Mint performs CONNECT and the TLS handshake as one proxied connection. + connect_in_task(target, timeouts, request_caller, proxy_options) end end - defp connect_in_task(target, timeouts, request_caller) do + defp connect_in_task(target, timeouts, request_caller, proxy_options) do parent = self() {:ok, connect_pid} = Task.start(fn -> parent_monitor = Process.monitor(parent) - result = - Mint.HTTP.connect(target.connect_scheme, target.host, target.port, + connect_options = + [ protocols: [:http1], transport_opts: websocket_transport_opts(target, timeouts) + ] + + connect_options = + connect_options ++ proxy_options_with_connect_timeout(proxy_options, timeouts) + + result = + Mint.HTTP.connect( + target.connect_scheme, + target.host, + target.port, + connect_options ) send(parent, {@connect_ready_tag, self(), result}) @@ -82,6 +103,16 @@ defmodule CodexPooler.Gateway.Transports.Websocket.UpstreamWebsocketSession.Conn await_connect(connect_pid, connect_monitor, timeouts, request_caller) end + defp proxy_options_with_connect_timeout(proxy_options, timeouts) do + Enum.map(proxy_options, fn + {:proxy, {scheme, host, port, _opts}} -> + {:proxy, {scheme, host, port, [transport_opts: [timeout: timeouts.connect_timeout_ms]]}} + + option -> + option + end) + end + defp upgrade_tls_connection(raw_conn, target, timeouts, request_caller) do watcher = start_tls_upgrade_caller_watcher(raw_conn, request_caller) @@ -326,7 +357,14 @@ defmodule CodexPooler.Gateway.Transports.Websocket.UpstreamWebsocketSession.Conn path = websocket_path(uri) {:ok, - %{connect_scheme: connect_scheme, ws_scheme: ws_scheme, host: host, port: port, path: path}} + %{ + connect_scheme: connect_scheme, + ws_scheme: ws_scheme, + host: host, + port: port, + path: path, + uri: uri + }} else _invalid -> {:error, :invalid_upstream_websocket_url} end diff --git a/lib/codex_pooler/platform/outbound_http.ex b/lib/codex_pooler/platform/outbound_http.ex index e61b298f6..5fc06d013 100644 --- a/lib/codex_pooler/platform/outbound_http.ex +++ b/lib/codex_pooler/platform/outbound_http.ex @@ -1,6 +1,8 @@ defmodule CodexPooler.Platform.OutboundHTTP do @moduledoc """ - Finch pool options for every outbound Req request Codex Pooler sends. + Finch pool options for every outbound Req request Codex Pooler sends, plus + the standard boot-time proxy environment shared by Req and upstream + WebSockets. Req runs a request that carries Finch pool options on a dedicated Finch instance keyed by those options, with one HTTP/1 pool per origin, instead of @@ -40,21 +42,21 @@ defmodule CodexPooler.Platform.OutboundHTTP do outside the bounds, which only a stale cache or a hand-edited row can carry, are clamped so a bad value cannot make Finch reject every outbound request. - Callers add `finch: pool_options()` and keep their own receive, retry, and - redirect options. Req starts one Finch instance per distinct `finch:` pool + Callers add `finch: pool_options_for_url(url)` and keep their own receive, + retry, and redirect options. Req starts one Finch instance per distinct `finch:` pool option tuple; `pool_timeout` and `receive_timeout` are per-request options outside that key. Every caller that needs no connect timeout of its own passes exactly these options, so they share one instance per saved value; a - caller that sets a connect timeout adds `conn_opts` and gets its own instance - per idle bound and connect timeout pair. `pool_max_idle_time` stays unset, so - no instance is ever stopped: stopping an idle per-origin pool can race a - request that has just looked it up, and stale connections are already - dropped at checkout. Each distinct combination of this setting and the - gateway connect timeout saved since boot therefore keeps its pools until - restart, so both settings are low-churn and must not be driven from - automation. Req refuses `finch:` together with `connect_options:`, so a - connect timeout travels as `conn_opts: [transport_opts: [timeout: ms]]` - inside `finch:`. + caller that sets a connect timeout passes it to `pool_options_for_url/2` and + gets its own instance per idle bound and connect timeout pair. The optional + selected boot-time proxy is also part of the pool tuple. `pool_max_idle_time` stays + unset, so no instance is ever stopped: stopping an idle per-origin pool can + race a request that has just looked it up, and stale connections are already + dropped at checkout. Each distinct combination of this setting, proxy, and + gateway connect timeout therefore keeps its pools until restart, so the live + settings are low-churn and must not be driven from automation. Req refuses + `finch:` together with `connect_options:`, so a connect timeout travels as + `conn_opts: [transport_opts: [timeout: ms]]` inside `finch:`. """ alias CodexPooler.InstanceSettings @@ -63,8 +65,9 @@ defmodule CodexPooler.Platform.OutboundHTTP do @conn_max_idle_time_default_ms 45_000 @conn_max_idle_time_min_ms 1_000 @conn_max_idle_time_max_ms 3_600_000 + @invalid_proxy_message "proxy environment variables must be http:// URLs containing only authority and optional basic credentials" - @type pool_options :: [conn_max_idle_time: non_neg_integer()] + @type pool_options :: keyword() @doc """ Finch pool options carrying the current outbound connection idle bound. @@ -79,7 +82,231 @@ defmodule CodexPooler.Platform.OutboundHTTP do @spec pool_options(non_neg_integer()) :: pool_options() def pool_options(conn_max_idle_time_ms) when is_integer(conn_max_idle_time_ms) and conn_max_idle_time_ms >= 0, - do: [conn_max_idle_time: conn_max_idle_time_ms] + do: pool_options_with_conn_opts(conn_max_idle_time_ms, []) + + @doc """ + Builds Finch pool options with caller-specific Mint connection options when + no target URL is available. Production outbound calls use + `pool_options_for_url/2` so scheme selection and `no_proxy` can be applied. + """ + @spec pool_options_with_conn_opts(keyword()) :: pool_options() + def pool_options_with_conn_opts(conn_opts) when is_list(conn_opts), + do: pool_options_with_conn_opts(current_conn_max_idle_time_ms(), conn_opts) + + @spec pool_options_with_conn_opts(non_neg_integer(), keyword()) :: pool_options() + def pool_options_with_conn_opts(conn_max_idle_time_ms, conn_opts) + when is_integer(conn_max_idle_time_ms) and conn_max_idle_time_ms >= 0 and + is_list(conn_opts) do + pool_options = [conn_max_idle_time: conn_max_idle_time_ms] + + if conn_opts == [], do: pool_options, else: [conn_opts: conn_opts] ++ pool_options + end + + @doc """ + Builds pool options for `url`, selecting `http_proxy` or `https_proxy` and + honoring `no_proxy`. + """ + @spec pool_options_for_url(String.t() | URI.t(), keyword()) :: pool_options() + def pool_options_for_url(url, conn_opts \\ []) do + pool_options_for_url(url, current_conn_max_idle_time_ms(), conn_opts) + end + + @spec pool_options_for_url(String.t() | URI.t(), non_neg_integer(), keyword()) :: pool_options() + def pool_options_for_url(url, conn_max_idle_time_ms, conn_opts) + when is_integer(conn_max_idle_time_ms) and conn_max_idle_time_ms >= 0 and + is_list(conn_opts) do + conn_opts = Keyword.merge(proxy_options_for_url(url, conn_opts), conn_opts) + pool_options = [conn_max_idle_time: conn_max_idle_time_ms] + + if conn_opts == [], do: pool_options, else: [conn_opts: conn_opts] ++ pool_options + end + + @doc "Returns Mint connection options selected for the target URL." + @spec proxy_options_for_url(String.t() | URI.t(), keyword()) :: keyword() + def proxy_options_for_url(url, connect_opts \\ []) when is_list(connect_opts) do + uri = if is_struct(url, URI), do: url, else: URI.parse(url) + config = proxy_config() + + if no_proxy?(uri, config.no_proxy) do + [] + else + uri.scheme + |> proxy_for_scheme(config) + |> proxy_with_connect_timeout(connect_opts) + end + end + + defp proxy_with_connect_timeout([], _connect_opts), do: [] + + defp proxy_with_connect_timeout(proxy_options, connect_opts) do + proxy_connect_opts = + case get_in(connect_opts, [:transport_opts, :timeout]) do + timeout when is_integer(timeout) -> [transport_opts: [timeout: timeout]] + _other -> [] + end + + Enum.map(proxy_options, fn + {:proxy, {scheme, host, port, _opts}} -> + {:proxy, {scheme, host, port, proxy_connect_opts}} + + option -> + option + end) + end + + @doc "Reads lowercase proxy variables, with uppercase variants as fallbacks." + @spec proxy_config_from_env!() :: map() + def proxy_config_from_env! do + %{ + http: parse_proxy_url!(proxy_env("http_proxy", "HTTP_PROXY")), + https: parse_proxy_url!(proxy_env("https_proxy", "HTTPS_PROXY")), + no_proxy: parse_no_proxy(proxy_env("no_proxy", "NO_PROXY")) + } + end + + @doc """ + Parses one standard proxy environment variable into Mint connection options. + + Only an `http://` proxy is accepted because Mint cannot tunnel an HTTPS + destination through an HTTPS proxy. + """ + @spec parse_proxy_url!(String.t() | nil) :: keyword() + def parse_proxy_url!(value) when value in [nil, ""], do: [] + + def parse_proxy_url!(value) when is_binary(value) do + try do + uri = URI.parse(value) + port = uri.port || 80 + + if uri.scheme != "http" or not is_binary(uri.host) or uri.host == "" or + not is_integer(port) or port not in 1..65_535 or uri.path not in [nil, "", "/"] or + not is_nil(uri.query) or not is_nil(uri.fragment) do + raise ArgumentError, @invalid_proxy_message + end + + proxy = {:http, uri.host, port, []} + + case proxy_authorization(uri.userinfo) do + nil -> [proxy: proxy] + authorization -> [proxy: proxy, proxy_headers: [{"proxy-authorization", authorization}]] + end + rescue + _invalid in [ArgumentError, URI.Error] -> raise ArgumentError, @invalid_proxy_message + end + end + + defp proxy_authorization(nil), do: nil + + defp proxy_authorization(userinfo) do + "Basic " <> Base.encode64(URI.decode(userinfo)) + end + + defp proxy_config do + Application.get_env(:codex_pooler, __MODULE__, []) + |> Keyword.get(:proxy_config, %{http: [], https: [], no_proxy: []}) + end + + defp proxy_env(lowercase, uppercase) do + case System.fetch_env(lowercase) do + {:ok, value} -> value + :error -> System.get_env(uppercase) + end + end + + defp parse_no_proxy(value) when value in [nil, ""], do: [] + + defp parse_no_proxy(value) do + value + |> String.split(",", trim: true) + |> Enum.map(&String.trim/1) + |> Enum.reject(&(&1 == "")) + end + + defp proxy_for_scheme("http", config), do: config.http + defp proxy_for_scheme("https", config), do: config.https + defp proxy_for_scheme(_scheme, _config), do: [] + + defp no_proxy?(%URI{host: host, port: port}, entries) when is_binary(host) do + host = normalize_host(host) + Enum.any?(entries, &no_proxy_entry?(&1, host, port)) + end + + defp no_proxy?(_uri, _entries), do: false + + defp no_proxy_entry?("*", _host, _port), do: true + + defp no_proxy_entry?(entry, host, port) do + {entry_host, entry_port} = split_no_proxy_entry(entry) + entry_host = normalize_host(entry_host) + port_matches? = is_nil(entry_port) or entry_port == port + + port_matches? and (cidr_match?(host, entry_host) or host_matches?(host, entry_host)) + end + + defp host_matches?(host, entry_host) do + host == entry_host or + (not ip_address?(host) and String.ends_with?(host, "." <> entry_host)) + end + + defp cidr_match?(host, entry) do + with [network, prefix] <- String.split(entry, "/", parts: 2), + {:ok, address} <- :inet.parse_address(String.to_charlist(host)), + {:ok, network_address} <- :inet.parse_address(String.to_charlist(network)), + true <- tuple_size(address) == tuple_size(network_address), + {prefix, ""} <- Integer.parse(prefix), + segment_bits = address_segment_bits(address), + bits = tuple_size(address) * segment_bits, + true <- prefix in 0..bits do + shift = bits - prefix + divisor = Integer.pow(2, shift) + + div(address_integer(address, segment_bits), divisor) == + div(address_integer(network_address, segment_bits), divisor) + else + _no_match -> false + end + end + + defp address_integer(address, segment_bits) do + address + |> Tuple.to_list() + |> Enum.reduce(0, fn segment, value -> value * Integer.pow(2, segment_bits) + segment end) + end + + defp address_segment_bits(address) when tuple_size(address) == 4, do: 8 + defp address_segment_bits(_address), do: 16 + + defp split_no_proxy_entry("[" <> rest = entry) do + case String.split(rest, "]", parts: 2) do + [host, ":" <> port] -> {host, parse_no_proxy_port(port)} + [host, ""] -> {host, nil} + _invalid -> {entry, nil} + end + end + + defp split_no_proxy_entry(entry) do + case String.split(entry, ":") do + [host, port] -> {host, parse_no_proxy_port(port)} + _host_or_ipv6 -> {entry, nil} + end + end + + defp parse_no_proxy_port(port) do + case Integer.parse(port) do + {value, ""} when value in 1..65_535 -> value + _invalid -> :invalid + end + end + + defp normalize_host(host) do + host + |> String.downcase() + |> String.trim_trailing(".") + |> String.trim_leading("*.") + |> String.trim_leading(".") + end + + defp ip_address?(host), do: match?({:ok, _address}, :inet.parse_address(String.to_charlist(host))) @doc """ The clamped outbound connection idle bound carried by `settings`. diff --git a/lib/codex_pooler/status/feed_client.ex b/lib/codex_pooler/status/feed_client.ex index 93b9ceeff..0974a66e5 100644 --- a/lib/codex_pooler/status/feed_client.ex +++ b/lib/codex_pooler/status/feed_client.ex @@ -8,6 +8,7 @@ defmodule CodexPooler.Status.FeedClient do @spec fetch(map(), keyword()) :: {:ok, map()} | {:not_modified, map()} | {:error, map()} def fetch(state \\ %{}, opts \\ []) do timeout = Keyword.get(opts, :timeout, 10_000) + url = Keyword.get(opts, :url, @url) headers = [] @@ -15,14 +16,15 @@ defmodule CodexPooler.Status.FeedClient do |> maybe_header("if-modified-since", Map.get(state, :last_modified)) request = [ - url: Keyword.get(opts, :url, @url), + url: url, headers: headers, decode_body: false, retry: false, receive_timeout: timeout, # Req refuses `connect_options` together with `finch`, so the connect # timeout travels as Finch `conn_opts` next to the idle bound. - finch: [conn_opts: [transport_opts: [timeout: timeout]]] ++ OutboundHTTP.pool_options(), + finch: + OutboundHTTP.pool_options_for_url(url, transport_opts: [timeout: timeout]), redirect: false ] diff --git a/lib/codex_pooler/upstreams/auth/codex_auth.ex b/lib/codex_pooler/upstreams/auth/codex_auth.ex index 693d7f300..70d617962 100644 --- a/lib/codex_pooler/upstreams/auth/codex_auth.ex +++ b/lib/codex_pooler/upstreams/auth/codex_auth.ex @@ -332,13 +332,15 @@ defmodule CodexPooler.Upstreams.Auth.CodexAuth do @spec request_device_code() :: CodexAuth.device_code_response() def request_device_code do + url = CodexAuth.issuer() <> "/api/accounts/deviceauth/usercode" + case Req.post( - CodexAuth.issuer() <> "/api/accounts/deviceauth/usercode", + url, headers: [{"content-type", "application/json"} | browser_request_headers()], body: CodexPooler.JSON.encode_to_iodata!(%{client_id: CodexAuth.client_id()}), retry: false, receive_timeout: 30_000, - finch: OutboundHTTP.pool_options() + finch: OutboundHTTP.pool_options_for_url(url) ) do {:ok, %{status: status, body: body}} when status in 200..299 -> decode_device_code(body) @@ -361,13 +363,14 @@ defmodule CodexPooler.Upstreams.Auth.CodexAuth do @spec poll_device_authorization(map()) :: CodexAuth.token_response() def poll_device_authorization(state) do body = %{device_auth_id: state["device_auth_id"], user_code: state["user_code"]} + url = CodexAuth.issuer() <> "/api/accounts/deviceauth/token" - case Req.post(CodexAuth.issuer() <> "/api/accounts/deviceauth/token", + case Req.post(url, headers: [{"content-type", "application/json"} | browser_request_headers()], body: CodexPooler.JSON.encode_to_iodata!(body), retry: false, receive_timeout: 30_000, - finch: OutboundHTTP.pool_options() + finch: OutboundHTTP.pool_options_for_url(url) ) do {:ok, %{status: status, body: body}} when status in 200..299 -> case body do @@ -391,6 +394,8 @@ defmodule CodexPooler.Upstreams.Auth.CodexAuth do end defp request_tokens_for_authorization_code(code, verifier, redirect_uri) do + url = CodexAuth.issuer() <> "/oauth/token" + form = [ grant_type: "authorization_code", code: code, @@ -399,12 +404,12 @@ defmodule CodexPooler.Upstreams.Auth.CodexAuth do code_verifier: verifier ] - case Req.post(CodexAuth.issuer() <> "/oauth/token", + case Req.post(url, headers: browser_request_headers(), form: form, retry: false, receive_timeout: 30_000, - finch: OutboundHTTP.pool_options() + finch: OutboundHTTP.pool_options_for_url(url) ) do {:ok, %{status: status, body: body}} when status in 200..299 -> decode_authorization_code_token_response(body) @@ -441,7 +446,7 @@ defmodule CodexPooler.Upstreams.Auth.CodexAuth do form: form, retry: false, receive_timeout: receive_timeout, - finch: OutboundHTTP.pool_options() + finch: OutboundHTTP.pool_options_for_url(token_url) ) do {:ok, %{status: status, body: body}} when status in 200..299 -> decode_refresh_token_response(body) diff --git a/lib/codex_pooler/upstreams/reconciliation/saved_reset_usage_enrichment.ex b/lib/codex_pooler/upstreams/reconciliation/saved_reset_usage_enrichment.ex index 03983ddad..792150072 100644 --- a/lib/codex_pooler/upstreams/reconciliation/saved_reset_usage_enrichment.ex +++ b/lib/codex_pooler/upstreams/reconciliation/saved_reset_usage_enrichment.ex @@ -85,7 +85,7 @@ defmodule CodexPooler.Upstreams.Reconciliation.SavedResetUsageEnrichment do into: &collect_bounded_body/2, retry: false, receive_timeout: timeout, - finch: OutboundHTTP.pool_options() + finch: OutboundHTTP.pool_options_for_url(url) ) do {:ok, %Req.Response{status: status} = response} when status in 200..299 -> handle_successful_reset_credits_response(url, response, observed_at) diff --git a/lib/codex_pooler/upstreams/reconciliation/usage_probe.ex b/lib/codex_pooler/upstreams/reconciliation/usage_probe.ex index 4e2215aba..928d38aa9 100644 --- a/lib/codex_pooler/upstreams/reconciliation/usage_probe.ex +++ b/lib/codex_pooler/upstreams/reconciliation/usage_probe.ex @@ -316,7 +316,7 @@ defmodule CodexPooler.Upstreams.Reconciliation.UsageProbe do headers: CloudflareCookies.request_headers(url, headers), retry: false, receive_timeout: timeout, - finch: OutboundHTTP.pool_options(), + finch: OutboundHTTP.pool_options_for_url(url), decode_body: false ) |> decode_usage_response() diff --git a/lib/codex_pooler/upstreams/saved_resets/redemption.ex b/lib/codex_pooler/upstreams/saved_resets/redemption.ex index ba0efb404..d2caad4e6 100644 --- a/lib/codex_pooler/upstreams/saved_resets/redemption.ex +++ b/lib/codex_pooler/upstreams/saved_resets/redemption.ex @@ -703,7 +703,7 @@ defmodule CodexPooler.Upstreams.SavedResetRedemption do ), retry: false, receive_timeout: recovery.receive_timeout, - finch: OutboundHTTP.pool_options() + finch: OutboundHTTP.pool_options_for_url(recovery.list_url) ) |> store_cloudflare_cookies(recovery.list_url) do {:ok, %{status: status, body: body}} when status in 200..299 and is_map(body) -> @@ -1012,7 +1012,7 @@ defmodule CodexPooler.Upstreams.SavedResetRedemption do body: CodexPooler.JSON.encode_to_iodata!(body), retry: false, receive_timeout: recovery.receive_timeout, - finch: OutboundHTTP.pool_options() + finch: OutboundHTTP.pool_options_for_url(recovery.consume_url) ) |> store_cloudflare_cookies(recovery.consume_url) do {:ok, %{status: status, body: response_body}} -> @@ -2282,7 +2282,7 @@ defmodule CodexPooler.Upstreams.SavedResetRedemption do ), retry: false, receive_timeout: receive_timeout, - finch: OutboundHTTP.pool_options() + finch: OutboundHTTP.pool_options_for_url(url) ) |> store_cloudflare_cookies(url) do {:ok, %{status: status, body: body}} when status in 200..299 and is_map(body) -> @@ -2385,7 +2385,7 @@ defmodule CodexPooler.Upstreams.SavedResetRedemption do body: CodexPooler.JSON.encode_to_iodata!(body), retry: false, receive_timeout: reserved_claim.receive_timeout, - finch: OutboundHTTP.pool_options() + finch: OutboundHTTP.pool_options_for_url(url) ) |> store_cloudflare_cookies(url) do {:ok, %{status: status, body: response_body}} -> diff --git a/scripts/self-host/generate-env.sh b/scripts/self-host/generate-env.sh index 1f94a8752..02dfc8fad 100755 --- a/scripts/self-host/generate-env.sh +++ b/scripts/self-host/generate-env.sh @@ -25,6 +25,9 @@ cat > "$target" < OutboundHTTP.pool_options(-1) end end + test "parse_proxy_url!/1 accepts an HTTP proxy and optional Basic credentials" do + assert OutboundHTTP.parse_proxy_url!(nil) == [] + assert OutboundHTTP.parse_proxy_url!("") == [] + + assert OutboundHTTP.parse_proxy_url!("http://proxy.example.com:3128") == + [proxy: {:http, "proxy.example.com", 3128, []}] + + assert OutboundHTTP.parse_proxy_url!("http://user:p%40ss@proxy.example.com/") == + [ + proxy: {:http, "proxy.example.com", 80, []}, + proxy_headers: [{"proxy-authorization", "Basic " <> Base.encode64("user:p@ss")}] + ] + end + + test "parse_proxy_url!/1 rejects unsupported or ambiguous proxy URLs" do + for value <- [ + "https://proxy.example.com", + "socks5://proxy.example.com:1080", + "http://proxy.example.com/path", + "http://proxy.example.com?mode=test", + "http://proxy.example.com:65536", + "proxy.example.com:3128" + ] do + assert_raise ArgumentError, fn -> OutboundHTTP.parse_proxy_url!(value) end + end + end + + test "pool options select proxies by target scheme and merge connection options" do + Application.put_env(:codex_pooler, OutboundHTTP, + proxy_config: %{ + http: [proxy: {:http, "http-proxy.example.com", 8080, []}], + https: [ + proxy: {:http, "proxy.example.com", 3128, []}, + proxy_headers: [{"proxy-authorization", "Basic dXNlcjpwYXNz"}] + ], + no_proxy: [] + } + ) + + assert OutboundHTTP.pool_options_for_url( + "https://api.example.com", + 30_000, + transport_opts: [timeout: 2_000] + ) == [ + conn_opts: [ + proxy: + {:http, "proxy.example.com", 3128, + [transport_opts: [timeout: 2_000]]}, + proxy_headers: [{"proxy-authorization", "Basic dXNlcjpwYXNz"}], + transport_opts: [timeout: 2_000] + ], + conn_max_idle_time: 30_000 + ] + + assert OutboundHTTP.proxy_options_for_url("http://api.example.com") == + [proxy: {:http, "http-proxy.example.com", 8080, []}] + end + + test "no_proxy bypasses domains, ports, IPs, CIDRs, and all hosts" do + proxy = [proxy: {:http, "proxy.example.com", 3128, []}] + + Application.put_env(:codex_pooler, OutboundHTTP, + proxy_config: %{ + http: proxy, + https: proxy, + no_proxy: [ + "localhost", + ".example.com", + "api.internal:8443", + "127.0.0.1", + "10.0.0.0/8", + "2001:db8::/32" + ] + } + ) + + assert OutboundHTTP.proxy_options_for_url("http://localhost/health") == [] + assert OutboundHTTP.proxy_options_for_url("https://example.com/v1") == [] + assert OutboundHTTP.proxy_options_for_url("https://sub.example.com/v1") == [] + assert OutboundHTTP.proxy_options_for_url("https://api.internal:8443/v1") == [] + assert OutboundHTTP.proxy_options_for_url("http://127.0.0.1/v1") == [] + assert OutboundHTTP.proxy_options_for_url("http://10.23.45.67/v1") == [] + assert OutboundHTTP.proxy_options_for_url("http://[2001:db8::1234]/v1") == [] + + assert OutboundHTTP.proxy_options_for_url("https://api.internal/v1") == proxy + assert OutboundHTTP.proxy_options_for_url("https://notexample.com/v1") == proxy + + Application.put_env(:codex_pooler, OutboundHTTP, + proxy_config: %{http: proxy, https: proxy, no_proxy: ["*"]} + ) + + assert OutboundHTTP.proxy_options_for_url("https://any.example/v1") == [] + end + + test "Req sends outbound HTTP through the configured proxy" do + {:ok, proxy} = FakeUpstream.start_link({:raw_body, 204, "", []}) + on_exit(fn -> FakeUpstream.stop(proxy) end) + proxy_uri = URI.parse(FakeUpstream.url(proxy)) + + Application.put_env(:codex_pooler, OutboundHTTP, + conn_max_idle_time_ms: 45_000, + proxy_config: %{ + http: [proxy: {:http, proxy_uri.host, proxy_uri.port, []}], + https: [], + no_proxy: [] + } + ) + + url = "http://unresolvable.invalid/proxy-check" + + assert {:ok, %Req.Response{status: 204}} = + Req.get(url, + retry: false, + finch: OutboundHTTP.pool_options_for_url(url) + ) + + assert FakeUpstream.count(proxy) == 1 + end + # Req 0.7.4 hashes the complete `finch:` pool option tuple into one Finch # instance under `Req.FinchSupervisor`; `pool_timeout`, `receive_timeout`, # `request_timeout`, and `pool_strategy` are per-request options outside the @@ -175,7 +294,7 @@ defmodule CodexPooler.Platform.OutboundHTTPTest do defp plain_request!(url) do assert {:ok, %Req.Response{status: 304}} = - Req.get(url: url, retry: false, finch: OutboundHTTP.pool_options()) + Req.get(url: url, retry: false, finch: OutboundHTTP.pool_options_for_url(url)) end defp feed_request!(url) do @@ -183,7 +302,7 @@ defmodule CodexPooler.Platform.OutboundHTTPTest do end defp gateway_request!(url) do - options = TransportEnvelope.req_timeout_options(TimeoutConfig.build([])) + options = TransportEnvelope.req_timeout_options(TimeoutConfig.build([]), url) assert {:ok, %Req.Response{status: 304}} = Req.get(url, [retry: false] ++ options) end diff --git a/test/codex_pooler/runtime_config_test.exs b/test/codex_pooler/runtime_config_test.exs index 1e4dd8352..87330d85c 100644 --- a/test/codex_pooler/runtime_config_test.exs +++ b/test/codex_pooler/runtime_config_test.exs @@ -128,6 +128,45 @@ defmodule CodexPooler.RuntimeConfigTest do end) end + test "standard proxy environment variables are parsed without retaining credentials in errors" do + env = + @required_env + |> Map.put("http_proxy", "http://http-proxy.example.com:8080") + |> Map.put("https_proxy", "http://user:p%40ss@proxy.example.com:3128") + |> Map.put("no_proxy", "localhost,.example.com") + |> Map.put("HTTP_PROXY", "http://ignored.example.com:9000") + |> Map.put("HTTPS_PROXY", "http://ignored.example.com:9001") + |> Map.put("NO_PROXY", "ignored.example.com") + + with_env(env, fn -> + config = Config.Reader.read!("config/runtime.exs", env: :prod) + + assert config[:codex_pooler][CodexPooler.Platform.OutboundHTTP][:proxy_config] == %{ + http: [proxy: {:http, "http-proxy.example.com", 8080, []}], + https: [ + proxy: {:http, "proxy.example.com", 3128, []}, + proxy_headers: [ + {"proxy-authorization", "Basic " <> Base.encode64("user:p@ss")} + ] + ], + no_proxy: ["localhost", ".example.com"] + } + end) + + invalid_proxy = "https://secret:password@proxy.example.com" + env = Map.put(@required_env, "https_proxy", invalid_proxy) + + with_env(env, fn -> + error = + assert_raise ArgumentError, fn -> + Config.Reader.read!("config/runtime.exs", env: :prod) + end + + refute Exception.message(error) =~ invalid_proxy + refute Exception.message(error) =~ "password" + end) + end + defp with_env(env, fun) do previous = Map.new(env, fn {key, _value} -> {key, System.get_env(key)} end)
CODEX_POOLER_HTTP_PORT Local host port, default 4000
http_proxy / HTTP_PROXYOptional proxy for HTTP and WS targets; lowercase takes precedence
https_proxy / HTTPS_PROXYOptional HTTP CONNECT proxy for HTTPS and WSS targets; lowercase takes precedence
no_proxy / NO_PROXYOptional comma-separated proxy bypass list; lowercase takes precedence
DATABASE_URL Postgres connection URL