auth: core primitives

Levels, principals, password hashing, cookies and login throttling. Pure
library code -- nothing imports it into a request path yet.

levels.py
  An ordered read < edit < manage scale. Ranks live in a separate mapping
  rather than as comparison operators on the enum: StrEnum inherits str's
  operators, so defining __lt__/__ge__ would shadow string ordering with a
  second alphabetical meaning that happens to agree today and would stop
  agreeing the moment a level is renamed. Callers use covers().

principal.py
  A frozen dataclass describing who is asking and what they may do. The
  absence of a credential is None rather than a Principal, so "nobody" can
  never be mistaken for a permission holder. Carries the tenant_id seam from
  the design doc: always None, present so auth functions already thread a
  tenant and adding tenancy later is a filter inside them rather than a
  change to forty routers. UNRESTRICTED is what gates return when auth is
  off, before any database access.

hashing.py
  scrypt at n=2**14, r=8, p=1 -- argon2-cffi and bcrypt ship compiled wheels
  that are painful on the armv7 image, and scrypt is memory-hard and in the
  standard library. OWASP's headline n=2**17 would be multi-second on the
  ARM boards Spoolman runs on; their memory-constrained variant raises p,
  which costs the same time without the memory hardness.

  Hashes embed their own parameters, so the cost can be raised later without
  a migration: verification re-derives with the stored parameters and
  needs_rehash() reports staleness for opportunistic upgrade at login.

  Deriving blocks for ~50 ms here and several times that on a Pi. Spoolman
  is a single process, so a synchronous call in a handler would stall every
  other request. Measured 4 hashes serially at 171 ms against 63 ms across
  four threads, confirming scrypt releases the GIL and the _async wrappers
  are a real offload. Handlers must use those.

  Session tokens are hashed with plain unsalted SHA-256, deliberately: they
  are 256 bits of cryptographic randomness, so there is no dictionary to
  attack and no work factor worth paying. The digest exists so a database
  leak does not hand over usable cookies.

cookies.py
  Session cookie is HttpOnly, so CSRF is handled by double submit against a
  second script-readable cookie. Secure is only ever concluded from positive
  evidence -- real TLS, or a proxy listed in SPOOLMAN_TRUSTED_PROXIES saying
  so -- because setting it on a plain-HTTP LAN deployment makes the browser
  discard the cookie and presents as a login that succeeds and does nothing.

ratelimit.py
  In-process sliding windows keyed by username and by client address. Sound
  because the deployment contract is one process, the same assumption the
  websocket manager and scheduler already make; counters reset on restart,
  which is why lockout is also persisted on auth_user. X-Forwarded-For and
  X-Forwarded-Proto are honoured only from a configured trusted proxy,
  otherwise any client could spoof them to evade the limiter.

Also adds AuthenticationRequiredError and PermissionDeniedError to
exceptions.py, matching the existing bare-Exception style.

Verified: level ordering and rejection of unknown names; principal frozen
and anonymous capped at read; hash round-trip, distinct salts, corrupt
hashes never validating, needs_rehash on stale parameters, and dummy_verify
costing 41 ms against a real verify's 42 ms so unknown usernames are
timing-indistinguishable; window limit/expiry/reset; untrusted
X-Forwarded-For ignored and untrusted proxies unable to force Secure; and
all nine CSRF cases including a lowercase-method bypass attempt.

spoolman/auth/cookies.py

@@ -0,0 +1,144 @@
+"""Session cookie and CSRF handling.
+
+The session cookie holds an opaque token and is ``HttpOnly``, so script cannot read it.
+Because it is sent automatically on same-origin requests it is an ambient credential and
+needs CSRF protection, which is done by double submit: a second, script-readable cookie
+carries a random value that the client must echo in a request header. An attacker's page
+can cause the browser to send the cookies but cannot read them to construct the header.
+
+API keys, once they arrive in phase 2, are exempt -- they are not ambient, so a
+cross-site page has no way to attach one.
+"""
+
+from datetime import timedelta
+from typing import Final
+
+from starlette.requests import Request
+from starlette.responses import Response
+from starlette.websockets import WebSocket
+
+from spoolman import env
+from spoolman.auth import ratelimit
+from spoolman.auth.hashing import tokens_equal
+
+SESSION_COOKIE: Final = "spoolman_session"
+CSRF_COOKIE: Final = "spoolman_csrf"
+CSRF_HEADER: Final = "X-CSRF-Token"
+
+# Sessions without "remember me" expire this long after they were last used.
+IDLE_LIFETIME: Final = timedelta(hours=12)
+
+# Sessions with "remember me" last this long, sliding forward as they are used.
+REMEMBER_LIFETIME: Final = timedelta(days=90)
+
+# Methods that do not change state, and so do not require a CSRF token.
+SAFE_METHODS: Final = frozenset({"GET", "HEAD", "OPTIONS"})
+
+
+def cookie_path() -> str:
+    """Get the path cookies are scoped to.
+
+    Returns:
+        str: The deployment's base path with a trailing slash.
+
+    """
+    return env.get_base_path() + "/"
+
+
+def cookie_secure(conn: Request | WebSocket) -> bool:
+    """Decide whether cookies should carry the Secure flag.
+
+    Setting Secure on a plain-HTTP deployment makes the browser silently discard the
+    cookie, which presents as a login that appears to succeed and then does nothing. The
+    automatic path therefore only concludes HTTPS from positive evidence: either the
+    request arrived over TLS, or a proxy we have been told to trust says it did.
+
+    Args:
+        conn: The request or websocket.
+
+    Returns:
+        bool: Whether to set the Secure flag.
+
+    """
+    override = env.get_cookie_secure_override()
+    if override is not None:
+        return override
+    if conn.url.scheme in {"https", "wss"}:
+        return True
+    return ratelimit.forwarded_proto(conn) == "https"
+
+
+def set_session_cookies(
+    response: Response,
+    conn: Request,
+    *,
+    token: str,
+    csrf: str,
+    remember: bool,
+) -> None:
+    """Attach the session and CSRF cookies to a response.
+
+    Args:
+        response: The response to modify.
+        conn: The request being answered, used to derive the Secure flag.
+        token: The plaintext session token.
+        csrf: The plaintext CSRF token.
+        remember: Whether the session should outlive the browser session.
+
+    """
+    secure = cookie_secure(conn)
+    path = cookie_path()
+    max_age = int(REMEMBER_LIFETIME.total_seconds()) if remember else None
+
+    response.set_cookie(
+        SESSION_COOKIE,
+        token,
+        max_age=max_age,
+        path=path,
+        secure=secure,
+        httponly=True,
+        samesite="lax",
+    )
+    # Readable by script on purpose: the client has to copy it into a request header.
+    response.set_cookie(
+        CSRF_COOKIE,
+        csrf,
+        max_age=max_age,
+        path=path,
+        secure=secure,
+        httponly=False,
+        samesite="lax",
+    )
+
+
+def clear_session_cookies(response: Response, conn: Request) -> None:
+    """Remove the session and CSRF cookies.
+
+    Args:
+        response: The response to modify.
+        conn: The request being answered, used to derive the Secure flag.
+
+    """
+    secure = cookie_secure(conn)
+    path = cookie_path()
+    for name in (SESSION_COOKIE, CSRF_COOKIE):
+        response.delete_cookie(name, path=path, secure=secure, httponly=name == SESSION_COOKIE, samesite="lax")
+
+
+def verify_csrf(request: Request) -> bool:
+    """Check the CSRF double-submit for a state-changing request.
+
+    Args:
+        request: The request to check.
+
+    Returns:
+        bool: True if the request may proceed.
+
+    """
+    if request.method.upper() in SAFE_METHODS:
+        return True
+    cookie = request.cookies.get(CSRF_COOKIE, "")
+    header = request.headers.get(CSRF_HEADER, "")
+    if not cookie or not header:
+        return False
+    return tokens_equal(cookie, header)

spoolman/auth/hashing.py

@@ -0,0 +1,243 @@
+"""Password and token hashing.
+
+Passwords use :func:`hashlib.scrypt`. The alternatives normally reached for --
+``argon2-cffi`` and ``bcrypt`` -- ship compiled wheels that are unavailable or painful
+on the armv7 image Spoolman still builds, whereas scrypt is in the standard library and
+is a memory-hard function in its own right.
+
+Every hash records the parameters it was produced with, so the cost can be raised later
+without a migration: :func:`verify_password` re-derives using the parameters embedded in
+the stored string, and :func:`needs_rehash` reports when a stored hash predates a change
+so the login path can opportunistically upgrade it.
+
+Deriving a hash blocks for tens to hundreds of milliseconds depending on hardware
+(measured at 49 ms on a desktop, several times that on a Raspberry Pi). Spoolman runs as
+a single process, so a synchronous call in a request handler would stall every other
+request and every websocket ping for that whole time. scrypt releases the GIL, so the
+``_async`` wrappers below genuinely offload to a worker thread; request handlers must
+use those and never the synchronous forms.
+"""
+
+import base64
+import hashlib
+import hmac
+import secrets
+from typing import Final
+
+import anyio.to_thread
+
+# Cost parameters. n is the CPU/memory cost, r the block size, p the parallelisation.
+# Memory use is roughly 128 * r * n bytes, so this costs ~16 MiB and ~50 ms per hash on
+# a desktop. OWASP's headline scrypt recommendation is n=2**17, which is eight times
+# this and would be multi-second on the low-power ARM boards that Spoolman commonly runs
+# on -- unacceptable for an interactive login. Their memory-constrained alternative
+# raises p instead, which costs the same wall-clock time without the memory hardness.
+SCRYPT_N: Final = 2**14
+SCRYPT_R: Final = 8
+SCRYPT_P: Final = 1
+SCRYPT_DKLEN: Final = 32
+SALT_BYTES: Final = 16
+
+# OpenSSL caps scrypt's allocation at 32 MiB when maxmem is left at 0, which n=2**14
+# fits under. It is passed explicitly so that raising SCRYPT_N later fails loudly at the
+# parameter change rather than obscurely inside OpenSSL.
+SCRYPT_MAXMEM: Final = 64 * 1024 * 1024
+
+ALGORITHM: Final = "scrypt"
+TOKEN_BYTES: Final = 32
+
+
+def _b64encode(raw: bytes) -> str:
+    """Encode bytes as unpadded url-safe base64."""
+    return base64.urlsafe_b64encode(raw).decode().rstrip("=")
+
+
+def _b64decode(value: str) -> bytes:
+    """Decode unpadded url-safe base64."""
+    padding = "=" * (-len(value) % 4)
+    return base64.urlsafe_b64decode(value + padding)
+
+
+def _derive(password: str, salt: bytes, n: int, r: int, p: int, dklen: int) -> bytes:
+    """Run scrypt with explicit parameters."""
+    return hashlib.scrypt(
+        password.encode(),
+        salt=salt,
+        n=n,
+        r=r,
+        p=p,
+        dklen=dklen,
+        maxmem=SCRYPT_MAXMEM,
+    )
+
+
+def _format(salt: bytes, derived: bytes, n: int, r: int, p: int) -> str:
+    """Build the stored representation of a hash."""
+    return f"{ALGORITHM}${n}${r}${p}${_b64encode(salt)}${_b64encode(derived)}"
+
+
+def _parse(encoded: str) -> tuple[int, int, int, bytes, bytes]:
+    """Split a stored hash into its parameters, salt and digest.
+
+    Args:
+        encoded: The stored hash string.
+
+    Raises:
+        ValueError: If the string is not a well-formed scrypt hash.
+
+    Returns:
+        tuple: n, r, p, salt and the expected digest.
+
+    """
+    parts = encoded.split("$")
+    expected_parts = 6
+    if len(parts) != expected_parts or parts[0] != ALGORITHM:
+        raise ValueError("Malformed password hash.")
+    try:
+        n, r, p = int(parts[1]), int(parts[2]), int(parts[3])
+        return n, r, p, _b64decode(parts[4]), _b64decode(parts[5])
+    except (ValueError, TypeError) as exc:
+        raise ValueError("Malformed password hash.") from exc
+
+
+def hash_password(password: str) -> str:
+    """Hash a password for storage.
+
+    Args:
+        password: The plaintext password.
+
+    Returns:
+        str: The encoded hash, including the parameters used.
+
+    """
+    salt = secrets.token_bytes(SALT_BYTES)
+    derived = _derive(password, salt, SCRYPT_N, SCRYPT_R, SCRYPT_P, SCRYPT_DKLEN)
+    return _format(salt, derived, SCRYPT_N, SCRYPT_R, SCRYPT_P)
+
+
+def verify_password(password: str, encoded: str) -> bool:
+    """Check a password against a stored hash.
+
+    Args:
+        password: The plaintext password to check.
+        encoded: The stored hash.
+
+    Returns:
+        bool: True if the password matches. False if it does not, and also if the stored
+        hash is unreadable -- a corrupt hash must never authenticate anyone.
+
+    """
+    try:
+        n, r, p, salt, expected = _parse(encoded)
+    except ValueError:
+        return False
+    derived = _derive(password, salt, n, r, p, len(expected))
+    return hmac.compare_digest(derived, expected)
+
+
+def needs_rehash(encoded: str) -> bool:
+    """Check whether a stored hash was produced with outdated parameters.
+
+    Args:
+        encoded: The stored hash.
+
+    Returns:
+        bool: True if the hash should be recomputed on the next successful login.
+
+    """
+    try:
+        n, r, p, _, expected = _parse(encoded)
+    except ValueError:
+        return True
+    return (n, r, p, len(expected)) != (SCRYPT_N, SCRYPT_R, SCRYPT_P, SCRYPT_DKLEN)
+
+
+# A real hash of a random password, derived once at import. Verifying against this costs
+# the same as verifying a real account, which is what lets the login path spend equal
+# time on unknown and known usernames.
+_DUMMY_ENCODED: Final = hash_password(secrets.token_urlsafe(32))
+
+
+def dummy_verify() -> None:
+    """Spend the same time a real verification would.
+
+    Called on login paths that fail before reaching a password check -- unknown
+    username, disabled account, or an account with no password set -- so that response
+    timing does not reveal which accounts exist.
+    """
+    verify_password("", _DUMMY_ENCODED)
+
+
+async def hash_password_async(password: str) -> str:
+    """Hash a password on a worker thread.
+
+    Args:
+        password: The plaintext password.
+
+    Returns:
+        str: The encoded hash.
+
+    """
+    return await anyio.to_thread.run_sync(hash_password, password)
+
+
+async def verify_password_async(password: str, encoded: str) -> bool:
+    """Check a password on a worker thread.
+
+    Args:
+        password: The plaintext password to check.
+        encoded: The stored hash.
+
+    Returns:
+        bool: True if the password matches.
+
+    """
+    return await anyio.to_thread.run_sync(verify_password, password, encoded)
+
+
+async def dummy_verify_async() -> None:
+    """Spend a verification's worth of time on a worker thread."""
+    await anyio.to_thread.run_sync(dummy_verify)
+
+
+def new_token() -> str:
+    """Generate an opaque credential.
+
+    Returns:
+        str: A url-safe 256-bit random token.
+
+    """
+    return secrets.token_urlsafe(TOKEN_BYTES)
+
+
+def hash_token(token: str) -> str:
+    """Hash a token for storage.
+
+    A plain SHA-256 with no salt and no stretching, unlike passwords. Tokens are 256
+    bits of output from a cryptographic random source, so there is no dictionary to
+    attack and nothing for a work factor to buy; the digest is here so that a database
+    leak does not hand over usable session cookies. The fixed 64-character output also
+    lets the column be indexed for lookup.
+
+    Args:
+        token: The plaintext token.
+
+    Returns:
+        str: The hex digest.
+
+    """
+    return hashlib.sha256(token.encode()).hexdigest()
+
+
+def tokens_equal(left: str, right: str) -> bool:
+    """Compare two tokens without leaking their contents through timing.
+
+    Args:
+        left: One token.
+        right: The other.
+
+    Returns:
+        bool: True if equal.
+
+    """
+    return hmac.compare_digest(left, right)

spoolman/auth/levels.py

@@ -0,0 +1,66 @@
+"""The permission level scale.
+
+Three ordered levels gate the API: ``read`` covers every GET and websocket
+subscription, ``edit`` adds modification of existing records, and ``manage`` adds
+creation and deletion.
+
+``PUT /spool/{id}/use`` deliberately sits at ``edit`` rather than ``manage``: it is
+Moonraker's and OctoPrint's hot path, so a printer can be given an ``edit`` credential
+that tracks filament usage but cannot delete anything.
+"""
+
+from enum import StrEnum
+
+
+class Level(StrEnum):
+    """A permission level."""
+
+    READ = "read"
+    EDIT = "edit"
+    MANAGE = "manage"
+
+
+# Ranks are kept in a separate mapping rather than expressed with comparison operators
+# on Level itself. StrEnum inherits str's operators, so defining __lt__/__ge__ would
+# silently shadow string ordering and give "edit" < "manage" a second, alphabetical
+# meaning that happens to agree here and would stop agreeing the moment a level is
+# renamed. Callers use covers() instead.
+_RANKS: dict[Level, int] = {
+    Level.READ: 0,
+    Level.EDIT: 1,
+    Level.MANAGE: 2,
+}
+
+
+def covers(actual: Level, required: Level) -> bool:
+    """Check whether a level satisfies a requirement.
+
+    Args:
+        actual: The level held.
+        required: The level demanded.
+
+    Returns:
+        bool: True if `actual` is at least as permissive as `required`.
+
+    """
+    return _RANKS[actual] >= _RANKS[required]
+
+
+def parse_level(value: str) -> Level:
+    """Parse a level name.
+
+    Args:
+        value: The level name, case-insensitive.
+
+    Raises:
+        ValueError: If the name is not a known level.
+
+    Returns:
+        Level: The parsed level.
+
+    """
+    try:
+        return Level(value.strip().lower())
+    except ValueError as exc:
+        known = ", ".join(level.value for level in Level)
+        raise ValueError(f"Unknown permission level '{value}'. Expected one of: {known}.") from exc

spoolman/auth/principal.py

@@ -0,0 +1,66 @@
+"""The authenticated identity behind a request."""
+
+from dataclasses import dataclass
+from enum import StrEnum
+from typing import Final
+
+from spoolman.auth.levels import Level, covers
+
+
+class PrincipalKind(StrEnum):
+    """How a principal was established."""
+
+    USER = "user"
+    APIKEY = "apikey"
+    CERT = "cert"
+    ANONYMOUS = "anonymous"
+    DISABLED = "disabled"
+
+
+@dataclass(frozen=True, slots=True)
+class Principal:
+    """Who is making a request, and what they may do.
+
+    The absence of a credential is represented by ``None`` rather than by a Principal,
+    so a caller can never accidentally treat "nobody" as a permission holder.
+    """
+
+    kind: PrincipalKind
+    level: Level
+    user_id: int | None = None
+    username: str | None = None
+    is_admin: bool = False
+    is_owner: bool = False
+    session_id: int | None = None
+    api_key_id: int | None = None
+    # Multi-tenancy seam (see AUTHENTICATION_PLAN.md). Always None today. It exists now
+    # so that every auth database function already threads a tenant through its
+    # signature, and adding tenancy later becomes a filter inside those functions
+    # rather than a change to forty routers.
+    tenant_id: int | None = None
+
+    def covers(self, required: Level) -> bool:
+        """Check whether this principal satisfies a level requirement.
+
+        Args:
+            required: The level demanded.
+
+        Returns:
+            bool: True if permitted.
+
+        """
+        return covers(self.level, required)
+
+
+# Returned by every gate when SPOOLMAN_AUTH_ENABLED is unset, before any database access
+# happens. This is what keeps the auth-disabled request path identical to an instance
+# built before authentication existed.
+UNRESTRICTED: Final = Principal(
+    kind=PrincipalKind.DISABLED,
+    level=Level.MANAGE,
+    is_admin=True,
+    is_owner=True,
+)
+
+# Granted when the auth_anonymous_read setting is on and no credential was presented.
+ANONYMOUS_READER: Final = Principal(kind=PrincipalKind.ANONYMOUS, level=Level.READ)

spoolman/auth/ratelimit.py

@@ -0,0 +1,177 @@
+"""Login throttling and client address resolution.
+
+The windows are in-process dictionaries. That is sound here because Spoolman's
+deployment contract is exactly one process -- the same assumption the websocket manager
+and the scheduler already make -- but it does mean the counters reset on restart. Account
+lockout is therefore also persisted on ``auth_user``; this module only provides the
+cheap first line of defence that never touches the database.
+"""
+
+import time
+from collections import deque
+from dataclasses import dataclass
+from datetime import timedelta
+from ipaddress import IPv4Network, IPv6Network, ip_address, ip_network
+from typing import Final
+
+from starlette.requests import Request
+from starlette.websockets import WebSocket
+
+from spoolman import env
+
+FORWARDED_FOR: Final = "x-forwarded-for"
+FORWARDED_PROTO: Final = "x-forwarded-proto"
+
+
+@dataclass(frozen=True, slots=True)
+class Decision:
+    """The outcome of a rate limit check."""
+
+    allowed: bool
+    retry_after: int = 0
+
+
+class SlidingWindow:
+    """Counts events per key over a moving time window."""
+
+    def __init__(self, *, limit: int, window: timedelta) -> None:
+        """Create a window.
+
+        Args:
+            limit: How many events are allowed within the window.
+            window: How far back the window reaches.
+
+        """
+        self._limit = limit
+        self._window = window.total_seconds()
+        self._events: dict[str, deque[float]] = {}
+
+    def _prune(self, key: str, now: float) -> deque[float] | None:
+        """Drop expired events for a key, forgetting the key entirely once empty."""
+        events = self._events.get(key)
+        if events is None:
+            return None
+        cutoff = now - self._window
+        while events and events[0] <= cutoff:
+            events.popleft()
+        if not events:
+            del self._events[key]
+            return None
+        return events
+
+    def check(self, key: str) -> Decision:
+        """Test whether an event would be allowed, without recording it.
+
+        Args:
+            key: The identity being limited.
+
+        Returns:
+            Decision: Whether to allow, and how long to wait if not.
+
+        """
+        now = time.monotonic()
+        events = self._prune(key, now)
+        if events is None or len(events) < self._limit:
+            return Decision(allowed=True)
+        retry_after = int(events[0] + self._window - now) + 1
+        return Decision(allowed=False, retry_after=max(retry_after, 1))
+
+    def hit(self, key: str) -> None:
+        """Record an event against a key.
+
+        Args:
+            key: The identity being limited.
+
+        """
+        now = time.monotonic()
+        self._prune(key, now)
+        self._events.setdefault(key, deque()).append(now)
+
+    def clear(self, key: str) -> None:
+        """Forget a key's history, called after a successful login.
+
+        Args:
+            key: The identity to reset.
+
+        """
+        self._events.pop(key, None)
+
+
+# Keyed by username, so guessing one account's password does not lock out the instance.
+user_window: Final = SlidingWindow(limit=5, window=timedelta(minutes=15))
+
+# Keyed by client address, to blunt spraying across many usernames from one source.
+ip_window: Final = SlidingWindow(limit=20, window=timedelta(minutes=15))
+
+
+def _parse_network(entry: str) -> IPv4Network | IPv6Network | None:
+    """Parse one trusted-proxy entry, returning None if it is not an address or network."""
+    try:
+        return ip_network(entry, strict=False)
+    except ValueError:
+        return None
+
+
+def _trusted_networks() -> list[IPv4Network | IPv6Network]:
+    """Parse SPOOLMAN_TRUSTED_PROXIES, skipping entries that are not addresses."""
+    parsed = (_parse_network(entry) for entry in env.get_trusted_proxies())
+    return [network for network in parsed if network is not None]
+
+
+def is_trusted_peer(peer: str | None) -> bool:
+    """Check whether forwarding headers from an address may be believed.
+
+    Args:
+        peer: The address the connection actually came from.
+
+    Returns:
+        bool: True if the address is covered by SPOOLMAN_TRUSTED_PROXIES.
+
+    """
+    if not peer:
+        return False
+    try:
+        address = ip_address(peer)
+    except ValueError:
+        return False
+    return any(address in network for network in _trusted_networks())
+
+
+def client_ip(conn: Request | WebSocket) -> str:
+    """Determine the address a request came from.
+
+    X-Forwarded-For is honoured only when the immediate peer is a configured trusted
+    proxy. Without that check any client could spoof the header and evade rate limiting
+    by varying it.
+
+    Args:
+        conn: The request or websocket.
+
+    Returns:
+        str: The client address, or an empty string if it cannot be determined.
+
+    """
+    peer = conn.client.host if conn.client else ""
+    if is_trusted_peer(peer):
+        forwarded = conn.headers.get(FORWARDED_FOR)
+        if forwarded:
+            return forwarded.split(",")[0].strip()
+    return peer
+
+
+def forwarded_proto(conn: Request | WebSocket) -> str | None:
+    """Read the originating scheme reported by a trusted proxy.
+
+    Args:
+        conn: The request or websocket.
+
+    Returns:
+        Optional[str]: The scheme, or None if there is no trustworthy header.
+
+    """
+    if not is_trusted_peer(conn.client.host if conn.client else ""):
+        return None
+    proto = conn.headers.get(FORWARDED_PROTO)
+    if not proto:
+        return None
+    return proto.split(",")[0].strip().lower()

spoolman/exceptions.py

@@ -15,3 +15,11 @@ class ItemCreateError(Exception):
 
 class SpoolMeasureError(Exception):
     pass
+
+
+class AuthenticationRequiredError(Exception):
+    """No usable credential was presented. Mapped to HTTP 401."""
+
+
+class PermissionDeniedError(Exception):
+    """A credential was presented but does not permit the operation. Mapped to HTTP 403."""