|
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