Compare commits
5 Commits
5e5147251f
...
069998f8ac
| Author | SHA1 | Date | |
|---|---|---|---|
| 069998f8ac | |||
| b1ef00b437 | |||
| 09ea29d6af | |||
| 8855fd0b01 | |||
| 8854cb393c |
+12
-3
@@ -25,6 +25,9 @@ if typing.TYPE_CHECKING:
|
||||
|
||||
|
||||
class BaseHandler(abc.ABC, http.server.BaseHTTPRequestHandler):
|
||||
SANITIZE_REGEX = re.compile(r"[^\x20-\x7F]+")
|
||||
timeout = 10
|
||||
|
||||
@typing.override
|
||||
def __init__(
|
||||
self,
|
||||
@@ -39,6 +42,8 @@ class BaseHandler(abc.ABC, http.server.BaseHTTPRequestHandler):
|
||||
self.__in_size: int | None = None
|
||||
self.https: bool = params.https
|
||||
super().__init__(*args, **kwargs)
|
||||
with contextlib.suppress(Exception):
|
||||
self.connection.close()
|
||||
|
||||
@typing.override
|
||||
def send_error(
|
||||
@@ -75,7 +80,7 @@ class BaseHandler(abc.ABC, http.server.BaseHTTPRequestHandler):
|
||||
|
||||
@typing.override
|
||||
def address_string(self) -> str: # pragma: no cover
|
||||
return re.sub(r"[^\x00-\x7F]+", "", super().address_string())
|
||||
return self.SANITIZE_REGEX.sub("?", super().address_string())
|
||||
|
||||
@typing.override
|
||||
def log_message(self, format: str, *args: typing.Any) -> None: # pragma: no cover
|
||||
@@ -106,7 +111,7 @@ class BaseHandler(abc.ABC, http.server.BaseHTTPRequestHandler):
|
||||
code,
|
||||
self.address_string(),
|
||||
self.host,
|
||||
re.sub(r"[^\x00-\x7F]+", "?", self.requestline),
|
||||
self.SANITIZE_REGEX.sub("?", self.requestline),
|
||||
)
|
||||
fmt = "→ %s - %s - %s - %s"
|
||||
if size != "":
|
||||
@@ -126,6 +131,7 @@ class BaseHandler(abc.ABC, http.server.BaseHTTPRequestHandler):
|
||||
self.send_response(code, message)
|
||||
self.send_header("Content-Type", f"{content_type}; charset=UTF-8")
|
||||
self.send_header("Content-Length", str(len(encoded)))
|
||||
self.send_header("Connection", "close")
|
||||
self.end_headers()
|
||||
self.wfile.write(encoded)
|
||||
self.close_connection = True
|
||||
@@ -140,6 +146,7 @@ class BaseHandler(abc.ABC, http.server.BaseHTTPRequestHandler):
|
||||
headers = {}
|
||||
self.send_response(code, message)
|
||||
self.send_header("Content-Length", "0")
|
||||
self.send_header("Connection", "close")
|
||||
for header, value in headers.items():
|
||||
self.send_header(header, value)
|
||||
self.end_headers()
|
||||
@@ -188,9 +195,11 @@ class BaseHandler(abc.ABC, http.server.BaseHTTPRequestHandler):
|
||||
"transfer-encoding",
|
||||
"server",
|
||||
"date",
|
||||
"connection",
|
||||
]:
|
||||
self.send_header(header, value.replace(target_host, self.host))
|
||||
self.send_header("Content-Length", str(out_size := len(response.content)))
|
||||
self.send_header("Connection", "close")
|
||||
self.end_headers()
|
||||
if out_size > 0:
|
||||
self.wfile.write(response.content)
|
||||
@@ -232,7 +241,7 @@ class BaseHandler(abc.ABC, http.server.BaseHTTPRequestHandler):
|
||||
"...",
|
||||
self.address_string(),
|
||||
self.host,
|
||||
re.sub(r"[^\x00-\x7F]+", "?", self.requestline),
|
||||
self.SANITIZE_REGEX.sub("?", self.requestline),
|
||||
)
|
||||
fmt = "← %s - %s - %s - %s"
|
||||
if self.in_size > 0:
|
||||
|
||||
@@ -60,7 +60,8 @@ class TestDataDir(BaseTestCase):
|
||||
self.assertIsNone(self.data_dir.get_file("test_1", ".value"))
|
||||
|
||||
def test_get_file_cannot_read(self) -> None:
|
||||
self.__create_path("test_1", {".value/.value": "value"})
|
||||
self.__create_path("test_1", {".value": "value"})
|
||||
(self.tmp_path / "test_1" / ".value").chmod(0o300)
|
||||
self.assertIsNone(self.data_dir.get_file("test_1", ".value"))
|
||||
|
||||
def test_get_file_invalid_path(self) -> None:
|
||||
|
||||
Reference in New Issue
Block a user