120 lines
4.1 KiB
Python
120 lines
4.1 KiB
Python
import base64
|
|
import json
|
|
import sqlite3
|
|
|
|
import pytest
|
|
|
|
from src.today_store import TodayStore
|
|
from src.state_encryption import (
|
|
PrivateStateCipher,
|
|
PrivateStateEncryptionError,
|
|
decode_private_state_encryption_keyring,
|
|
private_state_encryption_config,
|
|
)
|
|
|
|
|
|
def encoded(byte: bytes) -> str:
|
|
return base64.b64encode(byte * 32).decode()
|
|
|
|
|
|
def test_keyring_writes_active_versioned_envelope_and_reads_it_current():
|
|
keys, active = decode_private_state_encryption_keyring(
|
|
json.dumps({"old": encoded(b"o"), "next": encoded(b"n")}), "next"
|
|
)
|
|
|
|
cipher = PrivateStateCipher(keys, active_key_id=active, store="today")
|
|
payload = cipher.seal({"ids": [7]}, binding="timmy")
|
|
|
|
assert payload.startswith("v2:next:")
|
|
assert cipher.open(payload, binding="timmy") == ({"ids": [7]}, False)
|
|
|
|
|
|
def test_keyring_reads_legacy_v1_and_marks_it_for_rewrap():
|
|
legacy = PrivateStateCipher(b"o" * 32, store="today")
|
|
payload = legacy.seal({"ids": [7]}, binding="timmy")
|
|
rotating = PrivateStateCipher(
|
|
{"old": b"o" * 32, "next": b"n" * 32},
|
|
active_key_id="next",
|
|
store="today",
|
|
)
|
|
|
|
assert rotating.open(payload, binding="timmy") == ({"ids": [7]}, True)
|
|
|
|
|
|
def test_keyring_reads_inactive_v2_and_marks_it_for_rewrap():
|
|
old = PrivateStateCipher(
|
|
{"old": b"o" * 32, "next": b"n" * 32},
|
|
active_key_id="old",
|
|
store="today",
|
|
)
|
|
payload = old.seal({"ids": [7]}, binding="timmy")
|
|
rotating = PrivateStateCipher(
|
|
{"old": b"o" * 32, "next": b"n" * 32},
|
|
active_key_id="next",
|
|
store="today",
|
|
)
|
|
|
|
assert rotating.open(payload, binding="timmy") == ({"ids": [7]}, True)
|
|
|
|
|
|
def test_keyring_fails_closed_for_unknown_key_id():
|
|
cipher = PrivateStateCipher(
|
|
{"next": b"n" * 32}, active_key_id="next", store="today"
|
|
)
|
|
|
|
with pytest.raises(PrivateStateEncryptionError, match="could not be decrypted"):
|
|
cipher.open("v2:retired:AAAA", binding="timmy")
|
|
|
|
|
|
def test_keyring_configuration_is_bounded_and_requires_active_key():
|
|
with pytest.raises(PrivateStateEncryptionError, match="active key"):
|
|
decode_private_state_encryption_keyring(
|
|
json.dumps({"old": encoded(b"o")}), "missing"
|
|
)
|
|
with pytest.raises(PrivateStateEncryptionError, match="at most 4"):
|
|
decode_private_state_encryption_keyring(
|
|
json.dumps({str(i): encoded(bytes([65 + i])) for i in range(5)}), "0"
|
|
)
|
|
|
|
|
|
def test_environment_prefers_keyring_and_keeps_single_key_rollout(monkeypatch):
|
|
monkeypatch.setenv("STACKCHAIN_PRIVATE_STATE_ENCRYPTION_KEY", encoded(b"l"))
|
|
monkeypatch.delenv("STACKCHAIN_PRIVATE_STATE_ENCRYPTION_KEYS", raising=False)
|
|
monkeypatch.delenv("STACKCHAIN_PRIVATE_STATE_ACTIVE_KEY_ID", raising=False)
|
|
assert private_state_encryption_config() == b"l" * 32
|
|
|
|
monkeypatch.setenv(
|
|
"STACKCHAIN_PRIVATE_STATE_ENCRYPTION_KEYS",
|
|
json.dumps({"legacy": encoded(b"l"), "next": encoded(b"n")}),
|
|
)
|
|
monkeypatch.setenv("STACKCHAIN_PRIVATE_STATE_ACTIVE_KEY_ID", "next")
|
|
keys, active = private_state_encryption_config()
|
|
assert keys == {"legacy": b"l" * 32, "next": b"n" * 32}
|
|
assert active == "next"
|
|
|
|
|
|
def test_today_store_reads_and_lazily_rewraps_v2_state_during_rotation(tmp_path, monkeypatch):
|
|
path = tmp_path / "today.sqlite3"
|
|
legacy = TodayStore(path, encryption_key=b"l" * 32)
|
|
legacy.apply("timmy", "seed", "add", "issue:stackchain/dashboard#1237")
|
|
|
|
monkeypatch.setenv(
|
|
"STACKCHAIN_PRIVATE_STATE_ENCRYPTION_KEYS",
|
|
json.dumps({"legacy": encoded(b"l"), "next": encoded(b"n")}),
|
|
)
|
|
monkeypatch.setenv("STACKCHAIN_PRIVATE_STATE_ACTIVE_KEY_ID", "next")
|
|
rotating = TodayStore(path)
|
|
|
|
assert rotating.get("timmy")["ids"] == ["issue:stackchain/dashboard#1237"]
|
|
assert rotating.apply(
|
|
"timmy", "add-next", "add", "issue:stackchain/dashboard#1238"
|
|
)["ids"] == [
|
|
"issue:stackchain/dashboard#1237",
|
|
"issue:stackchain/dashboard#1238",
|
|
]
|
|
with sqlite3.connect(path) as connection:
|
|
payload = connection.execute(
|
|
"SELECT ids FROM today_plans WHERE login = 'timmy'"
|
|
).fetchone()[0]
|
|
assert payload.startswith("v2:next:")
|