"""This is a copy of the DigestAuth logic from httpx auth module with additions for cached auth. Base copied from: https://github.com/encode/httpx/blob/32d37cfdf1697f0dbacbc2233ee34606369be87b/httpx/_auth.py Issue tracking the need for the copy: https://github.com/encode/httpx/issues/1467 """ import hashlib import os import re import time import typing from urllib.request import parse_http_list from httpx import ProtocolError from httpx import Request, Response from httpx import Auth def to_bytes(value: typing.Union[str, bytes], encoding: str = "utf-8") -> bytes: return value.encode(encoding) if isinstance(value, str) else value def to_str(value: typing.Union[str, bytes], encoding: str = "utf-8") -> str: return value if isinstance(value, str) else value.decode(encoding) def unquote(value: str) -> str: return value[1:-1] if value[0] == value[-1] == '"' else value class CachedDigestAuth(Auth): """Digest authentication with a cached challange response. Known issues: * Only a single challenge is every cached, no checks is done for hostname being correct * Nonce counting is not implemented since philips TV's don't seem to require it. """ _ALGORITHM_TO_HASH_FUNCTION: typing.Dict[str, typing.Callable] = { "MD5": hashlib.md5, "MD5-SESS": hashlib.md5, "SHA": hashlib.sha1, "SHA-SESS": hashlib.sha1, "SHA-256": hashlib.sha256, "SHA-256-SESS": hashlib.sha256, "SHA-512": hashlib.sha512, "SHA-512-SESS": hashlib.sha512, } def __init__( self, username: typing.Union[str, bytes], password: typing.Union[str, bytes] ) -> None: self._challenge = None self._username = to_bytes(username) self._password = to_bytes(password) def auth_flow(self, request: Request) -> typing.Generator[Request, Response, None]: if self._challenge: request.headers["Authorization"] = self._build_auth_header(request, self._challenge) response = yield request if response.status_code != 401 or "www-authenticate" not in response.headers: # If the response is not a 401 then we don't # need to build an authenticated request. return for auth_header in response.headers.get_list("www-authenticate"): if auth_header.lower().startswith("digest "): break else: # If the response does not include a 'WWW-Authenticate: Digest ...' # header, then we don't need to build an authenticated request. return self._challenge = self._parse_challenge(request, response, auth_header) request.headers["Authorization"] = self._build_auth_header(request, self._challenge) yield request def _parse_challenge( self, request: Request, response: Response, auth_header: str ) -> "_DigestAuthChallenge": """ Returns a challenge from a Digest WWW-Authenticate header. These take the form of: `Digest realm="realm@host.com",qop="auth,auth-int",nonce="abc",opaque="xyz"` """ scheme, _, fields = auth_header.partition(" ") # This method should only ever have been called with a Digest auth header. assert scheme.lower() == "digest" header_dict: typing.Dict[str, str] = {} for field in parse_http_list(fields): key, value = field.strip().split("=", 1) header_dict[key] = unquote(value) try: realm = header_dict["realm"].encode() nonce = header_dict["nonce"].encode() algorithm = header_dict.get("algorithm", "MD5") opaque = header_dict["opaque"].encode() if "opaque" in header_dict else None qop = header_dict["qop"].encode() if "qop" in header_dict else None return _DigestAuthChallenge( realm=realm, nonce=nonce, algorithm=algorithm, opaque=opaque, qop=qop ) except KeyError as exc: message = "Malformed Digest WWW-Authenticate header" raise ProtocolError(message, request=request) from exc def _build_auth_header( self, request: Request, challenge: "_DigestAuthChallenge" ) -> str: hash_func = self._ALGORITHM_TO_HASH_FUNCTION[challenge.algorithm] def digest(data: bytes) -> bytes: return hash_func(data).hexdigest().encode() A1 = b":".join((self._username, challenge.realm, self._password)) path = request.url.raw_path A2 = b":".join((request.method.encode(), path)) # TODO: implement auth-int HA2 = digest(A2) nonce_count = 1 # TODO: implement nonce counting nc_value = b"%08x" % nonce_count cnonce = self._get_client_nonce(nonce_count, challenge.nonce) HA1 = digest(A1) if challenge.algorithm.lower().endswith("-sess"): HA1 = digest(b":".join((HA1, challenge.nonce, cnonce))) qop = self._resolve_qop(challenge.qop, request=request) if qop is None: digest_data = [HA1, challenge.nonce, HA2] else: digest_data = [challenge.nonce, nc_value, cnonce, qop, HA2] key_digest = b":".join(digest_data) format_args = { "username": self._username, "realm": challenge.realm, "nonce": challenge.nonce, "uri": path, "response": digest(b":".join((HA1, key_digest))), "algorithm": challenge.algorithm.encode(), } if challenge.opaque: format_args["opaque"] = challenge.opaque if qop: format_args["qop"] = b"auth" format_args["nc"] = nc_value format_args["cnonce"] = cnonce return "Digest " + self._get_header_value(format_args) def _get_client_nonce(self, nonce_count: int, nonce: bytes) -> bytes: s = str(nonce_count).encode() s += nonce s += time.ctime().encode() s += os.urandom(8) return hashlib.sha1(s).hexdigest()[:16].encode() def _get_header_value(self, header_fields: typing.Dict[str, bytes]) -> str: NON_QUOTED_FIELDS = ("algorithm", "qop", "nc") QUOTED_TEMPLATE = '{}="{}"' NON_QUOTED_TEMPLATE = "{}={}" header_value = "" for i, (field, value) in enumerate(header_fields.items()): if i > 0: header_value += ", " template = ( QUOTED_TEMPLATE if field not in NON_QUOTED_FIELDS else NON_QUOTED_TEMPLATE ) header_value += template.format(field, to_str(value)) return header_value def _resolve_qop( self, qop: typing.Optional[bytes], request: Request ) -> typing.Optional[bytes]: if qop is None: return None qops = re.split(b", ?", qop) if b"auth" in qops: return b"auth" if qops == [b"auth-int"]: raise NotImplementedError("Digest auth-int support is not yet implemented") message = f'Unexpected qop value "{qop!r}" in digest auth' raise ProtocolError(message, request=request) class _DigestAuthChallenge(typing.NamedTuple): realm: bytes nonce: bytes algorithm: str opaque: typing.Optional[bytes] qop: typing.Optional[bytes]