klemek/stapler
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 10 Wiki Activity

feat: better logging

Browse Source
This commit is contained in:
klemek
2026-04-12 14:55:47 +02:00
parent 64adac02a2
commit cd81abf785
11 changed files with 208 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Dockerfile
+1 -1
View File
@@ -5,7 +5,7 @@ WORKDIR /app
VOLUME [ "/data", "/etc/letsencrypt" ]
ENV PORT=8080
ENV HOST=localhost
ENV HOST=localhost:8080
ENV DATA_DIR=/data
ENV MAX_SIZE=2000000
ENV BIND=0.0.0.0
Makefile
+1 -1
View File
@@ -62,7 +62,7 @@ docker-build: ## docker build
.PHONY: docker-run
docker-run: docker-build ## docker run
@$(DOCKER) run -it -p $(PORT):8080 -v ./data:/data $(DOCKER_TAG) --token $(TOKEN)
@$(DOCKER) run -it -p $(PORT):8080 -v ./data:/data $(DOCKER_TAG) --token $(TOKEN) --host localhost:$(PORT) --debug
# ACTIONS
README.md
+2 -1
View File
@@ -75,7 +75,7 @@ curl -X DELETE \
- [x] cerbot install in container + path env/arg
- [x] redirect /.well-known/acme-challenge to specific path
- [ ] certbot/self-signed create/renew in specific dir
- [ ] better logger
- [x] better logger
- [ ] renew command
- [ ] https mode w/ multiple hosts
- [ ] restart command (on new/deleted host)
@@ -83,6 +83,7 @@ curl -X DELETE \
- [ ] log visits (and store accross sessions)
- [ ] deliver visits in /page/visits
- [x] better error page
- [ ] add favicon.ico + special path
### Makefile targets
main.py
+2
View File
@@ -1,9 +1,11 @@
from src.params import parse_parameters
from src.server import StaplerServer
from src.logs import setup_logs
def main():
params = parse_parameters()
setup_logs(params)
server = StaplerServer(params)
server.start()
src/data_dir.py
+6
View File
@@ -3,6 +3,7 @@ import io
import tarfile
import shutil
import re
import logging
class DataDir:
@@ -10,6 +11,7 @@ class DataDir:
PATH_REGEX = re.compile(r"^[\w-]+$")
def __init__(self, root_path: str):
self.logger = logging.getLogger(self.__class__.__name__)
self.root_path = root_path
def list_paths(self) -> list[str]:
@@ -29,6 +31,7 @@ class DataDir:
path_host = os.path.join(self.root_path, path, self.HOST_FILE)
with open(path_host, mode="w") as host_file:
host_file.write(host)
self.logger.debug("Wrote %s", path_host)
def has_index(self, path: str):
if self.__valid_path(path):
@@ -52,12 +55,15 @@ class DataDir:
with tarfile.open(fileobj=tar_bytes) as tar_file:
if os.path.exists(target_path):
shutil.rmtree(target_path)
self.logger.debug("Deleted %s", target_path)
tar_file.extractall(target_path)
self.logger.debug("Extracted tar to %s", target_path)
def remove(self, path: str):
if self.__valid_path(path):
target_path = os.path.join(self.root_path, path)
shutil.rmtree(target_path)
self.logger.debug("Deleted %s", target_path)
def exists(self, path: str):
return self.__valid_path(path)
src/handler.py
+94 -24
View File
@@ -4,11 +4,12 @@ import tarfile
import re
import io
import os
import logging
from . import project, params, registry, data_dir
from . import project, params, registry, data_dir, logs
class StaplerRequestHandler(http.server.SimpleHTTPRequestHandler):
class RequestHandler(http.server.SimpleHTTPRequestHandler):
protocol_version = "HTTP/2.0"
server_version = "StaplerServer/" + project.get_version()
CERTBOT_CHALLENGE_PATH = "/.well-known/acme-challenge"
@@ -17,14 +18,71 @@ class StaplerRequestHandler(http.server.SimpleHTTPRequestHandler):
def __init__(
self, *args, params: params.Parameters, registry: registry.Registry, **kwargs
):
self.logger = logging.getLogger(self.__class__.__name__)
self.default_host = params.host
self.token = params.token
self.data_dir = data_dir.DataDir(params.data_dir)
self.max_size_bytes = params.max_size_bytes
self.registry = registry
self.certbot_www = os.path.realpath(params.certbot_www)
self.out_size = 0
super().__init__(*args, directory=params.data_dir, **kwargs)
def log_message(self, format: str, *args):
self.logger.info("%s - " + format, self.address_string(), *args)
def log_error(self, format: str, *args):
self.logger.error("%s - " + format, self.address_string(), *args)
def log_request(self, code="?", size=""):
if isinstance(code, http.HTTPStatus):
color = logs.TermColor.RED
if 100 <= code < 200:
color = logs.TermColor.CYAN
if 200 <= code < 300:
color = logs.TermColor.GREEN
elif 300 <= code < 400:
color = logs.TermColor.BLUE
elif 400 <= code < 500:
color = logs.TermColor.YELLOW
code = color + str(code.value) + logs.TermColor.RESET
if size == "" and self.out_size > 0:
size = str(self.out_size)
if size != "":
self.logger.info(
"→ %s - %s - %s - %s - %s",
code,
self.address_string(),
self.__get_host(),
self.requestline,
size,
)
else:
self.logger.info(
"→ %s - %s - %s - %s",
code,
self.address_string(),
self.__get_host(),
self.requestline,
)
def pre_log_request(self):
if (size := self.__get_length()) > 0:
self.logger.debug(
"← ... - %s - %s - %s - %d",
self.address_string(),
self.__get_host(),
self.requestline,
size,
)
else:
self.logger.debug(
"← ... - %s - %s - %s",
self.address_string(),
self.__get_host(),
self.requestline,
)
def list_directory(self, *_, **__):
"""Disable default directory listing"""
self.send_error(http.HTTPStatus.NOT_FOUND, "File not found")
@@ -32,30 +90,32 @@ class StaplerRequestHandler(http.server.SimpleHTTPRequestHandler):
def translate_path(self, path: str) -> str:
if path.startswith(self.CERTBOT_CHALLENGE_PATH):
return self.certbot_www + path.removeprefix(self.CERTBOT_CHALLENGE_PATH)
if (page := self.registry.get_from_host(self.get_host())) is not None:
if (page := self.registry.get_from_host(self.__get_host())) is not None:
path = f"/{page.path}" + path
path = super().translate_path(path)
if self.get_subpath(match_full=False) is None: # not a valid path
if self.__get_subpath(match_full=False) is None: # not a valid path
return ""
if os.path.basename(path).startswith("."): # hidden files
return ""
return path
def do_HEAD(self):
self.pre_log_request()
super().do_HEAD()
def do_GET(self):
if self.path == "/" and self.get_host() == self.default_host:
return self.server_index()
self.pre_log_request()
if self.path == "/" and self.__get_host() == self.default_host:
return self.__server_index()
super().do_GET()
def do_PUT(self):
self.pre_log_request()
if self.headers["X-Token"] != self.token:
return self.send_error(http.HTTPStatus.UNAUTHORIZED, "Invalid token")
if (sub_path := self.get_subpath()) is None:
if (sub_path := self.__get_subpath()) is None:
return self.send_error(http.HTTPStatus.BAD_REQUEST, "Invalid path")
if not self.headers["Content-Length"]:
content_length = 0
else:
content_length = int(self.headers["Content-Length"])
if content_length == 0:
if (content_length := self.__get_length()) == 0:
return self.send_error(http.HTTPStatus.LENGTH_REQUIRED, "No body found")
if content_length > self.max_size_bytes:
return self.send_error(
@@ -68,15 +128,18 @@ class StaplerRequestHandler(http.server.SimpleHTTPRequestHandler):
return self.send_error(http.HTTPStatus.BAD_REQUEST, "Invalid tar archive")
except Exception as e:
return self.send_error(http.HTTPStatus.INTERNAL_SERVER_ERROR, str(e))
self.send_status_only(http.HTTPStatus.CREATED, f"Resource /{sub_path}/ updated")
self.__send_status_only(
http.HTTPStatus.CREATED, f"Resource /{sub_path}/ updated"
)
if self.headers["X-Host"] is not None:
self.registry.set_host(sub_path, self.headers["X-Host"])
self.registry.add(sub_path)
def do_DELETE(self):
self.pre_log_request()
if self.headers["X-Token"] != self.token:
return self.send_error(http.HTTPStatus.UNAUTHORIZED, "Invalid token")
if (sub_path := self.get_subpath()) is None:
if (sub_path := self.__get_subpath()) is None:
return self.send_error(http.HTTPStatus.BAD_REQUEST, "Invalid path")
if not self.data_dir.exists(sub_path):
return self.send_error(http.HTTPStatus.NOT_FOUND, "Not found")
@@ -84,12 +147,12 @@ class StaplerRequestHandler(http.server.SimpleHTTPRequestHandler):
self.data_dir.remove(sub_path)
except Exception as e:
return self.send_error(http.HTTPStatus.INTERNAL_SERVER_ERROR, str(e))
self.send_status_only(
self.__send_status_only(
http.HTTPStatus.NO_CONTENT, f"Resource /{sub_path}/ removed"
)
self.registry.remove(sub_path)
def get_subpath(self, match_full: bool = True) -> str | None:
def __get_subpath(self, match_full: bool = True) -> str | None:
if match_full:
match = self.PATH_REGEX.fullmatch(self.path)
else:
@@ -98,15 +161,21 @@ class StaplerRequestHandler(http.server.SimpleHTTPRequestHandler):
return match.group(1)
return None
def get_host(self) -> str:
def __get_host(self) -> str:
if self.headers["Host"] is None:
return self.default_host
return self.headers["Host"].split(":")[0]
return self.headers["Host"]
def server_index(self):
self.send_basic_body(self.server_version + "\n")
def __get_length(self) -> int:
if not self.headers["Content-Length"]:
return 0
else:
return int(self.headers["Content-Length"])
def send_basic_body(
def __server_index(self):
self.__send_basic_body(self.server_version + "\n")
def __send_basic_body(
self,
body: str,
content_type: str = "text/plain",
@@ -114,13 +183,14 @@ class StaplerRequestHandler(http.server.SimpleHTTPRequestHandler):
message: str | None = None,
):
encoded: bytes = body.encode()
self.out_size = len(encoded)
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.end_headers()
self.wfile.write(encoded)
def send_status_only(self, code: int, message: str | None = None):
def __send_status_only(self, code: int, message: str | None = None):
self.send_response(code, message)
self.send_header("Content-Length", "0")
self.end_headers()
@@ -134,10 +204,10 @@ class StaplerRequestHandler(http.server.SimpleHTTPRequestHandler):
if explain is None:
explain = longmsg
if "Accept" not in self.headers["Accept"] or "text/" in self.headers["Accept"]:
self.send_basic_body(
self.__send_basic_body(
f"{code} {message}\n{explain}\n{self.server_version}",
code=code,
message=message,
)
else:
self.send_status_only(code, message)
self.__send_status_only(code, message)
src/logs.py
+79
View File
@@ -0,0 +1,79 @@
import enum
import logging
from . import params
class TermColor(enum.StrEnum):
RESET = "0"
BOLD = "1"
FEINT = "2"
UNDERLINE = "4"
BLACK = "30"
RED = "31"
GREEN = "32"
YELLOW = "33"
BLUE = "34"
MAGENTA = "35"
CYAN = "36"
WHITE = "37"
GREY = "38"
def __str__(self) -> str:
return f"\033[{self.value}m"
def __add__(self, second):
return str(self) + str(second)
def __radd__(self, second):
return str(second) + str(self)
class ColoredLoggingFormatter(logging.Formatter):
pre_format = "%(asctime)s | "
level_format = "%(levelname)-8s"
post_format = " | [%(name)s] %(message)s"
trace_format = (
TermColor.FEINT
+ TermColor.GREY
+ " (%(filename)s:%(lineno)d)"
+ TermColor.RESET
)
FORMAT_COLORS = {
logging.DEBUG: TermColor.FEINT + TermColor.GREY,
logging.INFO: TermColor.GREEN,
logging.WARNING: TermColor.YELLOW,
logging.ERROR: TermColor.RED,
logging.CRITICAL: TermColor.RED,
}
def __init__(self, trace: bool):
self.trace = trace
def format(self, record):
log_color: TermColor = (
self.FORMAT_COLORS[record.levelno]
if record.levelno in self.FORMAT_COLORS
else TermColor.MAGENTA
)
formatter = logging.Formatter(
self.pre_format
+ log_color
+ TermColor.BOLD
+ self.level_format
+ TermColor.RESET
+ self.post_format
+ (self.trace_format if self.trace else "")
)
return formatter.format(record)
def setup_logs(params: params.Parameters):
stream_handler = logging.StreamHandler()
stream_handler.setFormatter(ColoredLoggingFormatter(trace=params.debug))
log_level = logging.INFO
if params.debug:
log_level = logging.DEBUG
logging.basicConfig(level=log_level, handlers=[stream_handler])
src/page.py
+1 -1
View File
@@ -13,7 +13,7 @@ class Page:
def __repr__(self) -> str:
out = self.get_url_path()
if self.host is not None:
out += f" ({self.host})"
out += f" [http://{self.host}/]"
if not self.with_index:
out += " (no index)"
return out
src/params.py
+3 -1
View File
@@ -16,6 +16,7 @@ class Parameters:
max_size_bytes: int
certbot_conf: str
certbot_www: str
debug: bool
@classmethod
def from_namespace(cls, args: argparse.Namespace) -> "Parameters":
@@ -83,7 +84,7 @@ def parse_parameters() -> Parameters:
parser,
"--host",
env_var="HOST",
default="localhost",
default="localhost:8080",
help="server default host",
)
__add_arg_str(
@@ -130,5 +131,6 @@ def parse_parameters() -> Parameters:
default=os.path.join(".", "data", ".certbot"),
help="Certbot www dir",
)
parser.add_argument("--debug", action=argparse.BooleanOptionalAction)
args = parser.parse_args()
return Parameters.from_namespace(args)
src/registry.py
+6 -3
View File
@@ -1,11 +1,14 @@
import logging
from . import params, page, data_dir
class Registry:
def __init__(self, params: params.Parameters):
self.logger = logging.getLogger(self.__class__.__name__)
self.pages: dict[str, page.Page] = {}
self.data_dir = data_dir.DataDir(params.data_dir)
self.prefix = f"http://{params.host}:{params.port}"
self.prefix = f"http://{params.host}"
def load_pages(self):
self.pages = {}
@@ -16,7 +19,7 @@ class Registry:
self.pages[path] = page.Page(
path, self.data_dir.has_index(path), self.data_dir.get_host(path)
)
print("Updated: " + self.prefix + str(self.pages[path]))
self.logger.info("Updated %s%s", self.prefix, str(self.pages[path]))
def set_host(self, path: str, host: str):
self.data_dir.set_host(path, host)
@@ -25,7 +28,7 @@ class Registry:
def remove(self, path: str):
page = self.pages[path]
del self.pages[path]
print("Removed: " + self.prefix + str(page))
self.logger.info("Removed %s%s", self.prefix, str(page))
def get_from_host(self, host: str) -> page.Page | None:
for p in self.pages.values():
src/server.py
+13 -9
View File
@@ -1,11 +1,13 @@
import http.server
import os
import logging
from . import params, handler, registry, project
class StaplerServer:
def __init__(self, params: params.Parameters):
self.logger = logging.getLogger(self.__class__.__name__)
self.params = params
self.registry = registry.Registry(params)
self.server = http.server.ThreadingHTTPServer(
@@ -14,25 +16,27 @@ class StaplerServer:
)
def request_handler(self, *args) -> http.server.BaseHTTPRequestHandler:
return handler.StaplerRequestHandler(
*args, params=self.params, registry=self.registry
)
def __repr__(self):
return f"StaplerServer ({project.get_version()})"
return handler.RequestHandler(*args, params=self.params, registry=self.registry)
def __init_certbot_www(self):
os.makedirs(self.params.certbot_www, exist_ok=True)
def __startup(self):
print(f"{self}: starting up...")
self.logger.info("Starting up...")
self.registry.load_pages()
self.__init_certbot_www()
def start(self):
self.logger.info("Version %s", project.get_version())
self.__startup()
print(
f"{self}: serving on http://{self.params.host}:{self.server.server_port}..."
self.logger.info(
"Listening on %s:%d...",
self.server.server_address[0],
self.server.server_port,
)
self.logger.info(
"Server up and ready on http://%s",
self.params.host,
)
try:
self.server.serve_forever()