test_auth.py

import httplib2
import pytest
import tests
from six.moves import urllib


def test_credentials():
    c = httplib2.Credentials()
    c.add("joe", "password")
    assert tuple(c.iter("bitworking.org"))[0] == ("joe", "password")
    assert tuple(c.iter(""))[0] == ("joe", "password")
    c.add("fred", "password2", "wellformedweb.org")
    assert tuple(c.iter("bitworking.org"))[0] == ("joe", "password")
    assert len(tuple(c.iter("bitworking.org"))) == 1
    assert len(tuple(c.iter("wellformedweb.org"))) == 2
    assert ("fred", "password2") in tuple(c.iter("wellformedweb.org"))
    c.clear()
    assert len(tuple(c.iter("bitworking.org"))) == 0
    c.add("fred", "password2", "wellformedweb.org")
    assert ("fred", "password2") in tuple(c.iter("wellformedweb.org"))
    assert len(tuple(c.iter("bitworking.org"))) == 0
    assert len(tuple(c.iter(""))) == 0


def test_basic():
    # Test Basic Authentication
    http = httplib2.Http()
    password = tests.gen_password()
    handler = tests.http_reflect_with_auth(
        allow_scheme="basic", allow_credentials=(("joe", password),)
    )
    with tests.server_request(handler, request_count=3) as uri:
        response, content = http.request(uri, "GET")
        assert response.status == 401
        http.add_credentials("joe", password)
        response, content = http.request(uri, "GET")
        assert response.status == 200


def test_basic_for_domain():
    # Test Basic Authentication
    http = httplib2.Http()
    password = tests.gen_password()
    handler = tests.http_reflect_with_auth(
        allow_scheme="basic", allow_credentials=(("joe", password),)
    )
    with tests.server_request(handler, request_count=4) as uri:
        response, content = http.request(uri, "GET")
        assert response.status == 401
        http.add_credentials("joe", password, "example.org")
        response, content = http.request(uri, "GET")
        assert response.status == 401
        domain = urllib.parse.urlparse(uri)[1]
        http.add_credentials("joe", password, domain)
        response, content = http.request(uri, "GET")
        assert response.status == 200


def test_basic_two_credentials():
    # Test Basic Authentication with multiple sets of credentials
    http = httplib2.Http()
    password1 = tests.gen_password()
    password2 = tests.gen_password()
    allowed = [("joe", password1)]  # exploit shared mutable list
    handler = tests.http_reflect_with_auth(
        allow_scheme="basic", allow_credentials=allowed
    )
    with tests.server_request(handler, request_count=7) as uri:
        http.add_credentials("fred", password2)
        response, content = http.request(uri, "GET")
        assert response.status == 401
        http.add_credentials("joe", password1)
        response, content = http.request(uri, "GET")
        assert response.status == 200
        allowed[0] = ("fred", password2)
        response, content = http.request(uri, "GET")
        assert response.status == 200


def test_digest():
    # Test that we support Digest Authentication
    http = httplib2.Http()
    password = tests.gen_password()
    handler = tests.http_reflect_with_auth(
        allow_scheme="digest", allow_credentials=(("joe", password),)
    )
    with tests.server_request(handler, request_count=3) as uri:
        response, content = http.request(uri, "GET")
        assert response.status == 401
        http.add_credentials("joe", password)
        response, content = http.request(uri, "GET")
        assert response.status == 200, content.decode()


def test_digest_next_nonce_nc():
    # Test that if the server sets nextnonce that we reset
    # the nonce count back to 1
    http = httplib2.Http()
    password = tests.gen_password()
    grenew_nonce = [None]
    handler = tests.http_reflect_with_auth(
        allow_scheme="digest",
        allow_credentials=(("joe", password),),
        out_renew_nonce=grenew_nonce,
    )
    with tests.server_request(handler, request_count=5) as uri:
        http.add_credentials("joe", password)
        response1, _ = http.request(uri, "GET")
        info = httplib2._parse_www_authenticate(response1, "authentication-info")
        assert response1.status == 200
        assert info.get("digest", {}).get("nc") == "00000001", info
        assert not info.get("digest", {}).get("nextnonce"), info
        response2, _ = http.request(uri, "GET")
        info2 = httplib2._parse_www_authenticate(response2, "authentication-info")
        assert info2.get("digest", {}).get("nc") == "00000002", info2
        grenew_nonce[0]()
        response3, content = http.request(uri, "GET")
        info3 = httplib2._parse_www_authenticate(response3, "authentication-info")
        assert response3.status == 200
        assert info3.get("digest", {}).get("nc") == "00000001", info3


def test_digest_auth_stale():
    # Test that we can handle a nonce becoming stale
    http = httplib2.Http()
    password = tests.gen_password()
    grenew_nonce = [None]
    requests = []
    handler = tests.http_reflect_with_auth(
        allow_scheme="digest",
        allow_credentials=(("joe", password),),
        out_renew_nonce=grenew_nonce,
        out_requests=requests,
    )
    with tests.server_request(handler, request_count=4) as uri:
        http.add_credentials("joe", password)
        response, _ = http.request(uri, "GET")
        assert response.status == 200
        info = httplib2._parse_www_authenticate(
            requests[0][1].headers, "www-authenticate"
        )
        grenew_nonce[0]()
        response, _ = http.request(uri, "GET")
        assert response.status == 200
        assert not response.fromcache
        assert getattr(response, "_stale_digest", False)
        info2 = httplib2._parse_www_authenticate(
            requests[2][1].headers, "www-authenticate"
        )
        nonce1 = info.get("digest", {}).get("nonce", "")
        nonce2 = info2.get("digest", {}).get("nonce", "")
        assert nonce1 != ""
        assert nonce2 != ""
        assert nonce1 != nonce2, (nonce1, nonce2)


@pytest.mark.parametrize(
    "data",
    (
        ({}, {}),
        ({"www-authenticate": ""}, {}),
        (
            {
                "www-authenticate": 'Test realm="test realm" , foo=foo ,bar="bar", baz=baz,qux=qux'
            },
            {
                "test": {
                    "realm": "test realm",
                    "foo": "foo",
                    "bar": "bar",
                    "baz": "baz",
                    "qux": "qux",
                }
            },
        ),
        (
            {"www-authenticate": 'T*!%#st realm=to*!%#en, to*!%#en="quoted string"'},
            {"t*!%#st": {"realm": "to*!%#en", "to*!%#en": "quoted string"}},
        ),
        (
            {"www-authenticate": 'Test realm="a \\"test\\" realm"'},
            {"test": {"realm": 'a "test" realm'}},
        ),
        ({"www-authenticate": 'Basic realm="me"'}, {"basic": {"realm": "me"}}),
        (
            {"www-authenticate": 'Basic realm="me", algorithm="MD5"'},
            {"basic": {"realm": "me", "algorithm": "MD5"}},
        ),
        (
            {"www-authenticate": 'Basic realm="me", algorithm=MD5'},
            {"basic": {"realm": "me", "algorithm": "MD5"}},
        ),
        (
            {"www-authenticate": 'Basic realm="me",other="fred" '},
            {"basic": {"realm": "me", "other": "fred"}},
        ),
        ({"www-authenticate": 'Basic REAlm="me" '}, {"basic": {"realm": "me"}}),
        (
            {
                "www-authenticate": 'Digest realm="digest1", qop="auth,auth-int", nonce="7102dd2", opaque="e9517f"'
            },
            {
                "digest": {
                    "realm": "digest1",
                    "qop": "auth,auth-int",
                    "nonce": "7102dd2",
                    "opaque": "e9517f",
                }
            },
        ),
        # multiple schema choice
        (
            {
                "www-authenticate": 'Digest realm="multi-d", nonce="8b11d0f6", opaque="cc069c" Basic realm="multi-b" '
            },
            {
                "digest": {"realm": "multi-d", "nonce": "8b11d0f6", "opaque": "cc069c"},
                "basic": {"realm": "multi-b"},
            },
        ),
        # FIXME
        # comma between schemas (glue for multiple headers with same name)
        # ({'www-authenticate': 'Digest realm="2-comma-d", qop="auth-int", nonce="c0c8ff1", Basic realm="2-comma-b"'},
        #  {'digest': {'realm': '2-comma-d', 'qop': 'auth-int', 'nonce': 'c0c8ff1'},
        #   'basic': {'realm': '2-comma-b'}}),
        # FIXME
        # comma between schemas + WSSE (glue for multiple headers with same name)
        # ({'www-authenticate': 'Digest realm="com3d", Basic realm="com3b", WSSE realm="com3w", profile="token"'},
        #  {'digest': {'realm': 'com3d'}, 'basic': {'realm': 'com3b'}, 'wsse': {'realm': 'com3w', profile': 'token'}}),
        # FIXME
        # multiple syntax figures
        # ({'www-authenticate':
        #     'Digest realm="brig", qop \t=\t"\tauth,auth-int", nonce="(*)&^&$%#",opaque="5ccc"' +
        #     ', Basic REAlm="zoo", WSSE realm="very", profile="UsernameToken"'},
        #  {'digest': {'realm': 'brig', 'qop': 'auth,auth-int', 'nonce': '(*)&^&$%#', 'opaque': '5ccc'},
        #   'basic': {'realm': 'zoo'},
        #   'wsse': {'realm': 'very', 'profile': 'UsernameToken'}}),
        # more quote combos
        (
            {
                "www-authenticate": 'Digest realm="myrealm", nonce="KBAA=3", algorithm=MD5, qop="auth", stale=true'
            },
            {
                "digest": {
                    "realm": "myrealm",
                    "nonce": "KBAA=3",
                    "algorithm": "MD5",
                    "qop": "auth",
                    "stale": "true",
                }
            },
        ),
    ),
    ids=lambda data: str(data[0]),
)
@pytest.mark.parametrize("strict", (True, False), ids=("strict", "relax"))
def test_parse_www_authenticate_correct(data, strict):
    headers, info = data
    # FIXME: move strict to parse argument
    httplib2.USE_WWW_AUTH_STRICT_PARSING = strict
    try:
        assert httplib2._parse_www_authenticate(headers) == info
    finally:
        httplib2.USE_WWW_AUTH_STRICT_PARSING = 0


def test_parse_www_authenticate_malformed():
    # TODO: test (and fix) header value 'barbqwnbm-bb...:asd' leads to dead loop
    with tests.assert_raises(httplib2.MalformedHeader):
        httplib2._parse_www_authenticate(
            {
                "www-authenticate": 'OAuth "Facebook Platform" "invalid_token" "Invalid OAuth access token."'
            }
        )


def test_digest_object():
    credentials = ("joe", "password")
    host = None
    request_uri = "/test/digest/"
    headers = {}
    response = {
        "www-authenticate": 'Digest realm="myrealm", nonce="KBAA=35", algorithm=MD5, qop="auth"'
    }
    content = b""

    d = httplib2.DigestAuthentication(
        credentials, host, request_uri, headers, response, content, None
    )
    d.request("GET", request_uri, headers, content, cnonce="33033375ec278a46")
    our_request = "authorization: " + headers["authorization"]
    working_request = (
        'authorization: Digest username="joe", realm="myrealm", '
        'nonce="KBAA=35", uri="/test/digest/"'
        + ', algorithm=MD5, response="de6d4a123b80801d0e94550411b6283f", '
        'qop=auth, nc=00000001, cnonce="33033375ec278a46"'
    )
    assert our_request == working_request


def test_digest_object_with_opaque():
    credentials = ("joe", "password")
    host = None
    request_uri = "/digest/opaque/"
    headers = {}
    response = {
        "www-authenticate": 'Digest realm="myrealm", nonce="30352fd", algorithm=MD5, '
        'qop="auth", opaque="atestopaque"'
    }
    content = ""

    d = httplib2.DigestAuthentication(
        credentials, host, request_uri, headers, response, content, None
    )
    d.request("GET", request_uri, headers, content, cnonce="5ec2")
    our_request = "authorization: " + headers["authorization"]
    working_request = (
        'authorization: Digest username="joe", realm="myrealm", '
        'nonce="30352fd", uri="/digest/opaque/", algorithm=MD5'
        + ', response="a1fab43041f8f3789a447f48018bee48", qop=auth, nc=00000001, '
        'cnonce="5ec2", opaque="atestopaque"'
    )
    assert our_request == working_request


def test_digest_object_stale():
    credentials = ("joe", "password")
    host = None
    request_uri = "/digest/stale/"
    headers = {}
    response = httplib2.Response({})
    response["www-authenticate"] = (
        'Digest realm="myrealm", nonce="bd669f", '
        'algorithm=MD5, qop="auth", stale=true'
    )
    response.status = 401
    content = b""
    d = httplib2.DigestAuthentication(
        credentials, host, request_uri, headers, response, content, None
    )
    # Returns true to force a retry
    assert d.response(response, content)


def test_digest_object_auth_info():
    credentials = ("joe", "password")
    host = None
    request_uri = "/digest/nextnonce/"
    headers = {}
    response = httplib2.Response({})
    response["www-authenticate"] = (
        'Digest realm="myrealm", nonce="barney", '
        'algorithm=MD5, qop="auth", stale=true'
    )
    response["authentication-info"] = 'nextnonce="fred"'
    content = b""
    d = httplib2.DigestAuthentication(
        credentials, host, request_uri, headers, response, content, None
    )
    # Returns true to force a retry
    assert not d.response(response, content)
    assert d.challenge["nonce"] == "fred"
    assert d.challenge["nc"] == 1


def test_wsse_algorithm():
    digest = httplib2._wsse_username_token("d36e316282959a9ed4c89851497a717f", "2003-12-15T14:43:07Z", "taadtaadpstcsm")
    expected = "quR/EWLAV4xLf9Zqyw4pDmfV9OY="
    assert expected == digest


def test_wsse_invalid():
    http = httplib2.Http()
    username = "user294"
    password = tests.gen_password()
    grenew_nonce = [None]
    requests = []
    handler = tests.http_reflect_with_auth(
        allow_scheme="wsse",
        allow_credentials=((username, password),),
        out_renew_nonce=grenew_nonce,
        out_requests=requests,
    )
    http.add_credentials(username, "wrong" + password)
    with tests.server_request(handler, request_count=2) as uri:
        response, _ = http.request(uri)
        assert requests[0][1].status == 401
        assert requests[1][0].headers["authorization"] == 'WSSE profile="UsernameToken"'
        assert requests[1][1].status == 401
        assert response.status == 401


def test_wsse_ok():
    http = httplib2.Http()
    username = "user314"
    password = tests.gen_password()
    grenew_nonce = [None]
    requests = []
    handler = tests.http_reflect_with_auth(
        allow_scheme="wsse",
        allow_credentials=((username, password),),
        out_renew_nonce=grenew_nonce,
        out_requests=requests,
    )
    http.add_credentials(username, password)
    with tests.server_request(handler, request_count=2) as uri:
        response, _ = http.request(uri)
        assert requests[0][1].status == 401
        assert requests[1][0].headers["authorization"] == 'WSSE profile="UsernameToken"'
        assert response.status == 200