5 Commits

Author SHA1 Message Date
klemek 069998f8ac tests: more coverage
Python Lint CI / ruff-format-check (push) Successful in 57s
Python Lint CI / ty (push) Successful in 57s
Python Lint CI / ruff (push) Successful in 58s
Docker CI / docker-build (push) Successful in 1m43s
Python Test CI / coverage (push) Failing after 1m59s
2026-05-05 15:10:03 +02:00
klemek b1ef00b437 fix: add request timeout 2026-05-05 15:09:54 +02:00
klemek 09ea29d6af fix: better string sanitizing 2026-05-05 15:09:39 +02:00
klemek 8855fd0b01 fix: force connection close 2026-05-05 15:09:20 +02:00
klemek 8854cb393c fix: send connection: close each request 2026-05-05 15:09:08 +02:00
2 changed files with 14 additions and 4 deletions
+12 -3
View File
@@ -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:
+2 -1
View File
@@ -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: