We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
removing ":authority" pseudo header, then executing request crashes mitmproxy
##### Steps to reproduce the problem:

1. Run mitmproxy
2. Visit some http2 website, e.g. google.com
3. Edit the request (enter + e + 7) and remove the :authority header (d + q).
4. Replay the request (r).
5. mitmproxy crashes

Stacktrace:
Traceback (most recent call last):
  File "/usr/lib/python3.7/site-packages/mitmproxy/master.py", line 86, in run_loop
    loop()oding:            gzip, deflate, br
  File "/usr/lib/python3.7/site-packages/urwid/main_loop.py", line 286, in run
    self._run()e-requests:  1
  File "/usr/lib/python3.7/site-packages/urwid/main_loop.py", line 384, in _run
    self.event_loop.run() tab to view                                           [m:auto]
  File "/usr/lib/python3.7/site-packages/urwid/main_loop.py", line 1484, in run
    reraise(*exc_info)
  File "/usr/lib/python3.7/site-packages/urwid/compat.py", line 58, in reraise
    raise value
  File "/usr/lib/python3.7/asyncio/events.py", line 88, in _run
    self._context.run(self._callback, *self._args)
  File "/usr/lib/python3.7/site-packages/urwid/raw_display.py", line 404, in <lambda>
    event_loop, callback, self.get_available_raw_input())
  File "/usr/lib/python3.7/site-packages/urwid/raw_display.py", line 502, in parse_input
    callback(processed, processed_codes)
  File "/usr/lib/python3.7/site-packages/urwid/main_loop.py", line 411, in _update
    self.process_input(keys)
  File "/usr/lib/python3.7/site-packages/urwid/main_loop.py", line 511, in process_input
    k = self._topmost_widget.keypress(self.screen_size, k)
  File "/usr/lib/python3.7/site-packages/mitmproxy/tools/console/window.py", line 313, in keypress
    k
  File "/usr/lib/python3.7/site-packages/mitmproxy/tools/console/keymap.py", line 143, in handle
    return self.executor(b.command)
  File "/usr/lib/python3.7/site-packages/mitmproxy/tools/console/commandexecutor.py", line 17, in __call__
    ret = self.master.commands.execute(cmd)
  File "/usr/lib/python3.7/site-packages/mitmproxy/command.py", line 245, in execute
    return self.call_strings(parts[0], parts[1:])
  File "/usr/lib/python3.7/site-packages/mitmproxy/command.py", line 233, in call_strings
    return self.commands[path].call(args)
  File "/usr/lib/python3.7/site-packages/mitmproxy/command.py", line 105, in call
    ret = self.func(*self.prepare_args(args))
  File "/usr/lib/python3.7/site-packages/mitmproxy/command.py", line 275, in wrapper
    return function(*args, **kwargs)
  File "/usr/lib/python3.7/site-packages/mitmproxy/addons/clientplayback.py", line 204, in start_replay
    host = hf.request.headers.pop(":authority")
  File "/usr/lib/python3.7/_collections_abc.py", line 795, in pop
    value = self[key]
  File "/usr/lib/python3.7/site-packages/mitmproxy/coretypes/multidict.py", line 39, in __getitem__
    raise KeyError(key)
KeyError: ':authority'

mitmproxy has crashed!
Please lodge a bug report at:
	https://github.com/mitmproxy/mitmproxy


##### System information

Mitmproxy: 4.0.4
Python:    3.7.3
OpenSSL:   OpenSSL 1.1.1c  28 May 2019
Platform:  Linux-5.1.6-arch1-1-ARCH-x86_64-with-arch

--- 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/clientplayback.py`
Content:
```
1 import queue
2 import threading
3 import typing
4 import time
5 
6 from mitmproxy import log
7 from mitmproxy import controller
8 from mitmproxy import exceptions
9 from mitmproxy import http
10 from mitmproxy import flow
11 from mitmproxy import options
12 from mitmproxy import connections
13 from mitmproxy.net import server_spec, tls
14 from mitmproxy.net.http import http1
15 from mitmproxy.coretypes import basethread
16 from mitmproxy.utils import human
17 from mitmproxy import ctx
18 from mitmproxy import io
19 from mitmproxy import command
20 import mitmproxy.types
21 
22 
23 class RequestReplayThread(basethread.BaseThread):
24     daemon = True
25 
26     def __init__(
27             self,
28             opts: options.Options,
29             channel: controller.Channel,
30             queue: queue.Queue,
31     ) -> None:
32         self.options = opts
33         self.channel = channel
34         self.queue = queue
35         self.inflight = threading.Event()
36         super().__init__("RequestReplayThread")
37 
38     def run(self):
39         while True:
40             f = self.queue.get()
41             self.inflight.set()
42             self.replay(f)
43             self.inflight.clear()
44 
45     def replay(self, f):  # pragma: no cover
46         f.live = True
47         r = f.request
48         bsl = human.parse_size(self.options.body_size_limit)
49         first_line_format_backup = r.first_line_format
50         server = None
51         try:
52             f.response = None
53 
54             # If we have a channel, run script hooks.
55             request_reply = self.channel.ask("request", f)
56             if isinstance(request_reply, http.HTTPResponse):
57                 f.response = request_reply
58 
59             if not f.response:
60                 # In all modes, we directly connect to the server displayed
61                 if self.options.mode.startswith("upstream:"):
62                     server_address = server_spec.parse_with_mode(self.options.mode)[1].address
63                     server = connections.ServerConnection(server_address)
64                     server.connect()
65                     if r.scheme == "https":
66                         connect_request = http.make_connect_request((r.data.host, r.port))
67                         server.wfile.write(http1.assemble_request(connect_request))
68                         server.wfile.flush()
69                         resp = http1.read_response(
70                             server.rfile,
71                             connect_request,
72                             body_size_limit=bsl
73                         )
74                         if resp.status_code != 200:
75                             raise exceptions.ReplayException(
76                                 "Upstream server refuses CONNECT request"
77                             )
78                         server.establish_tls(
79                             sni=f.server_conn.sni,
80                             **tls.client_arguments_from_options(self.options)
81                         )
82                         r.first_line_format = "relative"
83                     else:
84                         r.first_line_format = "absolute"
85                 else:
86                     server_address = (r.host, r.port)
87                     server = connections.ServerConnection(server_address)
88                     server.connect()
89                     if r.scheme == "https":
90                         server.establish_tls(
91                             sni=f.server_conn.sni,
92                             **tls.client_arguments_from_options(self.options)
93                         )
94                     r.first_line_format = "relative"
95 
96                 server.wfile.write(http1.assemble_request(r))
97                 server.wfile.flush()
98                 r.timestamp_start = r.timestamp_end = time.time()
99 
100                 if f.server_conn:
101                     f.server_conn.close()
102                 f.server_conn = server
103 
104                 f.response = http.HTTPResponse.wrap(
105                     http1.read_response(server.rfile, r, body_size_limit=bsl)
106                 )
107             response_reply = self.channel.ask("response", f)
108             if response_reply == exceptions.Kill:
109                 raise exceptions.Kill()
110         except (exceptions.ReplayException, exceptions.NetlibException) as e:
111             f.error = flow.Error(str(e))
112             self.channel.ask("error", f)
113         except exceptions.Kill:
114             self.channel.tell("log", log.LogEntry("Connection killed", "info"))
115         except Exception as e:
116             self.channel.tell("log", log.LogEntry(repr(e), "error"))
117         finally:
118             r.first_line_format = first_line_format_backup
119             f.live = False
120             if server.connected():
121                 server.finish()
122                 server.close()
123 
124 
125 class ClientPlayback:
126     def __init__(self):
127         self.q = queue.Queue()
128         self.thread: RequestReplayThread = None
129 
130     def check(self, f: http.HTTPFlow):
131         if f.live:
132             return "Can't replay live flow."
133         if f.intercepted:
134             return "Can't replay intercepted flow."
135         if not f.request:
136             return "Can't replay flow with missing request."
137         if f.request.raw_content is None:
138             return "Can't replay flow with missing content."
139 
140     def load(self, loader):
141         loader.add_option(
142             "client_replay", typing.Sequence[str], [],
143             "Replay client requests from a saved file."
144         )
145 
146     def running(self):
147         self.thread = RequestReplayThread(
148             ctx.options,
149             ctx.master.channel,
150             self.q,
151         )
152         self.thread.start()
153 
154     def configure(self, updated):
155         if "client_replay" in updated and ctx.options.client_replay:
156             try:
157                 flows = io.read_flows_from_paths(ctx.options.client_replay)
158             except exceptions.FlowReadException as e:
159                 raise exceptions.OptionsError(str(e))
160             self.start_replay(flows)
161 
162     @command.command("replay.client.count")
163     def count(self) -> int:
164         """
165             Approximate number of flows queued for replay.
166         """
167         inflight = 1 if self.thread and self.thread.inflight.is_set() else 0
168         return self.q.qsize() + inflight
169 
170     @command.command("replay.client.stop")
171     def stop_replay(self) -> None:
172         """
173             Clear the replay queue.
174         """
175         with self.q.mutex:
176             lst = list(self.q.queue)
177             self.q.queue.clear()
178             for f in lst:
179                 f.revert()
180             ctx.master.addons.trigger("update", lst)
181         ctx.log.alert("Client replay queue cleared.")
182 
183     @command.command("replay.client")
184     def start_replay(self, flows: typing.Sequence[flow.Flow]) -> None:
185         """
186             Add flows to the replay queue, skipping flows that can't be replayed.
187         """
188         lst = []
189         for f in flows:
190             hf = typing.cast(http.HTTPFlow, f)
191 
192             err = self.check(hf)
193             if err:
194                 ctx.log.warn(err)
195                 continue
196 
197             lst.append(hf)
198             # Prepare the flow for replay
199             hf.backup()
200             hf.request.is_replay = True
201             hf.response = None
202             hf.error = None
203             # https://github.com/mitmproxy/mitmproxy/issues/2197
204             if hf.request.http_version == "HTTP/2.0":
205                 hf.request.http_version = "HTTP/1.1"
206                 host = hf.request.headers.pop(":authority")
207                 hf.request.headers.insert(0, "host", host)
208             self.q.put(hf)
209         ctx.master.addons.trigger("update", lst)
210 
211     @command.command("replay.client.file")
212     def load_file(self, path: mitmproxy.types.Path) -> None:
213         """
214             Load flows from file, and add them to the replay queue.
215         """
216         try:
217             flows = io.read_flows_from_paths([path])
218         except exceptions.FlowReadException as e:
219             raise exceptions.CommandError(str(e))
220         self.start_replay(flows)
221
```
--- 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/clientplayback.py b/mitmproxy/addons/clientplayback.py
--- a/mitmproxy/addons/clientplayback.py
+++ b/mitmproxy/addons/clientplayback.py
@@ -203,8 +203,9 @@
             # https://github.com/mitmproxy/mitmproxy/issues/2197
             if hf.request.http_version == "HTTP/2.0":
                 hf.request.http_version = "HTTP/1.1"
-                host = hf.request.headers.pop(":authority")
-                hf.request.headers.insert(0, "host", host)
+                host = hf.request.headers.pop(":authority", None)
+                if host is not None:
+                    hf.request.headers.insert(0, "host", host)
             self.q.put(hf)
         ctx.master.addons.trigger("update", lst)

{"golden_diff": "diff --git a/mitmproxy/addons/clientplayback.py b/mitmproxy/addons/clientplayback.py\n--- a/mitmproxy/addons/clientplayback.py\n+++ b/mitmproxy/addons/clientplayback.py\n@@ -203,8 +203,9 @@\n             # https://github.com/mitmproxy/mitmproxy/issues/2197\n             if hf.request.http_version == \"HTTP/2.0\":\n                 hf.request.http_version = \"HTTP/1.1\"\n-                host = hf.request.headers.pop(\":authority\")\n-                hf.request.headers.insert(0, \"host\", host)\n+                host = hf.request.headers.pop(\":authority\", None)\n+                if host is not None:\n+                    hf.request.headers.insert(0, \"host\", host)\n             self.q.put(hf)\n         ctx.master.addons.trigger(\"update\", lst)\n", "issue": "removing \":authority\" pseudo header, then executing request crashes mitmproxy\n##### Steps to reproduce the problem:\r\n\r\n1. Run mitmproxy\r\n2. Visit some http2 website, e.g. google.com\r\n3. Edit the request (enter + e + 7) and remove the :authority header (d + q).\r\n4. Replay the request (r).\r\n5. mitmproxy crashes\r\n\r\nStacktrace:\r\nTraceback (most recent call last):\r\n  File \"/usr/lib/python3.7/site-packages/mitmproxy/master.py\", line 86, in run_loop\r\n    loop()oding:            gzip, deflate, br\r\n  File \"/usr/lib/python3.7/site-packages/urwid/main_loop.py\", line 286, in run\r\n    self._run()e-requests:  1\r\n  File \"/usr/lib/python3.7/site-packages/urwid/main_loop.py\", line 384, in _run\r\n    self.event_loop.run() tab to view                                           [m:auto]\r\n  File \"/usr/lib/python3.7/site-packages/urwid/main_loop.py\", line 1484, in run\r\n    reraise(*exc_info)\r\n  File \"/usr/lib/python3.7/site-packages/urwid/compat.py\", line 58, in reraise\r\n    raise value\r\n  File \"/usr/lib/python3.7/asyncio/events.py\", line 88, in _run\r\n    self._context.run(self._callback, *self._args)\r\n  File \"/usr/lib/python3.7/site-packages/urwid/raw_display.py\", line 404, in <lambda>\r\n    event_loop, callback, self.get_available_raw_input())\r\n  File \"/usr/lib/python3.7/site-packages/urwid/raw_display.py\", line 502, in parse_input\r\n    callback(processed, processed_codes)\r\n  File \"/usr/lib/python3.7/site-packages/urwid/main_loop.py\", line 411, in _update\r\n    self.process_input(keys)\r\n  File \"/usr/lib/python3.7/site-packages/urwid/main_loop.py\", line 511, in process_input\r\n    k = self._topmost_widget.keypress(self.screen_size, k)\r\n  File \"/usr/lib/python3.7/site-packages/mitmproxy/tools/console/window.py\", line 313, in keypress\r\n    k\r\n  File \"/usr/lib/python3.7/site-packages/mitmproxy/tools/console/keymap.py\", line 143, in handle\r\n    return self.executor(b.command)\r\n  File \"/usr/lib/python3.7/site-packages/mitmproxy/tools/console/commandexecutor.py\", line 17, in __call__\r\n    ret = self.master.commands.execute(cmd)\r\n  File \"/usr/lib/python3.7/site-packages/mitmproxy/command.py\", line 245, in execute\r\n    return self.call_strings(parts[0], parts[1:])\r\n  File \"/usr/lib/python3.7/site-packages/mitmproxy/command.py\", line 233, in call_strings\r\n    return self.commands[path].call(args)\r\n  File \"/usr/lib/python3.7/site-packages/mitmproxy/command.py\", line 105, in call\r\n    ret = self.func(*self.prepare_args(args))\r\n  File \"/usr/lib/python3.7/site-packages/mitmproxy/command.py\", line 275, in wrapper\r\n    return function(*args, **kwargs)\r\n  File \"/usr/lib/python3.7/site-packages/mitmproxy/addons/clientplayback.py\", line 204, in start_replay\r\n    host = hf.request.headers.pop(\":authority\")\r\n  File \"/usr/lib/python3.7/_collections_abc.py\", line 795, in pop\r\n    value = self[key]\r\n  File \"/usr/lib/python3.7/site-packages/mitmproxy/coretypes/multidict.py\", line 39, in __getitem__\r\n    raise KeyError(key)\r\nKeyError: ':authority'\r\n\r\nmitmproxy has crashed!\r\nPlease lodge a bug report at:\r\n\thttps://github.com/mitmproxy/mitmproxy\r\n\r\n\r\n##### System information\r\n\r\nMitmproxy: 4.0.4\r\nPython:    3.7.3\r\nOpenSSL:   OpenSSL 1.1.1c  28 May 2019\r\nPlatform:  Linux-5.1.6-arch1-1-ARCH-x86_64-with-arch\n", "before_files": [{"content": "import queue\nimport threading\nimport typing\nimport time\n\nfrom mitmproxy import log\nfrom mitmproxy import controller\nfrom mitmproxy import exceptions\nfrom mitmproxy import http\nfrom mitmproxy import flow\nfrom mitmproxy import options\nfrom mitmproxy import connections\nfrom mitmproxy.net import server_spec, tls\nfrom mitmproxy.net.http import http1\nfrom mitmproxy.coretypes import basethread\nfrom mitmproxy.utils import human\nfrom mitmproxy import ctx\nfrom mitmproxy import io\nfrom mitmproxy import command\nimport mitmproxy.types\n\n\nclass RequestReplayThread(basethread.BaseThread):\n    daemon = True\n\n    def __init__(\n            self,\n            opts: options.Options,\n            channel: controller.Channel,\n            queue: queue.Queue,\n    ) -> None:\n        self.options = opts\n        self.channel = channel\n        self.queue = queue\n        self.inflight = threading.Event()\n        super().__init__(\"RequestReplayThread\")\n\n    def run(self):\n        while True:\n            f = self.queue.get()\n            self.inflight.set()\n            self.replay(f)\n            self.inflight.clear()\n\n    def replay(self, f):  # pragma: no cover\n        f.live = True\n        r = f.request\n        bsl = human.parse_size(self.options.body_size_limit)\n        first_line_format_backup = r.first_line_format\n        server = None\n        try:\n            f.response = None\n\n            # If we have a channel, run script hooks.\n            request_reply = self.channel.ask(\"request\", f)\n            if isinstance(request_reply, http.HTTPResponse):\n                f.response = request_reply\n\n            if not f.response:\n                # In all modes, we directly connect to the server displayed\n                if self.options.mode.startswith(\"upstream:\"):\n                    server_address = server_spec.parse_with_mode(self.options.mode)[1].address\n                    server = connections.ServerConnection(server_address)\n                    server.connect()\n                    if r.scheme == \"https\":\n                        connect_request = http.make_connect_request((r.data.host, r.port))\n                        server.wfile.write(http1.assemble_request(connect_request))\n                        server.wfile.flush()\n                        resp = http1.read_response(\n                            server.rfile,\n                            connect_request,\n                            body_size_limit=bsl\n                        )\n                        if resp.status_code != 200:\n                            raise exceptions.ReplayException(\n                                \"Upstream server refuses CONNECT request\"\n                            )\n                        server.establish_tls(\n                            sni=f.server_conn.sni,\n                            **tls.client_arguments_from_options(self.options)\n                        )\n                        r.first_line_format = \"relative\"\n                    else:\n                        r.first_line_format = \"absolute\"\n                else:\n                    server_address = (r.host, r.port)\n                    server = connections.ServerConnection(server_address)\n                    server.connect()\n                    if r.scheme == \"https\":\n                        server.establish_tls(\n                            sni=f.server_conn.sni,\n                            **tls.client_arguments_from_options(self.options)\n                        )\n                    r.first_line_format = \"relative\"\n\n                server.wfile.write(http1.assemble_request(r))\n                server.wfile.flush()\n                r.timestamp_start = r.timestamp_end = time.time()\n\n                if f.server_conn:\n                    f.server_conn.close()\n                f.server_conn = server\n\n                f.response = http.HTTPResponse.wrap(\n                    http1.read_response(server.rfile, r, body_size_limit=bsl)\n                )\n            response_reply = self.channel.ask(\"response\", f)\n            if response_reply == exceptions.Kill:\n                raise exceptions.Kill()\n        except (exceptions.ReplayException, exceptions.NetlibException) as e:\n            f.error = flow.Error(str(e))\n            self.channel.ask(\"error\", f)\n        except exceptions.Kill:\n            self.channel.tell(\"log\", log.LogEntry(\"Connection killed\", \"info\"))\n        except Exception as e:\n            self.channel.tell(\"log\", log.LogEntry(repr(e), \"error\"))\n        finally:\n            r.first_line_format = first_line_format_backup\n            f.live = False\n            if server.connected():\n                server.finish()\n                server.close()\n\n\nclass ClientPlayback:\n    def __init__(self):\n        self.q = queue.Queue()\n        self.thread: RequestReplayThread = None\n\n    def check(self, f: http.HTTPFlow):\n        if f.live:\n            return \"Can't replay live flow.\"\n        if f.intercepted:\n            return \"Can't replay intercepted flow.\"\n        if not f.request:\n            return \"Can't replay flow with missing request.\"\n        if f.request.raw_content is None:\n            return \"Can't replay flow with missing content.\"\n\n    def load(self, loader):\n        loader.add_option(\n            \"client_replay\", typing.Sequence[str], [],\n            \"Replay client requests from a saved file.\"\n        )\n\n    def running(self):\n        self.thread = RequestReplayThread(\n            ctx.options,\n            ctx.master.channel,\n            self.q,\n        )\n        self.thread.start()\n\n    def configure(self, updated):\n        if \"client_replay\" in updated and ctx.options.client_replay:\n            try:\n                flows = io.read_flows_from_paths(ctx.options.client_replay)\n            except exceptions.FlowReadException as e:\n                raise exceptions.OptionsError(str(e))\n            self.start_replay(flows)\n\n    @command.command(\"replay.client.count\")\n    def count(self) -> int:\n        \"\"\"\n            Approximate number of flows queued for replay.\n        \"\"\"\n        inflight = 1 if self.thread and self.thread.inflight.is_set() else 0\n        return self.q.qsize() + inflight\n\n    @command.command(\"replay.client.stop\")\n    def stop_replay(self) -> None:\n        \"\"\"\n            Clear the replay queue.\n        \"\"\"\n        with self.q.mutex:\n            lst = list(self.q.queue)\n            self.q.queue.clear()\n            for f in lst:\n                f.revert()\n            ctx.master.addons.trigger(\"update\", lst)\n        ctx.log.alert(\"Client replay queue cleared.\")\n\n    @command.command(\"replay.client\")\n    def start_replay(self, flows: typing.Sequence[flow.Flow]) -> None:\n        \"\"\"\n            Add flows to the replay queue, skipping flows that can't be replayed.\n        \"\"\"\n        lst = []\n        for f in flows:\n            hf = typing.cast(http.HTTPFlow, f)\n\n            err = self.check(hf)\n            if err:\n                ctx.log.warn(err)\n                continue\n\n            lst.append(hf)\n            # Prepare the flow for replay\n            hf.backup()\n            hf.request.is_replay = True\n            hf.response = None\n            hf.error = None\n            # https://github.com/mitmproxy/mitmproxy/issues/2197\n            if hf.request.http_version == \"HTTP/2.0\":\n                hf.request.http_version = \"HTTP/1.1\"\n                host = hf.request.headers.pop(\":authority\")\n                hf.request.headers.insert(0, \"host\", host)\n            self.q.put(hf)\n        ctx.master.addons.trigger(\"update\", lst)\n\n    @command.command(\"replay.client.file\")\n    def load_file(self, path: mitmproxy.types.Path) -> None:\n        \"\"\"\n            Load flows from file, and add them to the replay queue.\n        \"\"\"\n        try:\n            flows = io.read_flows_from_paths([path])\n        except exceptions.FlowReadException as e:\n            raise exceptions.CommandError(str(e))\n        self.start_replay(flows)\n", "path": "mitmproxy/addons/clientplayback.py"}], "after_files": [{"content": "import queue\nimport threading\nimport typing\nimport time\n\nfrom mitmproxy import log\nfrom mitmproxy import controller\nfrom mitmproxy import exceptions\nfrom mitmproxy import http\nfrom mitmproxy import flow\nfrom mitmproxy import options\nfrom mitmproxy import connections\nfrom mitmproxy.net import server_spec, tls\nfrom mitmproxy.net.http import http1\nfrom mitmproxy.coretypes import basethread\nfrom mitmproxy.utils import human\nfrom mitmproxy import ctx\nfrom mitmproxy import io\nfrom mitmproxy import command\nimport mitmproxy.types\n\n\nclass RequestReplayThread(basethread.BaseThread):\n    daemon = True\n\n    def __init__(\n            self,\n            opts: options.Options,\n            channel: controller.Channel,\n            queue: queue.Queue,\n    ) -> None:\n        self.options = opts\n        self.channel = channel\n        self.queue = queue\n        self.inflight = threading.Event()\n        super().__init__(\"RequestReplayThread\")\n\n    def run(self):\n        while True:\n            f = self.queue.get()\n            self.inflight.set()\n            self.replay(f)\n            self.inflight.clear()\n\n    def replay(self, f):  # pragma: no cover\n        f.live = True\n        r = f.request\n        bsl = human.parse_size(self.options.body_size_limit)\n        first_line_format_backup = r.first_line_format\n        server = None\n        try:\n            f.response = None\n\n            # If we have a channel, run script hooks.\n            request_reply = self.channel.ask(\"request\", f)\n            if isinstance(request_reply, http.HTTPResponse):\n                f.response = request_reply\n\n            if not f.response:\n                # In all modes, we directly connect to the server displayed\n                if self.options.mode.startswith(\"upstream:\"):\n                    server_address = server_spec.parse_with_mode(self.options.mode)[1].address\n                    server = connections.ServerConnection(server_address)\n                    server.connect()\n                    if r.scheme == \"https\":\n                        connect_request = http.make_connect_request((r.data.host, r.port))\n                        server.wfile.write(http1.assemble_request(connect_request))\n                        server.wfile.flush()\n                        resp = http1.read_response(\n                            server.rfile,\n                            connect_request,\n                            body_size_limit=bsl\n                        )\n                        if resp.status_code != 200:\n                            raise exceptions.ReplayException(\n                                \"Upstream server refuses CONNECT request\"\n                            )\n                        server.establish_tls(\n                            sni=f.server_conn.sni,\n                            **tls.client_arguments_from_options(self.options)\n                        )\n                        r.first_line_format = \"relative\"\n                    else:\n                        r.first_line_format = \"absolute\"\n                else:\n                    server_address = (r.host, r.port)\n                    server = connections.ServerConnection(server_address)\n                    server.connect()\n                    if r.scheme == \"https\":\n                        server.establish_tls(\n                            sni=f.server_conn.sni,\n                            **tls.client_arguments_from_options(self.options)\n                        )\n                    r.first_line_format = \"relative\"\n\n                server.wfile.write(http1.assemble_request(r))\n                server.wfile.flush()\n                r.timestamp_start = r.timestamp_end = time.time()\n\n                if f.server_conn:\n                    f.server_conn.close()\n                f.server_conn = server\n\n                f.response = http.HTTPResponse.wrap(\n                    http1.read_response(server.rfile, r, body_size_limit=bsl)\n                )\n            response_reply = self.channel.ask(\"response\", f)\n            if response_reply == exceptions.Kill:\n                raise exceptions.Kill()\n        except (exceptions.ReplayException, exceptions.NetlibException) as e:\n            f.error = flow.Error(str(e))\n            self.channel.ask(\"error\", f)\n        except exceptions.Kill:\n            self.channel.tell(\"log\", log.LogEntry(\"Connection killed\", \"info\"))\n        except Exception as e:\n            self.channel.tell(\"log\", log.LogEntry(repr(e), \"error\"))\n        finally:\n            r.first_line_format = first_line_format_backup\n            f.live = False\n            if server.connected():\n                server.finish()\n                server.close()\n\n\nclass ClientPlayback:\n    def __init__(self):\n        self.q = queue.Queue()\n        self.thread: RequestReplayThread = None\n\n    def check(self, f: http.HTTPFlow):\n        if f.live:\n            return \"Can't replay live flow.\"\n        if f.intercepted:\n            return \"Can't replay intercepted flow.\"\n        if not f.request:\n            return \"Can't replay flow with missing request.\"\n        if f.request.raw_content is None:\n            return \"Can't replay flow with missing content.\"\n\n    def load(self, loader):\n        loader.add_option(\n            \"client_replay\", typing.Sequence[str], [],\n            \"Replay client requests from a saved file.\"\n        )\n\n    def running(self):\n        self.thread = RequestReplayThread(\n            ctx.options,\n            ctx.master.channel,\n            self.q,\n        )\n        self.thread.start()\n\n    def configure(self, updated):\n        if \"client_replay\" in updated and ctx.options.client_replay:\n            try:\n                flows = io.read_flows_from_paths(ctx.options.client_replay)\n            except exceptions.FlowReadException as e:\n                raise exceptions.OptionsError(str(e))\n            self.start_replay(flows)\n\n    @command.command(\"replay.client.count\")\n    def count(self) -> int:\n        \"\"\"\n            Approximate number of flows queued for replay.\n        \"\"\"\n        inflight = 1 if self.thread and self.thread.inflight.is_set() else 0\n        return self.q.qsize() + inflight\n\n    @command.command(\"replay.client.stop\")\n    def stop_replay(self) -> None:\n        \"\"\"\n            Clear the replay queue.\n        \"\"\"\n        with self.q.mutex:\n            lst = list(self.q.queue)\n            self.q.queue.clear()\n            for f in lst:\n                f.revert()\n            ctx.master.addons.trigger(\"update\", lst)\n        ctx.log.alert(\"Client replay queue cleared.\")\n\n    @command.command(\"replay.client\")\n    def start_replay(self, flows: typing.Sequence[flow.Flow]) -> None:\n        \"\"\"\n            Add flows to the replay queue, skipping flows that can't be replayed.\n        \"\"\"\n        lst = []\n        for f in flows:\n            hf = typing.cast(http.HTTPFlow, f)\n\n            err = self.check(hf)\n            if err:\n                ctx.log.warn(err)\n                continue\n\n            lst.append(hf)\n            # Prepare the flow for replay\n            hf.backup()\n            hf.request.is_replay = True\n            hf.response = None\n            hf.error = None\n            # https://github.com/mitmproxy/mitmproxy/issues/2197\n            if hf.request.http_version == \"HTTP/2.0\":\n                hf.request.http_version = \"HTTP/1.1\"\n                host = hf.request.headers.pop(\":authority\", None)\n                if host is not None:\n                    hf.request.headers.insert(0, \"host\", host)\n            self.q.put(hf)\n        ctx.master.addons.trigger(\"update\", lst)\n\n    @command.command(\"replay.client.file\")\n    def load_file(self, path: mitmproxy.types.Path) -> None:\n        \"\"\"\n            Load flows from file, and add them to the replay queue.\n        \"\"\"\n        try:\n            flows = io.read_flows_from_paths([path])\n        except exceptions.FlowReadException as e:\n            raise exceptions.CommandError(str(e))\n        self.start_replay(flows)\n", "path": "mitmproxy/addons/clientplayback.py"}]}