| 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 = b"quR/EWLAV4xLf9Zqyw4pDmfV9OY=" | |
| assert expected == digest |