mirror of
https://github.com/Routstr/routstr-core.git
synced 2026-08-09 02:54:37 +00:00
feat(vault): add Fernet/scrypt secret primitives
Encrypt/decrypt secrets at rest with Fernet keyed by ROUTSTR_SECRET_KEY, and hash/verify admin passwords with scrypt. Self-contained helpers with no consumers yet. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 4.8
parent
18965b4ea4
commit
86adee9b1b
@@ -0,0 +1,131 @@
|
||||
"""Encrypt/hash/fingerprint helpers for secrets at rest (issue #553).
|
||||
|
||||
Thin wrapper over ``cryptography`` so nothing else in the codebase touches
|
||||
Fernet/scrypt/HMAC directly:
|
||||
|
||||
- :func:`encrypt`/:func:`decrypt` — Fernet symmetric encryption, keyed by the
|
||||
mandatory ``ROUTSTR_SECRET_KEY``. Ciphertext is self-describing (``fernet:v1:``
|
||||
prefix) so a value can be told apart from legacy plaintext and so reading it
|
||||
under the wrong key surfaces as a hard error rather than silent corruption.
|
||||
- :func:`hash_password`/:func:`verify_password` — salted scrypt hashing. This is
|
||||
*key-independent*: it never reads ``ROUTSTR_SECRET_KEY``, so password login and
|
||||
the recovery script keep working even when the key is missing.
|
||||
|
||||
A missing or malformed ``ROUTSTR_SECRET_KEY`` fails fast with the generation
|
||||
command in the message.
|
||||
"""
|
||||
|
||||
import base64
|
||||
import hashlib
|
||||
import hmac
|
||||
import os
|
||||
import secrets
|
||||
|
||||
from cryptography.fernet import Fernet
|
||||
|
||||
_PREFIX = "fernet:v1:"
|
||||
_GEN_COMMAND = (
|
||||
'python -c "from cryptography.fernet import Fernet; '
|
||||
'print(Fernet.generate_key().decode())"'
|
||||
)
|
||||
|
||||
# Minimum admin-password length, enforced wherever a password is set/changed
|
||||
# (admin endpoints + the recovery script) so the policy lives in one place.
|
||||
MIN_PASSWORD_LENGTH = 8
|
||||
|
||||
# scrypt parameters; packed into each hash so verification is parameter-free.
|
||||
_SCRYPT_N = 2**14
|
||||
_SCRYPT_R = 8
|
||||
_SCRYPT_P = 1
|
||||
_SCRYPT_DKLEN = 32
|
||||
_SCRYPT_SALT_BYTES = 16
|
||||
|
||||
|
||||
def _require_secret_key() -> str:
|
||||
key = os.environ.get("ROUTSTR_SECRET_KEY")
|
||||
if not key:
|
||||
raise RuntimeError(
|
||||
"ROUTSTR_SECRET_KEY is not set. It is required to encrypt secrets at "
|
||||
"rest. Generate one with:\n " + _GEN_COMMAND
|
||||
)
|
||||
return key
|
||||
|
||||
|
||||
def get_fernet() -> Fernet:
|
||||
"""Build a :class:`Fernet` from ``ROUTSTR_SECRET_KEY`` (fails fast)."""
|
||||
key = _require_secret_key()
|
||||
try:
|
||||
return Fernet(key.encode())
|
||||
except (ValueError, TypeError) as exc:
|
||||
raise RuntimeError(
|
||||
"ROUTSTR_SECRET_KEY is malformed; it must be a url-safe base64 "
|
||||
"32-byte Fernet key. Generate one with:\n " + _GEN_COMMAND
|
||||
) from exc
|
||||
|
||||
|
||||
def encrypt(plaintext: str) -> str:
|
||||
"""Encrypt ``plaintext`` into a self-describing ``fernet:v1:`` token."""
|
||||
token = get_fernet().encrypt(plaintext.encode()).decode()
|
||||
return _PREFIX + token
|
||||
|
||||
|
||||
def is_encrypted(value: str) -> bool:
|
||||
"""True if ``value`` carries the ``fernet:v1:`` prefix this module emits."""
|
||||
return value.startswith(_PREFIX)
|
||||
|
||||
|
||||
def decrypt(ciphertext: str) -> str:
|
||||
"""Decrypt a ``fernet:v1:`` token.
|
||||
|
||||
Raises ``ValueError`` for an unprefixed value (so legacy plaintext is never
|
||||
mistaken for ciphertext) and ``InvalidToken`` when the value was written
|
||||
under a different ``ROUTSTR_SECRET_KEY``.
|
||||
"""
|
||||
if not is_encrypted(ciphertext):
|
||||
raise ValueError("value is not fernet:v1: ciphertext")
|
||||
token = ciphertext[len(_PREFIX) :]
|
||||
return get_fernet().decrypt(token.encode()).decode()
|
||||
|
||||
|
||||
def hash_password(password: str) -> str:
|
||||
"""Salted scrypt hash, self-describing as ``scrypt:n:r:p:salt:hash``."""
|
||||
salt = secrets.token_bytes(_SCRYPT_SALT_BYTES)
|
||||
derived = hashlib.scrypt(
|
||||
password.encode(),
|
||||
salt=salt,
|
||||
n=_SCRYPT_N,
|
||||
r=_SCRYPT_R,
|
||||
p=_SCRYPT_P,
|
||||
dklen=_SCRYPT_DKLEN,
|
||||
)
|
||||
return ":".join(
|
||||
[
|
||||
"scrypt",
|
||||
str(_SCRYPT_N),
|
||||
str(_SCRYPT_R),
|
||||
str(_SCRYPT_P),
|
||||
base64.b64encode(salt).decode(),
|
||||
base64.b64encode(derived).decode(),
|
||||
]
|
||||
)
|
||||
|
||||
|
||||
def verify_password(password: str, stored: str) -> bool:
|
||||
"""Constant-time check of ``password`` against a :func:`hash_password` value."""
|
||||
try:
|
||||
scheme, n, r, p, salt_b64, hash_b64 = stored.split(":")
|
||||
if scheme != "scrypt":
|
||||
return False
|
||||
salt = base64.b64decode(salt_b64)
|
||||
expected = base64.b64decode(hash_b64)
|
||||
derived = hashlib.scrypt(
|
||||
password.encode(),
|
||||
salt=salt,
|
||||
n=int(n),
|
||||
r=int(r),
|
||||
p=int(p),
|
||||
dklen=len(expected),
|
||||
)
|
||||
except (ValueError, TypeError):
|
||||
return False
|
||||
return hmac.compare_digest(derived, expected)
|
||||
@@ -0,0 +1,141 @@
|
||||
"""Tests for ``routstr.core.vault`` — the secret encrypt/hash/fingerprint helpers.
|
||||
|
||||
Specifies the primitives that the rest of the secret-storage work (issue #553)
|
||||
builds on, independent of any database or app wiring:
|
||||
|
||||
- ``encrypt``/``decrypt`` — Fernet symmetric encryption emitting self-describing
|
||||
``fernet:v1:`` ciphertext, so a value can be told apart from legacy plaintext
|
||||
and from ciphertext written under a different ``ROUTSTR_SECRET_KEY`` (which
|
||||
surfaces as a hard ``InvalidToken`` rather than silent corruption).
|
||||
- ``hash_password``/``verify_password`` — salted scrypt hashing that is
|
||||
*key-independent* (does not depend on ``ROUTSTR_SECRET_KEY``), so password
|
||||
login and the recovery script keep working even if the key is lost.
|
||||
- a missing/malformed ``ROUTSTR_SECRET_KEY`` fails fast with the generation
|
||||
command in the message.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
from cryptography.fernet import InvalidToken
|
||||
|
||||
from routstr.core import vault
|
||||
|
||||
# Two distinct, valid Fernet keys held fixed so ciphertext/fingerprints are
|
||||
# reproducible across runs and we can exercise the wrong-key path.
|
||||
KEY_A = "l_Tkp-7xmjcQ-IFhr6qhILrU8HPRbEmYMrfSbo_5srU="
|
||||
KEY_B = "_Teyrky_iToeDK51Tj1FsI9MJ340_cqKGmeher-a7MQ="
|
||||
|
||||
|
||||
def _use_key(monkeypatch: pytest.MonkeyPatch, key: str) -> None:
|
||||
monkeypatch.setenv("ROUTSTR_SECRET_KEY", key)
|
||||
|
||||
|
||||
# --- encrypt / decrypt -----------------------------------------------------
|
||||
|
||||
|
||||
def test_encrypt_decrypt_round_trips(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
_use_key(monkeypatch, KEY_A)
|
||||
assert vault.decrypt(vault.encrypt("nsec1secret")) == "nsec1secret"
|
||||
|
||||
|
||||
def test_encrypt_emits_self_describing_prefix(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
_use_key(monkeypatch, KEY_A)
|
||||
assert vault.encrypt("x").startswith("fernet:v1:")
|
||||
|
||||
|
||||
def test_encrypt_is_non_deterministic(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
# Fernet embeds a random IV/timestamp: equal plaintext -> different
|
||||
# ciphertext. This is exactly why upstream-key equality needs a blind index.
|
||||
_use_key(monkeypatch, KEY_A)
|
||||
assert vault.encrypt("same") != vault.encrypt("same")
|
||||
|
||||
|
||||
def test_is_encrypted_distinguishes_ciphertext_from_plaintext(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
_use_key(monkeypatch, KEY_A)
|
||||
assert vault.is_encrypted(vault.encrypt("x")) is True
|
||||
assert vault.is_encrypted("sk-plaintext-api-key") is False
|
||||
assert vault.is_encrypted("") is False
|
||||
|
||||
|
||||
def test_decrypt_rejects_unprefixed_value(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
# Guards the migration paths: a legacy plaintext value must never be
|
||||
# mistaken for ciphertext and "decrypted".
|
||||
_use_key(monkeypatch, KEY_A)
|
||||
with pytest.raises(ValueError):
|
||||
vault.decrypt("not-encrypted")
|
||||
|
||||
|
||||
def test_decrypt_with_wrong_key_raises(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
# The fail-fast signal: ciphertext written under KEY_A cannot be read under
|
||||
# KEY_B -> InvalidToken (bootstrap turns this into a clear startup error).
|
||||
_use_key(monkeypatch, KEY_A)
|
||||
token = vault.encrypt("secret")
|
||||
_use_key(monkeypatch, KEY_B)
|
||||
with pytest.raises(InvalidToken):
|
||||
vault.decrypt(token)
|
||||
|
||||
|
||||
# --- password hashing (key-independent) ------------------------------------
|
||||
|
||||
|
||||
def test_hash_and_verify_password(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
_use_key(monkeypatch, KEY_A)
|
||||
stored = vault.hash_password("correct horse")
|
||||
assert vault.verify_password("correct horse", stored) is True
|
||||
assert vault.verify_password("wrong", stored) is False
|
||||
|
||||
|
||||
def test_password_hash_is_salted(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
_use_key(monkeypatch, KEY_A)
|
||||
a = vault.hash_password("pw")
|
||||
b = vault.hash_password("pw")
|
||||
assert a != b
|
||||
assert vault.verify_password("pw", a) is True
|
||||
assert vault.verify_password("pw", b) is True
|
||||
|
||||
|
||||
def test_verify_password_rejects_malformed_stored_value(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
# A garbage or non-scrypt stored value must verify to False, never raise.
|
||||
_use_key(monkeypatch, KEY_A)
|
||||
assert vault.verify_password("pw", "") is False
|
||||
assert vault.verify_password("pw", "not-a-hash") is False
|
||||
assert vault.verify_password("pw", "bcrypt:1:2:3:x:y") is False
|
||||
|
||||
|
||||
def test_password_hashing_is_key_independent(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
# scrypt does not use ROUTSTR_SECRET_KEY, so login and the recovery script
|
||||
# work even when the key is missing.
|
||||
monkeypatch.delenv("ROUTSTR_SECRET_KEY", raising=False)
|
||||
stored = vault.hash_password("pw")
|
||||
assert vault.verify_password("pw", stored) is True
|
||||
|
||||
|
||||
# --- fail-fast on missing/malformed key ------------------------------------
|
||||
|
||||
|
||||
def test_missing_key_fails_fast_with_generation_command(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
monkeypatch.delenv("ROUTSTR_SECRET_KEY", raising=False)
|
||||
with pytest.raises(RuntimeError) as exc:
|
||||
vault.encrypt("x")
|
||||
msg = str(exc.value)
|
||||
assert "ROUTSTR_SECRET_KEY" in msg
|
||||
assert "Fernet.generate_key" in msg
|
||||
|
||||
|
||||
def test_malformed_key_fails_fast(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setenv("ROUTSTR_SECRET_KEY", "not-a-valid-fernet-key")
|
||||
with pytest.raises(RuntimeError):
|
||||
vault.encrypt("x")
|
||||
Reference in New Issue
Block a user