Compare commits
6
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
60e70ea0c0 | ||
|
|
9cefd6657a | ||
|
|
1e5f3ba986 | ||
|
|
d3d98bd9b2 | ||
|
|
b6d751a97a | ||
|
|
3f0490ebc9 |
+1
-1
@@ -1,6 +1,6 @@
|
|||||||
[project]
|
[project]
|
||||||
name = "stapler"
|
name = "stapler"
|
||||||
version = "1.3.1"
|
version = "1.4.0"
|
||||||
description = "Static pages as simple as a gzip file"
|
description = "Static pages as simple as a gzip file"
|
||||||
requires-python = ">=3.14"
|
requires-python = ">=3.14"
|
||||||
dependencies = [
|
dependencies = [
|
||||||
|
|||||||
@@ -185,11 +185,16 @@ class CertManager:
|
|||||||
return False
|
return False
|
||||||
return self.__exists_certbot(host)
|
return self.__exists_certbot(host)
|
||||||
|
|
||||||
def sni_callback(
|
def servername_callback(
|
||||||
self, socket: ssl.SSLObject, host: str | None, _: ssl.SSLContext, /
|
self,
|
||||||
|
socket: ssl.SSLSocket | ssl.SSLObject,
|
||||||
|
host: str | None,
|
||||||
|
_: ssl.SSLSocket,
|
||||||
|
/,
|
||||||
) -> None | int:
|
) -> None | int:
|
||||||
if host is None:
|
if host is None:
|
||||||
return None
|
return None
|
||||||
|
self.logger.debug("servername callback: %s", host)
|
||||||
if not self.exists(host) and not self.create_or_update(host):
|
if not self.exists(host) and not self.create_or_update(host):
|
||||||
return None
|
return None
|
||||||
cert_file = self.get_cert(host)
|
cert_file = self.get_cert(host)
|
||||||
|
|||||||
+24
-7
@@ -29,6 +29,10 @@ class BaseHandler(abc.ABC, http.server.BaseHTTPRequestHandler):
|
|||||||
timeout = 10
|
timeout = 10
|
||||||
protocol_version = "HTTP/1.1"
|
protocol_version = "HTTP/1.1"
|
||||||
REQUEST_COUNT = 0
|
REQUEST_COUNT = 0
|
||||||
|
RESPONSE_HEADERS: typing.ClassVar = {
|
||||||
|
"Connection": "close",
|
||||||
|
"Strict-Transport-Security": "max-age=31536000; includeSubDomains; preload",
|
||||||
|
}
|
||||||
|
|
||||||
@typing.override
|
@typing.override
|
||||||
def __init__(
|
def __init__(
|
||||||
@@ -48,6 +52,10 @@ class BaseHandler(abc.ABC, http.server.BaseHTTPRequestHandler):
|
|||||||
with contextlib.suppress(Exception):
|
with contextlib.suppress(Exception):
|
||||||
self.connection.close()
|
self.connection.close()
|
||||||
|
|
||||||
|
@typing.override
|
||||||
|
def version_string(self) -> str:
|
||||||
|
return self.server_version
|
||||||
|
|
||||||
@typing.override
|
@typing.override
|
||||||
def send_error(
|
def send_error(
|
||||||
self,
|
self,
|
||||||
@@ -151,9 +159,10 @@ class BaseHandler(abc.ABC, http.server.BaseHTTPRequestHandler):
|
|||||||
self.send_response(code, message)
|
self.send_response(code, message)
|
||||||
self.send_header("Content-Type", f"{content_type}; charset=UTF-8")
|
self.send_header("Content-Type", f"{content_type}; charset=UTF-8")
|
||||||
self.send_header("Content-Length", str(len(encoded)))
|
self.send_header("Content-Length", str(len(encoded)))
|
||||||
self.send_header("Connection", "close")
|
self._send_basic_headers()
|
||||||
self.end_headers()
|
self.end_headers()
|
||||||
self.wfile.write(encoded)
|
if self.command != http.HTTPMethod.HEAD:
|
||||||
|
self.wfile.write(encoded)
|
||||||
self.close_connection = True
|
self.close_connection = True
|
||||||
|
|
||||||
def send_status_only(
|
def send_status_only(
|
||||||
@@ -166,7 +175,7 @@ class BaseHandler(abc.ABC, http.server.BaseHTTPRequestHandler):
|
|||||||
headers = {}
|
headers = {}
|
||||||
self.send_response(code, message)
|
self.send_response(code, message)
|
||||||
self.send_header("Content-Length", "0")
|
self.send_header("Content-Length", "0")
|
||||||
self.send_header("Connection", "close")
|
self._send_basic_headers()
|
||||||
for header, value in headers.items():
|
for header, value in headers.items():
|
||||||
self.send_header(header, value)
|
self.send_header(header, value)
|
||||||
self.end_headers()
|
self.end_headers()
|
||||||
@@ -220,9 +229,9 @@ class BaseHandler(abc.ABC, http.server.BaseHTTPRequestHandler):
|
|||||||
]:
|
]:
|
||||||
self.send_header(header, value.replace(target_host, self.host))
|
self.send_header(header, value.replace(target_host, self.host))
|
||||||
self.send_header("Content-Length", str(out_size := len(response.content)))
|
self.send_header("Content-Length", str(out_size := len(response.content)))
|
||||||
self.send_header("Connection", "close")
|
self._send_basic_headers()
|
||||||
self.end_headers()
|
self.end_headers()
|
||||||
if out_size > 0:
|
if out_size > 0 and self.command != http.HTTPMethod.HEAD:
|
||||||
self.wfile.write(response.content)
|
self.wfile.write(response.content)
|
||||||
self.close_connection = True
|
self.close_connection = True
|
||||||
|
|
||||||
@@ -268,6 +277,10 @@ class BaseHandler(abc.ABC, http.server.BaseHTTPRequestHandler):
|
|||||||
self.send_error(http.HTTPStatus.INTERNAL_SERVER_ERROR, str(e))
|
self.send_error(http.HTTPStatus.INTERNAL_SERVER_ERROR, str(e))
|
||||||
self.logger.exception("Internal Server Error")
|
self.logger.exception("Internal Server Error")
|
||||||
|
|
||||||
|
def _send_basic_headers(self) -> None:
|
||||||
|
for header, value in self.RESPONSE_HEADERS.items():
|
||||||
|
self.send_header(header, value)
|
||||||
|
|
||||||
|
|
||||||
class RequestHandler(http.server.SimpleHTTPRequestHandler, BaseHandler):
|
class RequestHandler(http.server.SimpleHTTPRequestHandler, BaseHandler):
|
||||||
protocol_version = "HTTP/1.1"
|
protocol_version = "HTTP/1.1"
|
||||||
@@ -387,9 +400,13 @@ class RequestHandler(http.server.SimpleHTTPRequestHandler, BaseHandler):
|
|||||||
def do_HEAD(self) -> None:
|
def do_HEAD(self) -> None:
|
||||||
with self.handle_errors():
|
with self.handle_errors():
|
||||||
self._pre_log_request()
|
self._pre_log_request()
|
||||||
if not self._proxy_or_redirect():
|
if self._proxy_or_redirect():
|
||||||
super().do_HEAD()
|
return None
|
||||||
|
if self.path == "/" and self.host == self.default_host:
|
||||||
|
return self.send_basic_body(self.server_signature())
|
||||||
|
super().do_HEAD()
|
||||||
self.close_connection = True
|
self.close_connection = True
|
||||||
|
return None
|
||||||
|
|
||||||
@typing.override
|
@typing.override
|
||||||
def do_GET(self) -> None:
|
def do_GET(self) -> None:
|
||||||
|
|||||||
+1
-1
@@ -73,7 +73,7 @@ class StaplerServer:
|
|||||||
)
|
)
|
||||||
context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
|
context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
|
||||||
server.socket = context.wrap_socket(server.socket, server_side=True)
|
server.socket = context.wrap_socket(server.socket, server_side=True)
|
||||||
context.sni_callback = self.cert_manager.sni_callback
|
context.set_servername_callback(self.cert_manager.servername_callback)
|
||||||
else:
|
else:
|
||||||
server = http.server.ThreadingHTTPServer(
|
server = http.server.ThreadingHTTPServer(
|
||||||
(
|
(
|
||||||
|
|||||||
@@ -161,24 +161,26 @@ class TestRegistry(BaseTestCase):
|
|||||||
lambda: self.cert_manager.get_key("example.com"),
|
lambda: self.cert_manager.get_key("example.com"),
|
||||||
)
|
)
|
||||||
|
|
||||||
def test_sni_callback_no_host(self) -> None:
|
def test_servername_callback_no_host(self) -> None:
|
||||||
self._make_self_signed("example.com")
|
self._make_self_signed("example.com")
|
||||||
with (
|
with (
|
||||||
self.patch("ssl.create_default_context", count=0),
|
self.patch("ssl.create_default_context", count=0),
|
||||||
):
|
):
|
||||||
self.cert_manager.sni_callback(self.socket_mock, None, self.context_mock)
|
self.cert_manager.servername_callback(
|
||||||
|
self.socket_mock, None, self.context_mock
|
||||||
|
)
|
||||||
|
|
||||||
def test_sni_callback_fail(self) -> None:
|
def test_servername_callback_fail(self) -> None:
|
||||||
self._make_self_signed("example.com")
|
self._make_self_signed("example.com")
|
||||||
with (
|
with (
|
||||||
self.patch("shutil.which", count=3),
|
self.patch("shutil.which", count=3),
|
||||||
self.patch("ssl.create_default_context", count=0),
|
self.patch("ssl.create_default_context", count=0),
|
||||||
):
|
):
|
||||||
self.cert_manager.sni_callback(
|
self.cert_manager.servername_callback(
|
||||||
self.socket_mock, "example.fr", self.context_mock
|
self.socket_mock, "example.fr", self.context_mock
|
||||||
)
|
)
|
||||||
|
|
||||||
def test_sni_callback_create_context(self) -> None:
|
def test_servername_callback_create_context(self) -> None:
|
||||||
self._make_self_signed("example.com")
|
self._make_self_signed("example.com")
|
||||||
with (
|
with (
|
||||||
self.patch("ssl.create_default_context", return_value=self.context_mock),
|
self.patch("ssl.create_default_context", return_value=self.context_mock),
|
||||||
@@ -191,18 +193,18 @@ class TestRegistry(BaseTestCase):
|
|||||||
),
|
),
|
||||||
self.patch("shutil.which", count=0),
|
self.patch("shutil.which", count=0),
|
||||||
):
|
):
|
||||||
self.cert_manager.sni_callback(
|
self.cert_manager.servername_callback(
|
||||||
self.socket_mock, "example.com", self.context_mock
|
self.socket_mock, "example.com", self.context_mock
|
||||||
)
|
)
|
||||||
|
|
||||||
def test_sni_callback_create_context_fail(self) -> None:
|
def test_servername_callback_create_context_fail(self) -> None:
|
||||||
self._make_self_signed("example.com")
|
self._make_self_signed("example.com")
|
||||||
with (
|
with (
|
||||||
self.patch("ssl.create_default_context", return_value=self.context_mock),
|
self.patch("ssl.create_default_context", return_value=self.context_mock),
|
||||||
self.patch("shutil.which", count=0),
|
self.patch("shutil.which", count=0),
|
||||||
):
|
):
|
||||||
self.context_mock.load_cert_chain.side_effect = Exception
|
self.context_mock.load_cert_chain.side_effect = Exception
|
||||||
self.cert_manager.sni_callback(
|
self.cert_manager.servername_callback(
|
||||||
self.socket_mock, "example.com", self.context_mock
|
self.socket_mock, "example.com", self.context_mock
|
||||||
)
|
)
|
||||||
self.context_mock.load_cert_chain.assert_called_once_with(
|
self.context_mock.load_cert_chain.assert_called_once_with(
|
||||||
|
|||||||
+15
-2
@@ -36,6 +36,7 @@ class BaseHandlerTestCase(BaseTestCase, abc.ABC):
|
|||||||
code: int,
|
code: int,
|
||||||
message: str | None = None,
|
message: str | None = None,
|
||||||
headers: dict[str, str] | None = None,
|
headers: dict[str, str] | None = None,
|
||||||
|
content_length: int = 0,
|
||||||
) -> typing.Iterator[None]:
|
) -> typing.Iterator[None]:
|
||||||
if headers is None:
|
if headers is None:
|
||||||
headers = {}
|
headers = {}
|
||||||
@@ -46,7 +47,7 @@ class BaseHandlerTestCase(BaseTestCase, abc.ABC):
|
|||||||
send_response_mock.assert_called_once_with(code, message)
|
send_response_mock.assert_called_once_with(code, message)
|
||||||
send_header_mock.assert_has_calls(
|
send_header_mock.assert_has_calls(
|
||||||
[
|
[
|
||||||
unittest.mock.call("Content-Length", "0"),
|
unittest.mock.call("Content-Length", str(content_length)),
|
||||||
]
|
]
|
||||||
+ [unittest.mock.call(header, value) for header, value in headers.items()],
|
+ [unittest.mock.call(header, value) for header, value in headers.items()],
|
||||||
any_order=True,
|
any_order=True,
|
||||||
@@ -192,9 +193,21 @@ class TestRequestHandler(BaseHandlerTestCase):
|
|||||||
token_manager=self.token_manager,
|
token_manager=self.token_manager,
|
||||||
)
|
)
|
||||||
|
|
||||||
def test_do_head_forward(self) -> None:
|
def test_do_head_index(self) -> None:
|
||||||
handler = self._get_handler()
|
handler = self._get_handler()
|
||||||
with (
|
with (
|
||||||
|
self.expects_status_only(
|
||||||
|
handler, 200, content_length=len(handler.server_signature())
|
||||||
|
),
|
||||||
|
self.patch("http.server.SimpleHTTPRequestHandler.do_HEAD", count=0),
|
||||||
|
self.seal_mocks(),
|
||||||
|
):
|
||||||
|
handler.do_HEAD()
|
||||||
|
|
||||||
|
def test_do_head_forward(self) -> None:
|
||||||
|
handler = self._get_handler("/file")
|
||||||
|
with (
|
||||||
|
self.mock_call(self.registry.get_from_path, ["file"], Page("file")),
|
||||||
self.patch("http.server.SimpleHTTPRequestHandler.do_HEAD"),
|
self.patch("http.server.SimpleHTTPRequestHandler.do_HEAD"),
|
||||||
self.seal_mocks(),
|
self.seal_mocks(),
|
||||||
):
|
):
|
||||||
|
|||||||
@@ -65,7 +65,7 @@ class TestStaplerServer(BaseTestCase):
|
|||||||
|
|
||||||
def test_run_https(self) -> None:
|
def test_run_https(self) -> None:
|
||||||
self.token_manager.detect_file_change.side_effect = KeyboardInterrupt
|
self.token_manager.detect_file_change.side_effect = KeyboardInterrupt
|
||||||
self.cert_manager.sni_callback = unittest.mock.Mock()
|
self.cert_manager.servername_callback = unittest.mock.Mock()
|
||||||
with (
|
with (
|
||||||
self.mock_call(self.registry.load_pages),
|
self.mock_call(self.registry.load_pages),
|
||||||
self.mock_call(self.cert_manager.init),
|
self.mock_call(self.cert_manager.init),
|
||||||
@@ -74,6 +74,7 @@ class TestStaplerServer(BaseTestCase):
|
|||||||
self.patch("ssl.create_default_context", return_value=self.context_mock),
|
self.patch("ssl.create_default_context", return_value=self.context_mock),
|
||||||
self.patch("http.server.ThreadingHTTPServer", self.server_mock, 2),
|
self.patch("http.server.ThreadingHTTPServer", self.server_mock, 2),
|
||||||
self.mock_call_unchecked(self.context_mock.wrap_socket),
|
self.mock_call_unchecked(self.context_mock.wrap_socket),
|
||||||
|
self.mock_call_unchecked(self.context_mock.set_servername_callback),
|
||||||
self.mock_calls_unchecked(self.server_mock.serve_forever, 2),
|
self.mock_calls_unchecked(self.server_mock.serve_forever, 2),
|
||||||
self.mock_call(self.server_mock.shutdown),
|
self.mock_call(self.server_mock.shutdown),
|
||||||
self.seal_mocks(),
|
self.seal_mocks(),
|
||||||
|
|||||||
Reference in New Issue
Block a user