From 53428dc0bb47cb86d53450d34c881961ecdfe671 Mon Sep 17 00:00:00 2001
From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com>
Date: Sat, 30 May 2026 20:20:58 +0530
Subject: [PATCH 01/30] fix(dashboard): hide auth/secret keys from settings API

---
 py_src/taskito/dashboard/handlers/settings.py | 24 ++++++++++++++++---
 1 file changed, 21 insertions(+), 3 deletions(-)

diff --git a/py_src/taskito/dashboard/handlers/settings.py b/py_src/taskito/dashboard/handlers/settings.py
index 412c93fe..f02e4bfa 100644
--- a/py_src/taskito/dashboard/handlers/settings.py
+++ b/py_src/taskito/dashboard/handlers/settings.py
@@ -21,15 +21,27 @@
 _MAX_KEY_LENGTH = 256
 _MAX_VALUE_LENGTH = 64 * 1024  # 64 KiB — enough for any realistic dashboard config blob
 
+# Internal namespaces that must never be exposed or mutated through the public
+# settings API. ``auth:`` holds password hashes, live session records (with
+# their CSRF tokens), and the CSRF secret — reading or overwriting any of these
+# is a full auth bypass, so the settings endpoints treat them as if absent.
+_PROTECTED_PREFIXES = ("auth:",)
+
+
+def _is_protected(key: str) -> bool:
+    return any(key.startswith(p) for p in _PROTECTED_PREFIXES)
+
 
 def _validate_key(key: str) -> None:
-    """Reject empty / oversized / control-character keys."""
+    """Reject empty / oversized / control-character / protected keys."""
     if not key:
         raise _BadRequest("setting key must not be empty")
     if len(key) > _MAX_KEY_LENGTH:
         raise _BadRequest(f"setting key exceeds {_MAX_KEY_LENGTH} characters")
     if any(ord(c) < 32 or ord(c) == 127 for c in key):
         raise _BadRequest("setting key must not contain control characters")
+    if _is_protected(key):
+        raise _BadRequest("setting key is reserved")
 
 
 def _validate_value(value: str) -> None:
@@ -38,12 +50,16 @@ def _validate_value(value: str) -> None:
 
 
 def _handle_list_settings(queue: Queue, _qs: dict) -> dict[str, str]:
-    """Return all settings as a ``{key: value}`` dict."""
-    return queue.list_settings()
+    """Return all settings as a ``{key: value}`` dict, minus protected keys."""
+    return {k: v for k, v in queue.list_settings().items() if not _is_protected(k)}
 
 
 def _handle_get_setting(queue: Queue, _qs: dict, key: str) -> dict[str, Any]:
     """Return a single setting, or 404 if it does not exist."""
+    # Protected keys are reported as absent so the API never confirms their
+    # existence or leaks their contents.
+    if _is_protected(key):
+        raise _NotFound(f"setting '{key}' not found")
     value = queue.get_setting(key)
     if value is None:
         raise _NotFound(f"setting '{key}' not found")
@@ -66,4 +82,6 @@ def _handle_set_setting(queue: Queue, body: dict, key: str) -> dict[str, Any]:
 
 def _handle_delete_setting(queue: Queue, key: str) -> dict[str, bool]:
     """Delete a setting. Returns ``{deleted: bool}``."""
+    if _is_protected(key):
+        raise _NotFound(f"setting '{key}' not found")
     return {"deleted": queue.delete_setting(key)}

From 6bd77873286509c1caedba209eeaadec3fca88ca Mon Sep 17 00:00:00 2001
From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com>
Date: Sat, 30 May 2026 20:31:03 +0530
Subject: [PATCH 02/30] fix(dashboard): enforce admin role on mutating routes

---
 py_src/taskito/dashboard/routes.py | 19 +++++++++++++++++++
 py_src/taskito/dashboard/server.py |  7 +++++++
 2 files changed, 26 insertions(+)

diff --git a/py_src/taskito/dashboard/routes.py b/py_src/taskito/dashboard/routes.py
index 09f3cd51..8fa1871b 100644
--- a/py_src/taskito/dashboard/routes.py
+++ b/py_src/taskito/dashboard/routes.py
@@ -267,6 +267,25 @@ def is_state_changing_method(method: str) -> bool:
     return method in {"POST", "PUT", "DELETE", "PATCH"}
 
 
+# Mutating routes that any authenticated user may call against their own
+# session — these are intentionally NOT admin-gated.
+SELF_SERVICE_PATHS: frozenset[str] = frozenset({"/api/auth/logout", "/api/auth/change-password"})
+
+
+def requires_admin(path: str, method: str) -> bool:
+    """Whether a request needs the ``admin`` role.
+
+    Every state-changing API route (cancel/replay jobs, purge DLQ, pause
+    queues, manage webhooks/settings/overrides) is admin-only. ``viewer``
+    sessions keep read (GET) access and their own auth self-service routes.
+    """
+    if not is_state_changing_method(method):
+        return False
+    if not path.startswith("/api/"):
+        return False
+    return not (is_public_path(path) or path in SELF_SERVICE_PATHS)
+
+
 def is_csrf_exempt(path: str) -> bool:
     """Login and setup happen before a session exists, so they're CSRF-exempt.
 
diff --git a/py_src/taskito/dashboard/server.py b/py_src/taskito/dashboard/server.py
index f5d239ae..04267864 100644
--- a/py_src/taskito/dashboard/server.py
+++ b/py_src/taskito/dashboard/server.py
@@ -57,6 +57,7 @@
     is_csrf_exempt,
     is_public_path,
     is_state_changing_method,
+    requires_admin,
 )
 from taskito.dashboard.static import (
     IMMUTABLE_PREFIX,
@@ -428,6 +429,12 @@ def _authorize(self, path: str, method: str) -> tuple[RequestContext, bool]:
                 self._json_response({"error": "csrf_failed"}, status=403)
                 return ctx, True
 
+            # Authorization: mutating routes require the admin role. Viewers
+            # retain read access and their own auth self-service endpoints.
+            if requires_admin(path, method) and ctx.role != "admin":
+                self._json_response({"error": "forbidden"}, status=403)
+                return ctx, True
+
             return ctx, False
 
         def _build_context(self) -> RequestContext:

From e547db86bf4ebeaffabdafba8c9f7b2b60d2c79a Mon Sep 17 00:00:00 2001
From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com>
Date: Sat, 30 May 2026 20:37:59 +0530
Subject: [PATCH 03/30] fix(dashboard): Secure cookies and token-gate metrics
 endpoints

---
 py_src/taskito/cli.py              | 13 ++++++-
 py_src/taskito/dashboard/server.py | 60 +++++++++++++++++++++++-------
 2 files changed, 58 insertions(+), 15 deletions(-)

diff --git a/py_src/taskito/cli.py b/py_src/taskito/cli.py
index cd041990..11bf5e94 100644
--- a/py_src/taskito/cli.py
+++ b/py_src/taskito/cli.py
@@ -60,6 +60,12 @@ def _build_parser() -> argparse.ArgumentParser:
         "--host", default="127.0.0.1", help="Bind address (default: 127.0.0.1)"
     )
     dash_parser.add_argument("--port", type=int, default=8080, help="Bind port (default: 8080)")
+    dash_parser.add_argument(
+        "--insecure-cookies",
+        action="store_true",
+        default=False,
+        help="Drop the Secure flag on session cookies (local HTTP dev only)",
+    )
 
     # info subcommand
     info_parser = subparsers.add_parser("info", help="Show queue statistics")
@@ -280,7 +286,12 @@ def run_worker(args: argparse.Namespace) -> None:
 def run_dashboard(args: argparse.Namespace) -> None:
     """Start the web dashboard."""
     queue = _load_queue(args.app)
-    serve_dashboard(queue, host=args.host, port=args.port)
+    serve_dashboard(
+        queue,
+        host=args.host,
+        port=args.port,
+        secure_cookies=not args.insecure_cookies,
+    )
 
 
 def run_info(args: argparse.Namespace) -> None:
diff --git a/py_src/taskito/dashboard/server.py b/py_src/taskito/dashboard/server.py
index 04267864..543e915b 100644
--- a/py_src/taskito/dashboard/server.py
+++ b/py_src/taskito/dashboard/server.py
@@ -10,8 +10,10 @@
 
 from __future__ import annotations
 
+import hmac
 import json
 import logging
+import os
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from typing import TYPE_CHECKING, Any
 from urllib.parse import parse_qs, unquote, urlparse
@@ -92,16 +94,18 @@ def _safe_path(path: str) -> str:
     return path.translate(_LOG_UNSAFE_CHARS)[:_LOG_PATH_MAX]
 
 
-def _session_cookies(session: Any) -> tuple[str, ...]:
+def _session_cookies(session: Any, *, secure: bool = True) -> tuple[str, ...]:
     """Build the standard ``Set-Cookie`` headers for a freshly-created session.
 
     Used by both password login and OAuth callback so the cookie shape
-    stays in lockstep across login methods.
+    stays in lockstep across login methods. ``secure`` adds the ``Secure``
+    attribute so the cookies are never sent over plain HTTP.
     """
+    secure_attr = "; Secure" if secure else ""
     return (
-        f"{SESSION_COOKIE}={session.token}; HttpOnly; SameSite=Strict; Path=/; "
+        f"{SESSION_COOKIE}={session.token}; HttpOnly; SameSite=Strict; Path=/{secure_attr}; "
         f"Max-Age={DEFAULT_SESSION_TTL_SECONDS}",
-        f"{CSRF_COOKIE}={session.csrf_token}; SameSite=Strict; Path=/; "
+        f"{CSRF_COOKIE}={session.csrf_token}; SameSite=Strict; Path=/{secure_attr}; "
         f"Max-Age={DEFAULT_SESSION_TTL_SECONDS}",
     )
 
@@ -113,6 +117,7 @@ def serve_dashboard(
     *,
     static_assets: StaticAssets | None = None,
     oauth_flow: OAuthFlow | None = None,
+    secure_cookies: bool = True,
 ) -> None:
     """Start the dashboard HTTP server (blocking).
 
@@ -130,7 +135,12 @@ def serve_dashboard(
     bootstrap_admin_from_env(queue)
     if oauth_flow is None:
         oauth_flow = _build_oauth_flow_from_env(queue)
-    handler = _make_handler(queue, static_assets=static_assets, oauth_flow=oauth_flow)
+    handler = _make_handler(
+        queue,
+        static_assets=static_assets,
+        oauth_flow=oauth_flow,
+        secure_cookies=secure_cookies,
+    )
     server = ThreadingHTTPServer((host, port), handler)
     print(f"taskito dashboard → http://{host}:{port}")
     print("Press Ctrl+C to stop")
@@ -167,9 +177,11 @@ def _make_handler(
     *,
     static_assets: StaticAssets | None = None,
     oauth_flow: OAuthFlow | None = None,
+    secure_cookies: bool = True,
 ) -> type:
     """Create a request handler class bound to the given queue."""
     assets = static_assets if static_assets is not None else _get_default_assets()
+    cookie_secure_attr = "; Secure" if secure_cookies else ""
 
     class DashboardHandler(BaseHTTPRequestHandler):
         # ── Entry points ────────────────────────────────────────────
@@ -267,9 +279,11 @@ def _handle_get(self) -> None:
             if path == "/health":
                 self._json_response(check_health())
             elif path == "/readiness":
-                self._json_response(check_readiness(queue))
+                if self._metrics_token_ok():
+                    self._json_response(check_readiness(queue))
             elif path == "/metrics":
-                self._serve_prometheus_metrics()
+                if self._metrics_token_ok():
+                    self._serve_prometheus_metrics()
             else:
                 self._json_response({"error": "Not found"}, status=404)
 
@@ -460,18 +474,19 @@ def _set_login_cookies(self, response: dict[str, Any]) -> None:
                 return
             # 24-hour Max-Age matches the session TTL.
             self._extra_set_cookies = [
-                f"{SESSION_COOKIE}={token}; HttpOnly; SameSite=Strict; Path=/; "
-                f"Max-Age={DEFAULT_SESSION_TTL_SECONDS}",
-                f"{CSRF_COOKIE}={csrf}; SameSite=Strict; Path=/; "
-                f"Max-Age={DEFAULT_SESSION_TTL_SECONDS}",
+                f"{SESSION_COOKIE}={token}; HttpOnly; SameSite=Strict; "
+                f"Path=/{cookie_secure_attr}; Max-Age={DEFAULT_SESSION_TTL_SECONDS}",
+                f"{CSRF_COOKIE}={csrf}; SameSite=Strict; "
+                f"Path=/{cookie_secure_attr}; Max-Age={DEFAULT_SESSION_TTL_SECONDS}",
             ]
             # Don't leak the raw token in the JSON body — the cookie holds it.
             response["session"] = {k: v for k, v in session.items() if k != "token"}
 
         def _clear_login_cookies(self) -> None:
             self._extra_set_cookies = [
-                f"{SESSION_COOKIE}=; HttpOnly; SameSite=Strict; Path=/; Max-Age=0",
-                f"{CSRF_COOKIE}=; SameSite=Strict; Path=/; Max-Age=0",
+                f"{SESSION_COOKIE}=; HttpOnly; SameSite=Strict; "
+                f"Path=/{cookie_secure_attr}; Max-Age=0",
+                f"{CSRF_COOKIE}=; SameSite=Strict; Path=/{cookie_secure_attr}; Max-Age=0",
             ]
 
         # ── Dispatch helper ─────────────────────────────────────────
@@ -516,7 +531,7 @@ def _dispatch_oauth_redirect(
                 return
             cookies: list[str] = []
             if redirect.session is not None:
-                cookies = list(_session_cookies(redirect.session))
+                cookies = list(_session_cookies(redirect.session, secure=secure_cookies))
             self.send_response(redirect.status)
             self.send_header("Location", redirect.url)
             self.send_header("Content-Length", "0")
@@ -563,6 +578,23 @@ def _json_response(self, data: Any, status: int = 200) -> None:
             self.end_headers()
             self.wfile.write(body)
 
+        def _metrics_token_ok(self) -> bool:
+            """Gate ``/metrics`` and ``/readiness`` behind an optional bearer token.
+
+            When ``TASKITO_DASHBOARD_METRICS_TOKEN`` is unset the endpoints stay
+            public (probe-friendly default). When set, a matching
+            ``Authorization: Bearer <token>`` header is required. Writes a 401
+            and returns ``False`` when the check fails.
+            """
+            token = os.environ.get("TASKITO_DASHBOARD_METRICS_TOKEN")
+            if not token:
+                return True
+            header = self.headers.get("Authorization", "")
+            if hmac.compare_digest(header, f"Bearer {token}"):
+                return True
+            self._json_response({"error": "not_authenticated"}, status=401)
+            return False
+
         def _serve_prometheus_metrics(self) -> None:
             try:
                 from prometheus_client import generate_latest

From a40b68fac63a3fd938f76629f00bbda1c4b0bc22 Mon Sep 17 00:00:00 2001
From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com>
Date: Sat, 30 May 2026 20:38:23 +0530
Subject: [PATCH 04/30] fix(dashboard): clear admin password from environ after
 read

---
 py_src/taskito/dashboard/auth.py | 9 ++++++---
 1 file changed, 6 insertions(+), 3 deletions(-)

diff --git a/py_src/taskito/dashboard/auth.py b/py_src/taskito/dashboard/auth.py
index a7f2b6b2..40e96c7b 100644
--- a/py_src/taskito/dashboard/auth.py
+++ b/py_src/taskito/dashboard/auth.py
@@ -22,6 +22,7 @@
 import hmac
 import json
 import logging
+import os
 import secrets
 import time
 from dataclasses import asdict, dataclass
@@ -434,11 +435,13 @@ def bootstrap_admin_from_env(queue: Queue) -> User | None:
     If ``TASKITO_DASHBOARD_ADMIN_USER`` and ``TASKITO_DASHBOARD_ADMIN_PASSWORD``
     are set AND the user does not exist yet, create it. Safe to call on every
     startup — does nothing if the user already exists.
-    """
-    import os
 
+    The password is removed from ``os.environ`` immediately after it is read so
+    it cannot later be harvested via ``/proc/<pid>/environ``, ``ps``, or a
+    crash reporter.
+    """
     username = os.environ.get("TASKITO_DASHBOARD_ADMIN_USER")
-    password = os.environ.get("TASKITO_DASHBOARD_ADMIN_PASSWORD")
+    password = os.environ.pop("TASKITO_DASHBOARD_ADMIN_PASSWORD", None)
     if not username or not password:
         return None
     store = AuthStore(queue)

From 2dedc4b4dcced1eea2cda3e5eb339a7e2687e223 Mon Sep 17 00:00:00 2001
From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com>
Date: Sat, 30 May 2026 20:51:34 +0530
Subject: [PATCH 05/30] fix(webhooks): re-validate resolved IP at delivery and
 block redirects

---
 py_src/taskito/webhooks.py | 43 +++++++++++++++++++++++++++++++++++++-
 1 file changed, 42 insertions(+), 1 deletion(-)

diff --git a/py_src/taskito/webhooks.py b/py_src/taskito/webhooks.py
index b26d30e9..92cad79d 100644
--- a/py_src/taskito/webhooks.py
+++ b/py_src/taskito/webhooks.py
@@ -28,6 +28,7 @@
 from typing import TYPE_CHECKING, Any
 
 from taskito.dashboard.delivery_store import DeliveryStore
+from taskito.dashboard.url_safety import UnsafeWebhookUrl, validate_webhook_url
 from taskito.events import EventType
 
 if TYPE_CHECKING:
@@ -37,6 +38,21 @@
 logger = logging.getLogger("taskito.webhooks")
 
 
+class _NoRedirect(urllib.request.HTTPRedirectHandler):
+    """Redirect handler that refuses to follow 3xx responses.
+
+    Without this, a destination that passed the SSRF check could 30x-bounce
+    delivery to a private/metadata address that the validator never saw.
+    """
+
+    def redirect_request(self, *_args: Any, **_kwargs: Any) -> None:
+        return None
+
+
+# Module-level opener reused across deliveries — it follows no redirects.
+_NO_REDIRECT_OPENER = urllib.request.build_opener(_NoRedirect)
+
+
 class WebhookManager:
     """Delivers webhook POST requests for job events.
 
@@ -212,8 +228,15 @@ def _send(
         for attempt in range(max_retries):
             attempt_count = attempt + 1
             try:
+                # Re-validate at delivery time: the URL passed the check when
+                # the webhook was registered, but DNS may have been rebound
+                # since (or between retries) to a private/loopback/metadata
+                # address. A residual micro-race remains between this resolve
+                # and the socket connect below, but this closes the wide
+                # registration-to-delivery rebinding window.
+                validate_webhook_url(wh["url"])
                 req = urllib.request.Request(wh["url"], data=body, headers=headers, method="POST")
-                with urllib.request.urlopen(req, timeout=timeout) as resp:
+                with _NO_REDIRECT_OPENER.open(req, timeout=timeout) as resp:
                     last_status = int(resp.status)
                     last_response_body = self._read_response_body(resp)
                     if last_status < 400:
@@ -255,6 +278,24 @@ def _send(
                     return e.code
                 if write_to_log:
                     logger.warning("Webhook %s returned server error %d", wh["url"], e.code)
+            except UnsafeWebhookUrl as e:
+                # The target now resolves somewhere we won't deliver to.
+                # Don't retry — record once and stop.
+                last_error = f"blocked unsafe webhook URL: {e}"
+                if write_to_log:
+                    logger.warning("Webhook %s blocked: %s", wh["url"], e)
+                self._record(
+                    wh,
+                    payload,
+                    status="failed",
+                    attempts=attempt_count,
+                    response_code=None,
+                    response_body=None,
+                    latency_ms=int((time.monotonic() - started_at) * 1000),
+                    error=last_error,
+                    write_to_log=write_to_log,
+                )
+                return None
             except Exception as e:
                 last_error = f"{type(e).__name__}: {e}"
                 if write_to_log:

From 58f87323f6fb8a063fad71c0df8a9e40576ae126 Mon Sep 17 00:00:00 2001
From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com>
Date: Sat, 30 May 2026 20:52:23 +0530
Subject: [PATCH 06/30] test(webhooks): allow private host for local delivery
 tests

---
 tests/observability/test_webhooks.py | 11 +++++++++--
 1 file changed, 9 insertions(+), 2 deletions(-)

diff --git a/tests/observability/test_webhooks.py b/tests/observability/test_webhooks.py
index ac41b427..bd4c2408 100644
--- a/tests/observability/test_webhooks.py
+++ b/tests/observability/test_webhooks.py
@@ -17,8 +17,15 @@
 
 
 @pytest.fixture
-def webhook_server() -> Generator[tuple[str, list[dict[str, Any]]]]:
-    """Start a local HTTP server that records webhook deliveries."""
+def webhook_server(
+    monkeypatch: pytest.MonkeyPatch,
+) -> Generator[tuple[str, list[dict[str, Any]]]]:
+    """Start a local HTTP server that records webhook deliveries.
+
+    Delivery now validates the target URL and refuses private/loopback hosts,
+    so the 127.0.0.1 test server needs the documented dev escape hatch.
+    """
+    monkeypatch.setenv("TASKITO_WEBHOOKS_ALLOW_PRIVATE", "1")
     received: list[dict[str, Any]] = []
 
     class Handler(BaseHTTPRequestHandler):

From cd83c6d781173916b555f149d8acb02c9d4897f5 Mon Sep 17 00:00:00 2001
From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com>
Date: Sat, 30 May 2026 21:03:24 +0530
Subject: [PATCH 07/30] feat(serializers): add opt-in SignedSerializer (HMAC
 integrity)

---
 py_src/taskito/__init__.py    |  2 ++
 py_src/taskito/serializers.py | 52 +++++++++++++++++++++++++++++++++++
 2 files changed, 54 insertions(+)

diff --git a/py_src/taskito/__init__.py b/py_src/taskito/__init__.py
index 21931f23..2e2a6816 100644
--- a/py_src/taskito/__init__.py
+++ b/py_src/taskito/__init__.py
@@ -44,6 +44,7 @@
     JsonSerializer,
     MsgPackSerializer,
     Serializer,
+    SignedSerializer,
     SmartSerializer,
 )
 from taskito.task import TaskWrapper
@@ -84,6 +85,7 @@
     "SerializationError",
     "Serializer",
     "Signature",
+    "SignedSerializer",
     "SmartSerializer",
     "SoftTimeoutError",
     "TaskCancelledError",
diff --git a/py_src/taskito/serializers.py b/py_src/taskito/serializers.py
index 51c8f95d..f97cb2e8 100644
--- a/py_src/taskito/serializers.py
+++ b/py_src/taskito/serializers.py
@@ -2,6 +2,8 @@
 
 from __future__ import annotations
 
+import hashlib
+import hmac
 import json
 from typing import Any, Protocol, runtime_checkable
 
@@ -136,6 +138,56 @@ def loads(self, data: bytes) -> Any:
         return cloudpickle.loads(data)
 
 
+class SignedSerializer:
+    """Wraps another serializer with an HMAC-SHA256 integrity tag.
+
+    Task payloads and results are read back from the queue's storage and
+    deserialized on every worker. When the inner serializer can execute code
+    on load (``cloudpickle`` / the default ``SmartSerializer`` fallback),
+    anyone able to write to the storage backend can achieve remote code
+    execution. ``SignedSerializer`` prepends a keyed HMAC so the worker
+    refuses to deserialize bytes that were not produced with the shared key.
+
+    This authenticates but does **not** encrypt — pair it with
+    :class:`EncryptedSerializer` if confidentiality is also required.
+
+    Usage::
+
+        import os
+        from taskito import SignedSerializer, SmartSerializer
+
+        key = os.urandom(32)  # share across producers and workers
+        serializer = SignedSerializer(SmartSerializer(), key)
+        queue = Queue(serializer=serializer)
+    """
+
+    _DIGEST_SIZE = 32  # SHA-256 output length in bytes
+
+    def __init__(self, inner: Serializer, key: bytes):
+        if not isinstance(key, bytes):
+            raise TypeError(f"key must be bytes, got {type(key).__name__}")
+        if len(key) < 32:
+            raise ValueError(
+                f"key must be at least 32 bytes of CSPRNG output, got {len(key)} bytes"
+            )
+        self._inner = inner
+        self._key = key
+
+    def dumps(self, obj: Any) -> bytes:
+        body = self._inner.dumps(obj)
+        tag = hmac.new(self._key, body, hashlib.sha256).digest()
+        return tag + body
+
+    def loads(self, data: bytes) -> Any:
+        if len(data) < self._DIGEST_SIZE:
+            raise ValueError("Signed payload too short")
+        tag, body = data[: self._DIGEST_SIZE], data[self._DIGEST_SIZE :]
+        expected = hmac.new(self._key, body, hashlib.sha256).digest()
+        if not hmac.compare_digest(tag, expected):
+            raise ValueError("Signature verification failed: payload integrity check failed")
+        return self._inner.loads(body)
+
+
 class EncryptedSerializer:
     """Wraps another serializer with AES-256-GCM encryption at rest.
 

From 11819456fd74f5a9481362ccfe213168a3b4eb2a Mon Sep 17 00:00:00 2001
From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com>
Date: Sat, 30 May 2026 21:10:46 +0530
Subject: [PATCH 08/30] fix(proxies): harden recipe signing key and warn when
 unsigned

---
 py_src/taskito/proxies/reconstruct.py | 23 +++++++++++++++++++++++
 py_src/taskito/proxies/signing.py     | 20 ++++++++++++++++----
 2 files changed, 39 insertions(+), 4 deletions(-)

diff --git a/py_src/taskito/proxies/reconstruct.py b/py_src/taskito/proxies/reconstruct.py
index 62f5cd2a..1cd4c182 100644
--- a/py_src/taskito/proxies/reconstruct.py
+++ b/py_src/taskito/proxies/reconstruct.py
@@ -19,6 +19,29 @@
 
 logger = logging.getLogger("taskito.proxies")
 
+# Set once we've warned about reconstructing proxy recipes without a signing
+# key, so the warning fires at most once per process instead of per recipe.
+_UNSIGNED_WARNED = False
+
+
+def _warn_unsigned_once() -> None:
+    """Warn (once) that a proxy recipe was reconstructed without verification.
+
+    A proxy recipe travels inside the untrusted payload. With no signing key
+    its integrity is unchecked, so anyone able to write to the queue can forge
+    one. Surfacing this at the moment it actually happens avoids spamming every
+    Queue that merely has the built-in handlers registered.
+    """
+    global _UNSIGNED_WARNED
+    if _UNSIGNED_WARNED:
+        return
+    _UNSIGNED_WARNED = True
+    logger.warning(
+        "Reconstructing proxy recipes without integrity verification; set "
+        "recipe_signing_key= or TASKITO_RECIPE_SECRET to enable HMAC signing."
+    )
+
+
 _PROXY_KEY = "__taskito_proxy__"
 _REF_KEY = "__taskito_ref__"
 
diff --git a/py_src/taskito/proxies/signing.py b/py_src/taskito/proxies/signing.py
index be67a398..2be4c5b4 100644
--- a/py_src/taskito/proxies/signing.py
+++ b/py_src/taskito/proxies/signing.py
@@ -10,12 +10,24 @@
 from taskito.exceptions import ProxyReconstructionError
 
 
-def sign_recipe(handler_name: str, version: int, recipe: dict[str, Any], secret: str) -> str:
-    """Compute HMAC-SHA256 over handler_name + version + canonical JSON."""
+def _key_bytes(secret: str | bytes) -> bytes:
+    """Normalise the signing secret to bytes (UTF-8 for ``str``)."""
+    return secret if isinstance(secret, bytes) else secret.encode("utf-8")
+
+
+def sign_recipe(
+    handler_name: str, version: int, recipe: dict[str, Any], secret: str | bytes
+) -> str:
+    """Compute HMAC-SHA256 over handler_name + version + canonical JSON.
+
+    The signature covers the handler name, version, and recipe — but not the
+    owning job, so a valid signed recipe can be replayed into another job.
+    Treat the signing key as a shared secret and rotate it if leaked.
+    """
     canonical = f"{handler_name}:{version}:" + json.dumps(
         recipe, sort_keys=True, separators=(",", ":")
     )
-    return hmac.new(secret.encode(), canonical.encode(), hashlib.sha256).hexdigest()
+    return hmac.new(_key_bytes(secret), canonical.encode("utf-8"), hashlib.sha256).hexdigest()
 
 
 def verify_recipe(
@@ -23,7 +35,7 @@ def verify_recipe(
     version: int,
     recipe: dict[str, Any],
     checksum: str,
-    secret: str,
+    secret: str | bytes,
 ) -> bool:
     """Return True if checksum matches; raise ProxyReconstructionError if not."""
     expected = sign_recipe(handler_name, version, recipe, secret)

From 5a14567cdd48c4d4e4c1e0b8c3acde8d4309297a Mon Sep 17 00:00:00 2001
From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com>
Date: Sat, 30 May 2026 21:11:10 +0530
Subject: [PATCH 09/30] fix(proxies): close file allowlist prefix and symlink
 bypass

---
 py_src/taskito/proxies/handlers/file.py | 30 +++++++++++++++++++++----
 1 file changed, 26 insertions(+), 4 deletions(-)

diff --git a/py_src/taskito/proxies/handlers/file.py b/py_src/taskito/proxies/handlers/file.py
index dbb0c231..f4f50376 100644
--- a/py_src/taskito/proxies/handlers/file.py
+++ b/py_src/taskito/proxies/handlers/file.py
@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import io
+import os
 import pathlib
 import sys
 from typing import Any, ClassVar
@@ -57,22 +58,43 @@ def deconstruct(self, obj: Any) -> dict[str, Any]:
     def reconstruct(self, recipe: dict[str, Any], version: int) -> Any:
         path = recipe["path"]
 
-        # Path allowlist enforcement
+        # Path allowlist enforcement. Resolve once, check containment by path
+        # component (not string prefix — ``/var/data`` must not allow
+        # ``/var/data_exfil``), then open the *resolved* path so a symlink
+        # swapped in after the check can't redirect us outside the allowlist.
+        open_path = path
         if self._path_allowlist:
             resolved = str(pathlib.Path(path).resolve())
-            allowed = any(resolved.startswith(prefix) for prefix in self._path_allowlist)
-            if not allowed:
+            if not self._is_allowed(resolved):
                 raise ProxyReconstructionError(f"File path '{path}' is not in the allowed paths")
+            open_path = resolved
 
         kwargs: dict[str, Any] = {}
         if recipe.get("encoding") is not None:
             kwargs["encoding"] = recipe["encoding"]
-        f = open(path, recipe["mode"], **kwargs)  # noqa: SIM115
+        f = open(open_path, recipe["mode"], **kwargs)  # noqa: SIM115
         position = recipe.get("position", 0)
         if position and position > 0:
             f.seek(position)
         return f
 
+    def _is_allowed(self, resolved: str) -> bool:
+        """Whether a resolved path lies within an allowlisted directory.
+
+        Containment is by path component via ``os.path.commonpath`` so
+        ``/var/data`` does not admit ``/var/data_exfil``. Incomparable paths
+        (e.g. different drives) fail closed.
+        """
+        for prefix in self._path_allowlist:
+            if resolved == prefix:
+                return True
+            try:
+                if os.path.commonpath([resolved, prefix]) == prefix:
+                    return True
+            except ValueError:
+                continue
+        return False
+
     def cleanup(self, obj: Any) -> None:
         if not obj.closed:
             obj.close()

From 2a5c7bf8ec8dc8aa18ddcb47d0ce752154b422b6 Mon Sep 17 00:00:00 2001
From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com>
Date: Sat, 30 May 2026 21:12:10 +0530
Subject: [PATCH 10/30] fix(interception): restrict type_path reconstruction to
 safe kinds

---
 py_src/taskito/interception/converters.py | 26 ++++++++++++++++++++---
 1 file changed, 23 insertions(+), 3 deletions(-)

diff --git a/py_src/taskito/interception/converters.py b/py_src/taskito/interception/converters.py
index 6f328f2f..b440ec14 100644
--- a/py_src/taskito/interception/converters.py
+++ b/py_src/taskito/interception/converters.py
@@ -6,6 +6,7 @@
 import dataclasses
 import datetime
 import decimal
+import enum
 import importlib
 import pathlib
 import re
@@ -14,12 +15,21 @@
 
 
 def _import_type(type_path: str) -> type:
-    """Import a type from its fully-qualified dotted path."""
+    """Import a type from its fully-qualified dotted path.
+
+    The path comes from the (untrusted) payload, so the resolved object must
+    be a class — never an arbitrary callable like ``os.system``. Per-kind
+    checks in the reconstructors below further constrain *which* classes are
+    acceptable.
+    """
     module_path, _, class_name = type_path.rpartition(".")
     if not module_path:
         raise ValueError(f"Invalid type path: {type_path!r}")
     module = importlib.import_module(module_path)
-    return getattr(module, class_name)  # type: ignore[no-any-return]
+    obj = getattr(module, class_name)
+    if not isinstance(obj, type):
+        raise ValueError(f"Type path {type_path!r} did not resolve to a class")
+    return obj
 
 
 def _type_path(cls: type) -> str:
@@ -125,6 +135,8 @@ def convert_enum(obj: Any) -> dict[str, Any]:
 
 def reconstruct_enum(data: dict[str, Any]) -> Any:
     cls = _import_type(data["type_path"])
+    if not issubclass(cls, enum.Enum):
+        raise ValueError(f"{data['type_path']!r} is not an Enum")
     return cls(data["value"])
 
 
@@ -142,7 +154,11 @@ def convert_pydantic(obj: Any) -> dict[str, Any]:
 
 def reconstruct_pydantic(data: dict[str, Any]) -> Any:
     cls = _import_type(data["type_path"])
-    return cls.model_validate(data["value"])  # type: ignore[attr-defined]
+    # Duck-typed so pydantic stays an optional dependency: a genuine model
+    # class exposes ``model_validate`` / ``model_dump``.
+    if not (hasattr(cls, "model_validate") and hasattr(cls, "model_fields")):
+        raise ValueError(f"{data['type_path']!r} is not a pydantic model")
+    return cls.model_validate(data["value"])
 
 
 # -- dataclass --
@@ -159,6 +175,8 @@ def convert_dataclass(obj: Any) -> dict[str, Any]:
 
 def reconstruct_dataclass(data: dict[str, Any]) -> Any:
     cls = _import_type(data["type_path"])
+    if not dataclasses.is_dataclass(cls):
+        raise ValueError(f"{data['type_path']!r} is not a dataclass")
     return cls(**data["value"])
 
 
@@ -193,6 +211,8 @@ def convert_named_tuple(obj: Any) -> dict[str, Any]:
 
 def reconstruct_named_tuple(data: dict[str, Any]) -> Any:
     cls = _import_type(data["type_path"])
+    if not (issubclass(cls, tuple) and hasattr(cls, "_fields")):
+        raise ValueError(f"{data['type_path']!r} is not a NamedTuple")
     return cls(*data["fields"])
 
 

From 70cc81e946806e456bbdc8e42c8c5d77ccc5880a Mon Sep 17 00:00:00 2001
From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com>
Date: Sat, 30 May 2026 21:19:28 +0530
Subject: [PATCH 11/30] test(security): cover SignedSerializer, file allowlist,
 type_path guards

---
 tests/core/test_serializers.py       | 42 ++++++++++++++++++++++++++++
 tests/resources/test_interception.py | 27 ++++++++++++++++++
 tests/resources/test_proxies.py      | 16 +++++++++++
 3 files changed, 85 insertions(+)

diff --git a/tests/core/test_serializers.py b/tests/core/test_serializers.py
index f5da3935..88fa786d 100644
--- a/tests/core/test_serializers.py
+++ b/tests/core/test_serializers.py
@@ -9,6 +9,7 @@
     CloudpickleSerializer,
     JsonSerializer,
     Serializer,
+    SignedSerializer,
     SmartSerializer,
 )
 
@@ -217,3 +218,44 @@ def test_empty_payload_rejected(self) -> None:
 
     def test_satisfies_protocol(self) -> None:
         assert isinstance(SmartSerializer(), Serializer)
+
+
+class TestSignedSerializer:
+    def test_roundtrip(self) -> None:
+        s = SignedSerializer(SmartSerializer(), b"k" * 32)
+        assert s.loads(s.dumps({"hello": "world"})) == {"hello": "world"}
+
+    def test_requires_bytes_key(self) -> None:
+        with pytest.raises(TypeError, match="key must be bytes"):
+            SignedSerializer(CloudpickleSerializer(), "string-key")  # type: ignore[arg-type]
+
+    def test_rejects_short_key(self) -> None:
+        with pytest.raises(ValueError, match="at least 32 bytes"):
+            SignedSerializer(CloudpickleSerializer(), b"tooshort")
+
+    def test_wrong_key_rejected(self) -> None:
+        producer = SignedSerializer(CloudpickleSerializer(), b"a" * 32)
+        attacker = SignedSerializer(CloudpickleSerializer(), b"b" * 32)
+        data = producer.dumps({"x": 1})
+        with pytest.raises(ValueError, match="integrity check failed"):
+            attacker.loads(data)
+
+    def test_tamper_detected(self) -> None:
+        s = SignedSerializer(CloudpickleSerializer(), b"k" * 32)
+        data = s.dumps({"x": 1})
+        tampered = data[:-1] + bytes([data[-1] ^ 0xFF])
+        with pytest.raises(ValueError, match="integrity check failed"):
+            s.loads(tampered)
+
+    def test_truncated_payload_rejected(self) -> None:
+        s = SignedSerializer(CloudpickleSerializer(), b"k" * 32)
+        with pytest.raises(ValueError, match="too short"):
+            s.loads(b"short")
+
+    def test_blocks_forged_unsigned_body(self) -> None:
+        # An attacker who can write to storage but lacks the key cannot get a
+        # forged cloudpickle body past verification (the RCE vector).
+        s = SignedSerializer(CloudpickleSerializer(), b"k" * 32)
+        forged = CloudpickleSerializer().dumps({"evil": True})
+        with pytest.raises(ValueError, match="integrity check failed"):
+            s.loads(b"\x00" * 32 + forged)
diff --git a/tests/resources/test_interception.py b/tests/resources/test_interception.py
index 4f7db154..2ef882c7 100644
--- a/tests/resources/test_interception.py
+++ b/tests/resources/test_interception.py
@@ -444,3 +444,30 @@ def convert_money(obj: Money) -> dict[str, Any]:
         args, _ = strict.intercept((Money(100, "USD"),), {})
         assert args[0]["__taskito_convert__"] is True
         assert args[0]["value"]["amount"] == 100
+
+
+class TestMaliciousTypePathRejected:
+    """type_path comes from the untrusted payload; reconstruction must refuse
+    to import-and-call arbitrary objects like os.system / eval (H6)."""
+
+    def test_enum_rejects_non_enum(self) -> None:
+        data = {"type_key": "enum", "type_path": "os.system", "value": "id"}
+        with pytest.raises(ValueError):
+            reconstruct_converted(data)
+
+    def test_dataclass_rejects_non_dataclass(self) -> None:
+        # Target a function (not a class): _import_type must reject it before
+        # any call, proving an attacker can't invoke arbitrary callables.
+        data = {"type_key": "dataclass", "type_path": "subprocess.run", "value": {"args": "id"}}
+        with pytest.raises(ValueError):
+            reconstruct_converted(data)
+
+    def test_named_tuple_rejects_plain_class(self) -> None:
+        data = {"type_key": "named_tuple", "type_path": "subprocess.Popen", "fields": ["id"]}
+        with pytest.raises(ValueError):
+            reconstruct_converted(data)
+
+    def test_callable_path_rejected_as_non_class(self) -> None:
+        data = {"type_key": "enum", "type_path": "builtins.eval", "value": "1+1"}
+        with pytest.raises(ValueError):
+            reconstruct_converted(data)
diff --git a/tests/resources/test_proxies.py b/tests/resources/test_proxies.py
index 312ed4f9..30c848a0 100644
--- a/tests/resources/test_proxies.py
+++ b/tests/resources/test_proxies.py
@@ -484,3 +484,19 @@ def cleanup(self, obj: Any) -> None:
             reconstruct_proxies((marker,), {}, reg, max_timeout=1)
         # Surfaced on the 1s budget rather than blocking for the 5s sleep.
         assert time.monotonic() - started < 3.0
+
+
+def test_file_handler_rejects_sibling_prefix_bypass(tmp_path: Any) -> None:
+    """A string-prefix check would let '<allow>_exfil' through; component
+    containment must not (H4)."""
+    allowed = tmp_path / "data"
+    allowed.mkdir()
+    sibling = tmp_path / "data_exfil"
+    sibling.mkdir()
+    secret = sibling / "secret.txt"
+    secret.write_text("top-secret")
+
+    handler = FileHandler(path_allowlist=[str(allowed)])
+    recipe = {"path": str(secret), "mode": "r"}
+    with pytest.raises(ProxyReconstructionError, match="not in the allowed"):
+        handler.reconstruct(recipe, 1)

From 3477d796a5cae03107750761bf7b8ad7bf136524 Mon Sep 17 00:00:00 2001
From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com>
Date: Sat, 30 May 2026 21:29:03 +0530
Subject: [PATCH 12/30] fix(locks): mask foreign lock owner token in info()

---
 py_src/taskito/locks.py | 17 +++++++++++++++--
 1 file changed, 15 insertions(+), 2 deletions(-)

diff --git a/py_src/taskito/locks.py b/py_src/taskito/locks.py
index f6a2e9d3..ed5322fe 100644
--- a/py_src/taskito/locks.py
+++ b/py_src/taskito/locks.py
@@ -81,8 +81,21 @@ def extend(self, ttl: float | None = None) -> bool:
         return self._inner.extend_lock(self._name, self._owner_id, ttl_ms)
 
     def info(self) -> dict[str, Any] | None:
-        """Get lock info."""
-        return self._inner.get_lock_info(self._name)
+        """Get lock info.
+
+        ``owner_id`` is a bearer token: any caller that learns it can release
+        or extend another holder's lock. The real holder already knows its own
+        token, so it's only revealed when it matches this handle's owner; every
+        other caller (e.g. a peer polling lock status) sees it masked.
+        ``acquired_at`` / ``expires_at`` are left intact for monitoring.
+        """
+        info = self._inner.get_lock_info(self._name)
+        if not info:
+            return info
+        owner = info.get("owner_id")
+        if owner is not None and owner != self._owner_id:
+            info = {**info, "owner_id": "***"}
+        return info
 
     def _start_extend(self) -> None:
         """Start auto-extend background thread."""

From 8b5620d8ed3b891492f0b38777f05cb1f03ccc6d Mon Sep 17 00:00:00 2001
From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com>
Date: Sat, 30 May 2026 21:29:18 +0530
Subject: [PATCH 13/30] fix(scaler): bind to localhost by default

---
 py_src/taskito/cli.py    | 6 +++++-
 py_src/taskito/scaler.py | 7 ++++---
 2 files changed, 9 insertions(+), 4 deletions(-)

diff --git a/py_src/taskito/cli.py b/py_src/taskito/cli.py
index 11bf5e94..8ddd017b 100644
--- a/py_src/taskito/cli.py
+++ b/py_src/taskito/cli.py
@@ -106,7 +106,11 @@ def _build_parser() -> argparse.ArgumentParser:
         required=True,
         help="Python path to the Queue instance (e.g., 'myapp.tasks:queue')",
     )
-    scaler_parser.add_argument("--host", default="0.0.0.0", help="Bind address (default: 0.0.0.0)")
+    scaler_parser.add_argument(
+        "--host",
+        default="127.0.0.1",
+        help="Bind address (default: 127.0.0.1; use 0.0.0.0 only behind a network boundary)",
+    )
     scaler_parser.add_argument("--port", type=int, default=9091, help="Bind port (default: 9091)")
     scaler_parser.add_argument(
         "--target-queue-depth",
diff --git a/py_src/taskito/scaler.py b/py_src/taskito/scaler.py
index 21f2a13e..a3ded4ea 100644
--- a/py_src/taskito/scaler.py
+++ b/py_src/taskito/scaler.py
@@ -10,7 +10,7 @@
 Or programmatically::
 
     from taskito.scaler import serve_scaler
-    serve_scaler(queue, host="0.0.0.0", port=9091)
+    serve_scaler(queue, host="127.0.0.1", port=9091)
 """
 
 from __future__ import annotations
@@ -31,7 +31,7 @@
 
 def serve_scaler(
     queue: Queue,
-    host: str = "0.0.0.0",
+    host: str = "127.0.0.1",
     port: int = 9091,
     target_queue_depth: int = 10,
 ) -> None:
@@ -39,7 +39,8 @@ def serve_scaler(
 
     Args:
         queue: The Queue instance to monitor.
-        host: Bind address.
+        host: Bind address. Defaults to loopback; only bind to ``0.0.0.0``
+            behind a trusted network boundary — the endpoints are unauthenticated.
         port: Bind port.
         target_queue_depth: Default scaling target hint for KEDA.
     """

From e045a1a301373632ed17bedae7dec9595c07f538 Mon Sep 17 00:00:00 2001
From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com>
Date: Sat, 30 May 2026 21:40:30 +0530
Subject: [PATCH 14/30] fix(idempotency): use NUL domain separator in auto key

---
 py_src/taskito/app.py | 7 ++++++-
 1 file changed, 6 insertions(+), 1 deletion(-)

diff --git a/py_src/taskito/app.py b/py_src/taskito/app.py
index 805f7c1e..9fa40b8f 100644
--- a/py_src/taskito/app.py
+++ b/py_src/taskito/app.py
@@ -302,7 +302,12 @@ def _get_serializer(self, task_name: str) -> Serializer:
     @staticmethod
     def _auto_idempotency_key(task_name: str, payload: bytes) -> str:
         """Derive a deterministic dedup key from task name + serialized payload."""
-        digest = hashlib.sha256(task_name.encode("utf-8") + b"|" + payload).hexdigest()
+        # NUL separates the task name from the payload so the boundary is
+        # unambiguous. A printable separator like "|" can occur in both fields,
+        # letting one job's (name, payload) hash-collide with another's — a
+        # job-suppression vector when task names are caller-influenced. NUL
+        # cannot appear in a Python task name.
+        digest = hashlib.sha256(task_name.encode("utf-8") + b"\x00" + payload).hexdigest()
         return f"auto:{digest[:32]}"
 
     def _resolve_unique_key(

From d0aa59b28f1383231070bdbe03ef1c0c9af5c1e3 Mon Sep 17 00:00:00 2001
From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com>
Date: Sat, 30 May 2026 21:42:33 +0530
Subject: [PATCH 15/30] fix(queue): cap serialized payload size at enqueue

---
 py_src/taskito/app.py | 21 +++++++++++++++++++++
 1 file changed, 21 insertions(+)

diff --git a/py_src/taskito/app.py b/py_src/taskito/app.py
index 9fa40b8f..53f9cb0d 100644
--- a/py_src/taskito/app.py
+++ b/py_src/taskito/app.py
@@ -75,6 +75,12 @@ class QueueWorkflowMixin:  # type: ignore[no-redef]
 
 logger = logging.getLogger("taskito")
 
+# Hard cap on a single serialized task payload. Unbounded payloads let any
+# producer exhaust DB/Redis storage and spike per-job worker memory, so reject
+# oversized ones at enqueue time. Override per-Queue via ``max_payload_bytes``
+# (set to 0 to disable).
+DEFAULT_MAX_PAYLOAD_BYTES = 1 * 1024 * 1024  # 1 MiB
+
 
 class Queue(
     QueueDecoratorMixin,
@@ -262,6 +268,8 @@ def __init__(
         self._proxy_metrics = ProxyMetrics()
         self._recipe_signing_key = recipe_signing_key or os.environ.get("TASKITO_RECIPE_SECRET")
         self._max_reconstruction_timeout = max_reconstruction_timeout
+        # Reject oversized serialized payloads at enqueue time (0 disables).
+        self.max_payload_bytes: int = DEFAULT_MAX_PAYLOAD_BYTES
 
         # Argument interception
         self._interception_metrics: InterceptionMetrics | None = None
@@ -299,6 +307,15 @@ def _get_serializer(self, task_name: str) -> Serializer:
         """Get the serializer for a task (per-task or queue-level fallback)."""
         return self._task_serializers.get(task_name, self._serializer)
 
+    def _check_payload_size(self, task_name: str, size: int) -> None:
+        """Reject a serialized payload that exceeds ``max_payload_bytes``."""
+        limit = self.max_payload_bytes
+        if limit and size > limit:
+            raise ValueError(
+                f"Serialized payload for task '{task_name}' is {size} bytes, exceeding the "
+                f"{limit}-byte limit (adjust queue.max_payload_bytes)"
+            )
+
     @staticmethod
     def _auto_idempotency_key(task_name: str, payload: bytes) -> str:
         """Derive a deterministic dedup key from task name + serialized payload."""
@@ -366,6 +383,7 @@ def _dispatch_batched_payload(
         del config  # currently unused at dispatch time; reserved for telemetry
         task_serializer = self._get_serializer(task_name)
         payload = task_serializer.dumps(((items,), {}))
+        self._check_payload_size(task_name, len(payload))
         py_job = self._inner.enqueue(
             task_name=task_name,
             payload=payload,
@@ -535,6 +553,7 @@ def enqueue(
 
         task_serializer = self._get_serializer(task_name)
         payload = task_serializer.dumps((final_args, final_kwargs))
+        self._check_payload_size(task_name, len(payload))
 
         # Evaluate enqueue-time predicate (if registered). Outcome may
         # adjust the delay (Defer / False+defer), raise (Cancel /
@@ -758,6 +777,8 @@ def enqueue_many(
             payloads = [
                 task_serializer.dumps((a, kw)) for a, kw in zip(args_list, kw_list, strict=True)
             ]
+        for payload in payloads:
+            self._check_payload_size(task_name, len(payload))
         task_names = [task_name] * count
 
         per_job_unique_keys = [

From c0203cb188a8343a1ef2d092fa108860fcd10963 Mon Sep 17 00:00:00 2001
From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com>
Date: Sat, 30 May 2026 22:21:12 +0530
Subject: [PATCH 16/30] fix(worker): drop exception repr from failure log

---
 py_src/taskito/mixins/decorators.py | 7 +++++--
 1 file changed, 5 insertions(+), 2 deletions(-)

diff --git a/py_src/taskito/mixins/decorators.py b/py_src/taskito/mixins/decorators.py
index 11f88311..50c0832d 100644
--- a/py_src/taskito/mixins/decorators.py
+++ b/py_src/taskito/mixins/decorators.py
@@ -294,13 +294,16 @@ def wrapper(*args: Any, **kwargs: Any) -> Any:
                         exc,
                     )
                 else:
+                    # Log the exception type only; ``exc_info`` carries the full
+                    # detail. ``%r`` of the exception is avoided because its
+                    # message can echo task argument data into log sinks.
                     logger.error(
-                        "Task %s[%s] raised %s in %.3fs: %r",
+                        "Task %s[%s] raised %s in %.3fs",
                         task_name,
                         job_id,
                         type(exc).__name__,
                         elapsed,
-                        exc,
+                        exc_info=True,
                     )
                 for hook in hooks["on_failure"]:
                     hook(task_name, args, kwargs, exc)

From 762137cb5045d23d94624d4dfd189c3b2e06cba9 Mon Sep 17 00:00:00 2001
From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com>
Date: Sat, 30 May 2026 22:28:07 +0530
Subject: [PATCH 17/30] fix(dashboard): summarize tracebacks in job API
 responses

---
 py_src/taskito/result.py  | 23 ++++++++++++++++++++++-
 tests/core/test_result.py | 34 ++++++++++++++++++++++++++++++++++
 2 files changed, 56 insertions(+), 1 deletion(-)
 create mode 100644 tests/core/test_result.py

diff --git a/py_src/taskito/result.py b/py_src/taskito/result.py
index 46847b92..70a8c9c0 100644
--- a/py_src/taskito/result.py
+++ b/py_src/taskito/result.py
@@ -22,6 +22,27 @@
 
 log = logging.getLogger("taskito.result")
 
+# Cap on the error summary surfaced in the job list/detail API. The full
+# traceback (which can echo argument values and local context) stays available
+# only via the per-job ``/api/jobs/{id}/errors`` endpoint.
+_ERROR_SUMMARY_MAX = 500
+
+
+def _summarize_error(error: str | None) -> str | None:
+    """Reduce a stored traceback to its final line (exception type + message).
+
+    A Python traceback ends with the ``ExceptionType: message`` line; the
+    intervening frames carry file paths and source snippets we don't want in
+    the broadly-readable job list. Return that last non-empty line, capped.
+    """
+    if not error:
+        return error
+    last_line = next((ln for ln in reversed(error.splitlines()) if ln.strip()), error)
+    last_line = last_line.strip()
+    if len(last_line) > _ERROR_SUMMARY_MAX:
+        last_line = last_line[:_ERROR_SUMMARY_MAX] + "…"
+    return last_line
+
 
 class JobResult(AsyncJobResultMixin):
     """
@@ -260,7 +281,7 @@ def to_dict(self) -> dict[str, Any]:
             "scheduled_at": self._py_job.scheduled_at,
             "started_at": self._py_job.started_at,
             "completed_at": self._py_job.completed_at,
-            "error": self._py_job.error,
+            "error": _summarize_error(self._py_job.error),
             "timeout_ms": self._py_job.timeout_ms,
             "unique_key": self._py_job.unique_key,
             "metadata": self._py_job.metadata,
diff --git a/tests/core/test_result.py b/tests/core/test_result.py
new file mode 100644
index 00000000..ced233d2
--- /dev/null
+++ b/tests/core/test_result.py
@@ -0,0 +1,34 @@
+"""Tests for JobResult helpers."""
+
+from __future__ import annotations
+
+from taskito.result import _ERROR_SUMMARY_MAX, _summarize_error
+
+
+class TestErrorSummary:
+    """``to_dict`` exposes only the final traceback line so the job list API
+    doesn't broadcast frame source/locals to every viewer (M8)."""
+
+    def test_returns_last_line(self) -> None:
+        tb = (
+            "Traceback (most recent call last):\n"
+            '  File "x.py", line 3, in f\n'
+            '    do(secret="sk-abc123")\n'
+            "ValueError: bad value 42"
+        )
+        assert _summarize_error(tb) == "ValueError: bad value 42"
+
+    def test_none_passes_through(self) -> None:
+        assert _summarize_error(None) is None
+
+    def test_empty_passes_through(self) -> None:
+        assert _summarize_error("") == ""
+
+    def test_long_line_capped(self) -> None:
+        out = _summarize_error("E" * 5000)
+        assert out is not None
+        assert len(out) == _ERROR_SUMMARY_MAX + 1  # +1 for the ellipsis
+        assert out.endswith("…")
+
+    def test_single_line_unchanged(self) -> None:
+        assert _summarize_error("RuntimeError: boom") == "RuntimeError: boom"

From 4531706254dadddce849559065074280c1c2e239 Mon Sep 17 00:00:00 2001
From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com>
Date: Sat, 30 May 2026 22:35:19 +0530
Subject: [PATCH 18/30] fix(scheduler): skip dispatch when claim errors

---
 crates/taskito-core/src/scheduler/poller.rs | 18 ++++++++++--------
 1 file changed, 10 insertions(+), 8 deletions(-)

diff --git a/crates/taskito-core/src/scheduler/poller.rs b/crates/taskito-core/src/scheduler/poller.rs
index 46568d0d..b9fe101f 100644
--- a/crates/taskito-core/src/scheduler/poller.rs
+++ b/crates/taskito-core/src/scheduler/poller.rs
@@ -218,19 +218,21 @@ impl Scheduler {
         Ok(true)
     }
 
-    /// Try to claim exactly-once execution. Returns `Ok(true)` if the claim
-    /// was taken (or recoverably failed and the caller should still attempt
-    /// dispatch), `Ok(false)` if the job was already claimed by another
-    /// scheduler.
+    /// Try to claim exactly-once execution. Returns `Ok(true)` only if the
+    /// claim was actually taken; `Ok(false)` if it was already claimed by
+    /// another scheduler **or** the claim attempt errored.
     fn claim_for_dispatch(&self, job: &Job) -> Result<bool> {
         match self.storage.claim_execution(&job.id, SCHEDULER_CLAIM_OWNER) {
             Ok(true) => Ok(true),
             Ok(false) => Ok(false),
             Err(e) => {
-                // Don't drop the job on a transient claim error — proceed and
-                // let the worker handle the duplicate execution defensively.
-                warn!("claim_execution error for job {}: {e}", job.id);
-                Ok(true)
+                // Treat a claim error as "not claimed" and skip this tick. The
+                // job stays Running and the stale-reaper will requeue it. The
+                // previous behaviour (dispatch anyway) caused duplicate
+                // execution when several schedulers hit a transient storage
+                // error at once, since no claim row actually guarded the job.
+                warn!("claim_execution error for job {}; skipping: {e}", job.id);
+                Ok(false)
             }
         }
     }

From 798bef2e237fd2f2d663ec50f10cc9bad76e0bdb Mon Sep 17 00:00:00 2001
From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com>
Date: Sat, 30 May 2026 22:41:02 +0530
Subject: [PATCH 19/30] fix(prefork): recover from poisoned slot mutex

---
 crates/taskito-python/src/prefork/slot.rs | 23 ++++++++++++++++++-----
 1 file changed, 18 insertions(+), 5 deletions(-)

diff --git a/crates/taskito-python/src/prefork/slot.rs b/crates/taskito-python/src/prefork/slot.rs
index 3d93be63..27d2defe 100644
--- a/crates/taskito-python/src/prefork/slot.rs
+++ b/crates/taskito-python/src/prefork/slot.rs
@@ -8,9 +8,19 @@
 //! reader (child finished normally) and the watchdog (child exceeded its
 //! deadline).
 
-use std::sync::{Arc, Mutex};
+use std::sync::{Arc, Mutex, PoisonError};
 use std::time::Instant;
 
+/// Recover a slot guard from a poisoned mutex instead of panicking.
+///
+/// A panic in one reader/watchdog thread must not cascade into every later slot
+/// operation (which would crash the whole prefork pool and abandon in-flight
+/// jobs). The slot holds a plain `Option<ActiveJob>`, so reading through a
+/// poisoned lock is safe. Mirrors the recovery used in the scheduler poller.
+fn recover_poison<T>(poisoned: PoisonError<T>) -> T {
+    poisoned.into_inner()
+}
+
 /// Metadata about a job currently being executed by a child process.
 #[derive(Clone)]
 pub struct ActiveJob {
@@ -35,12 +45,15 @@ pub fn new_slots(n: usize) -> SlotState {
 /// Atomically install `job` in slot `idx`, returning any previous occupant
 /// (which would only happen on a programming error — children are sequential).
 pub fn set(slots: &SlotState, idx: usize, job: ActiveJob) -> Option<ActiveJob> {
-    slots[idx].lock().expect("slot mutex poisoned").replace(job)
+    slots[idx]
+        .lock()
+        .unwrap_or_else(recover_poison)
+        .replace(job)
 }
 
 /// Atomically take whatever is in slot `idx`, leaving it empty.
 pub fn take(slots: &SlotState, idx: usize) -> Option<ActiveJob> {
-    slots[idx].lock().expect("slot mutex poisoned").take()
+    slots[idx].lock().unwrap_or_else(recover_poison).take()
 }
 
 /// Atomically take the slot only if its deadline has passed at `now`.
@@ -49,7 +62,7 @@ pub fn take(slots: &SlotState, idx: usize) -> Option<ActiveJob> {
 /// race-free: if a result arrived between the watchdog's scan and its
 /// take, the reader will have cleared the slot first and we return `None`.
 pub fn take_if_expired(slots: &SlotState, idx: usize, now: Instant) -> Option<ActiveJob> {
-    let mut guard = slots[idx].lock().expect("slot mutex poisoned");
+    let mut guard = slots[idx].lock().unwrap_or_else(recover_poison);
     let expired = guard
         .as_ref()
         .and_then(|j| j.deadline)
@@ -68,7 +81,7 @@ pub fn take_if_expired(slots: &SlotState, idx: usize, now: Instant) -> Option<Ac
 /// the scan never serialises with dispatch or completion.
 pub fn find_by_job_id(slots: &SlotState, job_id: &str) -> Option<usize> {
     for (idx, slot) in slots.iter().enumerate() {
-        let guard = slot.lock().expect("slot mutex poisoned");
+        let guard = slot.lock().unwrap_or_else(recover_poison);
         if guard.as_ref().is_some_and(|j| j.job_id == job_id) {
             return Some(idx);
         }

From fc9c71eff68dd9957fa182fdeb50ac210c249874 Mon Sep 17 00:00:00 2001
From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com>
Date: Sat, 30 May 2026 22:44:13 +0530
Subject: [PATCH 20/30] fix(workflows): cap fan-out child count

---
 .../src/py_queue/workflow_ops/fan_out.rs              | 11 +++++++++++
 1 file changed, 11 insertions(+)

diff --git a/crates/taskito-python/src/py_queue/workflow_ops/fan_out.rs b/crates/taskito-python/src/py_queue/workflow_ops/fan_out.rs
index ea7dc85f..1ce46a09 100644
--- a/crates/taskito-python/src/py_queue/workflow_ops/fan_out.rs
+++ b/crates/taskito-python/src/py_queue/workflow_ops/fan_out.rs
@@ -11,6 +11,11 @@ use taskito_workflows::{WorkflowNode, WorkflowNodeStatus, WorkflowStorage};
 use crate::py_queue::workflow_ops::{build_metadata_json, workflow_storage};
 use crate::py_queue::PyQueue;
 
+/// Maximum number of children a single fan-out may expand into. Guards against
+/// a task returning an enormous list and flooding storage + memory in one
+/// transaction.
+const MAX_FAN_OUT: usize = 10_000;
+
 #[pymethods]
 impl PyQueue {
     /// Expand a fan-out node into N child nodes + jobs.
@@ -42,6 +47,12 @@ impl PyQueue {
                 "child_names and child_payloads must have the same length",
             ));
         }
+        if child_names.len() > MAX_FAN_OUT {
+            return Err(PyValueError::new_err(format!(
+                "fan-out of {} children exceeds the limit of {MAX_FAN_OUT}",
+                child_names.len()
+            )));
+        }
 
         let wf_storage = workflow_storage(self)?;
         let run_id_owned = run_id.to_string();

From c4ea93e39a29517ec04e490a83bfb24a8c817f8f Mon Sep 17 00:00:00 2001
From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com>
Date: Sat, 30 May 2026 22:44:20 +0530
Subject: [PATCH 21/30] fix(postgres): quote schema identifier in DDL

---
 .../taskito-core/src/storage/postgres/mod.rs  | 24 +++++++++++++++----
 1 file changed, 19 insertions(+), 5 deletions(-)

diff --git a/crates/taskito-core/src/storage/postgres/mod.rs b/crates/taskito-core/src/storage/postgres/mod.rs
index 7c819197..4e64e839 100644
--- a/crates/taskito-core/src/storage/postgres/mod.rs
+++ b/crates/taskito-core/src/storage/postgres/mod.rs
@@ -50,6 +50,14 @@ fn validate_schema_name(schema: &str) -> Result<()> {
     Ok(())
 }
 
+/// Quote a SQL identifier for safe interpolation. Postgres can't bind
+/// identifiers as parameters, and while `validate_schema_name` already
+/// restricts the schema to `[A-Za-z0-9_]`, quoting here makes the structural
+/// safety explicit rather than relying solely on the validator.
+fn pg_quote_ident(name: &str) -> String {
+    format!("\"{}\"", name.replace('"', "\"\""))
+}
+
 /// PostgreSQL-backed storage for the task queue, using Diesel ORM.
 #[derive(Clone)]
 pub struct PostgresStorage {
@@ -117,9 +125,12 @@ impl PostgresStorage {
 
     pub fn conn(&self) -> Result<diesel::r2d2::PooledConnection<ConnectionManager<PgConnection>>> {
         let mut conn = self.pool.get()?;
-        diesel::sql_query(format!("SET search_path TO {}", self.schema))
-            .execute(&mut conn)
-            .map_err(crate::error::QueueError::Storage)?;
+        diesel::sql_query(format!(
+            "SET search_path TO {}",
+            pg_quote_ident(&self.schema)
+        ))
+        .execute(&mut conn)
+        .map_err(crate::error::QueueError::Storage)?;
         Ok(conn)
     }
 
@@ -127,8 +138,11 @@ impl PostgresStorage {
         let mut conn = self.conn()?;
 
         // Postgres-only: ensure the target schema exists before any DDL runs.
-        diesel::sql_query(format!("CREATE SCHEMA IF NOT EXISTS {}", self.schema))
-            .execute(&mut conn)?;
+        diesel::sql_query(format!(
+            "CREATE SCHEMA IF NOT EXISTS {}",
+            pg_quote_ident(&self.schema)
+        ))
+        .execute(&mut conn)?;
 
         for sql in common_migrations::create_tables(&common_migrations::POSTGRES) {
             diesel::sql_query(&sql).execute(&mut conn)?;

From 8bb4ee9169df7f8da1d78fa0061d9db3c0334f59 Mon Sep 17 00:00:00 2001
From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com>
Date: Sat, 30 May 2026 22:44:25 +0530
Subject: [PATCH 22/30] fix(redis): set native lock TTL and atomic reap

---
 .../src/storage/redis_backend/locks.rs        | 25 +++++++++++++------
 1 file changed, 18 insertions(+), 7 deletions(-)

diff --git a/crates/taskito-core/src/storage/redis_backend/locks.rs b/crates/taskito-core/src/storage/redis_backend/locks.rs
index 57a19137..346e47d0 100644
--- a/crates/taskito-core/src/storage/redis_backend/locks.rs
+++ b/crates/taskito-core/src/storage/redis_backend/locks.rs
@@ -25,6 +25,7 @@ const EXTEND_LOCK_SCRIPT: &str = r#"
     local current = redis.call('HGET', key, 'owner_id')
     if current == owner then
         redis.call('HSET', key, 'expires_at', new_expires)
+        redis.call('PEXPIREAT', key, tonumber(new_expires))
         return 1
     end
     return 0
@@ -43,9 +44,23 @@ const ACQUIRE_LOCK_SCRIPT: &str = r#"
     end
     redis.call('HSET', key, 'lock_name', KEYS[2], 'owner_id', owner,
                'acquired_at', acquired_at, 'expires_at', expires_at)
+    redis.call('PEXPIREAT', key, tonumber(expires_at))
     return 1
 "#;
 
+/// Lua script: delete a lock only if it is still expired at delete time.
+/// Re-checking inside the script closes the TOCTOU window where the SCAN-driven
+/// reaper would HGET an expired lock, another client re-acquires it, and the
+/// reaper then DELs the now-valid lock.
+const REAP_LOCK_SCRIPT: &str = r#"
+    local exp = redis.call('HGET', KEYS[1], 'expires_at')
+    if exp and tonumber(exp) <= tonumber(ARGV[1]) then
+        redis.call('DEL', KEYS[1])
+        return 1
+    end
+    return 0
+"#;
+
 impl RedisStorage {
     pub fn acquire_lock(&self, lock_name: &str, owner_id: &str, ttl_ms: i64) -> Result<bool> {
         let mut conn = self.conn()?;
@@ -139,14 +154,10 @@ impl RedisStorage {
                 .query(&mut conn)
                 .map_err(map_err)?;
 
+            let reap = redis::Script::new(REAP_LOCK_SCRIPT);
             for key in keys {
-                let expires_at: Option<i64> = conn.hget(&key, "expires_at").map_err(map_err)?;
-                if let Some(exp) = expires_at {
-                    if exp <= now {
-                        conn.del::<_, ()>(&key).map_err(map_err)?;
-                        count += 1;
-                    }
-                }
+                let deleted: i64 = reap.key(&key).arg(now).invoke(&mut conn).map_err(map_err)?;
+                count += deleted as u64;
             }
 
             cursor = next_cursor;

From 0bf3f63914b0bf65c9a8f482431db578b733c97f Mon Sep 17 00:00:00 2001
From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com>
Date: Sat, 30 May 2026 23:06:45 +0530
Subject: [PATCH 23/30] docs(security): add security guide and SignedSerializer

---
 docs/content/docs/guides/operations/meta.json |   1 +
 .../docs/guides/operations/security.mdx       | 146 ++++++++++++++++++
 2 files changed, 147 insertions(+)
 create mode 100644 docs/content/docs/guides/operations/security.mdx

diff --git a/docs/content/docs/guides/operations/meta.json b/docs/content/docs/guides/operations/meta.json
index 8ad0b92d..565e81a6 100644
--- a/docs/content/docs/guides/operations/meta.json
+++ b/docs/content/docs/guides/operations/meta.json
@@ -5,6 +5,7 @@
     "testing",
     "job-management",
     "troubleshooting",
+    "security",
     "deployment",
     "autoscaler",
     "keda",
diff --git a/docs/content/docs/guides/operations/security.mdx b/docs/content/docs/guides/operations/security.mdx
new file mode 100644
index 00000000..6a5f88f1
--- /dev/null
+++ b/docs/content/docs/guides/operations/security.mdx
@@ -0,0 +1,146 @@
+---
+title: Security
+description: "Trust model, payload authentication, dashboard auth, webhook SSRF, resource limits, and the production hardening checklist."
+---
+
+import { Callout } from "fumadocs-ui/components/callout";
+
+taskito is a library you embed in your own services, so most of its security
+posture is a function of how you deploy it. This page describes the trust
+model and the controls available to harden a deployment.
+
+## Trust model
+
+The single most important assumption to understand:
+
+<Callout type="warn" title="Storage write access ≈ code execution">
+  taskito trusts its backing store. The default serializer can execute code
+  on load (cloudpickle handles lambdas and arbitrary objects), so anyone who
+  can write to the SQLite file, the Postgres tables, or the Redis keyspace
+  can craft a job payload that runs arbitrary code on every worker that
+  dequeues it. Treat database/Redis write access as equivalent to shell
+  access on your workers.
+</Callout>
+
+Mitigations, in order of strength:
+
+1. **Lock down the store.** Network-isolate Postgres/Redis, use least-privilege
+   credentials, and never expose the SQLite file to untrusted writers.
+2. **Authenticate payloads** with `SignedSerializer` (below) so a forged
+   payload is rejected before it is ever deserialized.
+3. **Avoid cloudpickle** entirely by using `JsonSerializer` or
+   `MsgPackSerializer` when your task arguments are plain data.
+
+## Payload authentication
+
+`SignedSerializer` prepends an HMAC-SHA256 tag to every payload and verifies
+it (constant-time) before deserializing. A worker refuses any bytes not
+produced with the shared key.
+
+```python
+import os
+from taskito import Queue
+from taskito.serializers import SignedSerializer, SmartSerializer
+
+key = os.urandom(32)  # 32+ bytes, identical on producers and workers
+queue = Queue(serializer=SignedSerializer(SmartSerializer(), key))
+```
+
+For confidentiality as well, wrap `EncryptedSerializer`. See
+[Pluggable Serializers](/docs/guides/extensibility/serializers) for the full
+matrix.
+
+### Payloads at rest
+
+By default, payloads are stored **unencrypted** (msgpack/cloudpickle bytes).
+Anyone with read access to the database can recover task arguments. Use
+`EncryptedSerializer` when tasks carry secrets or PII.
+
+### Payload size limit
+
+Every serialized payload is capped at `max_payload_bytes` (default **1 MiB**)
+and rejected at enqueue time, so a single producer can't exhaust storage or
+spike worker memory. Adjust or disable it per queue:
+
+```python
+queue.max_payload_bytes = 4 * 1024 * 1024  # 4 MiB
+queue.max_payload_bytes = 0                 # disable the cap
+```
+
+## Dashboard
+
+The dashboard enforces authentication once at least one user exists; until
+then every API route returns `503 setup_required`. Bootstrap the first admin
+with `TASKITO_DASHBOARD_ADMIN_USER` / `TASKITO_DASHBOARD_ADMIN_PASSWORD`
+(the password is removed from the environment after first use).
+
+- **Roles.** Mutating routes (cancel/replay jobs, purge dead-letters, pause
+  queues, manage webhooks/settings/overrides) require the `admin` role.
+  `viewer` sessions keep read access and their own logout/change-password.
+- **Cookies.** Session and CSRF cookies carry `HttpOnly` (session),
+  `SameSite=Strict`, and `Secure`. Behind a TLS-terminating proxy that speaks
+  plain HTTP to the backend you may need `serve_dashboard(secure_cookies=False)`
+  or `taskito dashboard --insecure-cookies` — only on a trusted network.
+- **Metrics.** `/metrics` and `/readiness` are public by default for probes.
+  Set `TASKITO_DASHBOARD_METRICS_TOKEN` to require an
+  `Authorization: Bearer <token>` header on them; `/health` stays open.
+- **Settings.** The settings API never exposes or accepts keys under the
+  internal `auth:` namespace (password hashes, sessions, CSRF secret).
+
+<Callout type="warn" title="Contrib routers are not auth-gated by default">
+  The FastAPI/Flask routers in `taskito.contrib` expose job-control endpoints
+  with **no authentication** unless you attach your own. Always mount them
+  behind your application's auth (e.g. FastAPI `dependencies=[Depends(...)]`).
+</Callout>
+
+## Webhooks
+
+Outbound webhook URLs are validated against private/loopback/link-local and
+cloud-metadata addresses **at delivery time** (not just registration), which
+closes the DNS-rebinding SSRF window, and redirects are not followed. For
+local development against `http://localhost`, set
+`TASKITO_WEBHOOKS_ALLOW_PRIVATE=1`.
+
+Webhook payloads are signed with HMAC-SHA256 in the `X-Taskito-Signature`
+header (`sha256=<hex>`). Receivers must verify it with a constant-time
+comparison (`hmac.compare_digest`), never `==`.
+
+## Proxies and interception
+
+These features reconstruct objects from the (untrusted) payload, so they have
+their own guards:
+
+- **Recipe signing.** Set `recipe_signing_key=` (or `TASKITO_RECIPE_SECRET`)
+  whenever proxy handlers are registered. Without a key, recipes are
+  reconstructed unverified and a one-time warning is logged.
+- **File proxy allowlist.** `file_path_allowlist` is enforced by path-component
+  containment (no `/var/data` → `/var/data_exfil` bypass) against the resolved
+  path.
+- **Type reconstruction.** Interception only reconstructs known-safe kinds
+  (enums, dataclasses, pydantic models, named tuples); arbitrary
+  import-and-call targets like `os.system` are rejected.
+
+## Distributed locks
+
+`DistributedLock.info()` masks the `owner_id` of locks held by other owners,
+so a peer can't read another holder's token and forge a release. Redis locks
+set a native key TTL, so a crashed holder's lock is reclaimed even if the
+maintenance reaper is down.
+
+## Scaler bind address
+
+`taskito scaler` and `serve_scaler()` bind to `127.0.0.1` by default. The
+scaler endpoints are unauthenticated — only bind to `0.0.0.0` behind a trusted
+network boundary (e.g. a KEDA sidecar on a private network).
+
+## Production hardening checklist
+
+- [ ] Network-isolate Postgres/Redis; least-privilege credentials.
+- [ ] Use `SignedSerializer` (and/or `EncryptedSerializer`) for the queue.
+- [ ] Set `recipe_signing_key` if any proxy handler is registered.
+- [ ] Create a dashboard admin; serve the dashboard over TLS.
+- [ ] Set `TASKITO_DASHBOARD_METRICS_TOKEN` if metrics are network-reachable.
+- [ ] Auth-gate any mounted FastAPI/Flask contrib routers.
+- [ ] Leave `max_payload_bytes` enabled (default 1 MiB) unless you need larger.
+- [ ] Keep the scaler bound to loopback (or behind a network policy).
+- [ ] Do **not** set `TASKITO_WEBHOOKS_ALLOW_PRIVATE` in production.

From c350c0f475a40210a8724ae4ba474fc22d97fb5a Mon Sep 17 00:00:00 2001
From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com>
Date: Sat, 30 May 2026 23:10:33 +0530
Subject: [PATCH 24/30] test(dashboard): assert settings API hides auth keys

---
 tests/dashboard/test_dashboard_settings.py | 10 ++++++----
 1 file changed, 6 insertions(+), 4 deletions(-)

diff --git a/tests/dashboard/test_dashboard_settings.py b/tests/dashboard/test_dashboard_settings.py
index 9ba4ba24..917e8c7e 100644
--- a/tests/dashboard/test_dashboard_settings.py
+++ b/tests/dashboard/test_dashboard_settings.py
@@ -94,12 +94,14 @@ def dashboard_server(queue: Queue) -> Generator[tuple[AuthedClient, Queue]]:
         server.shutdown()
 
 
-def test_get_settings_returns_empty_dict(dashboard_server: tuple[AuthedClient, Queue]) -> None:
+def test_get_settings_hides_auth_keys(dashboard_server: tuple[AuthedClient, Queue]) -> None:
     client, _ = dashboard_server
-    # The admin user setting is the only one populated by the seed helper.
     snapshot = client.get("/api/settings")
-    assert "auth:users" in snapshot
-    # No dashboard.* keys yet.
+    # The settings API must never expose the internal ``auth:`` namespace
+    # (password hashes, sessions, CSRF secret), even though the seed helper
+    # populated ``auth:users``.
+    assert not any(k.startswith("auth:") for k in snapshot)
+    # No dashboard.* keys yet either.
     assert not any(k.startswith("dashboard.") for k in snapshot)
 
 

From 890b7cc5434691bf714f6d7de6c605dfe4777afc Mon Sep 17 00:00:00 2001
From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com>
Date: Sat, 30 May 2026 23:13:15 +0530
Subject: [PATCH 25/30] docs(serializers): document SignedSerializer

---
 .../docs/guides/extensibility/serializers.mdx | 34 +++++++++++++++++++
 1 file changed, 34 insertions(+)

diff --git a/docs/content/docs/guides/extensibility/serializers.mdx b/docs/content/docs/guides/extensibility/serializers.mdx
index 1cdfd8c4..3e8fef7d 100644
--- a/docs/content/docs/guides/extensibility/serializers.mdx
+++ b/docs/content/docs/guides/extensibility/serializers.mdx
@@ -87,6 +87,40 @@ from taskito.serializers import EncryptedSerializer, MsgPackSerializer
 queue = Queue(serializer=EncryptedSerializer(key=key, inner=MsgPackSerializer()))
 ```
 
+### SignedSerializer
+
+HMAC-SHA256 integrity tag for task arguments and results. Unlike
+`EncryptedSerializer`, it does **not** hide the payload — it authenticates it.
+A worker refuses to deserialize any bytes that were not produced with the
+shared key, so an attacker who can write to the queue's storage (the SQLite
+file, a Postgres table, or Redis) cannot smuggle in a forged payload.
+
+```python
+import os
+from taskito.serializers import SignedSerializer, SmartSerializer
+
+key = os.urandom(32)  # share this across producers and workers
+queue = Queue(serializer=SignedSerializer(SmartSerializer(), key))
+```
+
+<Callout type="warn" title="Why this matters">
+  The default serializer can execute code on load (cloudpickle handles
+  lambdas and arbitrary objects). Without signing, anyone able to write to
+  the backing store can achieve remote code execution on every worker that
+  dequeues a crafted job. `SignedSerializer` closes that path. The key must
+  be at least 32 bytes of CSPRNG output and identical on producers and
+  workers.
+</Callout>
+
+For both confidentiality **and** integrity, wrap one in the other:
+
+```python
+from taskito.serializers import EncryptedSerializer, SignedSerializer, SmartSerializer
+
+inner = EncryptedSerializer(SmartSerializer(), key=enc_key)
+queue = Queue(serializer=SignedSerializer(inner, sign_key))
+```
+
 ## When to use each
 
 |   | CloudpickleSerializer | JsonSerializer | MsgPackSerializer | EncryptedSerializer |

From 68b1435235a4ff6e9f9a505f46a62f19956e40ff Mon Sep 17 00:00:00 2001
From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com>
Date: Sat, 30 May 2026 23:34:29 +0530
Subject: [PATCH 26/30] Revert "fix(worker): drop exception repr from failure
 log"

This reverts commit c0203cb188a8343a1ef2d092fa108860fcd10963.
---
 py_src/taskito/mixins/decorators.py | 7 ++-----
 1 file changed, 2 insertions(+), 5 deletions(-)

diff --git a/py_src/taskito/mixins/decorators.py b/py_src/taskito/mixins/decorators.py
index 50c0832d..11f88311 100644
--- a/py_src/taskito/mixins/decorators.py
+++ b/py_src/taskito/mixins/decorators.py
@@ -294,16 +294,13 @@ def wrapper(*args: Any, **kwargs: Any) -> Any:
                         exc,
                     )
                 else:
-                    # Log the exception type only; ``exc_info`` carries the full
-                    # detail. ``%r`` of the exception is avoided because its
-                    # message can echo task argument data into log sinks.
                     logger.error(
-                        "Task %s[%s] raised %s in %.3fs",
+                        "Task %s[%s] raised %s in %.3fs: %r",
                         task_name,
                         job_id,
                         type(exc).__name__,
                         elapsed,
-                        exc_info=True,
+                        exc,
                     )
                 for hook in hooks["on_failure"]:
                     hook(task_name, args, kwargs, exc)

From ce0c7869f95f600dcaf2eddc3f96bf85a639b157 Mon Sep 17 00:00:00 2001
From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com>
Date: Sat, 30 May 2026 23:50:26 +0530
Subject: [PATCH 27/30] docs(dashboard): document role enforcement, Secure
 cookies, metrics token

---
 .../docs/guides/dashboard/authentication.mdx  | 25 +++++++++++++++----
 1 file changed, 20 insertions(+), 5 deletions(-)

diff --git a/docs/content/docs/guides/dashboard/authentication.mdx b/docs/content/docs/guides/dashboard/authentication.mdx
index 5fb972cc..6aa16fd7 100644
--- a/docs/content/docs/guides/dashboard/authentication.mdx
+++ b/docs/content/docs/guides/dashboard/authentication.mdx
@@ -22,7 +22,10 @@ token on every state-changing request.
   baseline). No third-party crypto dependency.
 - **Sessions** are stored server-side under
   `auth:session:<random_token>` with a 24-hour TTL. The token rides in
-  an `HttpOnly` + `SameSite=Strict` cookie named `taskito_session`.
+  an `HttpOnly` + `SameSite=Strict` + `Secure` cookie named
+  `taskito_session` (the `Secure` flag means HTTPS-only; for local HTTP
+  dev use `serve_dashboard(secure_cookies=False)` or
+  `taskito dashboard --insecure-cookies`).
 - **CSRF** uses the double-submit pattern: a non-HttpOnly cookie named
   `taskito_csrf` carries a per-session token that the SPA reads and
   echoes back via the `X-CSRF-Token` header on POST/PUT/DELETE. The
@@ -100,7 +103,9 @@ All routes live under `/api/auth/`:
 
 Every other route under `/api/` is auth-gated. Public exceptions:
 `/health`, `/readiness`, `/metrics` (Prometheus), and the static SPA
-assets.
+assets. Set `TASKITO_DASHBOARD_METRICS_TOKEN` to additionally require an
+`Authorization: Bearer <token>` header on `/metrics` and `/readiness`
+(`/health` stays open for liveness probes).
 
 ## Headless requests
 
@@ -154,10 +159,20 @@ password auth — see [SSO (OAuth & OIDC)](/guides/dashboard/sso).
 Operators can mix-and-match providers or run an OAuth-only deployment
 by setting `TASKITO_DASHBOARD_PASSWORD_AUTH_ENABLED=false`.
 
+## Roles
+
+Two roles exist, and the role is **enforced server-side**:
+
+- **admin** — full access. Every state-changing API route (cancel/replay
+  jobs, purge dead-letters, pause/resume queues, manage webhooks, edit
+  settings, manage users, edit task/queue overrides) requires this role.
+- **viewer** — read-only access to dashboards and job data, plus their own
+  logout and change-password. Any mutating route returns `403 forbidden`.
+
+The first user is always an admin; additional users can be created with
+either role.
+
 ## Limitations
 
-- **One role** today (`admin`). Read-only viewers and per-route
-  permissions are planned; the column already exists on the user
-  record.
 - **Password rotation** has an endpoint but no UI yet — invoke
   `POST /api/auth/change-password` directly.

From 8b1fc02247c9f638552e3eadc045f6856b9fc21d Mon Sep 17 00:00:00 2001
From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com>
Date: Sat, 30 May 2026 23:50:49 +0530
Subject: [PATCH 28/30] docs(webhooks): note delivery-time SSRF revalidation
 and no redirects

---
 docs/content/docs/guides/extensibility/events-webhooks.mdx | 7 +++++--
 1 file changed, 5 insertions(+), 2 deletions(-)

diff --git a/docs/content/docs/guides/extensibility/events-webhooks.mdx b/docs/content/docs/guides/extensibility/events-webhooks.mdx
index 83237535..3107f089 100644
--- a/docs/content/docs/guides/extensibility/events-webhooks.mdx
+++ b/docs/content/docs/guides/extensibility/events-webhooks.mdx
@@ -165,8 +165,11 @@ your business logic.
 
 ### SSRF guard
 
-Outbound webhook URLs are validated before the manager queues any
-delivery. The guard rejects:
+Outbound webhook URLs are validated both when a webhook is registered and
+again **at delivery time**, so a hostname that is rebound to an internal
+address after registration (DNS rebinding) is still refused on the actual
+request. Redirects are not followed, so a `30x` response can't bounce
+delivery to an internal host either. The guard rejects:
 
 - Non-`http` / `https` schemes
 - `localhost`, `*.local`, `*.internal`, `*.intranet`, `*.lan`,

From a4730b849734e2856f2ebdb3207ce01ecdda28cd Mon Sep 17 00:00:00 2001
From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com>
Date: Sat, 30 May 2026 23:52:32 +0530
Subject: [PATCH 29/30] docs(locking): note owner_id masking via info()

---
 docs/content/docs/guides/reliability/locking.mdx | 8 ++++++++
 1 file changed, 8 insertions(+)

diff --git a/docs/content/docs/guides/reliability/locking.mdx b/docs/content/docs/guides/reliability/locking.mdx
index 1c4a329e..0fa2afd0 100644
--- a/docs/content/docs/guides/reliability/locking.mdx
+++ b/docs/content/docs/guides/reliability/locking.mdx
@@ -146,3 +146,11 @@ queue._inner.release_lock("my-lock", lock_id)
   The low-level API skips auto-extension and does not release on exception.
   Prefer the context manager (`lock()` / `alock()`) for production code.
 </Callout>
+
+<Callout type="info" title="Owner token is sensitive">
+  `owner_id` is a bearer token — whoever knows it can release or extend the
+  lock. The low-level `get_lock_info` returns it raw; the high-level
+  `queue.lock(name).info()` masks it to `"***"` for any caller that isn't the
+  current holder, so prefer `info()` when surfacing lock status to other
+  parties (e.g. a dashboard or another tenant).
+</Callout>

From 557e5251dab23fb6f0ac7798fe806524d10f5f93 Mon Sep 17 00:00:00 2001
From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com>
Date: Sun, 31 May 2026 00:30:15 +0530
Subject: [PATCH 30/30] fix(proxies): emit unsigned-recipe warning via else
 branch

---
 py_src/taskito/proxies/reconstruct.py | 4 ++++
 1 file changed, 4 insertions(+)

diff --git a/py_src/taskito/proxies/reconstruct.py b/py_src/taskito/proxies/reconstruct.py
index 1cd4c182..4e9ad604 100644
--- a/py_src/taskito/proxies/reconstruct.py
+++ b/py_src/taskito/proxies/reconstruct.py
@@ -218,6 +218,10 @@ def _reconstruct_one(
             if metrics is not None:
                 metrics.record_checksum_failure(handler_name)
             raise
+    else:
+        # No signing key: the recipe's integrity is unverified. Warn once so an
+        # unsigned-reconstruction deployment is a deliberate choice, not silent.
+        _warn_unsigned_once()
 
     # Schema validation
     handler_schema = getattr(handler, "schema", None)