Archived
StasonJatham
5c46b47937
Preserve the GitHub history and the newer deployed source from ct_134_karldns. Exclude runtime data, credentials, AppleDouble files and local backup copies. Make Gitea the canonical source for documentation, package metadata and project links.
229 lines
9.8 KiB
Python
229 lines
9.8 KiB
Python
from __future__ import annotations
|
|
|
|
import logging
|
|
import datetime
|
|
import ipaddress
|
|
import json
|
|
import time
|
|
import re
|
|
import uuid
|
|
from urllib.parse import unquote_plus
|
|
from collections.abc import Awaitable, Callable
|
|
|
|
from fastapi import FastAPI, Request
|
|
from fastapi.responses import PlainTextResponse
|
|
from starlette.concurrency import run_in_threadpool
|
|
from starlette.responses import Response
|
|
from starlette.types import ASGIApp, Message, Receive, Scope, Send
|
|
|
|
from ..security import client_ip, is_rate_limited_path, rate_limit_bucket
|
|
from ..settings import DdnsSettings
|
|
from ..store import DdnsStore
|
|
|
|
ACCESS_LOGGER = logging.getLogger("karldns.access")
|
|
SECURITY_LOGGER = logging.getLogger("karldns.security")
|
|
SECRET_RESPONSE_PREFIXES = (
|
|
"/m/",
|
|
"/api/v1/hostnames",
|
|
"/api/v1/router/",
|
|
"/api/v1/admin/",
|
|
)
|
|
|
|
|
|
class RequestBodyTooLargeError(Exception):
|
|
pass
|
|
|
|
|
|
class RequestBodyLimitMiddleware:
|
|
def __init__(self, app: ASGIApp, max_body_bytes: int) -> None:
|
|
self.app = app
|
|
self.max_body_bytes = max_body_bytes
|
|
|
|
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
|
|
if scope["type"] != "http":
|
|
await self.app(scope, receive, send)
|
|
return
|
|
|
|
received = 0
|
|
|
|
async def limited_receive() -> Message:
|
|
nonlocal received
|
|
message = await receive()
|
|
if message["type"] == "http.request":
|
|
received += len(message.get("body", b""))
|
|
if received > self.max_body_bytes:
|
|
raise RequestBodyTooLargeError
|
|
return message
|
|
|
|
try:
|
|
await self.app(scope, limited_receive, send)
|
|
except RequestBodyTooLargeError:
|
|
response = PlainTextResponse("request body too large", status_code=413)
|
|
await response(scope, receive, send)
|
|
|
|
|
|
def install_security_middleware(app: FastAPI, settings: DdnsSettings, store: DdnsStore) -> None:
|
|
@app.middleware("http")
|
|
async def security_middleware(request: Request, call_next: Callable[[Request], Awaitable[Response]]) -> Response:
|
|
started_at = time.perf_counter()
|
|
creation = request.method == "POST" and request.url.path in ("/", "/api/v1/hostnames")
|
|
if creation:
|
|
supplied_id = request.headers.get("x-request-id", "")
|
|
request.state.audit_request_id = supplied_id if re.fullmatch(r"[a-fA-F0-9]{32}", supplied_id) else uuid.uuid4().hex
|
|
try:
|
|
content_length = int(request.headers.get("content-length", "0") or "0")
|
|
except ValueError:
|
|
content_length = 0
|
|
if content_length > settings.max_request_body_bytes:
|
|
return PlainTextResponse("request body too large", status_code=413)
|
|
|
|
if is_rate_limited_path(request.url.path) or is_admin_path(request.url.path, settings):
|
|
limit = (
|
|
settings.admin_rate_limit_per_minute
|
|
if is_admin_path(request.url.path, settings)
|
|
else settings.rate_limit_per_minute
|
|
)
|
|
key = f"{client_ip(request, settings)}:{rate_limit_bucket(request.url.path)}"
|
|
allowed = await run_in_threadpool(store.allow_rate_limit, key, limit)
|
|
if not allowed:
|
|
await run_in_threadpool(store.increment_metric, "rate_limit_hits_total")
|
|
emit_security_event(request, settings, 429)
|
|
return PlainTextResponse("rate limit exceeded", status_code=429)
|
|
|
|
if creation:
|
|
body = await request.body()
|
|
decoded = unquote_plus(body.decode("utf-8", errors="replace")).lower()
|
|
# Indicators trigger review; arbitrary nonempty input alone is not an attack verdict.
|
|
request.state.creation_audit = {
|
|
"body_bytes": len(body),
|
|
"unexpected_body": bool(body),
|
|
"php_code_marker": "<?php" in decoded or "<?=" in decoded,
|
|
"shell_substitution_marker": "$(" in decoded or "`" in decoded,
|
|
"sql_union_marker": bool(re.search(r"\bunion\s+(?:all\s+)?select\b", decoded)),
|
|
}
|
|
response = await call_next(request)
|
|
if creation:
|
|
response.headers["X-Request-ID"] = request.state.audit_request_id
|
|
emit_security_event(request, settings, response.status_code)
|
|
if is_secret_response(request):
|
|
response.headers["Cache-Control"] = "no-store"
|
|
response.headers["Pragma"] = "no-cache"
|
|
if is_noindex_response(request):
|
|
response.headers["X-Robots-Tag"] = "noindex, nofollow"
|
|
response.headers["X-Content-Type-Options"] = "nosniff"
|
|
response.headers["X-Frame-Options"] = "DENY"
|
|
response.headers["Referrer-Policy"] = "same-origin"
|
|
response.headers["Permissions-Policy"] = "geolocation=(), microphone=(), camera=()"
|
|
response.headers["Content-Security-Policy"] = content_security_policy(request.url.path)
|
|
if settings.public_base_url.startswith("https://"):
|
|
response.headers["Strict-Transport-Security"] = "max-age=31536000; includeSubDomains"
|
|
if settings.redacted_access_log:
|
|
elapsed_ms = (time.perf_counter() - started_at) * 1000
|
|
ACCESS_LOGGER.info(
|
|
"%s %s status=%s duration_ms=%.2f",
|
|
request.method,
|
|
redacted_access_path(request.url.path),
|
|
response.status_code,
|
|
elapsed_ms,
|
|
)
|
|
return response
|
|
|
|
|
|
def emit_security_event(request: Request, settings: DdnsSettings, status: int) -> None:
|
|
"""Metadata only: never persist paths, bodies, account names or credential values."""
|
|
path = request.url.path
|
|
admin = is_admin_path(path, settings)
|
|
unlock = request.method == "POST" and path.startswith("/m/") and path.endswith("/unlock")
|
|
action, category, outcome = "", "authentication", "unknown"
|
|
if status == 429:
|
|
action, category, outcome = "request_rate_limited", "web", "failure"
|
|
elif status == 403:
|
|
action, category, outcome = "request_rejected", "web", "failure"
|
|
elif status == 401:
|
|
attempted = unlock or admin and bool(request.headers.get("authorization"))
|
|
action = "login_failure" if attempted else "authentication_required"
|
|
outcome = "failure" if attempted else "unknown"
|
|
elif admin and 200 <= status < 400:
|
|
action, outcome = "admin_authenticated_request", "success"
|
|
elif unlock and status == 303:
|
|
action, outcome = "dashboard_unlocked", "success"
|
|
creation = request.method == "POST" and path in ("/", "/api/v1/hostnames")
|
|
if creation:
|
|
action, category = "public_creation_request", "web"
|
|
outcome = "success" if getattr(request.state, "account_created", False) else "failure"
|
|
if not action:
|
|
return
|
|
row = {"@timestamp": datetime.datetime.now(datetime.timezone.utc).isoformat(),
|
|
"event": {"dataset": "karldns.security", "kind": "event", "category": [category],
|
|
"type": ["access" if outcome == "success" else "info"],
|
|
"action": "karldns." + action, "outcome": outcome, "provider": "karldns-native"},
|
|
"http": {"response": {"status_code": status}, "request": {"method": request.method}},
|
|
"labels": {"auth_scope": "admin" if admin else "dashboard" if path.startswith("/m/") else "request",
|
|
"actor_identity": "not_available"}, "message": "KarlDNS " + action}
|
|
if creation:
|
|
row["http"]["request"]["id"] = request.state.audit_request_id
|
|
row["url"] = {"path": path}
|
|
row["labels"].update({"auth_scope": "public_creation", "account_created": getattr(request.state, "account_created", False),
|
|
"dns_activation": False, **getattr(request.state, "creation_audit", {})})
|
|
row["labels"]["indicator_verdict"] = "review_only"
|
|
try:
|
|
ip = str(ipaddress.ip_address(client_ip(request, settings)))
|
|
row["source"] = {"ip": ip}
|
|
row["labels"]["source_ip_basis"] = "trusted_proxy" if request.client and ip != request.client.host else "socket_peer"
|
|
except ValueError:
|
|
pass
|
|
SECURITY_LOGGER.warning("KARLDNSSEC %s", json.dumps(row, separators=(",", ":")))
|
|
|
|
|
|
def is_admin_path(path: str, settings: DdnsSettings) -> bool:
|
|
admin_path = settings.admin_path.rstrip("/")
|
|
return path == admin_path or path.startswith(f"{admin_path}/") or path.startswith("/api/v1/admin/")
|
|
|
|
|
|
def is_secret_response(request: Request) -> bool:
|
|
path = request.url.path
|
|
return (
|
|
is_admin_path(path, request.app.state.settings)
|
|
or path.startswith(SECRET_RESPONSE_PREFIXES)
|
|
or request.method == "POST"
|
|
and path == "/"
|
|
)
|
|
|
|
|
|
def is_noindex_response(request: Request) -> bool:
|
|
path = request.url.path
|
|
if request.method != "GET":
|
|
return True
|
|
if path in {"/openapi.json", "/healthz"}:
|
|
return True
|
|
if path.startswith(("/api/", "/m/")):
|
|
return True
|
|
return is_admin_path(path, request.app.state.settings)
|
|
|
|
|
|
def content_security_policy(path: str) -> str:
|
|
script_src = "script-src 'self' 'unsafe-inline'; " if path in {"/", "/privacy"} else "script-src 'self'; "
|
|
return (
|
|
"default-src 'self'; "
|
|
f"{script_src}"
|
|
"style-src 'self'; "
|
|
"img-src 'self' data:; "
|
|
"font-src 'self'; "
|
|
"connect-src 'self' https://api4.ipify.org https://api6.ipify.org; "
|
|
"form-action 'self'; "
|
|
"frame-ancestors 'none'"
|
|
)
|
|
|
|
|
|
def redacted_access_path(path: str) -> str:
|
|
secret_prefixes = {
|
|
"/m": "/m/<secret>",
|
|
"/api/v1/hostnames": "/api/v1/hostnames/<secret>",
|
|
"/api/v1/router/register": "/api/v1/router/register/<secret>",
|
|
"/api/v1/router/update": "/api/v1/router/update/<secret>",
|
|
}
|
|
for prefix, replacement in secret_prefixes.items():
|
|
if path == prefix or path.startswith(f"{prefix}/"):
|
|
return replacement
|
|
return path
|