Merge pull request #2327 from omarbounawarapy/bugfix/http-proxy-credentials-encoding

Percent-encode proxy credentials in the HTTP crawler proxy URL

Author: ntohidi

crawl4ai/async_configs.py

@@ -3,6 +3,7 @@
 import importlib
 import os
 import warnings
+from urllib.parse import unquote
 import requests
 from .config import (
     DEFAULT_PROVIDER,
@@ -711,8 +712,8 @@ def from_string(proxy_str: str) -> "ProxyConfig":
                 username, password = credentials.split(":", 1)
                 return ProxyConfig(
                     server=f"{protocol}://{server_part}",
-                    username=username,
-                    password=password,
+                    username=unquote(username),
+                    password=unquote(password),
                 )
         # URL without credentials (keep scheme)
         if "://" in s and "@" not in s:

crawl4ai/async_crawler_strategy.py

@@ -28,7 +28,7 @@
 import aiohttp
 import chardet
 from aiohttp.client import ClientTimeout
-from urllib.parse import urlparse
+from urllib.parse import quote, urlparse
 from types import MappingProxyType
 import contextlib
 from functools import partial
@@ -2642,6 +2642,8 @@ def _format_proxy_url(self, proxy_config) -> str:
 
         if username and password:
             # Insert credentials into URL: http://user:pass@host:port
+            username = quote(username, safe='')
+            password = quote(password, safe='')
             if '://' in server:
                 protocol, rest = server.split('://', 1)
                 return f"{protocol}://{username}:{password}@{rest}"

tests/proxy/test_http_proxy_url_encoding.py

@@ -0,0 +1,27 @@
+import pytest
+from yarl import URL
+
+from crawl4ai.async_configs import ProxyConfig
+from crawl4ai.async_crawler_strategy import AsyncHTTPCrawlerStrategy
+
+
+@pytest.mark.parametrize(
+    "password", ["plainpass", "p@ss", "p#ss", "pa/ss", "p?ss", "p%41ss"]
+)
+def test_proxy_url_roundtrips_credentials(password):
+    strategy = AsyncHTTPCrawlerStrategy()
+    config = ProxyConfig(server="http://host:8080", username="us:er", password=password)
+    url = URL(strategy._format_proxy_url(config))
+    assert url.host == "host"
+    assert url.port == 8080
+    assert url.user == "us:er"
+    assert url.password == password
+
+
+def test_encoded_proxy_string_is_sent_once_encoded():
+    strategy = AsyncHTTPCrawlerStrategy()
+    config = ProxyConfig.from_string("http://user:p%40ss@host:8080")
+    assert config.password == "p@ss"
+    url = URL(strategy._format_proxy_url(config))
+    assert url.user == "user"
+    assert url.password == "p@ss"