We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
DigestAuth should raise a clear error if the request cannot replay.
Our DigestAuth implementation cannot work with non-replayable requests.
We ought to raise a nice clear error if `request.stream.is_replayable` is not True.

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `httpx/auth.py`
Content:
```
1 import hashlib
2 import os
3 import re
4 import time
5 import typing
6 from base64 import b64encode
7 from urllib.request import parse_http_list
8 
9 from .exceptions import ProtocolError
10 from .models import Request, Response
11 from .utils import to_bytes, to_str, unquote
12 
13 AuthFlow = typing.Generator[Request, Response, None]
14 
15 AuthTypes = typing.Union[
16     typing.Tuple[typing.Union[str, bytes], typing.Union[str, bytes]],
17     typing.Callable[["Request"], "Request"],
18     "Auth",
19 ]
20 
21 
22 class Auth:
23     """
24     Base class for all authentication schemes.
25     """
26 
27     def __call__(self, request: Request) -> AuthFlow:
28         """
29         Execute the authentication flow.
30 
31         To dispatch a request, `yield` it:
32 
33         ```
34         yield request
35         ```
36 
37         The client will `.send()` the response back into the flow generator. You can
38         access it like so:
39 
40         ```
41         response = yield request
42         ```
43 
44         A `return` (or reaching the end of the generator) will result in the
45         client returning the last response obtained from the server.
46 
47         You can dispatch as many requests as is necessary.
48         """
49         yield request
50 
51 
52 class FunctionAuth(Auth):
53     """
54     Allows the 'auth' argument to be passed as a simple callable function,
55     that takes the request, and returns a new, modified request.
56     """
57 
58     def __init__(self, func: typing.Callable[[Request], Request]) -> None:
59         self.func = func
60 
61     def __call__(self, request: Request) -> AuthFlow:
62         yield self.func(request)
63 
64 
65 class BasicAuth(Auth):
66     """
67     Allows the 'auth' argument to be passed as a (username, password) pair,
68     and uses HTTP Basic authentication.
69     """
70 
71     def __init__(
72         self, username: typing.Union[str, bytes], password: typing.Union[str, bytes]
73     ):
74         self.auth_header = self.build_auth_header(username, password)
75 
76     def __call__(self, request: Request) -> AuthFlow:
77         request.headers["Authorization"] = self.auth_header
78         yield request
79 
80     def build_auth_header(
81         self, username: typing.Union[str, bytes], password: typing.Union[str, bytes]
82     ) -> str:
83         userpass = b":".join((to_bytes(username), to_bytes(password)))
84         token = b64encode(userpass).decode().strip()
85         return f"Basic {token}"
86 
87 
88 class DigestAuth(Auth):
89     ALGORITHM_TO_HASH_FUNCTION: typing.Dict[str, typing.Callable] = {
90         "MD5": hashlib.md5,
91         "MD5-SESS": hashlib.md5,
92         "SHA": hashlib.sha1,
93         "SHA-SESS": hashlib.sha1,
94         "SHA-256": hashlib.sha256,
95         "SHA-256-SESS": hashlib.sha256,
96         "SHA-512": hashlib.sha512,
97         "SHA-512-SESS": hashlib.sha512,
98     }
99 
100     def __init__(
101         self, username: typing.Union[str, bytes], password: typing.Union[str, bytes]
102     ) -> None:
103         self.username = to_bytes(username)
104         self.password = to_bytes(password)
105 
106     def __call__(self, request: Request) -> AuthFlow:
107         response = yield request
108 
109         if response.status_code != 401 or "www-authenticate" not in response.headers:
110             # If the response is not a 401 WWW-Authenticate, then we don't
111             # need to build an authenticated request.
112             return
113 
114         header = response.headers["www-authenticate"]
115         try:
116             challenge = DigestAuthChallenge.from_header(header)
117         except ValueError:
118             raise ProtocolError("Malformed Digest authentication header")
119 
120         request.headers["Authorization"] = self._build_auth_header(request, challenge)
121         yield request
122 
123     def _build_auth_header(
124         self, request: Request, challenge: "DigestAuthChallenge"
125     ) -> str:
126         hash_func = self.ALGORITHM_TO_HASH_FUNCTION[challenge.algorithm]
127 
128         def digest(data: bytes) -> bytes:
129             return hash_func(data).hexdigest().encode()
130 
131         A1 = b":".join((self.username, challenge.realm, self.password))
132 
133         path = request.url.full_path.encode("utf-8")
134         A2 = b":".join((request.method.encode(), path))
135         # TODO: implement auth-int
136         HA2 = digest(A2)
137 
138         nonce_count = 1  # TODO: implement nonce counting
139         nc_value = b"%08x" % nonce_count
140         cnonce = self._get_client_nonce(nonce_count, challenge.nonce)
141 
142         HA1 = digest(A1)
143         if challenge.algorithm.lower().endswith("-sess"):
144             HA1 = digest(b":".join((HA1, challenge.nonce, cnonce)))
145 
146         qop = self._resolve_qop(challenge.qop)
147         if qop is None:
148             digest_data = [HA1, challenge.nonce, HA2]
149         else:
150             digest_data = [challenge.nonce, nc_value, cnonce, qop, HA2]
151         key_digest = b":".join(digest_data)
152 
153         format_args = {
154             "username": self.username,
155             "realm": challenge.realm,
156             "nonce": challenge.nonce,
157             "uri": path,
158             "response": digest(b":".join((HA1, key_digest))),
159             "algorithm": challenge.algorithm.encode(),
160         }
161         if challenge.opaque:
162             format_args["opaque"] = challenge.opaque
163         if qop:
164             format_args["qop"] = b"auth"
165             format_args["nc"] = nc_value
166             format_args["cnonce"] = cnonce
167 
168         return "Digest " + self._get_header_value(format_args)
169 
170     def _get_client_nonce(self, nonce_count: int, nonce: bytes) -> bytes:
171         s = str(nonce_count).encode()
172         s += nonce
173         s += time.ctime().encode()
174         s += os.urandom(8)
175 
176         return hashlib.sha1(s).hexdigest()[:16].encode()
177 
178     def _get_header_value(self, header_fields: typing.Dict[str, bytes]) -> str:
179         NON_QUOTED_FIELDS = ("algorithm", "qop", "nc")
180         QUOTED_TEMPLATE = '{}="{}"'
181         NON_QUOTED_TEMPLATE = "{}={}"
182 
183         header_value = ""
184         for i, (field, value) in enumerate(header_fields.items()):
185             if i > 0:
186                 header_value += ", "
187             template = (
188                 QUOTED_TEMPLATE
189                 if field not in NON_QUOTED_FIELDS
190                 else NON_QUOTED_TEMPLATE
191             )
192             header_value += template.format(field, to_str(value))
193 
194         return header_value
195 
196     def _resolve_qop(self, qop: typing.Optional[bytes]) -> typing.Optional[bytes]:
197         if qop is None:
198             return None
199         qops = re.split(b", ?", qop)
200         if b"auth" in qops:
201             return b"auth"
202 
203         if qops == [b"auth-int"]:
204             raise NotImplementedError("Digest auth-int support is not yet implemented")
205 
206         raise ProtocolError(f'Unexpected qop value "{qop!r}" in digest auth')
207 
208 
209 class DigestAuthChallenge:
210     def __init__(
211         self,
212         realm: bytes,
213         nonce: bytes,
214         algorithm: str = None,
215         opaque: typing.Optional[bytes] = None,
216         qop: typing.Optional[bytes] = None,
217     ) -> None:
218         self.realm = realm
219         self.nonce = nonce
220         self.algorithm = algorithm or "MD5"
221         self.opaque = opaque
222         self.qop = qop
223 
224     @classmethod
225     def from_header(cls, header: str) -> "DigestAuthChallenge":
226         """Returns a challenge from a Digest WWW-Authenticate header.
227         These take the form of:
228         `Digest realm="realm@host.com",qop="auth,auth-int",nonce="abc",opaque="xyz"`
229         """
230         scheme, _, fields = header.partition(" ")
231         if scheme.lower() != "digest":
232             raise ValueError("Header does not start with 'Digest'")
233 
234         header_dict: typing.Dict[str, str] = {}
235         for field in parse_http_list(fields):
236             key, value = field.strip().split("=", 1)
237             header_dict[key] = unquote(value)
238 
239         try:
240             return cls.from_header_dict(header_dict)
241         except KeyError as exc:
242             raise ValueError("Malformed Digest WWW-Authenticate header") from exc
243 
244     @classmethod
245     def from_header_dict(cls, header_dict: dict) -> "DigestAuthChallenge":
246         realm = header_dict["realm"].encode()
247         nonce = header_dict["nonce"].encode()
248         qop = header_dict["qop"].encode() if "qop" in header_dict else None
249         opaque = header_dict["opaque"].encode() if "opaque" in header_dict else None
250         algorithm = header_dict.get("algorithm")
251         return cls(
252             realm=realm, nonce=nonce, qop=qop, opaque=opaque, algorithm=algorithm
253         )
254
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/httpx/auth.py b/httpx/auth.py
--- a/httpx/auth.py
+++ b/httpx/auth.py
@@ -6,7 +6,7 @@
 from base64 import b64encode
 from urllib.request import parse_http_list
 
-from .exceptions import ProtocolError
+from .exceptions import ProtocolError, RequestBodyUnavailable
 from .models import Request, Response
 from .utils import to_bytes, to_str, unquote
 
@@ -104,6 +104,8 @@
         self.password = to_bytes(password)
 
     def __call__(self, request: Request) -> AuthFlow:
+        if not request.stream.can_replay():
+            raise RequestBodyUnavailable("Request body is no longer available.")
         response = yield request
 
         if response.status_code != 401 or "www-authenticate" not in response.headers:

{"golden_diff": "diff --git a/httpx/auth.py b/httpx/auth.py\n--- a/httpx/auth.py\n+++ b/httpx/auth.py\n@@ -6,7 +6,7 @@\n from base64 import b64encode\n from urllib.request import parse_http_list\n \n-from .exceptions import ProtocolError\n+from .exceptions import ProtocolError, RequestBodyUnavailable\n from .models import Request, Response\n from .utils import to_bytes, to_str, unquote\n \n@@ -104,6 +104,8 @@\n         self.password = to_bytes(password)\n \n     def __call__(self, request: Request) -> AuthFlow:\n+        if not request.stream.can_replay():\n+            raise RequestBodyUnavailable(\"Request body is no longer available.\")\n         response = yield request\n \n         if response.status_code != 401 or \"www-authenticate\" not in response.headers:\n", "issue": "DigestAuth should raise a clear error if the request cannot replay.\nOur DigestAuth implementation cannot work with non-replayable requests.\r\nWe ought to raise a nice clear error if `request.stream.is_replayable` is not True.\n", "before_files": [{"content": "import hashlib\nimport os\nimport re\nimport time\nimport typing\nfrom base64 import b64encode\nfrom urllib.request import parse_http_list\n\nfrom .exceptions import ProtocolError\nfrom .models import Request, Response\nfrom .utils import to_bytes, to_str, unquote\n\nAuthFlow = typing.Generator[Request, Response, None]\n\nAuthTypes = typing.Union[\n    typing.Tuple[typing.Union[str, bytes], typing.Union[str, bytes]],\n    typing.Callable[[\"Request\"], \"Request\"],\n    \"Auth\",\n]\n\n\nclass Auth:\n    \"\"\"\n    Base class for all authentication schemes.\n    \"\"\"\n\n    def __call__(self, request: Request) -> AuthFlow:\n        \"\"\"\n        Execute the authentication flow.\n\n        To dispatch a request, `yield` it:\n\n        ```\n        yield request\n        ```\n\n        The client will `.send()` the response back into the flow generator. You can\n        access it like so:\n\n        ```\n        response = yield request\n        ```\n\n        A `return` (or reaching the end of the generator) will result in the\n        client returning the last response obtained from the server.\n\n        You can dispatch as many requests as is necessary.\n        \"\"\"\n        yield request\n\n\nclass FunctionAuth(Auth):\n    \"\"\"\n    Allows the 'auth' argument to be passed as a simple callable function,\n    that takes the request, and returns a new, modified request.\n    \"\"\"\n\n    def __init__(self, func: typing.Callable[[Request], Request]) -> None:\n        self.func = func\n\n    def __call__(self, request: Request) -> AuthFlow:\n        yield self.func(request)\n\n\nclass BasicAuth(Auth):\n    \"\"\"\n    Allows the 'auth' argument to be passed as a (username, password) pair,\n    and uses HTTP Basic authentication.\n    \"\"\"\n\n    def __init__(\n        self, username: typing.Union[str, bytes], password: typing.Union[str, bytes]\n    ):\n        self.auth_header = self.build_auth_header(username, password)\n\n    def __call__(self, request: Request) -> AuthFlow:\n        request.headers[\"Authorization\"] = self.auth_header\n        yield request\n\n    def build_auth_header(\n        self, username: typing.Union[str, bytes], password: typing.Union[str, bytes]\n    ) -> str:\n        userpass = b\":\".join((to_bytes(username), to_bytes(password)))\n        token = b64encode(userpass).decode().strip()\n        return f\"Basic {token}\"\n\n\nclass DigestAuth(Auth):\n    ALGORITHM_TO_HASH_FUNCTION: typing.Dict[str, typing.Callable] = {\n        \"MD5\": hashlib.md5,\n        \"MD5-SESS\": hashlib.md5,\n        \"SHA\": hashlib.sha1,\n        \"SHA-SESS\": hashlib.sha1,\n        \"SHA-256\": hashlib.sha256,\n        \"SHA-256-SESS\": hashlib.sha256,\n        \"SHA-512\": hashlib.sha512,\n        \"SHA-512-SESS\": hashlib.sha512,\n    }\n\n    def __init__(\n        self, username: typing.Union[str, bytes], password: typing.Union[str, bytes]\n    ) -> None:\n        self.username = to_bytes(username)\n        self.password = to_bytes(password)\n\n    def __call__(self, request: Request) -> AuthFlow:\n        response = yield request\n\n        if response.status_code != 401 or \"www-authenticate\" not in response.headers:\n            # If the response is not a 401 WWW-Authenticate, then we don't\n            # need to build an authenticated request.\n            return\n\n        header = response.headers[\"www-authenticate\"]\n        try:\n            challenge = DigestAuthChallenge.from_header(header)\n        except ValueError:\n            raise ProtocolError(\"Malformed Digest authentication header\")\n\n        request.headers[\"Authorization\"] = self._build_auth_header(request, challenge)\n        yield request\n\n    def _build_auth_header(\n        self, request: Request, challenge: \"DigestAuthChallenge\"\n    ) -> str:\n        hash_func = self.ALGORITHM_TO_HASH_FUNCTION[challenge.algorithm]\n\n        def digest(data: bytes) -> bytes:\n            return hash_func(data).hexdigest().encode()\n\n        A1 = b\":\".join((self.username, challenge.realm, self.password))\n\n        path = request.url.full_path.encode(\"utf-8\")\n        A2 = b\":\".join((request.method.encode(), path))\n        # TODO: implement auth-int\n        HA2 = digest(A2)\n\n        nonce_count = 1  # TODO: implement nonce counting\n        nc_value = b\"%08x\" % nonce_count\n        cnonce = self._get_client_nonce(nonce_count, challenge.nonce)\n\n        HA1 = digest(A1)\n        if challenge.algorithm.lower().endswith(\"-sess\"):\n            HA1 = digest(b\":\".join((HA1, challenge.nonce, cnonce)))\n\n        qop = self._resolve_qop(challenge.qop)\n        if qop is None:\n            digest_data = [HA1, challenge.nonce, HA2]\n        else:\n            digest_data = [challenge.nonce, nc_value, cnonce, qop, HA2]\n        key_digest = b\":\".join(digest_data)\n\n        format_args = {\n            \"username\": self.username,\n            \"realm\": challenge.realm,\n            \"nonce\": challenge.nonce,\n            \"uri\": path,\n            \"response\": digest(b\":\".join((HA1, key_digest))),\n            \"algorithm\": challenge.algorithm.encode(),\n        }\n        if challenge.opaque:\n            format_args[\"opaque\"] = challenge.opaque\n        if qop:\n            format_args[\"qop\"] = b\"auth\"\n            format_args[\"nc\"] = nc_value\n            format_args[\"cnonce\"] = cnonce\n\n        return \"Digest \" + self._get_header_value(format_args)\n\n    def _get_client_nonce(self, nonce_count: int, nonce: bytes) -> bytes:\n        s = str(nonce_count).encode()\n        s += nonce\n        s += time.ctime().encode()\n        s += os.urandom(8)\n\n        return hashlib.sha1(s).hexdigest()[:16].encode()\n\n    def _get_header_value(self, header_fields: typing.Dict[str, bytes]) -> str:\n        NON_QUOTED_FIELDS = (\"algorithm\", \"qop\", \"nc\")\n        QUOTED_TEMPLATE = '{}=\"{}\"'\n        NON_QUOTED_TEMPLATE = \"{}={}\"\n\n        header_value = \"\"\n        for i, (field, value) in enumerate(header_fields.items()):\n            if i > 0:\n                header_value += \", \"\n            template = (\n                QUOTED_TEMPLATE\n                if field not in NON_QUOTED_FIELDS\n                else NON_QUOTED_TEMPLATE\n            )\n            header_value += template.format(field, to_str(value))\n\n        return header_value\n\n    def _resolve_qop(self, qop: typing.Optional[bytes]) -> typing.Optional[bytes]:\n        if qop is None:\n            return None\n        qops = re.split(b\", ?\", qop)\n        if b\"auth\" in qops:\n            return b\"auth\"\n\n        if qops == [b\"auth-int\"]:\n            raise NotImplementedError(\"Digest auth-int support is not yet implemented\")\n\n        raise ProtocolError(f'Unexpected qop value \"{qop!r}\" in digest auth')\n\n\nclass DigestAuthChallenge:\n    def __init__(\n        self,\n        realm: bytes,\n        nonce: bytes,\n        algorithm: str = None,\n        opaque: typing.Optional[bytes] = None,\n        qop: typing.Optional[bytes] = None,\n    ) -> None:\n        self.realm = realm\n        self.nonce = nonce\n        self.algorithm = algorithm or \"MD5\"\n        self.opaque = opaque\n        self.qop = qop\n\n    @classmethod\n    def from_header(cls, header: str) -> \"DigestAuthChallenge\":\n        \"\"\"Returns a challenge from a Digest WWW-Authenticate header.\n        These take the form of:\n        `Digest realm=\"realm@host.com\",qop=\"auth,auth-int\",nonce=\"abc\",opaque=\"xyz\"`\n        \"\"\"\n        scheme, _, fields = header.partition(\" \")\n        if scheme.lower() != \"digest\":\n            raise ValueError(\"Header does not start with 'Digest'\")\n\n        header_dict: typing.Dict[str, str] = {}\n        for field in parse_http_list(fields):\n            key, value = field.strip().split(\"=\", 1)\n            header_dict[key] = unquote(value)\n\n        try:\n            return cls.from_header_dict(header_dict)\n        except KeyError as exc:\n            raise ValueError(\"Malformed Digest WWW-Authenticate header\") from exc\n\n    @classmethod\n    def from_header_dict(cls, header_dict: dict) -> \"DigestAuthChallenge\":\n        realm = header_dict[\"realm\"].encode()\n        nonce = header_dict[\"nonce\"].encode()\n        qop = header_dict[\"qop\"].encode() if \"qop\" in header_dict else None\n        opaque = header_dict[\"opaque\"].encode() if \"opaque\" in header_dict else None\n        algorithm = header_dict.get(\"algorithm\")\n        return cls(\n            realm=realm, nonce=nonce, qop=qop, opaque=opaque, algorithm=algorithm\n        )\n", "path": "httpx/auth.py"}], "after_files": [{"content": "import hashlib\nimport os\nimport re\nimport time\nimport typing\nfrom base64 import b64encode\nfrom urllib.request import parse_http_list\n\nfrom .exceptions import ProtocolError, RequestBodyUnavailable\nfrom .models import Request, Response\nfrom .utils import to_bytes, to_str, unquote\n\nAuthFlow = typing.Generator[Request, Response, None]\n\nAuthTypes = typing.Union[\n    typing.Tuple[typing.Union[str, bytes], typing.Union[str, bytes]],\n    typing.Callable[[\"Request\"], \"Request\"],\n    \"Auth\",\n]\n\n\nclass Auth:\n    \"\"\"\n    Base class for all authentication schemes.\n    \"\"\"\n\n    def __call__(self, request: Request) -> AuthFlow:\n        \"\"\"\n        Execute the authentication flow.\n\n        To dispatch a request, `yield` it:\n\n        ```\n        yield request\n        ```\n\n        The client will `.send()` the response back into the flow generator. You can\n        access it like so:\n\n        ```\n        response = yield request\n        ```\n\n        A `return` (or reaching the end of the generator) will result in the\n        client returning the last response obtained from the server.\n\n        You can dispatch as many requests as is necessary.\n        \"\"\"\n        yield request\n\n\nclass FunctionAuth(Auth):\n    \"\"\"\n    Allows the 'auth' argument to be passed as a simple callable function,\n    that takes the request, and returns a new, modified request.\n    \"\"\"\n\n    def __init__(self, func: typing.Callable[[Request], Request]) -> None:\n        self.func = func\n\n    def __call__(self, request: Request) -> AuthFlow:\n        yield self.func(request)\n\n\nclass BasicAuth(Auth):\n    \"\"\"\n    Allows the 'auth' argument to be passed as a (username, password) pair,\n    and uses HTTP Basic authentication.\n    \"\"\"\n\n    def __init__(\n        self, username: typing.Union[str, bytes], password: typing.Union[str, bytes]\n    ):\n        self.auth_header = self.build_auth_header(username, password)\n\n    def __call__(self, request: Request) -> AuthFlow:\n        request.headers[\"Authorization\"] = self.auth_header\n        yield request\n\n    def build_auth_header(\n        self, username: typing.Union[str, bytes], password: typing.Union[str, bytes]\n    ) -> str:\n        userpass = b\":\".join((to_bytes(username), to_bytes(password)))\n        token = b64encode(userpass).decode().strip()\n        return f\"Basic {token}\"\n\n\nclass DigestAuth(Auth):\n    ALGORITHM_TO_HASH_FUNCTION: typing.Dict[str, typing.Callable] = {\n        \"MD5\": hashlib.md5,\n        \"MD5-SESS\": hashlib.md5,\n        \"SHA\": hashlib.sha1,\n        \"SHA-SESS\": hashlib.sha1,\n        \"SHA-256\": hashlib.sha256,\n        \"SHA-256-SESS\": hashlib.sha256,\n        \"SHA-512\": hashlib.sha512,\n        \"SHA-512-SESS\": hashlib.sha512,\n    }\n\n    def __init__(\n        self, username: typing.Union[str, bytes], password: typing.Union[str, bytes]\n    ) -> None:\n        self.username = to_bytes(username)\n        self.password = to_bytes(password)\n\n    def __call__(self, request: Request) -> AuthFlow:\n        if not request.stream.can_replay():\n            raise RequestBodyUnavailable(\"Request body is no longer available.\")\n        response = yield request\n\n        if response.status_code != 401 or \"www-authenticate\" not in response.headers:\n            # If the response is not a 401 WWW-Authenticate, then we don't\n            # need to build an authenticated request.\n            return\n\n        header = response.headers[\"www-authenticate\"]\n        try:\n            challenge = DigestAuthChallenge.from_header(header)\n        except ValueError:\n            raise ProtocolError(\"Malformed Digest authentication header\")\n\n        request.headers[\"Authorization\"] = self._build_auth_header(request, challenge)\n        yield request\n\n    def _build_auth_header(\n        self, request: Request, challenge: \"DigestAuthChallenge\"\n    ) -> str:\n        hash_func = self.ALGORITHM_TO_HASH_FUNCTION[challenge.algorithm]\n\n        def digest(data: bytes) -> bytes:\n            return hash_func(data).hexdigest().encode()\n\n        A1 = b\":\".join((self.username, challenge.realm, self.password))\n\n        path = request.url.full_path.encode(\"utf-8\")\n        A2 = b\":\".join((request.method.encode(), path))\n        # TODO: implement auth-int\n        HA2 = digest(A2)\n\n        nonce_count = 1  # TODO: implement nonce counting\n        nc_value = b\"%08x\" % nonce_count\n        cnonce = self._get_client_nonce(nonce_count, challenge.nonce)\n\n        HA1 = digest(A1)\n        if challenge.algorithm.lower().endswith(\"-sess\"):\n            HA1 = digest(b\":\".join((HA1, challenge.nonce, cnonce)))\n\n        qop = self._resolve_qop(challenge.qop)\n        if qop is None:\n            digest_data = [HA1, challenge.nonce, HA2]\n        else:\n            digest_data = [challenge.nonce, nc_value, cnonce, qop, HA2]\n        key_digest = b\":\".join(digest_data)\n\n        format_args = {\n            \"username\": self.username,\n            \"realm\": challenge.realm,\n            \"nonce\": challenge.nonce,\n            \"uri\": path,\n            \"response\": digest(b\":\".join((HA1, key_digest))),\n            \"algorithm\": challenge.algorithm.encode(),\n        }\n        if challenge.opaque:\n            format_args[\"opaque\"] = challenge.opaque\n        if qop:\n            format_args[\"qop\"] = b\"auth\"\n            format_args[\"nc\"] = nc_value\n            format_args[\"cnonce\"] = cnonce\n\n        return \"Digest \" + self._get_header_value(format_args)\n\n    def _get_client_nonce(self, nonce_count: int, nonce: bytes) -> bytes:\n        s = str(nonce_count).encode()\n        s += nonce\n        s += time.ctime().encode()\n        s += os.urandom(8)\n\n        return hashlib.sha1(s).hexdigest()[:16].encode()\n\n    def _get_header_value(self, header_fields: typing.Dict[str, bytes]) -> str:\n        NON_QUOTED_FIELDS = (\"algorithm\", \"qop\", \"nc\")\n        QUOTED_TEMPLATE = '{}=\"{}\"'\n        NON_QUOTED_TEMPLATE = \"{}={}\"\n\n        header_value = \"\"\n        for i, (field, value) in enumerate(header_fields.items()):\n            if i > 0:\n                header_value += \", \"\n            template = (\n                QUOTED_TEMPLATE\n                if field not in NON_QUOTED_FIELDS\n                else NON_QUOTED_TEMPLATE\n            )\n            header_value += template.format(field, to_str(value))\n\n        return header_value\n\n    def _resolve_qop(self, qop: typing.Optional[bytes]) -> typing.Optional[bytes]:\n        if qop is None:\n            return None\n        qops = re.split(b\", ?\", qop)\n        if b\"auth\" in qops:\n            return b\"auth\"\n\n        if qops == [b\"auth-int\"]:\n            raise NotImplementedError(\"Digest auth-int support is not yet implemented\")\n\n        raise ProtocolError(f'Unexpected qop value \"{qop!r}\" in digest auth')\n\n\nclass DigestAuthChallenge:\n    def __init__(\n        self,\n        realm: bytes,\n        nonce: bytes,\n        algorithm: str = None,\n        opaque: typing.Optional[bytes] = None,\n        qop: typing.Optional[bytes] = None,\n    ) -> None:\n        self.realm = realm\n        self.nonce = nonce\n        self.algorithm = algorithm or \"MD5\"\n        self.opaque = opaque\n        self.qop = qop\n\n    @classmethod\n    def from_header(cls, header: str) -> \"DigestAuthChallenge\":\n        \"\"\"Returns a challenge from a Digest WWW-Authenticate header.\n        These take the form of:\n        `Digest realm=\"realm@host.com\",qop=\"auth,auth-int\",nonce=\"abc\",opaque=\"xyz\"`\n        \"\"\"\n        scheme, _, fields = header.partition(\" \")\n        if scheme.lower() != \"digest\":\n            raise ValueError(\"Header does not start with 'Digest'\")\n\n        header_dict: typing.Dict[str, str] = {}\n        for field in parse_http_list(fields):\n            key, value = field.strip().split(\"=\", 1)\n            header_dict[key] = unquote(value)\n\n        try:\n            return cls.from_header_dict(header_dict)\n        except KeyError as exc:\n            raise ValueError(\"Malformed Digest WWW-Authenticate header\") from exc\n\n    @classmethod\n    def from_header_dict(cls, header_dict: dict) -> \"DigestAuthChallenge\":\n        realm = header_dict[\"realm\"].encode()\n        nonce = header_dict[\"nonce\"].encode()\n        qop = header_dict[\"qop\"].encode() if \"qop\" in header_dict else None\n        opaque = header_dict[\"opaque\"].encode() if \"opaque\" in header_dict else None\n        algorithm = header_dict.get(\"algorithm\")\n        return cls(\n            realm=realm, nonce=nonce, qop=qop, opaque=opaque, algorithm=algorithm\n        )\n", "path": "httpx/auth.py"}]}