| 
15 | 15 | # specific language governing permissions and limitations  | 
16 | 16 | # under the License.  | 
17 | 17 | 
 
  | 
 | 18 | +import base64  | 
18 | 19 | import os  | 
19 | 20 | from unittest.mock import patch  | 
20 | 21 | from urllib import parse  | 
@@ -544,3 +545,58 @@ def test_connection_manager_with_custom_args_via_client_config():  | 
544 | 545 |     assert isinstance(conn, PoolManager)  | 
545 | 546 |     assert conn.connection_pool_kw["retries"] == retries  | 
546 | 547 |     assert conn.connection_pool_kw["timeout"] == timeout  | 
 | 548 | + | 
 | 549 | + | 
 | 550 | +def test_proxy_auth_with_special_characters_url_encoded():  | 
 | 551 | +    proxy_url = "http://user:passw%[email protected]:8080"  | 
 | 552 | +    client_config = ClientConfig(  | 
 | 553 | +        remote_server_addr="http://localhost:4444",  | 
 | 554 | +        keep_alive=False,  | 
 | 555 | +        proxy=Proxy({"proxyType": ProxyType.MANUAL, "httpProxy": proxy_url}),  | 
 | 556 | +    )  | 
 | 557 | +    remote_connection = RemoteConnection(client_config=client_config)  | 
 | 558 | + | 
 | 559 | +    proxy_without_auth, basic_auth = remote_connection._separate_http_proxy_auth()  | 
 | 560 | + | 
 | 561 | +    assert proxy_without_auth == "http://proxy.example.com:8080"  | 
 | 562 | +    assert basic_auth == "user:passw%23rd"  # Still URL-encoded  | 
 | 563 | + | 
 | 564 | +    conn = remote_connection._get_connection_manager()  | 
 | 565 | +    assert isinstance(conn, ProxyManager)  | 
 | 566 | + | 
 | 567 | +    expected_auth = base64.b64encode("user:passw#rd".encode()).decode()  # Decoded password  | 
 | 568 | +    expected_headers = make_headers(proxy_basic_auth="user:passw#rd")  # Unquoted password  | 
 | 569 | + | 
 | 570 | +    assert conn.proxy_headers == expected_headers  | 
 | 571 | +    assert conn.proxy_headers["proxy-authorization"] == f"Basic {expected_auth}"  | 
 | 572 | + | 
 | 573 | + | 
 | 574 | +def test_proxy_auth_with_multiple_special_characters():  | 
 | 575 | +    test_cases = [  | 
 | 576 | +        ("passw%23rd", "passw#rd"),  # # character  | 
 | 577 | +        ("passw%40rd", "passw@rd"),  # @ character  | 
 | 578 | +        ("passw%26rd", "passw&rd"),  # & character  | 
 | 579 | +        ("passw%3Drd", "passw=rd"),  # = character  | 
 | 580 | +        ("passw%2Brd", "passw+rd"),  # + character  | 
 | 581 | +        ("passw%20rd", "passw rd"),  # space character  | 
 | 582 | +        ("passw%21%40%23%24", "passw!@#$"),  # Multiple special chars  | 
 | 583 | +    ]  | 
 | 584 | + | 
 | 585 | +    for encoded_password, decoded_password in test_cases:  | 
 | 586 | +        proxy_url = f"http://testuser:{encoded_password}@proxy.example.com:8080"  | 
 | 587 | +        client_config = ClientConfig(  | 
 | 588 | +            remote_server_addr="http://localhost:4444",  | 
 | 589 | +            keep_alive=False,  | 
 | 590 | +            proxy=Proxy({"proxyType": ProxyType.MANUAL, "httpProxy": proxy_url}),  | 
 | 591 | +        )  | 
 | 592 | +        remote_connection = RemoteConnection(client_config=client_config)  | 
 | 593 | + | 
 | 594 | +        proxy_without_auth, basic_auth = remote_connection._separate_http_proxy_auth()  | 
 | 595 | +        assert basic_auth == f"testuser:{encoded_password}"  | 
 | 596 | + | 
 | 597 | +        conn = remote_connection._get_connection_manager()  | 
 | 598 | +        expected_auth = base64.b64encode(f"testuser:{decoded_password}".encode()).decode()  | 
 | 599 | +        expected_headers = make_headers(proxy_basic_auth=f"testuser:{decoded_password}")  | 
 | 600 | + | 
 | 601 | +        assert conn.proxy_headers == expected_headers  | 
 | 602 | +        assert conn.proxy_headers["proxy-authorization"] == f"Basic {expected_auth}"  | 
0 commit comments