(self)
| 80 | post_body: Optional[bytes] = None |
| 81 | |
| 82 | def process(self) -> None: |
| 83 | server = self.channel.factory.server_instance |
| 84 | if self.content: |
| 85 | self.post_body = self.content.read() |
| 86 | self.content.seek(0, 0) |
| 87 | else: |
| 88 | self.post_body = None |
| 89 | path = urlparse(self.uri.decode()).path |
| 90 | |
| 91 | request_subscriber = server.request_subscribers.get(path) |
| 92 | if request_subscriber: |
| 93 | request_subscriber._loop.call_soon_threadsafe( |
| 94 | request_subscriber.set_result, self |
| 95 | ) |
| 96 | server.request_subscribers.pop(path) |
| 97 | |
| 98 | if path == "/ws": |
| 99 | server._ws_resource.render(self) |
| 100 | return |
| 101 | |
| 102 | if server.auth.get(path): |
| 103 | authorization_header = self.requestHeaders.getRawHeaders("authorization") |
| 104 | creds_correct = False |
| 105 | if authorization_header: |
| 106 | creds_correct = server.auth.get(path) == ( |
| 107 | self.getUser().decode(), |
| 108 | self.getPassword().decode(), |
| 109 | ) |
| 110 | if not creds_correct: |
| 111 | self.setHeader(b"www-authenticate", 'Basic realm="Secure Area"') |
| 112 | self.setResponseCode(HTTPStatus.UNAUTHORIZED) |
| 113 | self.write(b"HTTP Error 401 Unauthorized: Access is denied") |
| 114 | self.finish() |
| 115 | return |
| 116 | if server.csp.get(path): |
| 117 | self.setHeader(b"Content-Security-Policy", server.csp[path]) |
| 118 | if server.routes.get(path): |
| 119 | server.routes[path](self) |
| 120 | return |
| 121 | |
| 122 | self._serve_file(server.static_path / path[1:], path) |
| 123 | |
| 124 | def serve_file(self, path: pathlib.Path) -> None: |
| 125 | return self._serve_file(path, urlparse(self.uri.decode()).path) |
nothing calls this directly
no test coverage detected