495 lines
19 KiB
Python
495 lines
19 KiB
Python
# Copyright 2026 Umbra Tools (umbratools.com)
|
|
# SPDX-License-Identifier: Apache-2.0
|
|
|
|
"""Password, token, and redaction helpers for the account slice.
|
|
|
|
Only the Argon2id construction is used for local passwords. Session and CSRF
|
|
tokens are generated with :mod:`secrets`, handed to the client exactly once,
|
|
and stored exclusively as a hash (optionally keyed with a server-side pepper).
|
|
|
|
Nothing in this module logs. The one logging-related helper is
|
|
:class:`SecretRedactionFilter`, which scrubs well-known secret patterns from
|
|
records emitted by the ``umbratrans_web`` logger tree so that an accidental
|
|
interpolation cannot leak a credential into a log stream.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import hmac
|
|
import logging
|
|
import re
|
|
import secrets
|
|
import string
|
|
from dataclasses import dataclass, field
|
|
from datetime import datetime, timedelta, timezone
|
|
from uuid import UUID
|
|
|
|
from argon2 import PasswordHasher, Type
|
|
from argon2.exceptions import InvalidHashError, VerificationError, VerifyMismatchError
|
|
|
|
from .errors import PasswordPolicyError
|
|
from .models import ALLOWED_AUDIT_DETAIL_KEYS
|
|
|
|
# Argon2id parameters. The defaults are the OWASP "second recommended" profile
|
|
# (64 MiB, t=3, p=4) and are the values recorded in the reviewed migration
|
|
# comments. Tests may lower them through ``Settings`` to stay fast; the
|
|
# production profile is asserted by a dedicated test.
|
|
ARGON2_MEMORY_COST_KIB = 64 * 1024
|
|
ARGON2_TIME_COST = 3
|
|
ARGON2_PARALLELISM = 4
|
|
ARGON2_HASH_LENGTH = 32
|
|
ARGON2_SALT_LENGTH = 16
|
|
|
|
MIN_PASSWORD_LENGTH = 14
|
|
MAX_PASSWORD_LENGTH = 1024
|
|
MAX_FAILED_LOGIN_ATTEMPTS = 5
|
|
DEFAULT_LOCKOUT = timedelta(minutes=15)
|
|
SESSION_TOKEN_BYTES = 32
|
|
CSRF_TOKEN_BYTES = 32
|
|
MIN_PEPPER_BYTES = 32
|
|
|
|
#: Length of a server-generated one-time enrollment password.
|
|
TEMPORARY_PASSWORD_LENGTH = 20
|
|
# Ambiguous glyphs (0/O, 1/l/I, 5/S, 2/Z) are left out: a one-time password is
|
|
# read once by an administrator and typed once by its owner, so transcription
|
|
# errors are the main failure mode worth designing out.
|
|
_AMBIGUOUS_GLYPHS = frozenset("0Oo1lI5S2Z")
|
|
_TEMPORARY_LOWERCASE = "".join(character for character in string.ascii_lowercase if character not in _AMBIGUOUS_GLYPHS)
|
|
_TEMPORARY_UPPERCASE = "".join(character for character in string.ascii_uppercase if character not in _AMBIGUOUS_GLYPHS)
|
|
_TEMPORARY_DIGITS = "".join(character for character in string.digits if character not in _AMBIGUOUS_GLYPHS)
|
|
_TEMPORARY_SYMBOLS = "!@#$%^&*-_=+"
|
|
_TEMPORARY_ALPHABET = _TEMPORARY_LOWERCASE + _TEMPORARY_UPPERCASE + _TEMPORARY_DIGITS + _TEMPORARY_SYMBOLS
|
|
|
|
MAX_AUDIT_VALUE_LENGTH = 200
|
|
REDACTED = "[redacted]"
|
|
|
|
_SECRET_KEY_PATTERN = (
|
|
r"(?:password|passwd|pwd|token|secret|credential|passphrase|private[_-]?key|"
|
|
r"api[_-]?key|authorization|cookie|session[_-]?id|hash|pepper|salt)"
|
|
)
|
|
_SECRET_ASSIGNMENT_RE = re.compile(
|
|
rf"(?i)([\"']?\b[A-Za-z0-9_-]*{_SECRET_KEY_PATTERN}\b[\"']?\s*[:=]\s*)"
|
|
r"(\"[^\"]*\"|'[^']*'|[^\s,;}\]]+)"
|
|
)
|
|
_BEARER_RE = re.compile(r"(?i)(\bBearer\s+)[A-Za-z0-9._~+/=-]+")
|
|
_BASIC_RE = re.compile(r"(?i)(\bBasic\s+)[A-Za-z0-9+/=]+")
|
|
_PRIVATE_KEY_RE = re.compile(
|
|
r"-----BEGIN [^-]*(?:PRIVATE|SECRET) KEY(?: BLOCK)?-----.*?"
|
|
r"-----END [^-]*(?:PRIVATE|SECRET) KEY(?: BLOCK)?-----",
|
|
re.DOTALL,
|
|
)
|
|
_ARGON2_HASH_RE = re.compile(r"\$argon2(?:id|i|d)\$[^\s\"']+")
|
|
_JWT_RE = re.compile(r"\beyJ[A-Za-z0-9_-]{6,}\.[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+")
|
|
_CONTROL_CHARS_RE = re.compile(r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]")
|
|
|
|
|
|
def utc_now() -> datetime:
|
|
"""Return an aware UTC timestamp. The single clock used by the slice."""
|
|
|
|
return datetime.now(timezone.utc)
|
|
|
|
|
|
def generate_token(nbytes: int = SESSION_TOKEN_BYTES) -> str:
|
|
"""Return a URL-safe random token with at least ``nbytes * 8`` bits."""
|
|
|
|
return secrets.token_urlsafe(nbytes)
|
|
|
|
|
|
def generate_temporary_password(*, username: str | None = None, length: int = TEMPORARY_PASSWORD_LENGTH) -> str:
|
|
"""Return a one-time enrollment password generated inside the server.
|
|
|
|
The value is handed to the caller exactly once, hashed immediately, and
|
|
never logged, audited, or persisted. One character from each class is
|
|
guaranteed so the result satisfies the local policy while still being drawn
|
|
from :mod:`secrets`. Generation is retried if the candidate would contain
|
|
the account username, which the policy forbids.
|
|
"""
|
|
|
|
for _ in range(8):
|
|
characters = [
|
|
secrets.choice(_TEMPORARY_LOWERCASE),
|
|
secrets.choice(_TEMPORARY_UPPERCASE),
|
|
secrets.choice(_TEMPORARY_DIGITS),
|
|
secrets.choice(_TEMPORARY_SYMBOLS),
|
|
]
|
|
characters.extend(secrets.choice(_TEMPORARY_ALPHABET) for _ in range(max(length, 8) - len(characters)))
|
|
candidate = "".join(characters)
|
|
if username and username.casefold() in candidate.casefold():
|
|
continue
|
|
return candidate
|
|
# Unreachable in practice: the username would have to occupy the whole
|
|
# candidate on eight consecutive draws.
|
|
raise PasswordPolicyError(("unmet",)) # pragma: no cover
|
|
|
|
|
|
def hash_session_token(token: str, *, pepper: bytes | None = None) -> str:
|
|
"""Return the stored form of a session token.
|
|
|
|
A 256-bit random token does not need a slow KDF, so a single SHA-256 (or
|
|
HMAC-SHA256 when a server-side pepper is configured) is used. The plain
|
|
token is never stored.
|
|
"""
|
|
|
|
if not token:
|
|
raise ValueError("token must not be empty")
|
|
payload = token.encode("utf-8")
|
|
if pepper:
|
|
return hmac.new(pepper, payload, hashlib.sha256).hexdigest()
|
|
return hashlib.sha256(payload).hexdigest()
|
|
|
|
|
|
def tokens_match(candidate: str, expected: str) -> bool:
|
|
"""Constant-time comparison for token and CSRF values."""
|
|
|
|
return hmac.compare_digest(candidate.encode("utf-8"), expected.encode("utf-8"))
|
|
|
|
|
|
def redact_text(value: str) -> str:
|
|
"""Remove credential-shaped substrings from free text.
|
|
|
|
Used for log records and for operator-supplied audit reasons. It errs
|
|
towards over-redaction; false positives cost readability, false negatives
|
|
cost a credential.
|
|
"""
|
|
|
|
text = _PRIVATE_KEY_RE.sub(REDACTED, value)
|
|
text = _ARGON2_HASH_RE.sub(REDACTED, text)
|
|
text = _JWT_RE.sub(REDACTED, text)
|
|
text = _BEARER_RE.sub(rf"\1{REDACTED}", text)
|
|
text = _BASIC_RE.sub(rf"\1{REDACTED}", text)
|
|
text = _SECRET_ASSIGNMENT_RE.sub(rf"\1{REDACTED}", text)
|
|
return _CONTROL_CHARS_RE.sub(" ", text)
|
|
|
|
|
|
def sanitize_audit_detail(detail: dict[str, object] | None) -> dict[str, str]:
|
|
"""Reduce an audit detail mapping to allow-listed, redacted strings.
|
|
|
|
Keys outside :data:`umbratrans_web.models.ALLOWED_AUDIT_DETAIL_KEYS` are
|
|
dropped, so a future caller cannot record secret material simply by adding
|
|
a new key. Values are stringified, redacted, and length-bounded.
|
|
"""
|
|
|
|
sanitized: dict[str, str] = {}
|
|
for key, raw in (detail or {}).items():
|
|
if key not in ALLOWED_AUDIT_DETAIL_KEYS:
|
|
continue
|
|
if raw is None:
|
|
continue
|
|
if isinstance(raw, bool):
|
|
value = "true" if raw else "false"
|
|
elif isinstance(raw, (str, int, float, UUID)):
|
|
value = str(raw)
|
|
else:
|
|
value = f"<redacted:{type(raw).__name__}>"
|
|
value = redact_text(str(value)).strip()
|
|
if not value:
|
|
continue
|
|
sanitized[key] = value[:MAX_AUDIT_VALUE_LENGTH]
|
|
return sanitized
|
|
|
|
|
|
class SecretRedactionFilter(logging.Filter):
|
|
"""Scrub secret patterns from log records of the application logger tree.
|
|
|
|
Installed by :func:`install_secret_redaction_filter`. It is a defence in
|
|
depth measure: application code is still expected to log identifiers only.
|
|
|
|
Arguments are redacted in place and the format string is left alone, so the
|
|
handler's own formatter still runs normally. The earlier version rendered the
|
|
record, redacted the result, and cleared the arguments, which scrubbed the
|
|
secret correctly but also destroyed the surrounding format: uvicorn's access
|
|
lines lost their level prefix and their trailing status text, because the
|
|
formatter no longer had the structure it expected. A log nobody can grep is
|
|
not much better than no log, so the arguments are rewritten instead.
|
|
|
|
One case still collapses the record: when the *format string itself* names a
|
|
secret, as in ``"created password=%s"``. Redacting the format string would
|
|
consume the ``%s`` and break the substitution, so there is no way to keep
|
|
both. It renders, redacts and clears instead, which is the older behaviour
|
|
and is confined to application loggers, since the server's own format strings
|
|
carry no secret-shaped text.
|
|
|
|
Limitation: the filter is pattern based. It removes secret-shaped text
|
|
(``password=...``, ``Authorization: Bearer ...``, an Argon2id PHC string, a
|
|
private-key block, a JWT, ``token=...``). It cannot recognise a secret
|
|
published under an innocuous key, so no call site may log a submitted value in
|
|
the first place.
|
|
"""
|
|
|
|
def filter(self, record: logging.LogRecord) -> bool:
|
|
args = record.args
|
|
if isinstance(args, tuple) or isinstance(args, dict):
|
|
if re.search(_SECRET_KEY_PATTERN, str(record.msg), re.IGNORECASE):
|
|
# The format string names a secret, so it cannot survive
|
|
# redaction intact. Render first, then scrub the whole line.
|
|
try:
|
|
rendered = record.getMessage()
|
|
except Exception: # noqa: BLE001 - never break logging on a bad record
|
|
return True
|
|
record.msg = redact_text(rendered)
|
|
record.args = None
|
|
return True
|
|
if isinstance(args, tuple):
|
|
record.args = tuple(self._redact_argument(value) for value in args)
|
|
else:
|
|
record.args = {key: self._redact_argument(value) for key, value in args.items()}
|
|
return True
|
|
try:
|
|
record.msg = redact_text(str(record.msg))
|
|
except Exception: # noqa: BLE001 - never break logging on a bad record
|
|
pass
|
|
return True
|
|
|
|
@staticmethod
|
|
def _redact_argument(value: object) -> object:
|
|
"""Redact a single argument, leaving non-strings for the formatter.
|
|
|
|
``%d`` and friends have to keep receiving an int, so this only touches
|
|
text. A ``%`` inside a value is safe to leave: the formatter substitutes
|
|
arguments into the format string rather than re-scanning the result.
|
|
"""
|
|
|
|
return redact_text(value) if isinstance(value, str) else value
|
|
|
|
|
|
#: Root of the application logger tree.
|
|
LOGGER_NAMESPACE = "umbratrans_web"
|
|
|
|
|
|
#: Namespaces whose loggers are scrubbed. Widened to the server's own loggers
|
|
#: because uvicorn's access log writes the full request line, query string
|
|
#: included, and a verification link carries its token in the query string.
|
|
REDACTION_NAMESPACES: tuple[str, ...] = (
|
|
LOGGER_NAMESPACE,
|
|
"uvicorn",
|
|
"uvicorn.access",
|
|
"uvicorn.error",
|
|
"gunicorn.access",
|
|
)
|
|
|
|
|
|
def _in_redaction_scope(name: str) -> bool:
|
|
return any(name == prefix or name.startswith(f"{prefix}.") for prefix in REDACTION_NAMESPACES)
|
|
|
|
|
|
class _RedactingLogger(logging.Logger):
|
|
"""A logger that scrubs its own records.
|
|
|
|
Python does not apply a parent's filters to a child's records, so patching
|
|
the loggers that exist at install time leaves any logger created later
|
|
unprotected. Modules are imported lazily often enough for that to matter, and
|
|
the failure is silent: the line is simply written in the clear.
|
|
|
|
Constructing a logger in a redacting scope therefore adds the filter
|
|
immediately, so a namespace cannot grow an unprotected member. Loggers outside
|
|
the scope are untouched, because this is a behaviour this application wants
|
|
and no other library does.
|
|
"""
|
|
|
|
def __init__(self, name: str, level: int = logging.NOTSET) -> None:
|
|
super().__init__(name, level)
|
|
if _in_redaction_scope(name):
|
|
self.addFilter(SecretRedactionFilter())
|
|
|
|
|
|
def install_secret_redaction_filter(namespace: str = LOGGER_NAMESPACE) -> int:
|
|
"""Attach :class:`SecretRedactionFilter` to the loggers that can carry secrets.
|
|
|
|
Covers the application logger tree and the server's own loggers.
|
|
|
|
The server's loggers are not optional. Uvicorn's access log records the full
|
|
request line, so a request to ``/verify-email.html?token=<secret>`` writes the
|
|
verification token to the journal verbatim. Anyone able to read the journal
|
|
could then confirm an address they do not control, which is the one thing
|
|
email verification exists to prevent. The token is single-use and expires, so
|
|
the window is short, but it is a window and it does not need to exist.
|
|
|
|
The filter handles this: it renders the record with its arguments before
|
|
redacting, and the request line arrives as a ``%``-style argument, so the
|
|
scrubbed value is what gets written.
|
|
|
|
Two things are patched, because either alone leaves a gap. Loggers that exist
|
|
now are given the filter directly, and :class:`_RedactingLogger` is installed
|
|
as the logger class so loggers created later are covered too. Returns the
|
|
number of existing loggers patched, for tests and diagnostics.
|
|
"""
|
|
|
|
global REDACTION_NAMESPACES
|
|
if namespace != LOGGER_NAMESPACE and namespace not in REDACTION_NAMESPACES:
|
|
REDACTION_NAMESPACES = (*REDACTION_NAMESPACES, namespace)
|
|
|
|
# From here on, any logger constructed in scope arrives already filtered.
|
|
if not issubclass(logging.getLoggerClass(), _RedactingLogger):
|
|
logging.setLoggerClass(_RedactingLogger)
|
|
|
|
names = {name for name in logging.Logger.manager.loggerDict if _in_redaction_scope(name)}
|
|
names.update(REDACTION_NAMESPACES)
|
|
|
|
patched = 0
|
|
for name in sorted(names):
|
|
candidate = logging.getLogger(name)
|
|
if any(isinstance(item, SecretRedactionFilter) for item in candidate.filters):
|
|
continue
|
|
candidate.addFilter(SecretRedactionFilter())
|
|
patched += 1
|
|
return patched
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class PasswordPolicy:
|
|
"""Local password policy.
|
|
|
|
The policy is intentionally length-first: a long passphrase is the main
|
|
defence, and composition rules push users toward predictable substitutions.
|
|
"""
|
|
|
|
min_length: int = MIN_PASSWORD_LENGTH
|
|
max_length: int = MAX_PASSWORD_LENGTH
|
|
forbid_username: bool = True
|
|
|
|
def validate(self, password: str, *, username: str | None = None) -> None:
|
|
"""Raise :class:`PasswordPolicyError` listing unmet rule names only.
|
|
|
|
The candidate password never appears in the exception, its message, or
|
|
its attributes.
|
|
"""
|
|
|
|
unmet: list[str] = []
|
|
if not isinstance(password, str) or not password:
|
|
unmet.append("not_empty")
|
|
else:
|
|
if len(password) < self.min_length:
|
|
unmet.append("min_length")
|
|
if len(password) > self.max_length:
|
|
unmet.append("max_length")
|
|
if password.strip() != password:
|
|
unmet.append("no_surrounding_whitespace")
|
|
if self.forbid_username and username:
|
|
lowered = password.casefold()
|
|
if username.casefold() in lowered:
|
|
unmet.append("must_not_contain_username")
|
|
if unmet:
|
|
raise PasswordPolicyError(tuple(unmet))
|
|
|
|
def describe(self) -> dict[str, object]:
|
|
"""Return a non-secret description of the policy for operator docs."""
|
|
|
|
return {"min_length": self.min_length, "max_length": self.max_length, "forbid_username": self.forbid_username}
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class PasswordService:
|
|
"""Argon2id hashing and verification with an explicit, reviewable profile."""
|
|
|
|
memory_cost_kib: int = ARGON2_MEMORY_COST_KIB
|
|
time_cost: int = ARGON2_TIME_COST
|
|
parallelism: int = ARGON2_PARALLELISM
|
|
hash_length: int = ARGON2_HASH_LENGTH
|
|
salt_length: int = ARGON2_SALT_LENGTH
|
|
policy: PasswordPolicy = field(default_factory=PasswordPolicy)
|
|
_hasher: PasswordHasher = field(init=False, repr=False, compare=False)
|
|
_dummy_hash: str = field(init=False, repr=False, compare=False)
|
|
|
|
def __post_init__(self) -> None:
|
|
hasher = PasswordHasher(
|
|
time_cost=self.time_cost,
|
|
memory_cost=self.memory_cost_kib,
|
|
parallelism=self.parallelism,
|
|
hash_len=self.hash_length,
|
|
salt_len=self.salt_length,
|
|
type=Type.ID,
|
|
)
|
|
object.__setattr__(self, "_hasher", hasher)
|
|
# A fixed dummy hash keeps the unknown-user path comparable in cost to
|
|
# the real verification path. It is a hash of a random value that is
|
|
# discarded immediately and is not a usable credential.
|
|
object.__setattr__(self, "_dummy_hash", hasher.hash(secrets.token_urlsafe(32)))
|
|
|
|
def hash_password(self, password: str) -> str:
|
|
"""Return an Argon2id hash. The plaintext is not retained anywhere."""
|
|
|
|
return self._hasher.hash(password)
|
|
|
|
def generate_temporary_password(self, *, username: str | None = None) -> str:
|
|
"""Return a one-time password that satisfies this service's policy.
|
|
|
|
Used by administrator-driven enrollment, where the plaintext is created
|
|
in-process, shown to the administrator once, hashed immediately, and
|
|
never written to a log, an audit record, or the repository.
|
|
"""
|
|
|
|
candidate = generate_temporary_password(
|
|
username=username,
|
|
length=max(TEMPORARY_PASSWORD_LENGTH, self.policy.min_length + 4),
|
|
)
|
|
self.policy.validate(candidate, username=username)
|
|
return candidate
|
|
|
|
def verify_password(self, password_hash: str | None, password: str) -> bool:
|
|
"""Verify a password in constant time, returning ``False`` on any error."""
|
|
|
|
if not password_hash:
|
|
self.verify_dummy(password)
|
|
return False
|
|
try:
|
|
return bool(self._hasher.verify(password_hash, password))
|
|
except (VerifyMismatchError, VerificationError, InvalidHashError):
|
|
return False
|
|
|
|
def verify_dummy(self, password: str) -> None:
|
|
"""Burn a comparable amount of CPU for an unknown account."""
|
|
|
|
self.verify_password(self._dummy_hash, password)
|
|
|
|
def needs_rehash(self, password_hash: str) -> bool:
|
|
try:
|
|
return bool(self._hasher.check_needs_rehash(password_hash))
|
|
except (InvalidHashError, VerificationError):
|
|
return True
|
|
|
|
def identifier(self, password_hash: str) -> str:
|
|
"""Return a non-reversible short label for logs, e.g. ``argon2id:m=65536``."""
|
|
|
|
parts = password_hash.split("$")
|
|
if len(parts) < 4 or parts[1] != "argon2id":
|
|
return "unknown"
|
|
return f"argon2id:{parts[3]}"
|
|
|
|
def parameters(self) -> dict[str, int | str]:
|
|
return {
|
|
"type": "argon2id",
|
|
"memory_cost_kib": self.memory_cost_kib,
|
|
"time_cost": self.time_cost,
|
|
"parallelism": self.parallelism,
|
|
"hash_length": self.hash_length,
|
|
"salt_length": self.salt_length,
|
|
}
|
|
|
|
|
|
__all__ = [
|
|
"ARGON2_HASH_LENGTH",
|
|
"ARGON2_MEMORY_COST_KIB",
|
|
"ARGON2_PARALLELISM",
|
|
"ARGON2_SALT_LENGTH",
|
|
"ARGON2_TIME_COST",
|
|
"CSRF_TOKEN_BYTES",
|
|
"DEFAULT_LOCKOUT",
|
|
"LOGGER_NAMESPACE",
|
|
"MAX_FAILED_LOGIN_ATTEMPTS",
|
|
"MAX_PASSWORD_LENGTH",
|
|
"MIN_PASSWORD_LENGTH",
|
|
"MIN_PEPPER_BYTES",
|
|
"SESSION_TOKEN_BYTES",
|
|
"PasswordPolicy",
|
|
"PasswordService",
|
|
"SecretRedactionFilter",
|
|
"generate_token",
|
|
"hash_session_token",
|
|
"install_secret_redaction_filter",
|
|
"redact_text",
|
|
"sanitize_audit_detail",
|
|
"tokens_match",
|
|
"utc_now",
|
|
]
|