split_and_join_proxy_url.py

from urllib3.util import parse_url, Url


proxy_urls = [
    (
        "http://user:pass@host.com:9090",
        {
            "scheme": "http",
            "url": "host.com",
            "username": "user",
            "port": 9090,
            "password": "pass",
        },
        "http://host.com:9090",
    ),
    (
        "https://user:pass@host.com:9090",
        {
            "scheme": "https",
            "url": "host.com",
            "username": "user",
            "password": "pass",
            "port": 9090,
        },
        "https://host.com:9090",
    ),
    (
        "http://user@192.168.0.1:9090",
        {
            "scheme": "http",
            "url": "192.168.0.1",
            "username": "user",
            "port": 9090,
            "password": None,
        },
        "http://192.168.0.1:9090",
    ),
    (
        "https://user@192.168.0.1:9090",
        {
            "scheme": "https",
            "url": "192.168.0.1",
            "username": "user",
            "port": 9090,
            "password": None,
        },
        "https://192.168.0.1:9090",
    ),
    (
        "http://host.com:9090",
        {
            "scheme": "http",
            "url": "host.com",
            "username": None,
            "port": 9090,
            "password": None,
        },
        "http://host.com:9090",
    ),
    (
        "https://host.com:9090",
        {
            "scheme": "https",
            "url": "host.com",
            "username": None,
            "port": 9090,
            "password": None,
        },
        "https://host.com:9090",
    ),
    (
        "http://host.com/foo/bar/",
        {
            "scheme": "http",
            "url": "host.com/foo/bar/",
            "username": None,
            "port": None,
            "password": None,
        },
        "http://host.com/foo/bar/",
    ),
    (
        "https://host.com/foo/bar/",
        {
            "scheme": "https",
            "url": "host.com/foo/bar/",
            "username": None,
            "port": None,
            "password": None,
        },
        "https://host.com/foo/bar/",
    ),
]


def get_parsed_url(parsed: Url, auth_items: list[str]):
    """
    parsed: Url instance
    auth_items: Optional list of str ["user": "password"]
    returns: str e.g: http://user:pass@foo.bar:9999
    """
    return Url(
        scheme=parsed.scheme,
        auth=auth_items and ":".join(auth_items) or None,
        host=parsed.host,
        port=parsed.port,
        path=parsed.path,
        query=parsed.query,
        fragment=parsed.fragment,
    ).url


def strip_auth_from_url(url: str):
    """
    Gets a full proxy address and strips its auth
    # url: http://bruno:1234@foo.bar:9999/zaz/traz/?a#e;w
    # returns: tuple, e.g: "http://foo.bar:9999/zaz/traz/?a#e;w", bruno, 1234
    """
    parsed = parse_url(url)
    auth = parsed.auth and parsed.auth.split(":")
    username = auth and auth[0]
    password = auth and len(auth) > 1 and auth[1] or None
    return (
        get_parsed_url(parsed, None),
        username,
        password,
    )


def join_proxy_url(address: str, username: str, password: str):
    """
    Gets a splitted address, username, password
    returns: Joined URL forming proxy_url
    """
    # http://username:password@address
    parsed = parse_url(address)
    auth_items = []
    if username:
        auth_items.append(str(username))
    if password:
        auth_items.append(str(password))
    return get_parsed_url(parsed, auth_items)


for proxy_url, data, expected in proxy_urls:

    print("On database there is proxy_url", proxy_url)
    # Represent url and username in different fields
    address, username, password = strip_auth_from_url(proxy_url)
    assert address == expected, (address, expected)
    assert data["username"] == username, (data, username)
    assert data["password"] == password, (data, password)
    print("proxy_address got:", address)
    print("proxy_username got:", username)
    print("proxy_password (hidden):", password)

    # Rejoin a splitted data into a single URL
    print("API received", address, username, password)
    joined = join_proxy_url(address, data["username"], data["password"])
    assert joined == proxy_url, (data, joined, proxy_url)
    print("saved to database:", joined)
    print("_" * 30)