9 Commits

Author SHA1 Message Date
klemek d9b559d13d chore: release 1.3.0
Python Lint CI / ruff (push) Successful in 2m51s
Docker CI / docker-build (push) Successful in 3m23s
Python Lint CI / ruff-format-check (push) Successful in 1m50s
Python Lint CI / ty (push) Successful in 3m33s
Python Test CI / coverage (push) Successful in 3m4s
2026-05-06 14:32:00 +02:00
klemek b0d98dd48b feat: add robots.txt and favicon by default 2026-05-06 14:31:33 +02:00
klemek 64f45e9779 fix: dont init certificates with self-signed by default 2026-05-06 14:17:34 +02:00
klemek 2dd48042e7 chore: release 1.2.8
Python Lint CI / ruff (push) Successful in 1m57s
Python Lint CI / ruff-format-check (push) Successful in 1m52s
Docker CI / docker-build (push) Successful in 3m1s
Python Lint CI / ty (push) Successful in 2m47s
Python Test CI / coverage (push) Successful in 1m49s
2026-05-06 14:00:40 +02:00
klemek 74ceb0f677 fix: better host detection and cerbot only on valid hosts 2026-05-06 14:00:34 +02:00
klemek e7abe7924f chore: release 1.2.7
Python Lint CI / ruff (push) Successful in 1m27s
Python Lint CI / ruff-format-check (push) Successful in 1m19s
Docker CI / docker-build (push) Successful in 1m43s
Python Lint CI / ty (push) Successful in 4m22s
Python Test CI / coverage (push) Successful in 4m1s
2026-05-06 10:50:01 +02:00
klemek 8f7e4c8a91 fix: add debug request count
Docker CI / docker-build (push) Has been cancelled
Python Lint CI / ruff (push) Has been cancelled
Python Lint CI / ruff-format-check (push) Has been cancelled
Python Lint CI / ty (push) Has been cancelled
Python Test CI / coverage (push) Has been cancelled
2026-05-06 10:49:42 +02:00
klemek ab6879d54f chore: release 1.2.6
Python Lint CI / ruff (push) Successful in 3m52s
Python Lint CI / ruff-format-check (push) Successful in 1m41s
Docker CI / docker-build (push) Successful in 4m7s
Python Lint CI / ty (push) Successful in 2m11s
Python Test CI / coverage (push) Successful in 2m13s
2026-05-06 09:36:47 +02:00
klemek 8c93b9a015 fix: dont log BrokenPipeError and ConnectionResetError 2026-05-06 09:36:31 +02:00
12 changed files with 177 additions and 96 deletions
+2 -1
View File
@@ -1,6 +1,6 @@
[project] [project]
name = "stapler" name = "stapler"
version = "1.2.5" version = "1.3.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 = [
@@ -21,6 +21,7 @@ module-name = "stapler"
[dependency-groups] [dependency-groups]
dev = [ dev = [
"coverage>=7.13.5", "coverage>=7.13.5",
"parameterized>=0.9.0",
"pytest>=9.0.3", "pytest>=9.0.3",
"ruff>=0.15.10", "ruff>=0.15.10",
"ty>=0.0.29", "ty>=0.0.29",
+4 -4
View File
@@ -5,6 +5,8 @@ import ssl
import subprocess import subprocess
import typing import typing
from stapler.strings import valid_host
if typing.TYPE_CHECKING: if typing.TYPE_CHECKING:
from .params import Parameters from .params import Parameters
@@ -35,7 +37,7 @@ class CertManager:
self.with_certbot: bool = params.with_certbot self.with_certbot: bool = params.with_certbot
self.last_file_change: int | float = 0 self.last_file_change: int | float = 0
def init(self, hosts: list[str]) -> None: def init(self) -> None:
self.logger.debug("Initializing...") self.logger.debug("Initializing...")
if not self.certbot_www.exists(): if not self.certbot_www.exists():
self.certbot_www.mkdir(parents=True) self.certbot_www.mkdir(parents=True)
@@ -43,8 +45,6 @@ class CertManager:
if not self.self_signed_path.exists(): if not self.self_signed_path.exists():
self.self_signed_path.mkdir(parents=True) self.self_signed_path.mkdir(parents=True)
self.logger.debug("Created %s", self.self_signed_path) self.logger.debug("Created %s", self.self_signed_path)
for host in hosts:
self.init_cert(host)
def exists(self, host: str) -> bool: def exists(self, host: str) -> bool:
return self.__exists_certbot(host) or self.__exists_self_signed(host) return self.__exists_certbot(host) or self.__exists_self_signed(host)
@@ -57,7 +57,7 @@ class CertManager:
def create_or_update(self, host: str) -> bool: def create_or_update(self, host: str) -> bool:
created = self.init_cert(host) created = self.init_cert(host)
if self.with_certbot and self.__create_certbot(host): if self.with_certbot and valid_host(host) and self.__create_certbot(host):
return True return True
return created or self.__create_self_signed(host) return created or self.__create_self_signed(host)
+1 -1
View File
@@ -16,7 +16,7 @@ class DataDir:
] ]
PATH_REGEX = re.compile(r"^[\w-]+$") PATH_REGEX = re.compile(r"^[\w-]+$")
NEEDED_FILES: typing.ClassVar[list[str]] = ["favicon.ico"] NEEDED_FILES: typing.ClassVar[list[str]] = ["favicon.ico", "robots.txt"]
def __init__(self, root_path: str) -> None: def __init__(self, root_path: str) -> None:
self.logger: logging.Logger = logging.getLogger(self.__class__.__name__) self.logger: logging.Logger = logging.getLogger(self.__class__.__name__)
+30 -25
View File
@@ -16,6 +16,7 @@ import requests
from . import PKG_VERSION, STAPLER_ASCII, logs from . import PKG_VERSION, STAPLER_ASCII, logs
from .data_dir import DataDir from .data_dir import DataDir
from .strings import sanitize_string, valid_host
if typing.TYPE_CHECKING: if typing.TYPE_CHECKING:
from .page import Page from .page import Page
@@ -25,8 +26,8 @@ if typing.TYPE_CHECKING:
class BaseHandler(abc.ABC, http.server.BaseHTTPRequestHandler): class BaseHandler(abc.ABC, http.server.BaseHTTPRequestHandler):
SANITIZE_REGEX = re.compile(r"[^\x20-\x7F]+")
timeout = 10 timeout = 10
REQUEST_COUNT = 0
@typing.override @typing.override
def __init__( def __init__(
@@ -41,6 +42,7 @@ class BaseHandler(abc.ABC, http.server.BaseHTTPRequestHandler):
self.__host: str | None = None self.__host: str | None = None
self.__in_size: int | None = None self.__in_size: int | None = None
self.https: bool = params.https self.https: bool = params.https
self.__class__.REQUEST_COUNT += 1
super().__init__(*args, **kwargs) super().__init__(*args, **kwargs)
with contextlib.suppress(Exception): with contextlib.suppress(Exception):
self.connection.close() self.connection.close()
@@ -80,7 +82,7 @@ class BaseHandler(abc.ABC, http.server.BaseHTTPRequestHandler):
@typing.override @typing.override
def address_string(self) -> str: # pragma: no cover def address_string(self) -> str: # pragma: no cover
return self.SANITIZE_REGEX.sub("?", super().address_string()) return sanitize_string(super().address_string())
@typing.override @typing.override
def log_message(self, format: str, *args: typing.Any) -> None: # pragma: no cover def log_message(self, format: str, *args: typing.Any) -> None: # pragma: no cover
@@ -92,6 +94,20 @@ class BaseHandler(abc.ABC, http.server.BaseHTTPRequestHandler):
fmt = "%s - " + format fmt = "%s - " + format
self.logger.error(fmt, self.address_string(), *args) self.logger.error(fmt, self.address_string(), *args)
def _pre_log_request(self) -> None: # pragma: no cover
args = (
"...",
self.address_string(),
self.host,
format(self.__class__.REQUEST_COUNT, "07_d"),
sanitize_string(self.requestline),
)
fmt = "%s - %s - %s - %s - %s"
if self.in_size > 0:
args = (*args, self.in_size)
fmt += " - %s"
self.logger.debug(fmt, *args)
@typing.override @typing.override
def log_request(self, code: str = "?", size: str = "-") -> None: # ty:ignore[invalid-method-override] # pragma: no cover def log_request(self, code: str = "?", size: str = "-") -> None: # ty:ignore[invalid-method-override] # pragma: no cover
if isinstance(code, http.HTTPStatus): if isinstance(code, http.HTTPStatus):
@@ -111,9 +127,10 @@ class BaseHandler(abc.ABC, http.server.BaseHTTPRequestHandler):
code, code,
self.address_string(), self.address_string(),
self.host, self.host,
self.SANITIZE_REGEX.sub("?", self.requestline), format(self.__class__.REQUEST_COUNT, "07_d"),
sanitize_string(self.requestline),
) )
fmt = "%s - %s - %s - %s" fmt = "%s - %s - %s - %s - %s"
if size != "": if size != "":
args = (*args, size) args = (*args, size)
fmt += " - %s" fmt += " - %s"
@@ -236,19 +253,6 @@ class BaseHandler(abc.ABC, http.server.BaseHTTPRequestHandler):
and len(self.headers[key]) > 0 and len(self.headers[key]) > 0
) )
def _pre_log_request(self) -> None: # pragma: no cover
args = (
"...",
self.address_string(),
self.host,
self.SANITIZE_REGEX.sub("?", self.requestline),
)
fmt = "%s - %s - %s - %s"
if self.in_size > 0:
args = (*args, self.in_size)
fmt += " - %s"
self.logger.debug(fmt, *args)
def server_signature(self) -> str: def server_signature(self) -> str:
return self.server_version + "\n\n" + STAPLER_ASCII + "\n" return self.server_version + "\n\n" + STAPLER_ASCII + "\n"
@@ -268,7 +272,7 @@ class RequestHandler(http.server.SimpleHTTPRequestHandler, BaseHandler):
UPDATE_PATH_REGEX = re.compile(r"^\/([\w-]+)\/?$") UPDATE_PATH_REGEX = re.compile(r"^\/([\w-]+)\/?$")
GET_PATH_REGEX = re.compile(r"^\/([\w-]+)($|\/)") GET_PATH_REGEX = re.compile(r"^\/([\w-]+)($|\/)")
HOST_PART_REGEX = re.compile(r"^([a-z0-9]|[a-z0-9][a-z0-9-]{,61}[a-z0-9])$") HOST_PART_REGEX = re.compile(r"^([a-z0-9]|[a-z0-9][a-z0-9-]{,61}[a-z0-9])$")
AUTHORIZED_PATHS: typing.ClassVar[list[str]] = ["/favicon.ico"] AUTHORIZED_PATHS: typing.ClassVar[list[str]] = ["/favicon.ico", "/robots.txt"]
TOKEN_HEADER = "X-Token" # noqa: S105 TOKEN_HEADER = "X-Token" # noqa: S105
HOST_HEADER = "X-Host" HOST_HEADER = "X-Host"
HOST_ONLY_HEADER = "X-Host-Only" HOST_ONLY_HEADER = "X-Host-Only"
@@ -300,6 +304,8 @@ class RequestHandler(http.server.SimpleHTTPRequestHandler, BaseHandler):
self.__target_spa: str | None = None self.__target_spa: str | None = None
try: try:
super().__init__(*args, directory=params.data_dir, **kwargs, params=params) # ty:ignore[unknown-argument] super().__init__(*args, directory=params.data_dir, **kwargs, params=params) # ty:ignore[unknown-argument]
except (BrokenPipeError, ConnectionResetError) as e:
self.logger.error("Connection lost: %s", str(e)) # noqa: TRY400
except: except:
self.logger.exception("Could not handle request") self.logger.exception("Could not handle request")
@@ -536,6 +542,11 @@ class RequestHandler(http.server.SimpleHTTPRequestHandler, BaseHandler):
return super().translate_path(path) return super().translate_path(path)
return "" return ""
if self.host != self.default_host: if self.host != self.default_host:
if (
not (self.root_path / page.path / path).is_file()
and path in self.AUTHORIZED_PATHS
):
return super().translate_path(path)
path = f"/{page.path}" + path path = f"/{page.path}" + path
if pathlib.Path(path).name.startswith("."): # hidden files if pathlib.Path(path).name.startswith("."): # hidden files
return "" return ""
@@ -573,7 +584,7 @@ class RequestHandler(http.server.SimpleHTTPRequestHandler, BaseHandler):
f"Cannot use {self.HOST_ONLY_HEADER} with {self.HOST_HEADER}", f"Cannot use {self.HOST_ONLY_HEADER} with {self.HOST_HEADER}",
) )
return None return None
if self.has_target_host and not self.__valid_host(self.target_host): if self.has_target_host and not valid_host(self.target_host):
self.send_error(http.HTTPStatus.BAD_REQUEST, "Invalid requested host") self.send_error(http.HTTPStatus.BAD_REQUEST, "Invalid requested host")
return None return None
if self.has_target_proxy and self.has_target_redirect: if self.has_target_proxy and self.has_target_redirect:
@@ -596,12 +607,6 @@ class RequestHandler(http.server.SimpleHTTPRequestHandler, BaseHandler):
return match.group(1) return match.group(1)
return None return None
def __valid_host(self, host: str) -> bool:
return (
all(self.HOST_PART_REGEX.fullmatch(part) for part in host.split("."))
and len(host) < 256
)
def __get_page(self, src_path: str) -> Page | None: def __get_page(self, src_path: str) -> Page | None:
if self.host == self.default_host: if self.host == self.default_host:
if ( if (
+2
View File
@@ -0,0 +1,2 @@
User-agent: *
Disallow: /
+2 -2
View File
@@ -50,7 +50,7 @@ class StaplerServer:
self.logger.info("Starting up...") self.logger.info("Starting up...")
self.registry.load_pages() self.registry.load_pages()
if self.params.with_certificates: if self.params.with_certificates:
self.cert_manager.init(self.__get_all_hosts()) self.cert_manager.init()
self.data_dir.init() self.data_dir.init()
self.token_manager.init() self.token_manager.init()
@@ -152,7 +152,7 @@ class StaplerServer:
self.logger.warning("Cannot renew without certificates") self.logger.warning("Cannot renew without certificates")
return 1 return 1
self.registry.load_pages() self.registry.load_pages()
self.cert_manager.init(self.__get_all_hosts()) self.cert_manager.init()
for host in self.__get_all_hosts(): for host in self.__get_all_hosts():
self.cert_manager.create_or_update(host) self.cert_manager.create_or_update(host)
return 0 return 0
+19
View File
@@ -0,0 +1,19 @@
import re
__HOST_PART_REGEX = re.compile(r"^([a-z0-9]|[a-z0-9][a-z0-9-]{,61}[a-z0-9])$")
__SANITIZE_REGEX = re.compile(r"[^\x20-\x7F]")
def valid_host(host: str) -> bool:
parts = host.split(".")
return (
len(parts) > 1
and len(parts[-1]) > 1
and all(__HOST_PART_REGEX.fullmatch(part) for part in parts)
and not all(part.isnumeric() for part in parts)
and len(host) < 256
)
def sanitize_string(raw: str) -> str:
return __SANITIZE_REGEX.sub("?", raw)
+50 -56
View File
@@ -35,41 +35,33 @@ class TestRegistry(BaseTestCase):
self.patch("shutil.which", count=0), self.patch("shutil.which", count=0),
self.patch("subprocess.check_output", count=0), self.patch("subprocess.check_output", count=0),
): ):
self.cert_manager.init([]) self.cert_manager.init()
assert self.self_signed_path.is_dir() assert self.self_signed_path.is_dir()
assert self.certbot_www.is_dir() assert self.certbot_www.is_dir()
def test_init_with_hosts(self) -> None:
with (
self.patch("shutil.which", count=0),
self.patch("subprocess.check_output", count=0),
):
self._make_self_signed("localhost")
self.cert_manager.init(["localhost"])
def test_exists_self_signed(self) -> None: def test_exists_self_signed(self) -> None:
self._make_self_signed("localhost") self._make_self_signed("example.com")
assert self.cert_manager.exists("localhost") assert self.cert_manager.exists("example.com")
def test_exists_certbot(self) -> None: def test_exists_certbot(self) -> None:
self._make_certbot("localhost") self._make_certbot("example.com")
assert self.cert_manager.exists("localhost") assert self.cert_manager.exists("example.com")
def test_exists_fail(self) -> None: def test_exists_fail(self) -> None:
assert not self.cert_manager.exists("localhost") assert not self.cert_manager.exists("example.com")
def test_exists_fail_without_certbot(self) -> None: def test_exists_fail_without_certbot(self) -> None:
self.cert_manager.with_certbot = False self.cert_manager.with_certbot = False
self._make_certbot("localhost") self._make_certbot("example.com")
assert not self.cert_manager.exists("localhost") assert not self.cert_manager.exists("example.com")
def test_init_cert_existing(self) -> None: def test_init_cert_existing(self) -> None:
with ( with (
self.patch("shutil.which", count=0), self.patch("shutil.which", count=0),
self.patch("subprocess.check_output", count=0), self.patch("subprocess.check_output", count=0),
): ):
self._make_self_signed("localhost") self._make_self_signed("example.com")
assert not self.cert_manager.init_cert("localhost") assert not self.cert_manager.init_cert("example.com")
def test_init_cert_fail(self) -> None: def test_init_cert_fail(self) -> None:
with ( with (
@@ -77,7 +69,7 @@ class TestRegistry(BaseTestCase):
self.patch("subprocess.check_output") as process_mock, self.patch("subprocess.check_output") as process_mock,
): ):
process_mock.side_effect = subprocess.CalledProcessError(1, "", output=b"") process_mock.side_effect = subprocess.CalledProcessError(1, "", output=b"")
assert not self.cert_manager.init_cert("localhost") assert not self.cert_manager.init_cert("example.com")
def test_init_cert_new(self) -> None: def test_init_cert_new(self) -> None:
with ( with (
@@ -85,135 +77,137 @@ class TestRegistry(BaseTestCase):
self.patch("subprocess.check_output") as process_mock, self.patch("subprocess.check_output") as process_mock,
): ):
process_mock.side_effect = lambda *_, **__: self._make_self_signed( process_mock.side_effect = lambda *_, **__: self._make_self_signed(
"localhost" "example.com"
) )
assert self.cert_manager.init_cert("localhost") assert self.cert_manager.init_cert("example.com")
def test_create_or_update_existing_no_certbot(self) -> None: def test_create_or_update_existing_no_certbot(self) -> None:
self._make_self_signed("localhost") self._make_self_signed("example.com")
self.cert_manager.with_certbot = False self.cert_manager.with_certbot = False
with ( with (
self.patch("shutil.which", return_value=""), self.patch("shutil.which", return_value=""),
self.patch("subprocess.check_output") as process_mock, self.patch("subprocess.check_output") as process_mock,
): ):
process_mock.side_effect = lambda *_, **__: self._make_self_signed( process_mock.side_effect = lambda *_, **__: self._make_self_signed(
"localhost" "example.com"
) )
assert self.cert_manager.create_or_update("localhost") assert self.cert_manager.create_or_update("example.com")
def test_create_or_update_existing_certbot(self) -> None: def test_create_or_update_existing_certbot(self) -> None:
self._make_certbot("localhost") self._make_certbot("example.com")
with ( with (
self.patch("shutil.which", return_value=""), self.patch("shutil.which", return_value=""),
self.patch("subprocess.check_output") as process_mock, self.patch("subprocess.check_output") as process_mock,
): ):
process_mock.side_effect = lambda *_, **__: self._make_certbot("localhost") process_mock.side_effect = lambda *_, **__: self._make_certbot(
assert self.cert_manager.create_or_update("localhost") "example.com"
)
assert self.cert_manager.create_or_update("example.com")
def test_create_or_update_existing_fail_both(self) -> None: def test_create_or_update_existing_fail_both(self) -> None:
self._make_certbot("localhost") self._make_certbot("example.com")
with ( with (
self.patch("shutil.which", return_value="", count=2), self.patch("shutil.which", return_value="", count=2),
self.patch("subprocess.check_output", count=2) as process_mock, self.patch("subprocess.check_output", count=2) as process_mock,
): ):
process_mock.side_effect = subprocess.CalledProcessError(1, "", output=b"") process_mock.side_effect = subprocess.CalledProcessError(1, "", output=b"")
assert not self.cert_manager.create_or_update("localhost") assert not self.cert_manager.create_or_update("example.com")
def test_create_or_update_existing_fail_both_binary(self) -> None: def test_create_or_update_existing_fail_both_binary(self) -> None:
self._make_certbot("localhost") self._make_certbot("example.com")
with ( with (
self.patch("shutil.which", count=2), self.patch("shutil.which", count=2),
self.patch("subprocess.check_output", count=0), self.patch("subprocess.check_output", count=0),
): ):
assert not self.cert_manager.create_or_update("localhost") assert not self.cert_manager.create_or_update("example.com")
def test_get_cert_certbot(self) -> None: def test_get_cert_certbot(self) -> None:
self._make_certbot("localhost") self._make_certbot("example.com")
self.assertEqual( self.assertEqual(
self.cert_manager.get_cert("localhost"), self.cert_manager.get_cert("example.com"),
self.certbot_conf / "live" / "localhost" / CertManager.CRT_FILE, self.certbot_conf / "live" / "example.com" / CertManager.CRT_FILE,
) )
def test_get_cert_self_signed(self) -> None: def test_get_cert_self_signed(self) -> None:
self._make_self_signed("localhost") self._make_self_signed("example.com")
self.assertEqual( self.assertEqual(
self.cert_manager.get_cert("localhost"), self.cert_manager.get_cert("example.com"),
self.self_signed_path / "localhost" / CertManager.CRT_FILE, self.self_signed_path / "example.com" / CertManager.CRT_FILE,
) )
def test_get_cert_fail(self) -> None: def test_get_cert_fail(self) -> None:
self.assertRaises( self.assertRaises(
CertManagerError, CertManagerError,
lambda: self.cert_manager.get_cert("localhost"), lambda: self.cert_manager.get_cert("example.com"),
) )
def test_get_key_certbot(self) -> None: def test_get_key_certbot(self) -> None:
self._make_certbot("localhost") self._make_certbot("example.com")
self.assertEqual( self.assertEqual(
self.cert_manager.get_key("localhost"), self.cert_manager.get_key("example.com"),
self.certbot_conf / "live" / "localhost" / CertManager.KEY_FILE, self.certbot_conf / "live" / "example.com" / CertManager.KEY_FILE,
) )
def test_get_key_self_signed(self) -> None: def test_get_key_self_signed(self) -> None:
self._make_self_signed("localhost") self._make_self_signed("example.com")
self.assertEqual( self.assertEqual(
self.cert_manager.get_key("localhost"), self.cert_manager.get_key("example.com"),
self.self_signed_path / "localhost" / CertManager.KEY_FILE, self.self_signed_path / "example.com" / CertManager.KEY_FILE,
) )
def test_get_key_fail(self) -> None: def test_get_key_fail(self) -> None:
self.assertRaises( self.assertRaises(
CertManagerError, CertManagerError,
lambda: self.cert_manager.get_key("localhost"), lambda: self.cert_manager.get_key("example.com"),
) )
def test_sni_callback_no_host(self) -> None: def test_sni_callback_no_host(self) -> None:
self._make_self_signed("localhost") 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.sni_callback(self.socket_mock, None, self.context_mock)
def test_sni_callback_fail(self) -> None: def test_sni_callback_fail(self) -> None:
self._make_self_signed("localhost") 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.sni_callback(
self.socket_mock, "new_host", self.context_mock self.socket_mock, "example.fr", self.context_mock
) )
def test_sni_callback_create_context(self) -> None: def test_sni_callback_create_context(self) -> None:
self._make_self_signed("localhost") 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.mock_call( self.mock_call(
self.context_mock.load_cert_chain, self.context_mock.load_cert_chain,
[ [
self.self_signed_path / "localhost" / CertManager.CRT_FILE, self.self_signed_path / "example.com" / CertManager.CRT_FILE,
self.self_signed_path / "localhost" / CertManager.KEY_FILE, self.self_signed_path / "example.com" / CertManager.KEY_FILE,
], ],
), ),
self.patch("shutil.which", count=0), self.patch("shutil.which", count=0),
): ):
self.cert_manager.sni_callback( self.cert_manager.sni_callback(
self.socket_mock, "localhost", self.context_mock self.socket_mock, "example.com", self.context_mock
) )
def test_sni_callback_create_context_fail(self) -> None: def test_sni_callback_create_context_fail(self) -> None:
self._make_self_signed("localhost") 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.sni_callback(
self.socket_mock, "localhost", 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(
self.self_signed_path / "localhost" / CertManager.CRT_FILE, self.self_signed_path / "example.com" / CertManager.CRT_FILE,
self.self_signed_path / "localhost" / CertManager.KEY_FILE, self.self_signed_path / "example.com" / CertManager.KEY_FILE,
) )
def _make_self_signed(self, host: str) -> None: def _make_self_signed(self, host: str) -> None:
+30
View File
@@ -177,6 +177,21 @@ class TestRequestHandler(BaseHandlerTestCase):
token_manager=self.token_manager, token_manager=self.token_manager,
) )
def test_handle_disconnect_silently(self) -> None:
with self.patch("http.server.BaseHTTPRequestHandler.__init__") as mock:
mock.side_effect = BrokenPipeError
logging.basicConfig(level=logging.CRITICAL)
RequestHandler(
unittest.mock.MagicMock(),
"127.0.0.1",
unittest.mock.MagicMock(),
params=Parameters(
data_dir=self.get_tmp_dir(), certbot_www=str(self.certbot_www)
),
registry=self.registry,
token_manager=self.token_manager,
)
def test_do_head_forward(self) -> None: def test_do_head_forward(self) -> None:
handler = self._get_handler() handler = self._get_handler()
with ( with (
@@ -1177,6 +1192,21 @@ class TestRequestHandler(BaseHandlerTestCase):
None, None,
) )
def test_translate_path_with_host_favicon(self) -> None:
handler = self._get_handler(headers={"Host": "example.com"})
with (
self.mock_call(self.registry.get_from_host, ["example.com"], Page("path")),
self.patch_call(
"http.server.SimpleHTTPRequestHandler.translate_path",
["/favicon.ico"],
),
self.seal_mocks(),
):
self.assertEqual(
handler.translate_path("/favicon.ico"),
None,
)
def test_translate_path_default_host(self) -> None: def test_translate_path_default_host(self) -> None:
handler = self._get_handler() handler = self._get_handler()
with ( with (
+3 -6
View File
@@ -26,10 +26,8 @@ class TestStaplerServer(BaseTestCase):
def test_renew(self) -> None: def test_renew(self) -> None:
with ( with (
self.mock_call(self.registry.load_pages), self.mock_call(self.registry.load_pages),
self.mock_calls( self.mock_calls(self.registry.get_hosts, [[]], [["host_1"]]),
self.registry.get_hosts, [[], []], [["host_1"], ["host_1"]] self.mock_call(self.cert_manager.init),
),
self.mock_call(self.cert_manager.init, [["localhost", "host_1"]]),
self.mock_calls( self.mock_calls(
self.cert_manager.create_or_update, [["localhost"], ["host_1"]] self.cert_manager.create_or_update, [["localhost"], ["host_1"]]
), ),
@@ -70,8 +68,7 @@ class TestStaplerServer(BaseTestCase):
self.cert_manager.sni_callback = unittest.mock.Mock() self.cert_manager.sni_callback = unittest.mock.Mock()
with ( with (
self.mock_call(self.registry.load_pages), self.mock_call(self.registry.load_pages),
self.mock_call(self.registry.get_hosts, [], []), self.mock_call(self.cert_manager.init),
self.mock_call(self.cert_manager.init, [["localhost"]]),
self.mock_call(self.data_dir.init), self.mock_call(self.data_dir.init),
self.mock_call(self.token_manager.init), self.mock_call(self.token_manager.init),
self.patch("ssl.create_default_context", return_value=self.context_mock), self.patch("ssl.create_default_context", return_value=self.context_mock),
+22
View File
@@ -0,0 +1,22 @@
import parameterized
from stapler.strings import sanitize_string, valid_host
from . import BaseTestCase
class TestStrings(BaseTestCase):
def test_sanitize(self) -> None:
self.assertEqual("??A??", sanitize_string("\n\tA\x00\x99"))
@parameterized.parameterized.expand(
[("example.com"), ("test-test.com"), ("subdomain.example.com")]
)
def test_valid_host(self, host: str) -> None:
self.assertTrue(valid_host(host), host)
@parameterized.parameterized.expand(
[("example.c"), ("localhost"), ("127.0.0.1"), ("test..com"), ("www-.test.com")]
)
def test_invalid_host(self, host: str) -> None:
self.assertFalse(valid_host(host), host)
Generated
+12 -1
View File
@@ -127,6 +127,15 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/df/b2/87e62e8c3e2f4b32e5fe99e0b86d576da1312593b39f47d8ceef365e95ed/packaging-26.2-py3-none-any.whl", hash = "sha256:5fc45236b9446107ff2415ce77c807cee2862cb6fac22b8a73826d0693b0980e", size = 100195, upload-time = "2026-04-24T20:15:22.081Z" }, { url = "https://files.pythonhosted.org/packages/df/b2/87e62e8c3e2f4b32e5fe99e0b86d576da1312593b39f47d8ceef365e95ed/packaging-26.2-py3-none-any.whl", hash = "sha256:5fc45236b9446107ff2415ce77c807cee2862cb6fac22b8a73826d0693b0980e", size = 100195, upload-time = "2026-04-24T20:15:22.081Z" },
] ]
[[package]]
name = "parameterized"
version = "0.9.0"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/ea/49/00c0c0cc24ff4266025a53e41336b79adaa5a4ebfad214f433d623f9865e/parameterized-0.9.0.tar.gz", hash = "sha256:7fc905272cefa4f364c1a3429cbbe9c0f98b793988efb5bf90aac80f08db09b1", size = 24351, upload-time = "2023-03-27T02:01:11.592Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/00/2f/804f58f0b856ab3bf21617cccf5b39206e6c4c94c2cd227bde125ea6105f/parameterized-0.9.0-py2.py3-none-any.whl", hash = "sha256:4e0758e3d41bea3bbd05ec14fc2c24736723f243b28d702081aef438c9372b1b", size = 20475, upload-time = "2023-03-27T02:01:09.31Z" },
]
[[package]] [[package]]
name = "pluggy" name = "pluggy"
version = "1.6.0" version = "1.6.0"
@@ -203,7 +212,7 @@ wheels = [
[[package]] [[package]]
name = "stapler" name = "stapler"
version = "1.2.5" version = "1.3.0"
source = { editable = "." } source = { editable = "." }
dependencies = [ dependencies = [
{ name = "requests" }, { name = "requests" },
@@ -212,6 +221,7 @@ dependencies = [
[package.dev-dependencies] [package.dev-dependencies]
dev = [ dev = [
{ name = "coverage" }, { name = "coverage" },
{ name = "parameterized" },
{ name = "pytest" }, { name = "pytest" },
{ name = "ruff" }, { name = "ruff" },
{ name = "ty" }, { name = "ty" },
@@ -223,6 +233,7 @@ requires-dist = [{ name = "requests", specifier = ">=2.33.1" }]
[package.metadata.requires-dev] [package.metadata.requires-dev]
dev = [ dev = [
{ name = "coverage", specifier = ">=7.13.5" }, { name = "coverage", specifier = ">=7.13.5" },
{ name = "parameterized", specifier = ">=0.9.0" },
{ name = "pytest", specifier = ">=9.0.3" }, { name = "pytest", specifier = ">=9.0.3" },
{ name = "ruff", specifier = ">=0.15.10" }, { name = "ruff", specifier = ">=0.15.10" },
{ name = "ty", specifier = ">=0.0.29" }, { name = "ty", specifier = ">=0.0.29" },