248 lines
8.5 KiB
Python
248 lines
8.5 KiB
Python
"""
|
|
tests/test_fernet_rotation.py
|
|
-----------------------------
|
|
MT-23: control-plane Fernet key rotation.
|
|
|
|
The property that matters is continuity of access: at no point during a
|
|
rotation may a stored tenant credential become unreadable. A key change that
|
|
leaves one row encrypted under a discarded key locks that tenant out of its
|
|
own database, and nothing surfaces it until the next request for that tenant.
|
|
|
|
These tests exercise the rotation the way the runbook does — old key alone,
|
|
then both keys, then new key alone — and assert the credential survives each
|
|
step. The last of those is the one that catches a premature retirement.
|
|
|
|
`control.crypto` reads its keys from the environment on every call, so the
|
|
fixtures set os.environ directly rather than patching the module.
|
|
"""
|
|
|
|
import os
|
|
|
|
import pytest
|
|
from cryptography.fernet import Fernet, InvalidToken
|
|
|
|
from control import crypto
|
|
|
|
|
|
KEY_A = Fernet.generate_key().decode() # "old" key
|
|
KEY_B = Fernet.generate_key().decode() # "new" key
|
|
KEY_C = Fernet.generate_key().decode() # a key never used to encrypt
|
|
|
|
|
|
@pytest.fixture
|
|
def clean_env():
|
|
"""Isolate the crypto env vars and restore whatever was there."""
|
|
saved = {k: os.environ.get(k) for k in (
|
|
'CONTROL_FERNET_KEY', 'CONTROL_FERNET_KEYS_OLD',
|
|
'CONTROL_FERNET_KEY_VERSION',
|
|
)}
|
|
try:
|
|
yield
|
|
finally:
|
|
for k, v in saved.items():
|
|
if v is None:
|
|
os.environ.pop(k, None)
|
|
else:
|
|
os.environ[k] = v
|
|
|
|
|
|
def _configure(primary, old=(), version=1):
|
|
os.environ['CONTROL_FERNET_KEY'] = primary
|
|
os.environ['CONTROL_FERNET_KEYS_OLD'] = ','.join(old)
|
|
os.environ['CONTROL_FERNET_KEY_VERSION'] = str(version)
|
|
|
|
|
|
# ── Baseline behaviour ───────────────────────────────────────────────────────
|
|
|
|
def test_roundtrip_with_a_single_key(clean_env):
|
|
_configure(KEY_A)
|
|
token = crypto.encrypt('s3cret-pw')
|
|
assert token != 's3cret-pw'
|
|
assert crypto.decrypt(token) == 's3cret-pw'
|
|
|
|
|
|
def test_none_passes_through(clean_env):
|
|
_configure(KEY_A)
|
|
assert crypto.encrypt(None) is None
|
|
assert crypto.decrypt(None) is None
|
|
assert crypto.rotate(None) is None
|
|
assert crypto.can_decrypt(None) is True
|
|
|
|
|
|
def test_missing_key_raises_a_clear_error(clean_env):
|
|
os.environ.pop('CONTROL_FERNET_KEY', None)
|
|
with pytest.raises(RuntimeError) as exc:
|
|
crypto.encrypt('x')
|
|
assert 'CONTROL_FERNET_KEY' in str(exc.value)
|
|
|
|
|
|
def test_unknown_key_cannot_decrypt(clean_env):
|
|
_configure(KEY_A)
|
|
token = crypto.encrypt('pw')
|
|
_configure(KEY_C) # unrelated key, no old keys
|
|
assert crypto.can_decrypt(token) is False
|
|
with pytest.raises(InvalidToken):
|
|
crypto.decrypt(token)
|
|
|
|
|
|
# ── The rotation itself ──────────────────────────────────────────────────────
|
|
|
|
def test_old_ciphertext_readable_after_key_change(clean_env):
|
|
"""Step 2 of the runbook: new primary, old key retained for decrypt."""
|
|
_configure(KEY_A, version=1)
|
|
token = crypto.encrypt('tenant-pw')
|
|
|
|
_configure(KEY_B, old=[KEY_A], version=2)
|
|
assert crypto.decrypt(token) == 'tenant-pw'
|
|
assert crypto.can_decrypt(token) is True
|
|
|
|
|
|
def test_new_writes_use_the_primary_key_only(clean_env):
|
|
"""A credential written after rotation must not depend on the old key."""
|
|
_configure(KEY_B, old=[KEY_A], version=2)
|
|
token = crypto.encrypt('fresh-pw')
|
|
|
|
# Old key alone must NOT open it — proves the primary did the encrypting.
|
|
_configure(KEY_A, version=1)
|
|
assert crypto.can_decrypt(token) is False
|
|
|
|
|
|
def test_rotate_rewrites_under_the_primary_key(clean_env):
|
|
"""Step 4: after rotate(), the old key is no longer needed."""
|
|
_configure(KEY_A, version=1)
|
|
token = crypto.encrypt('tenant-pw')
|
|
|
|
_configure(KEY_B, old=[KEY_A], version=2)
|
|
rotated = crypto.rotate(token)
|
|
assert rotated != token
|
|
assert crypto.decrypt(rotated) == 'tenant-pw'
|
|
|
|
# Step 5: retire KEY_A. The rotated token must still open.
|
|
_configure(KEY_B, version=2)
|
|
assert crypto.decrypt(rotated) == 'tenant-pw'
|
|
# And the un-rotated one must not — this is what verify-keys catches.
|
|
assert crypto.can_decrypt(token) is False
|
|
|
|
|
|
def test_rotation_preserves_plaintext_across_two_generations(clean_env):
|
|
"""A credential must survive being rotated twice, A -> B -> C."""
|
|
_configure(KEY_A, version=1)
|
|
token = crypto.encrypt('long-lived-pw')
|
|
|
|
_configure(KEY_B, old=[KEY_A], version=2)
|
|
token = crypto.rotate(token)
|
|
|
|
_configure(KEY_C, old=[KEY_B], version=3) # KEY_A already retired
|
|
token = crypto.rotate(token)
|
|
|
|
_configure(KEY_C, version=3) # all predecessors retired
|
|
assert crypto.decrypt(token) == 'long-lived-pw'
|
|
|
|
|
|
def test_stale_primary_in_old_list_is_tolerated(clean_env):
|
|
"""An operator leaving the primary in CONTROL_FERNET_KEYS_OLD is harmless."""
|
|
_configure(KEY_B, old=[KEY_B, KEY_A], version=2)
|
|
token = crypto.encrypt('pw')
|
|
assert crypto.decrypt(token) == 'pw'
|
|
|
|
|
|
def test_blank_entries_in_old_key_list_are_ignored(clean_env):
|
|
"""Trailing commas and whitespace must not break key loading."""
|
|
os.environ['CONTROL_FERNET_KEY'] = KEY_B
|
|
os.environ['CONTROL_FERNET_KEYS_OLD'] = f' {KEY_A} , , '
|
|
os.environ['CONTROL_FERNET_KEY_VERSION'] = '2'
|
|
assert crypto.old_keys() == [KEY_A.encode()]
|
|
_configure(KEY_A, version=1)
|
|
token = crypto.encrypt('pw')
|
|
os.environ['CONTROL_FERNET_KEY'] = KEY_B
|
|
os.environ['CONTROL_FERNET_KEYS_OLD'] = f' {KEY_A} , , '
|
|
assert crypto.decrypt(token) == 'pw'
|
|
|
|
|
|
# ── Version stamping ─────────────────────────────────────────────────────────
|
|
|
|
def test_key_version_defaults_to_one(clean_env):
|
|
_configure(KEY_A)
|
|
os.environ.pop('CONTROL_FERNET_KEY_VERSION', None)
|
|
assert crypto.current_key_version() == 1
|
|
|
|
|
|
def test_key_version_survives_a_garbage_value(clean_env):
|
|
_configure(KEY_A)
|
|
os.environ['CONTROL_FERNET_KEY_VERSION'] = 'not-a-number'
|
|
assert crypto.current_key_version() == 1
|
|
|
|
|
|
def test_fingerprint_identifies_a_key_without_revealing_it(clean_env):
|
|
_configure(KEY_A)
|
|
fp = crypto.key_fingerprint()
|
|
assert len(fp) == 12
|
|
assert KEY_A not in fp
|
|
assert fp == crypto.key_fingerprint(KEY_A)
|
|
assert fp != crypto.key_fingerprint(KEY_B)
|
|
|
|
|
|
# ── Model-level stamping (no DB required) ────────────────────────────────────
|
|
|
|
def test_tenant_set_password_stamps_the_current_version(clean_env):
|
|
from control.models import Tenant
|
|
|
|
_configure(KEY_A, version=1)
|
|
t = Tenant()
|
|
t.set_db_password('pw-one')
|
|
assert t.key_version == 1
|
|
assert t.db_password == 'pw-one'
|
|
|
|
_configure(KEY_B, old=[KEY_A], version=2)
|
|
t.set_db_password('pw-two')
|
|
assert t.key_version == 2
|
|
assert t.db_password == 'pw-two'
|
|
|
|
|
|
def test_tenant_rotate_updates_ciphertext_and_version(clean_env):
|
|
from control.models import Tenant
|
|
|
|
_configure(KEY_A, version=1)
|
|
t = Tenant()
|
|
t.set_db_password('rotate-me')
|
|
before = t.db_password_enc
|
|
|
|
_configure(KEY_B, old=[KEY_A], version=2)
|
|
assert t.rotate_db_password_key() is True
|
|
assert t.db_password_enc != before
|
|
assert t.key_version == 2
|
|
assert t.db_password == 'rotate-me'
|
|
|
|
# Retire the old key — the rotated row must still open.
|
|
_configure(KEY_B, version=2)
|
|
assert t.db_password == 'rotate-me'
|
|
|
|
|
|
def test_tenant_without_credential_is_stamped_not_rotated(clean_env):
|
|
from control.models import Tenant
|
|
|
|
_configure(KEY_B, old=[KEY_A], version=2)
|
|
t = Tenant()
|
|
assert t.db_password_enc is None
|
|
assert t.rotate_db_password_key() is False
|
|
# Stamped anyway, so it stops appearing as outstanding work.
|
|
assert t.key_version == 2
|
|
|
|
|
|
def test_db_uri_still_builds_after_rotation(clean_env):
|
|
"""Rotation must not disturb the URI the routing layer depends on."""
|
|
from control.models import Tenant
|
|
|
|
_configure(KEY_A, version=1)
|
|
t = Tenant(db_host='localhost', db_port=3306,
|
|
db_name='jqc_t1', db_user='jqc_t1')
|
|
t.set_db_password('p@ss word//')
|
|
uri_before = t.db_uri
|
|
|
|
_configure(KEY_B, old=[KEY_A], version=2)
|
|
t.rotate_db_password_key()
|
|
|
|
_configure(KEY_B, version=2)
|
|
assert t.db_uri == uri_before
|
|
assert 'p%40ss+word%2F%2F' in t.db_uri # still URL-encoded
|