We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
CVE-2021-40978 - Path Traversal.
Hey!

We have verified a security flaw in the current version of MKdocs, a path traversal failure affecting the built-in dev-server.

That flaw turns the server susceptible to providing data outside the scope of the application allowing anyone to request sensitive files.

If you need further information, don't hesitate to get in touch with me.

https://cve.mitre.org/cgi-bin/cvename.cgi?name=CVE-2021-40978
https://github.com/nisdn/CVE-2021-40978


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `mkdocs/livereload/__init__.py`
Content:
```
1 import functools
2 import io
3 import logging
4 import mimetypes
5 import os
6 import os.path
7 import re
8 import socketserver
9 import threading
10 import time
11 import warnings
12 import wsgiref.simple_server
13 
14 import watchdog.events
15 import watchdog.observers.polling
16 
17 
18 class _LoggerAdapter(logging.LoggerAdapter):
19     def process(self, msg, kwargs):
20         return time.strftime("[%H:%M:%S] ") + msg, kwargs
21 
22 
23 log = _LoggerAdapter(logging.getLogger(__name__), {})
24 
25 
26 class LiveReloadServer(socketserver.ThreadingMixIn, wsgiref.simple_server.WSGIServer):
27     daemon_threads = True
28     poll_response_timeout = 60
29 
30     def __init__(
31         self,
32         builder,
33         host,
34         port,
35         root,
36         mount_path="/",
37         polling_interval=0.5,
38         shutdown_delay=0.25,
39         **kwargs,
40     ):
41         self.builder = builder
42         self.server_name = host
43         self.server_port = port
44         self.root = os.path.abspath(root)
45         self.mount_path = ("/" + mount_path.lstrip("/")).rstrip("/") + "/"
46         self.url = f"http://{self.server_name}:{self.server_port}{self.mount_path}"
47         self.build_delay = 0.1
48         self.shutdown_delay = shutdown_delay
49         # To allow custom error pages.
50         self.error_handler = lambda code: None
51 
52         super().__init__((host, port), _Handler, **kwargs)
53         self.set_app(self.serve_request)
54 
55         self._wanted_epoch = _timestamp()  # The version of the site that started building.
56         self._visible_epoch = self._wanted_epoch  # Latest fully built version of the site.
57         self._epoch_cond = threading.Condition()  # Must be held when accessing _visible_epoch.
58 
59         self._to_rebuild = {}  # Used as an ordered set of functions to call.
60         self._rebuild_cond = threading.Condition()  # Must be held when accessing _to_rebuild.
61 
62         self._shutdown = False
63         self.serve_thread = threading.Thread(target=lambda: self.serve_forever(shutdown_delay))
64         self.observer = watchdog.observers.polling.PollingObserver(timeout=polling_interval)
65 
66     def watch(self, path, func=None, recursive=True):
67         """Add the 'path' to watched paths, call the function and reload when any file changes under it."""
68         path = os.path.abspath(path)
69         if func in (None, self.builder):
70             func = self.builder
71         else:
72             warnings.warn(
73                 "Plugins should not pass the 'func' parameter of watch(). "
74                 "The ability to execute custom callbacks will be removed soon.",
75                 DeprecationWarning,
76                 stacklevel=2,
77             )
78 
79         def callback(event):
80             if event.is_directory:
81                 return
82             log.debug(str(event))
83             with self._rebuild_cond:
84                 self._to_rebuild[func] = True
85                 self._rebuild_cond.notify_all()
86 
87         handler = watchdog.events.FileSystemEventHandler()
88         handler.on_any_event = callback
89         log.debug(f"Watching '{path}'")
90         self.observer.schedule(handler, path, recursive=recursive)
91 
92     def serve(self):
93         self.observer.start()
94 
95         log.info(f"Serving on {self.url}")
96         self.serve_thread.start()
97 
98         self._build_loop()
99 
100     def _build_loop(self):
101         while True:
102             with self._rebuild_cond:
103                 while not self._rebuild_cond.wait_for(
104                     lambda: self._to_rebuild or self._shutdown, timeout=self.shutdown_delay
105                 ):
106                     # We could have used just one wait instead of a loop + timeout, but we need
107                     # occasional breaks, otherwise on Windows we can't receive KeyboardInterrupt.
108                     pass
109                 if self._shutdown:
110                     break
111                 log.info("Detected file changes")
112                 while self._rebuild_cond.wait(timeout=self.build_delay):
113                     log.debug("Waiting for file changes to stop happening")
114 
115                 self._wanted_epoch = _timestamp()
116                 funcs = list(self._to_rebuild)
117                 self._to_rebuild.clear()
118 
119             for func in funcs:
120                 func()
121 
122             with self._epoch_cond:
123                 log.info("Reloading browsers")
124                 self._visible_epoch = self._wanted_epoch
125                 self._epoch_cond.notify_all()
126 
127     def shutdown(self):
128         self.observer.stop()
129         with self._rebuild_cond:
130             self._shutdown = True
131             self._rebuild_cond.notify_all()
132 
133         if self.serve_thread.is_alive():
134             super().shutdown()
135             self.serve_thread.join()
136             self.observer.join()
137 
138     def serve_request(self, environ, start_response):
139         try:
140             result = self._serve_request(environ, start_response)
141         except Exception:
142             code = 500
143             msg = "500 Internal Server Error"
144             log.exception(msg)
145         else:
146             if result is not None:
147                 return result
148             code = 404
149             msg = "404 Not Found"
150 
151         error_content = None
152         try:
153             error_content = self.error_handler(code)
154         except Exception:
155             log.exception("Failed to render an error message!")
156         if error_content is None:
157             error_content = msg.encode()
158 
159         start_response(msg, [("Content-Type", "text/html")])
160         return [error_content]
161 
162     def _serve_request(self, environ, start_response):
163         # https://bugs.python.org/issue16679
164         # https://github.com/bottlepy/bottle/blob/f9b1849db4/bottle.py#L984
165         path = environ["PATH_INFO"].encode("latin-1").decode("utf-8", "ignore")
166 
167         m = re.fullmatch(r"/livereload/([0-9]+)/[0-9]+", path)
168         if m:
169             epoch = int(m[1])
170             start_response("200 OK", [("Content-Type", "text/plain")])
171 
172             def condition():
173                 return self._visible_epoch > epoch
174 
175             with self._epoch_cond:
176                 if not condition():
177                     # Stall the browser, respond as soon as there's something new.
178                     # If there's not, respond anyway after a minute.
179                     self._log_poll_request(environ.get("HTTP_REFERER"), request_id=path)
180                     self._epoch_cond.wait_for(condition, timeout=self.poll_response_timeout)
181                 return [b"%d" % self._visible_epoch]
182 
183         if path == "/js/livereload.js":
184             file_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "livereload.js")
185         elif path.startswith(self.mount_path):
186             rel_file_path = path[len(self.mount_path):].lstrip("/")
187             if path.endswith("/"):
188                 rel_file_path += "index.html"
189             file_path = os.path.join(self.root, rel_file_path)
190         elif path == "/":
191             start_response("302 Found", [("Location", self.mount_path)])
192             return []
193         else:
194             return None  # Not found
195 
196         # Wait until the ongoing rebuild (if any) finishes, so we're not serving a half-built site.
197         with self._epoch_cond:
198             self._epoch_cond.wait_for(lambda: self._visible_epoch == self._wanted_epoch)
199             epoch = self._visible_epoch
200 
201         try:
202             file = open(file_path, "rb")
203         except OSError:
204             if not path.endswith("/") and os.path.isfile(os.path.join(file_path, "index.html")):
205                 start_response("302 Found", [("Location", path + "/")])
206                 return []
207             return None  # Not found
208 
209         if file_path.endswith(".html"):
210             with file:
211                 content = file.read()
212             content = self._inject_js_into_html(content, epoch)
213             file = io.BytesIO(content)
214             content_length = len(content)
215         else:
216             content_length = os.path.getsize(file_path)
217 
218         content_type = self._guess_type(file_path)
219         start_response(
220             "200 OK", [("Content-Type", content_type), ("Content-Length", str(content_length))]
221         )
222         return wsgiref.util.FileWrapper(file)
223 
224     @classmethod
225     def _inject_js_into_html(cls, content, epoch):
226         try:
227             body_end = content.rindex(b"</body>")
228         except ValueError:
229             body_end = len(content)
230         # The page will reload if the livereload poller returns a newer epoch than what it knows.
231         # The other timestamp becomes just a unique identifier for the initiating page.
232         return (
233             b'%b<script src="/js/livereload.js"></script><script>livereload(%d, %d);</script>%b'
234             % (content[:body_end], epoch, _timestamp(), content[body_end:])
235         )
236 
237     @classmethod
238     @functools.lru_cache()  # "Cache" to not repeat the same message for the same browser tab.
239     def _log_poll_request(cls, url, request_id):
240         log.info(f"Browser connected: {url}")
241 
242     def _guess_type(cls, path):
243         # MkDocs only ensures a few common types (as seen in livereload_tests.py::test_mime_types).
244         # Other uncommon types will not be accepted.
245         if path.endswith((".js", ".JS")):
246             return "application/javascript"
247         if path.endswith(".gz"):
248             return "application/gzip"
249 
250         guess, _ = mimetypes.guess_type(path)
251         if guess:
252             return guess
253         return "application/octet-stream"
254 
255 
256 class _Handler(wsgiref.simple_server.WSGIRequestHandler):
257     def log_request(self, code="-", size="-"):
258         level = logging.DEBUG if str(code) == "200" else logging.WARNING
259         log.log(level, f'"{self.requestline}" code {code}')
260 
261     def log_message(self, format, *args):
262         log.debug(format, *args)
263 
264 
265 def _timestamp():
266     return round(time.monotonic() * 1000)
267
```
--- 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/mkdocs/livereload/__init__.py b/mkdocs/livereload/__init__.py
--- a/mkdocs/livereload/__init__.py
+++ b/mkdocs/livereload/__init__.py
@@ -4,6 +4,7 @@
 import mimetypes
 import os
 import os.path
+import posixpath
 import re
 import socketserver
 import threading
@@ -183,9 +184,11 @@
         if path == "/js/livereload.js":
             file_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "livereload.js")
         elif path.startswith(self.mount_path):
-            rel_file_path = path[len(self.mount_path):].lstrip("/")
+            rel_file_path = path[len(self.mount_path):]
             if path.endswith("/"):
                 rel_file_path += "index.html"
+            # Prevent directory traversal - normalize the path.
+            rel_file_path = posixpath.normpath("/" + rel_file_path).lstrip("/")
             file_path = os.path.join(self.root, rel_file_path)
         elif path == "/":
             start_response("302 Found", [("Location", self.mount_path)])

{"golden_diff": "diff --git a/mkdocs/livereload/__init__.py b/mkdocs/livereload/__init__.py\n--- a/mkdocs/livereload/__init__.py\n+++ b/mkdocs/livereload/__init__.py\n@@ -4,6 +4,7 @@\n import mimetypes\n import os\n import os.path\n+import posixpath\n import re\n import socketserver\n import threading\n@@ -183,9 +184,11 @@\n         if path == \"/js/livereload.js\":\n             file_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), \"livereload.js\")\n         elif path.startswith(self.mount_path):\n-            rel_file_path = path[len(self.mount_path):].lstrip(\"/\")\n+            rel_file_path = path[len(self.mount_path):]\n             if path.endswith(\"/\"):\n                 rel_file_path += \"index.html\"\n+            # Prevent directory traversal - normalize the path.\n+            rel_file_path = posixpath.normpath(\"/\" + rel_file_path).lstrip(\"/\")\n             file_path = os.path.join(self.root, rel_file_path)\n         elif path == \"/\":\n             start_response(\"302 Found\", [(\"Location\", self.mount_path)])\n", "issue": "CVE-2021-40978 - Path Traversal.\nHey!\r\n\r\nWe have verified a security flaw in the current version of MKdocs, a path traversal failure affecting the built-in dev-server.\r\n\r\nThat flaw turns the server susceptible to providing data outside the scope of the application allowing anyone to request sensitive files.\r\n\r\nIf you need further information, don't hesitate to get in touch with me.\r\n\r\nhttps://cve.mitre.org/cgi-bin/cvename.cgi?name=CVE-2021-40978\r\nhttps://github.com/nisdn/CVE-2021-40978\r\n\n", "before_files": [{"content": "import functools\nimport io\nimport logging\nimport mimetypes\nimport os\nimport os.path\nimport re\nimport socketserver\nimport threading\nimport time\nimport warnings\nimport wsgiref.simple_server\n\nimport watchdog.events\nimport watchdog.observers.polling\n\n\nclass _LoggerAdapter(logging.LoggerAdapter):\n    def process(self, msg, kwargs):\n        return time.strftime(\"[%H:%M:%S] \") + msg, kwargs\n\n\nlog = _LoggerAdapter(logging.getLogger(__name__), {})\n\n\nclass LiveReloadServer(socketserver.ThreadingMixIn, wsgiref.simple_server.WSGIServer):\n    daemon_threads = True\n    poll_response_timeout = 60\n\n    def __init__(\n        self,\n        builder,\n        host,\n        port,\n        root,\n        mount_path=\"/\",\n        polling_interval=0.5,\n        shutdown_delay=0.25,\n        **kwargs,\n    ):\n        self.builder = builder\n        self.server_name = host\n        self.server_port = port\n        self.root = os.path.abspath(root)\n        self.mount_path = (\"/\" + mount_path.lstrip(\"/\")).rstrip(\"/\") + \"/\"\n        self.url = f\"http://{self.server_name}:{self.server_port}{self.mount_path}\"\n        self.build_delay = 0.1\n        self.shutdown_delay = shutdown_delay\n        # To allow custom error pages.\n        self.error_handler = lambda code: None\n\n        super().__init__((host, port), _Handler, **kwargs)\n        self.set_app(self.serve_request)\n\n        self._wanted_epoch = _timestamp()  # The version of the site that started building.\n        self._visible_epoch = self._wanted_epoch  # Latest fully built version of the site.\n        self._epoch_cond = threading.Condition()  # Must be held when accessing _visible_epoch.\n\n        self._to_rebuild = {}  # Used as an ordered set of functions to call.\n        self._rebuild_cond = threading.Condition()  # Must be held when accessing _to_rebuild.\n\n        self._shutdown = False\n        self.serve_thread = threading.Thread(target=lambda: self.serve_forever(shutdown_delay))\n        self.observer = watchdog.observers.polling.PollingObserver(timeout=polling_interval)\n\n    def watch(self, path, func=None, recursive=True):\n        \"\"\"Add the 'path' to watched paths, call the function and reload when any file changes under it.\"\"\"\n        path = os.path.abspath(path)\n        if func in (None, self.builder):\n            func = self.builder\n        else:\n            warnings.warn(\n                \"Plugins should not pass the 'func' parameter of watch(). \"\n                \"The ability to execute custom callbacks will be removed soon.\",\n                DeprecationWarning,\n                stacklevel=2,\n            )\n\n        def callback(event):\n            if event.is_directory:\n                return\n            log.debug(str(event))\n            with self._rebuild_cond:\n                self._to_rebuild[func] = True\n                self._rebuild_cond.notify_all()\n\n        handler = watchdog.events.FileSystemEventHandler()\n        handler.on_any_event = callback\n        log.debug(f\"Watching '{path}'\")\n        self.observer.schedule(handler, path, recursive=recursive)\n\n    def serve(self):\n        self.observer.start()\n\n        log.info(f\"Serving on {self.url}\")\n        self.serve_thread.start()\n\n        self._build_loop()\n\n    def _build_loop(self):\n        while True:\n            with self._rebuild_cond:\n                while not self._rebuild_cond.wait_for(\n                    lambda: self._to_rebuild or self._shutdown, timeout=self.shutdown_delay\n                ):\n                    # We could have used just one wait instead of a loop + timeout, but we need\n                    # occasional breaks, otherwise on Windows we can't receive KeyboardInterrupt.\n                    pass\n                if self._shutdown:\n                    break\n                log.info(\"Detected file changes\")\n                while self._rebuild_cond.wait(timeout=self.build_delay):\n                    log.debug(\"Waiting for file changes to stop happening\")\n\n                self._wanted_epoch = _timestamp()\n                funcs = list(self._to_rebuild)\n                self._to_rebuild.clear()\n\n            for func in funcs:\n                func()\n\n            with self._epoch_cond:\n                log.info(\"Reloading browsers\")\n                self._visible_epoch = self._wanted_epoch\n                self._epoch_cond.notify_all()\n\n    def shutdown(self):\n        self.observer.stop()\n        with self._rebuild_cond:\n            self._shutdown = True\n            self._rebuild_cond.notify_all()\n\n        if self.serve_thread.is_alive():\n            super().shutdown()\n            self.serve_thread.join()\n            self.observer.join()\n\n    def serve_request(self, environ, start_response):\n        try:\n            result = self._serve_request(environ, start_response)\n        except Exception:\n            code = 500\n            msg = \"500 Internal Server Error\"\n            log.exception(msg)\n        else:\n            if result is not None:\n                return result\n            code = 404\n            msg = \"404 Not Found\"\n\n        error_content = None\n        try:\n            error_content = self.error_handler(code)\n        except Exception:\n            log.exception(\"Failed to render an error message!\")\n        if error_content is None:\n            error_content = msg.encode()\n\n        start_response(msg, [(\"Content-Type\", \"text/html\")])\n        return [error_content]\n\n    def _serve_request(self, environ, start_response):\n        # https://bugs.python.org/issue16679\n        # https://github.com/bottlepy/bottle/blob/f9b1849db4/bottle.py#L984\n        path = environ[\"PATH_INFO\"].encode(\"latin-1\").decode(\"utf-8\", \"ignore\")\n\n        m = re.fullmatch(r\"/livereload/([0-9]+)/[0-9]+\", path)\n        if m:\n            epoch = int(m[1])\n            start_response(\"200 OK\", [(\"Content-Type\", \"text/plain\")])\n\n            def condition():\n                return self._visible_epoch > epoch\n\n            with self._epoch_cond:\n                if not condition():\n                    # Stall the browser, respond as soon as there's something new.\n                    # If there's not, respond anyway after a minute.\n                    self._log_poll_request(environ.get(\"HTTP_REFERER\"), request_id=path)\n                    self._epoch_cond.wait_for(condition, timeout=self.poll_response_timeout)\n                return [b\"%d\" % self._visible_epoch]\n\n        if path == \"/js/livereload.js\":\n            file_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), \"livereload.js\")\n        elif path.startswith(self.mount_path):\n            rel_file_path = path[len(self.mount_path):].lstrip(\"/\")\n            if path.endswith(\"/\"):\n                rel_file_path += \"index.html\"\n            file_path = os.path.join(self.root, rel_file_path)\n        elif path == \"/\":\n            start_response(\"302 Found\", [(\"Location\", self.mount_path)])\n            return []\n        else:\n            return None  # Not found\n\n        # Wait until the ongoing rebuild (if any) finishes, so we're not serving a half-built site.\n        with self._epoch_cond:\n            self._epoch_cond.wait_for(lambda: self._visible_epoch == self._wanted_epoch)\n            epoch = self._visible_epoch\n\n        try:\n            file = open(file_path, \"rb\")\n        except OSError:\n            if not path.endswith(\"/\") and os.path.isfile(os.path.join(file_path, \"index.html\")):\n                start_response(\"302 Found\", [(\"Location\", path + \"/\")])\n                return []\n            return None  # Not found\n\n        if file_path.endswith(\".html\"):\n            with file:\n                content = file.read()\n            content = self._inject_js_into_html(content, epoch)\n            file = io.BytesIO(content)\n            content_length = len(content)\n        else:\n            content_length = os.path.getsize(file_path)\n\n        content_type = self._guess_type(file_path)\n        start_response(\n            \"200 OK\", [(\"Content-Type\", content_type), (\"Content-Length\", str(content_length))]\n        )\n        return wsgiref.util.FileWrapper(file)\n\n    @classmethod\n    def _inject_js_into_html(cls, content, epoch):\n        try:\n            body_end = content.rindex(b\"</body>\")\n        except ValueError:\n            body_end = len(content)\n        # The page will reload if the livereload poller returns a newer epoch than what it knows.\n        # The other timestamp becomes just a unique identifier for the initiating page.\n        return (\n            b'%b<script src=\"/js/livereload.js\"></script><script>livereload(%d, %d);</script>%b'\n            % (content[:body_end], epoch, _timestamp(), content[body_end:])\n        )\n\n    @classmethod\n    @functools.lru_cache()  # \"Cache\" to not repeat the same message for the same browser tab.\n    def _log_poll_request(cls, url, request_id):\n        log.info(f\"Browser connected: {url}\")\n\n    def _guess_type(cls, path):\n        # MkDocs only ensures a few common types (as seen in livereload_tests.py::test_mime_types).\n        # Other uncommon types will not be accepted.\n        if path.endswith((\".js\", \".JS\")):\n            return \"application/javascript\"\n        if path.endswith(\".gz\"):\n            return \"application/gzip\"\n\n        guess, _ = mimetypes.guess_type(path)\n        if guess:\n            return guess\n        return \"application/octet-stream\"\n\n\nclass _Handler(wsgiref.simple_server.WSGIRequestHandler):\n    def log_request(self, code=\"-\", size=\"-\"):\n        level = logging.DEBUG if str(code) == \"200\" else logging.WARNING\n        log.log(level, f'\"{self.requestline}\" code {code}')\n\n    def log_message(self, format, *args):\n        log.debug(format, *args)\n\n\ndef _timestamp():\n    return round(time.monotonic() * 1000)\n", "path": "mkdocs/livereload/__init__.py"}], "after_files": [{"content": "import functools\nimport io\nimport logging\nimport mimetypes\nimport os\nimport os.path\nimport posixpath\nimport re\nimport socketserver\nimport threading\nimport time\nimport warnings\nimport wsgiref.simple_server\n\nimport watchdog.events\nimport watchdog.observers.polling\n\n\nclass _LoggerAdapter(logging.LoggerAdapter):\n    def process(self, msg, kwargs):\n        return time.strftime(\"[%H:%M:%S] \") + msg, kwargs\n\n\nlog = _LoggerAdapter(logging.getLogger(__name__), {})\n\n\nclass LiveReloadServer(socketserver.ThreadingMixIn, wsgiref.simple_server.WSGIServer):\n    daemon_threads = True\n    poll_response_timeout = 60\n\n    def __init__(\n        self,\n        builder,\n        host,\n        port,\n        root,\n        mount_path=\"/\",\n        polling_interval=0.5,\n        shutdown_delay=0.25,\n        **kwargs,\n    ):\n        self.builder = builder\n        self.server_name = host\n        self.server_port = port\n        self.root = os.path.abspath(root)\n        self.mount_path = (\"/\" + mount_path.lstrip(\"/\")).rstrip(\"/\") + \"/\"\n        self.url = f\"http://{self.server_name}:{self.server_port}{self.mount_path}\"\n        self.build_delay = 0.1\n        self.shutdown_delay = shutdown_delay\n        # To allow custom error pages.\n        self.error_handler = lambda code: None\n\n        super().__init__((host, port), _Handler, **kwargs)\n        self.set_app(self.serve_request)\n\n        self._wanted_epoch = _timestamp()  # The version of the site that started building.\n        self._visible_epoch = self._wanted_epoch  # Latest fully built version of the site.\n        self._epoch_cond = threading.Condition()  # Must be held when accessing _visible_epoch.\n\n        self._to_rebuild = {}  # Used as an ordered set of functions to call.\n        self._rebuild_cond = threading.Condition()  # Must be held when accessing _to_rebuild.\n\n        self._shutdown = False\n        self.serve_thread = threading.Thread(target=lambda: self.serve_forever(shutdown_delay))\n        self.observer = watchdog.observers.polling.PollingObserver(timeout=polling_interval)\n\n    def watch(self, path, func=None, recursive=True):\n        \"\"\"Add the 'path' to watched paths, call the function and reload when any file changes under it.\"\"\"\n        path = os.path.abspath(path)\n        if func in (None, self.builder):\n            func = self.builder\n        else:\n            warnings.warn(\n                \"Plugins should not pass the 'func' parameter of watch(). \"\n                \"The ability to execute custom callbacks will be removed soon.\",\n                DeprecationWarning,\n                stacklevel=2,\n            )\n\n        def callback(event):\n            if event.is_directory:\n                return\n            log.debug(str(event))\n            with self._rebuild_cond:\n                self._to_rebuild[func] = True\n                self._rebuild_cond.notify_all()\n\n        handler = watchdog.events.FileSystemEventHandler()\n        handler.on_any_event = callback\n        log.debug(f\"Watching '{path}'\")\n        self.observer.schedule(handler, path, recursive=recursive)\n\n    def serve(self):\n        self.observer.start()\n\n        log.info(f\"Serving on {self.url}\")\n        self.serve_thread.start()\n\n        self._build_loop()\n\n    def _build_loop(self):\n        while True:\n            with self._rebuild_cond:\n                while not self._rebuild_cond.wait_for(\n                    lambda: self._to_rebuild or self._shutdown, timeout=self.shutdown_delay\n                ):\n                    # We could have used just one wait instead of a loop + timeout, but we need\n                    # occasional breaks, otherwise on Windows we can't receive KeyboardInterrupt.\n                    pass\n                if self._shutdown:\n                    break\n                log.info(\"Detected file changes\")\n                while self._rebuild_cond.wait(timeout=self.build_delay):\n                    log.debug(\"Waiting for file changes to stop happening\")\n\n                self._wanted_epoch = _timestamp()\n                funcs = list(self._to_rebuild)\n                self._to_rebuild.clear()\n\n            for func in funcs:\n                func()\n\n            with self._epoch_cond:\n                log.info(\"Reloading browsers\")\n                self._visible_epoch = self._wanted_epoch\n                self._epoch_cond.notify_all()\n\n    def shutdown(self):\n        self.observer.stop()\n        with self._rebuild_cond:\n            self._shutdown = True\n            self._rebuild_cond.notify_all()\n\n        if self.serve_thread.is_alive():\n            super().shutdown()\n            self.serve_thread.join()\n            self.observer.join()\n\n    def serve_request(self, environ, start_response):\n        try:\n            result = self._serve_request(environ, start_response)\n        except Exception:\n            code = 500\n            msg = \"500 Internal Server Error\"\n            log.exception(msg)\n        else:\n            if result is not None:\n                return result\n            code = 404\n            msg = \"404 Not Found\"\n\n        error_content = None\n        try:\n            error_content = self.error_handler(code)\n        except Exception:\n            log.exception(\"Failed to render an error message!\")\n        if error_content is None:\n            error_content = msg.encode()\n\n        start_response(msg, [(\"Content-Type\", \"text/html\")])\n        return [error_content]\n\n    def _serve_request(self, environ, start_response):\n        # https://bugs.python.org/issue16679\n        # https://github.com/bottlepy/bottle/blob/f9b1849db4/bottle.py#L984\n        path = environ[\"PATH_INFO\"].encode(\"latin-1\").decode(\"utf-8\", \"ignore\")\n\n        m = re.fullmatch(r\"/livereload/([0-9]+)/[0-9]+\", path)\n        if m:\n            epoch = int(m[1])\n            start_response(\"200 OK\", [(\"Content-Type\", \"text/plain\")])\n\n            def condition():\n                return self._visible_epoch > epoch\n\n            with self._epoch_cond:\n                if not condition():\n                    # Stall the browser, respond as soon as there's something new.\n                    # If there's not, respond anyway after a minute.\n                    self._log_poll_request(environ.get(\"HTTP_REFERER\"), request_id=path)\n                    self._epoch_cond.wait_for(condition, timeout=self.poll_response_timeout)\n                return [b\"%d\" % self._visible_epoch]\n\n        if path == \"/js/livereload.js\":\n            file_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), \"livereload.js\")\n        elif path.startswith(self.mount_path):\n            rel_file_path = path[len(self.mount_path):]\n            if path.endswith(\"/\"):\n                rel_file_path += \"index.html\"\n            # Prevent directory traversal - normalize the path.\n            rel_file_path = posixpath.normpath(\"/\" + rel_file_path).lstrip(\"/\")\n            file_path = os.path.join(self.root, rel_file_path)\n        elif path == \"/\":\n            start_response(\"302 Found\", [(\"Location\", self.mount_path)])\n            return []\n        else:\n            return None  # Not found\n\n        # Wait until the ongoing rebuild (if any) finishes, so we're not serving a half-built site.\n        with self._epoch_cond:\n            self._epoch_cond.wait_for(lambda: self._visible_epoch == self._wanted_epoch)\n            epoch = self._visible_epoch\n\n        try:\n            file = open(file_path, \"rb\")\n        except OSError:\n            if not path.endswith(\"/\") and os.path.isfile(os.path.join(file_path, \"index.html\")):\n                start_response(\"302 Found\", [(\"Location\", path + \"/\")])\n                return []\n            return None  # Not found\n\n        if file_path.endswith(\".html\"):\n            with file:\n                content = file.read()\n            content = self._inject_js_into_html(content, epoch)\n            file = io.BytesIO(content)\n            content_length = len(content)\n        else:\n            content_length = os.path.getsize(file_path)\n\n        content_type = self._guess_type(file_path)\n        start_response(\n            \"200 OK\", [(\"Content-Type\", content_type), (\"Content-Length\", str(content_length))]\n        )\n        return wsgiref.util.FileWrapper(file)\n\n    @classmethod\n    def _inject_js_into_html(cls, content, epoch):\n        try:\n            body_end = content.rindex(b\"</body>\")\n        except ValueError:\n            body_end = len(content)\n        # The page will reload if the livereload poller returns a newer epoch than what it knows.\n        # The other timestamp becomes just a unique identifier for the initiating page.\n        return (\n            b'%b<script src=\"/js/livereload.js\"></script><script>livereload(%d, %d);</script>%b'\n            % (content[:body_end], epoch, _timestamp(), content[body_end:])\n        )\n\n    @classmethod\n    @functools.lru_cache()  # \"Cache\" to not repeat the same message for the same browser tab.\n    def _log_poll_request(cls, url, request_id):\n        log.info(f\"Browser connected: {url}\")\n\n    def _guess_type(cls, path):\n        # MkDocs only ensures a few common types (as seen in livereload_tests.py::test_mime_types).\n        # Other uncommon types will not be accepted.\n        if path.endswith((\".js\", \".JS\")):\n            return \"application/javascript\"\n        if path.endswith(\".gz\"):\n            return \"application/gzip\"\n\n        guess, _ = mimetypes.guess_type(path)\n        if guess:\n            return guess\n        return \"application/octet-stream\"\n\n\nclass _Handler(wsgiref.simple_server.WSGIRequestHandler):\n    def log_request(self, code=\"-\", size=\"-\"):\n        level = logging.DEBUG if str(code) == \"200\" else logging.WARNING\n        log.log(level, f'\"{self.requestline}\" code {code}')\n\n    def log_message(self, format, *args):\n        log.debug(format, *args)\n\n\ndef _timestamp():\n    return round(time.monotonic() * 1000)\n", "path": "mkdocs/livereload/__init__.py"}]}