We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
httpauth is not attached to replay request 
#### Problem Description
I set mitmproxy to run in reverse mode as a proxy to real server, and then protect mitmproxy with a pair of user:pass in the proxyauth option. A regular request would go through, but a reply of that same request would return 401 Unauthorized

#### Steps to reproduce the behavior:
1. turn on reverse mode in mitmweb
2. set basic auth in proxyauth in 'username:pass' format
3. initiate a success request
4. replay the request

#### System Information
Mitmproxy: 10.1.5
Python:    3.11.6
OpenSSL:   OpenSSL 3.1.4 24 Oct 2023
Platform:  Linux-4.14.276-211.499.amzn2.x86_64-x86_64-with-glibc2.31


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `mitmproxy/addons/proxyauth.py`
Content:
```
1 from __future__ import annotations
2 
3 import binascii
4 import weakref
5 from abc import ABC
6 from abc import abstractmethod
7 from collections.abc import MutableMapping
8 from typing import Optional
9 
10 import ldap3
11 import passlib.apache
12 
13 from mitmproxy import connection
14 from mitmproxy import ctx
15 from mitmproxy import exceptions
16 from mitmproxy import http
17 from mitmproxy.net.http import status_codes
18 from mitmproxy.proxy import mode_specs
19 from mitmproxy.proxy.layers import modes
20 
21 REALM = "mitmproxy"
22 
23 
24 class ProxyAuth:
25     validator: Validator | None = None
26 
27     def __init__(self) -> None:
28         self.authenticated: MutableMapping[connection.Client, tuple[str, str]] = (
29             weakref.WeakKeyDictionary()
30         )
31         """Contains all connections that are permanently authenticated after an HTTP CONNECT"""
32 
33     def load(self, loader):
34         loader.add_option(
35             "proxyauth",
36             Optional[str],
37             None,
38             """
39             Require proxy authentication. Format:
40             "username:pass",
41             "any" to accept any user/pass combination,
42             "@path" to use an Apache htpasswd file,
43             or "ldap[s]:url_server_ldap[:port]:dn_auth:password:dn_subtree[?search_filter_key=...]" for LDAP authentication.
44             """,
45         )
46 
47     def configure(self, updated):
48         if "proxyauth" in updated:
49             auth = ctx.options.proxyauth
50             if auth:
51                 if auth == "any":
52                     self.validator = AcceptAll()
53                 elif auth.startswith("@"):
54                     self.validator = Htpasswd(auth)
55                 elif ctx.options.proxyauth.startswith("ldap"):
56                     self.validator = Ldap(auth)
57                 elif ":" in ctx.options.proxyauth:
58                     self.validator = SingleUser(auth)
59                 else:
60                     raise exceptions.OptionsError("Invalid proxyauth specification.")
61             else:
62                 self.validator = None
63 
64     def socks5_auth(self, data: modes.Socks5AuthData) -> None:
65         if self.validator and self.validator(data.username, data.password):
66             data.valid = True
67             self.authenticated[data.client_conn] = data.username, data.password
68 
69     def http_connect(self, f: http.HTTPFlow) -> None:
70         if self.validator and self.authenticate_http(f):
71             # Make a note that all further requests over this connection are ok.
72             self.authenticated[f.client_conn] = f.metadata["proxyauth"]
73 
74     def requestheaders(self, f: http.HTTPFlow) -> None:
75         if self.validator:
76             # Is this connection authenticated by a previous HTTP CONNECT?
77             if f.client_conn in self.authenticated:
78                 f.metadata["proxyauth"] = self.authenticated[f.client_conn]
79             else:
80                 self.authenticate_http(f)
81 
82     def authenticate_http(self, f: http.HTTPFlow) -> bool:
83         """
84         Authenticate an HTTP request, returns if authentication was successful.
85 
86         If valid credentials are found, the matching authentication header is removed.
87         In no or invalid credentials are found, flow.response is set to an error page.
88         """
89         assert self.validator
90         username = None
91         password = None
92         is_valid = False
93 
94         is_proxy = is_http_proxy(f)
95         auth_header = http_auth_header(is_proxy)
96         try:
97             auth_value = f.request.headers.get(auth_header, "")
98             scheme, username, password = parse_http_basic_auth(auth_value)
99             is_valid = self.validator(username, password)
100         except Exception:
101             pass
102 
103         if is_valid:
104             f.metadata["proxyauth"] = (username, password)
105             del f.request.headers[auth_header]
106             return True
107         else:
108             f.response = make_auth_required_response(is_proxy)
109             return False
110 
111 
112 def make_auth_required_response(is_proxy: bool) -> http.Response:
113     if is_proxy:
114         status_code = status_codes.PROXY_AUTH_REQUIRED
115         headers = {"Proxy-Authenticate": f'Basic realm="{REALM}"'}
116     else:
117         status_code = status_codes.UNAUTHORIZED
118         headers = {"WWW-Authenticate": f'Basic realm="{REALM}"'}
119 
120     reason = http.status_codes.RESPONSES[status_code]
121     return http.Response.make(
122         status_code,
123         (
124             f"<html>"
125             f"<head><title>{status_code} {reason}</title></head>"
126             f"<body><h1>{status_code} {reason}</h1></body>"
127             f"</html>"
128         ),
129         headers,
130     )
131 
132 
133 def http_auth_header(is_proxy: bool) -> str:
134     if is_proxy:
135         return "Proxy-Authorization"
136     else:
137         return "Authorization"
138 
139 
140 def is_http_proxy(f: http.HTTPFlow) -> bool:
141     """
142     Returns:
143         - True, if authentication is done as if mitmproxy is a proxy
144         - False, if authentication is done as if mitmproxy is an HTTP server
145     """
146     return isinstance(
147         f.client_conn.proxy_mode, (mode_specs.RegularMode, mode_specs.UpstreamMode)
148     )
149 
150 
151 def mkauth(username: str, password: str, scheme: str = "basic") -> str:
152     """
153     Craft a basic auth string
154     """
155     v = binascii.b2a_base64((username + ":" + password).encode("utf8")).decode("ascii")
156     return scheme + " " + v
157 
158 
159 def parse_http_basic_auth(s: str) -> tuple[str, str, str]:
160     """
161     Parse a basic auth header.
162     Raises a ValueError if the input is invalid.
163     """
164     scheme, authinfo = s.split()
165     if scheme.lower() != "basic":
166         raise ValueError("Unknown scheme")
167     try:
168         user, password = (
169             binascii.a2b_base64(authinfo.encode()).decode("utf8", "replace").split(":")
170         )
171     except binascii.Error as e:
172         raise ValueError(str(e))
173     return scheme, user, password
174 
175 
176 class Validator(ABC):
177     """Base class for all username/password validators."""
178 
179     @abstractmethod
180     def __call__(self, username: str, password: str) -> bool:
181         raise NotImplementedError
182 
183 
184 class AcceptAll(Validator):
185     def __call__(self, username: str, password: str) -> bool:
186         return True
187 
188 
189 class SingleUser(Validator):
190     def __init__(self, proxyauth: str):
191         try:
192             self.username, self.password = proxyauth.split(":")
193         except ValueError:
194             raise exceptions.OptionsError("Invalid single-user auth specification.")
195 
196     def __call__(self, username: str, password: str) -> bool:
197         return self.username == username and self.password == password
198 
199 
200 class Htpasswd(Validator):
201     def __init__(self, proxyauth: str):
202         path = proxyauth[1:]
203         try:
204             self.htpasswd = passlib.apache.HtpasswdFile(path)
205         except (ValueError, OSError):
206             raise exceptions.OptionsError(f"Could not open htpasswd file: {path}")
207 
208     def __call__(self, username: str, password: str) -> bool:
209         return self.htpasswd.check_password(username, password)
210 
211 
212 class Ldap(Validator):
213     conn: ldap3.Connection
214     server: ldap3.Server
215     dn_subtree: str
216     filter_key: str
217 
218     def __init__(self, proxyauth: str):
219         (
220             use_ssl,
221             url,
222             port,
223             ldap_user,
224             ldap_pass,
225             self.dn_subtree,
226             self.filter_key,
227         ) = self.parse_spec(proxyauth)
228         server = ldap3.Server(url, port=port, use_ssl=use_ssl)
229         conn = ldap3.Connection(server, ldap_user, ldap_pass, auto_bind=True)
230         self.conn = conn
231         self.server = server
232 
233     @staticmethod
234     def parse_spec(spec: str) -> tuple[bool, str, int | None, str, str, str, str]:
235         try:
236             if spec.count(":") > 4:
237                 (
238                     security,
239                     url,
240                     port_str,
241                     ldap_user,
242                     ldap_pass,
243                     dn_subtree,
244                 ) = spec.split(":")
245                 port = int(port_str)
246             else:
247                 security, url, ldap_user, ldap_pass, dn_subtree = spec.split(":")
248                 port = None
249 
250             if "?" in dn_subtree:
251                 dn_subtree, search_str = dn_subtree.split("?")
252                 key, value = search_str.split("=")
253                 if key == "search_filter_key":
254                     search_filter_key = value
255                 else:
256                     raise ValueError
257             else:
258                 search_filter_key = "cn"
259 
260             if security == "ldaps":
261                 use_ssl = True
262             elif security == "ldap":
263                 use_ssl = False
264             else:
265                 raise ValueError
266 
267             return (
268                 use_ssl,
269                 url,
270                 port,
271                 ldap_user,
272                 ldap_pass,
273                 dn_subtree,
274                 search_filter_key,
275             )
276         except ValueError:
277             raise exceptions.OptionsError(f"Invalid LDAP specification: {spec}")
278 
279     def __call__(self, username: str, password: str) -> bool:
280         if not username or not password:
281             return False
282         self.conn.search(self.dn_subtree, f"({self.filter_key}={username})")
283         if self.conn.response:
284             c = ldap3.Connection(
285                 self.server, self.conn.response[0]["dn"], password, auto_bind=True
286             )
287             if c:
288                 return True
289         return False
290
```
--- 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/mitmproxy/addons/proxyauth.py b/mitmproxy/addons/proxyauth.py
--- a/mitmproxy/addons/proxyauth.py
+++ b/mitmproxy/addons/proxyauth.py
@@ -76,6 +76,8 @@
             # Is this connection authenticated by a previous HTTP CONNECT?
             if f.client_conn in self.authenticated:
                 f.metadata["proxyauth"] = self.authenticated[f.client_conn]
+            elif f.is_replay:
+                pass
             else:
                 self.authenticate_http(f)

{"golden_diff": "diff --git a/mitmproxy/addons/proxyauth.py b/mitmproxy/addons/proxyauth.py\n--- a/mitmproxy/addons/proxyauth.py\n+++ b/mitmproxy/addons/proxyauth.py\n@@ -76,6 +76,8 @@\n             # Is this connection authenticated by a previous HTTP CONNECT?\n             if f.client_conn in self.authenticated:\n                 f.metadata[\"proxyauth\"] = self.authenticated[f.client_conn]\n+            elif f.is_replay:\n+                pass\n             else:\n                 self.authenticate_http(f)\n", "issue": "httpauth is not attached to replay request \n#### Problem Description\r\nI set mitmproxy to run in reverse mode as a proxy to real server, and then protect mitmproxy with a pair of user:pass in the proxyauth option. A regular request would go through, but a reply of that same request would return 401 Unauthorized\r\n\r\n#### Steps to reproduce the behavior:\r\n1. turn on reverse mode in mitmweb\r\n2. set basic auth in proxyauth in 'username:pass' format\r\n3. initiate a success request\r\n4. replay the request\r\n\r\n#### System Information\r\nMitmproxy: 10.1.5\r\nPython:    3.11.6\r\nOpenSSL:   OpenSSL 3.1.4 24 Oct 2023\r\nPlatform:  Linux-4.14.276-211.499.amzn2.x86_64-x86_64-with-glibc2.31\r\n\n", "before_files": [{"content": "from __future__ import annotations\n\nimport binascii\nimport weakref\nfrom abc import ABC\nfrom abc import abstractmethod\nfrom collections.abc import MutableMapping\nfrom typing import Optional\n\nimport ldap3\nimport passlib.apache\n\nfrom mitmproxy import connection\nfrom mitmproxy import ctx\nfrom mitmproxy import exceptions\nfrom mitmproxy import http\nfrom mitmproxy.net.http import status_codes\nfrom mitmproxy.proxy import mode_specs\nfrom mitmproxy.proxy.layers import modes\n\nREALM = \"mitmproxy\"\n\n\nclass ProxyAuth:\n    validator: Validator | None = None\n\n    def __init__(self) -> None:\n        self.authenticated: MutableMapping[connection.Client, tuple[str, str]] = (\n            weakref.WeakKeyDictionary()\n        )\n        \"\"\"Contains all connections that are permanently authenticated after an HTTP CONNECT\"\"\"\n\n    def load(self, loader):\n        loader.add_option(\n            \"proxyauth\",\n            Optional[str],\n            None,\n            \"\"\"\n            Require proxy authentication. Format:\n            \"username:pass\",\n            \"any\" to accept any user/pass combination,\n            \"@path\" to use an Apache htpasswd file,\n            or \"ldap[s]:url_server_ldap[:port]:dn_auth:password:dn_subtree[?search_filter_key=...]\" for LDAP authentication.\n            \"\"\",\n        )\n\n    def configure(self, updated):\n        if \"proxyauth\" in updated:\n            auth = ctx.options.proxyauth\n            if auth:\n                if auth == \"any\":\n                    self.validator = AcceptAll()\n                elif auth.startswith(\"@\"):\n                    self.validator = Htpasswd(auth)\n                elif ctx.options.proxyauth.startswith(\"ldap\"):\n                    self.validator = Ldap(auth)\n                elif \":\" in ctx.options.proxyauth:\n                    self.validator = SingleUser(auth)\n                else:\n                    raise exceptions.OptionsError(\"Invalid proxyauth specification.\")\n            else:\n                self.validator = None\n\n    def socks5_auth(self, data: modes.Socks5AuthData) -> None:\n        if self.validator and self.validator(data.username, data.password):\n            data.valid = True\n            self.authenticated[data.client_conn] = data.username, data.password\n\n    def http_connect(self, f: http.HTTPFlow) -> None:\n        if self.validator and self.authenticate_http(f):\n            # Make a note that all further requests over this connection are ok.\n            self.authenticated[f.client_conn] = f.metadata[\"proxyauth\"]\n\n    def requestheaders(self, f: http.HTTPFlow) -> None:\n        if self.validator:\n            # Is this connection authenticated by a previous HTTP CONNECT?\n            if f.client_conn in self.authenticated:\n                f.metadata[\"proxyauth\"] = self.authenticated[f.client_conn]\n            else:\n                self.authenticate_http(f)\n\n    def authenticate_http(self, f: http.HTTPFlow) -> bool:\n        \"\"\"\n        Authenticate an HTTP request, returns if authentication was successful.\n\n        If valid credentials are found, the matching authentication header is removed.\n        In no or invalid credentials are found, flow.response is set to an error page.\n        \"\"\"\n        assert self.validator\n        username = None\n        password = None\n        is_valid = False\n\n        is_proxy = is_http_proxy(f)\n        auth_header = http_auth_header(is_proxy)\n        try:\n            auth_value = f.request.headers.get(auth_header, \"\")\n            scheme, username, password = parse_http_basic_auth(auth_value)\n            is_valid = self.validator(username, password)\n        except Exception:\n            pass\n\n        if is_valid:\n            f.metadata[\"proxyauth\"] = (username, password)\n            del f.request.headers[auth_header]\n            return True\n        else:\n            f.response = make_auth_required_response(is_proxy)\n            return False\n\n\ndef make_auth_required_response(is_proxy: bool) -> http.Response:\n    if is_proxy:\n        status_code = status_codes.PROXY_AUTH_REQUIRED\n        headers = {\"Proxy-Authenticate\": f'Basic realm=\"{REALM}\"'}\n    else:\n        status_code = status_codes.UNAUTHORIZED\n        headers = {\"WWW-Authenticate\": f'Basic realm=\"{REALM}\"'}\n\n    reason = http.status_codes.RESPONSES[status_code]\n    return http.Response.make(\n        status_code,\n        (\n            f\"<html>\"\n            f\"<head><title>{status_code} {reason}</title></head>\"\n            f\"<body><h1>{status_code} {reason}</h1></body>\"\n            f\"</html>\"\n        ),\n        headers,\n    )\n\n\ndef http_auth_header(is_proxy: bool) -> str:\n    if is_proxy:\n        return \"Proxy-Authorization\"\n    else:\n        return \"Authorization\"\n\n\ndef is_http_proxy(f: http.HTTPFlow) -> bool:\n    \"\"\"\n    Returns:\n        - True, if authentication is done as if mitmproxy is a proxy\n        - False, if authentication is done as if mitmproxy is an HTTP server\n    \"\"\"\n    return isinstance(\n        f.client_conn.proxy_mode, (mode_specs.RegularMode, mode_specs.UpstreamMode)\n    )\n\n\ndef mkauth(username: str, password: str, scheme: str = \"basic\") -> str:\n    \"\"\"\n    Craft a basic auth string\n    \"\"\"\n    v = binascii.b2a_base64((username + \":\" + password).encode(\"utf8\")).decode(\"ascii\")\n    return scheme + \" \" + v\n\n\ndef parse_http_basic_auth(s: str) -> tuple[str, str, str]:\n    \"\"\"\n    Parse a basic auth header.\n    Raises a ValueError if the input is invalid.\n    \"\"\"\n    scheme, authinfo = s.split()\n    if scheme.lower() != \"basic\":\n        raise ValueError(\"Unknown scheme\")\n    try:\n        user, password = (\n            binascii.a2b_base64(authinfo.encode()).decode(\"utf8\", \"replace\").split(\":\")\n        )\n    except binascii.Error as e:\n        raise ValueError(str(e))\n    return scheme, user, password\n\n\nclass Validator(ABC):\n    \"\"\"Base class for all username/password validators.\"\"\"\n\n    @abstractmethod\n    def __call__(self, username: str, password: str) -> bool:\n        raise NotImplementedError\n\n\nclass AcceptAll(Validator):\n    def __call__(self, username: str, password: str) -> bool:\n        return True\n\n\nclass SingleUser(Validator):\n    def __init__(self, proxyauth: str):\n        try:\n            self.username, self.password = proxyauth.split(\":\")\n        except ValueError:\n            raise exceptions.OptionsError(\"Invalid single-user auth specification.\")\n\n    def __call__(self, username: str, password: str) -> bool:\n        return self.username == username and self.password == password\n\n\nclass Htpasswd(Validator):\n    def __init__(self, proxyauth: str):\n        path = proxyauth[1:]\n        try:\n            self.htpasswd = passlib.apache.HtpasswdFile(path)\n        except (ValueError, OSError):\n            raise exceptions.OptionsError(f\"Could not open htpasswd file: {path}\")\n\n    def __call__(self, username: str, password: str) -> bool:\n        return self.htpasswd.check_password(username, password)\n\n\nclass Ldap(Validator):\n    conn: ldap3.Connection\n    server: ldap3.Server\n    dn_subtree: str\n    filter_key: str\n\n    def __init__(self, proxyauth: str):\n        (\n            use_ssl,\n            url,\n            port,\n            ldap_user,\n            ldap_pass,\n            self.dn_subtree,\n            self.filter_key,\n        ) = self.parse_spec(proxyauth)\n        server = ldap3.Server(url, port=port, use_ssl=use_ssl)\n        conn = ldap3.Connection(server, ldap_user, ldap_pass, auto_bind=True)\n        self.conn = conn\n        self.server = server\n\n    @staticmethod\n    def parse_spec(spec: str) -> tuple[bool, str, int | None, str, str, str, str]:\n        try:\n            if spec.count(\":\") > 4:\n                (\n                    security,\n                    url,\n                    port_str,\n                    ldap_user,\n                    ldap_pass,\n                    dn_subtree,\n                ) = spec.split(\":\")\n                port = int(port_str)\n            else:\n                security, url, ldap_user, ldap_pass, dn_subtree = spec.split(\":\")\n                port = None\n\n            if \"?\" in dn_subtree:\n                dn_subtree, search_str = dn_subtree.split(\"?\")\n                key, value = search_str.split(\"=\")\n                if key == \"search_filter_key\":\n                    search_filter_key = value\n                else:\n                    raise ValueError\n            else:\n                search_filter_key = \"cn\"\n\n            if security == \"ldaps\":\n                use_ssl = True\n            elif security == \"ldap\":\n                use_ssl = False\n            else:\n                raise ValueError\n\n            return (\n                use_ssl,\n                url,\n                port,\n                ldap_user,\n                ldap_pass,\n                dn_subtree,\n                search_filter_key,\n            )\n        except ValueError:\n            raise exceptions.OptionsError(f\"Invalid LDAP specification: {spec}\")\n\n    def __call__(self, username: str, password: str) -> bool:\n        if not username or not password:\n            return False\n        self.conn.search(self.dn_subtree, f\"({self.filter_key}={username})\")\n        if self.conn.response:\n            c = ldap3.Connection(\n                self.server, self.conn.response[0][\"dn\"], password, auto_bind=True\n            )\n            if c:\n                return True\n        return False\n", "path": "mitmproxy/addons/proxyauth.py"}], "after_files": [{"content": "from __future__ import annotations\n\nimport binascii\nimport weakref\nfrom abc import ABC\nfrom abc import abstractmethod\nfrom collections.abc import MutableMapping\nfrom typing import Optional\n\nimport ldap3\nimport passlib.apache\n\nfrom mitmproxy import connection\nfrom mitmproxy import ctx\nfrom mitmproxy import exceptions\nfrom mitmproxy import http\nfrom mitmproxy.net.http import status_codes\nfrom mitmproxy.proxy import mode_specs\nfrom mitmproxy.proxy.layers import modes\n\nREALM = \"mitmproxy\"\n\n\nclass ProxyAuth:\n    validator: Validator | None = None\n\n    def __init__(self) -> None:\n        self.authenticated: MutableMapping[connection.Client, tuple[str, str]] = (\n            weakref.WeakKeyDictionary()\n        )\n        \"\"\"Contains all connections that are permanently authenticated after an HTTP CONNECT\"\"\"\n\n    def load(self, loader):\n        loader.add_option(\n            \"proxyauth\",\n            Optional[str],\n            None,\n            \"\"\"\n            Require proxy authentication. Format:\n            \"username:pass\",\n            \"any\" to accept any user/pass combination,\n            \"@path\" to use an Apache htpasswd file,\n            or \"ldap[s]:url_server_ldap[:port]:dn_auth:password:dn_subtree[?search_filter_key=...]\" for LDAP authentication.\n            \"\"\",\n        )\n\n    def configure(self, updated):\n        if \"proxyauth\" in updated:\n            auth = ctx.options.proxyauth\n            if auth:\n                if auth == \"any\":\n                    self.validator = AcceptAll()\n                elif auth.startswith(\"@\"):\n                    self.validator = Htpasswd(auth)\n                elif ctx.options.proxyauth.startswith(\"ldap\"):\n                    self.validator = Ldap(auth)\n                elif \":\" in ctx.options.proxyauth:\n                    self.validator = SingleUser(auth)\n                else:\n                    raise exceptions.OptionsError(\"Invalid proxyauth specification.\")\n            else:\n                self.validator = None\n\n    def socks5_auth(self, data: modes.Socks5AuthData) -> None:\n        if self.validator and self.validator(data.username, data.password):\n            data.valid = True\n            self.authenticated[data.client_conn] = data.username, data.password\n\n    def http_connect(self, f: http.HTTPFlow) -> None:\n        if self.validator and self.authenticate_http(f):\n            # Make a note that all further requests over this connection are ok.\n            self.authenticated[f.client_conn] = f.metadata[\"proxyauth\"]\n\n    def requestheaders(self, f: http.HTTPFlow) -> None:\n        if self.validator:\n            # Is this connection authenticated by a previous HTTP CONNECT?\n            if f.client_conn in self.authenticated:\n                f.metadata[\"proxyauth\"] = self.authenticated[f.client_conn]\n            elif f.is_replay:\n                pass\n            else:\n                self.authenticate_http(f)\n\n    def authenticate_http(self, f: http.HTTPFlow) -> bool:\n        \"\"\"\n        Authenticate an HTTP request, returns if authentication was successful.\n\n        If valid credentials are found, the matching authentication header is removed.\n        In no or invalid credentials are found, flow.response is set to an error page.\n        \"\"\"\n        assert self.validator\n        username = None\n        password = None\n        is_valid = False\n\n        is_proxy = is_http_proxy(f)\n        auth_header = http_auth_header(is_proxy)\n        try:\n            auth_value = f.request.headers.get(auth_header, \"\")\n            scheme, username, password = parse_http_basic_auth(auth_value)\n            is_valid = self.validator(username, password)\n        except Exception:\n            pass\n\n        if is_valid:\n            f.metadata[\"proxyauth\"] = (username, password)\n            del f.request.headers[auth_header]\n            return True\n        else:\n            f.response = make_auth_required_response(is_proxy)\n            return False\n\n\ndef make_auth_required_response(is_proxy: bool) -> http.Response:\n    if is_proxy:\n        status_code = status_codes.PROXY_AUTH_REQUIRED\n        headers = {\"Proxy-Authenticate\": f'Basic realm=\"{REALM}\"'}\n    else:\n        status_code = status_codes.UNAUTHORIZED\n        headers = {\"WWW-Authenticate\": f'Basic realm=\"{REALM}\"'}\n\n    reason = http.status_codes.RESPONSES[status_code]\n    return http.Response.make(\n        status_code,\n        (\n            f\"<html>\"\n            f\"<head><title>{status_code} {reason}</title></head>\"\n            f\"<body><h1>{status_code} {reason}</h1></body>\"\n            f\"</html>\"\n        ),\n        headers,\n    )\n\n\ndef http_auth_header(is_proxy: bool) -> str:\n    if is_proxy:\n        return \"Proxy-Authorization\"\n    else:\n        return \"Authorization\"\n\n\ndef is_http_proxy(f: http.HTTPFlow) -> bool:\n    \"\"\"\n    Returns:\n        - True, if authentication is done as if mitmproxy is a proxy\n        - False, if authentication is done as if mitmproxy is an HTTP server\n    \"\"\"\n    return isinstance(\n        f.client_conn.proxy_mode, (mode_specs.RegularMode, mode_specs.UpstreamMode)\n    )\n\n\ndef mkauth(username: str, password: str, scheme: str = \"basic\") -> str:\n    \"\"\"\n    Craft a basic auth string\n    \"\"\"\n    v = binascii.b2a_base64((username + \":\" + password).encode(\"utf8\")).decode(\"ascii\")\n    return scheme + \" \" + v\n\n\ndef parse_http_basic_auth(s: str) -> tuple[str, str, str]:\n    \"\"\"\n    Parse a basic auth header.\n    Raises a ValueError if the input is invalid.\n    \"\"\"\n    scheme, authinfo = s.split()\n    if scheme.lower() != \"basic\":\n        raise ValueError(\"Unknown scheme\")\n    try:\n        user, password = (\n            binascii.a2b_base64(authinfo.encode()).decode(\"utf8\", \"replace\").split(\":\")\n        )\n    except binascii.Error as e:\n        raise ValueError(str(e))\n    return scheme, user, password\n\n\nclass Validator(ABC):\n    \"\"\"Base class for all username/password validators.\"\"\"\n\n    @abstractmethod\n    def __call__(self, username: str, password: str) -> bool:\n        raise NotImplementedError\n\n\nclass AcceptAll(Validator):\n    def __call__(self, username: str, password: str) -> bool:\n        return True\n\n\nclass SingleUser(Validator):\n    def __init__(self, proxyauth: str):\n        try:\n            self.username, self.password = proxyauth.split(\":\")\n        except ValueError:\n            raise exceptions.OptionsError(\"Invalid single-user auth specification.\")\n\n    def __call__(self, username: str, password: str) -> bool:\n        return self.username == username and self.password == password\n\n\nclass Htpasswd(Validator):\n    def __init__(self, proxyauth: str):\n        path = proxyauth[1:]\n        try:\n            self.htpasswd = passlib.apache.HtpasswdFile(path)\n        except (ValueError, OSError):\n            raise exceptions.OptionsError(f\"Could not open htpasswd file: {path}\")\n\n    def __call__(self, username: str, password: str) -> bool:\n        return self.htpasswd.check_password(username, password)\n\n\nclass Ldap(Validator):\n    conn: ldap3.Connection\n    server: ldap3.Server\n    dn_subtree: str\n    filter_key: str\n\n    def __init__(self, proxyauth: str):\n        (\n            use_ssl,\n            url,\n            port,\n            ldap_user,\n            ldap_pass,\n            self.dn_subtree,\n            self.filter_key,\n        ) = self.parse_spec(proxyauth)\n        server = ldap3.Server(url, port=port, use_ssl=use_ssl)\n        conn = ldap3.Connection(server, ldap_user, ldap_pass, auto_bind=True)\n        self.conn = conn\n        self.server = server\n\n    @staticmethod\n    def parse_spec(spec: str) -> tuple[bool, str, int | None, str, str, str, str]:\n        try:\n            if spec.count(\":\") > 4:\n                (\n                    security,\n                    url,\n                    port_str,\n                    ldap_user,\n                    ldap_pass,\n                    dn_subtree,\n                ) = spec.split(\":\")\n                port = int(port_str)\n            else:\n                security, url, ldap_user, ldap_pass, dn_subtree = spec.split(\":\")\n                port = None\n\n            if \"?\" in dn_subtree:\n                dn_subtree, search_str = dn_subtree.split(\"?\")\n                key, value = search_str.split(\"=\")\n                if key == \"search_filter_key\":\n                    search_filter_key = value\n                else:\n                    raise ValueError\n            else:\n                search_filter_key = \"cn\"\n\n            if security == \"ldaps\":\n                use_ssl = True\n            elif security == \"ldap\":\n                use_ssl = False\n            else:\n                raise ValueError\n\n            return (\n                use_ssl,\n                url,\n                port,\n                ldap_user,\n                ldap_pass,\n                dn_subtree,\n                search_filter_key,\n            )\n        except ValueError:\n            raise exceptions.OptionsError(f\"Invalid LDAP specification: {spec}\")\n\n    def __call__(self, username: str, password: str) -> bool:\n        if not username or not password:\n            return False\n        self.conn.search(self.dn_subtree, f\"({self.filter_key}={username})\")\n        if self.conn.response:\n            c = ldap3.Connection(\n                self.server, self.conn.response[0][\"dn\"], password, auto_bind=True\n            )\n            if c:\n                return True\n        return False\n", "path": "mitmproxy/addons/proxyauth.py"}]}