114 lines
3.8 KiB
Python
114 lines
3.8 KiB
Python
import base64
|
|
import json
|
|
import os
|
|
import sqlite3
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
from src.completed_filed_review_store import CompletedFiledReviewStore
|
|
from src.saved_search_store import SavedSearchStore
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
SCRIPT = ROOT / "scripts" / "rotate_private_state.py"
|
|
|
|
|
|
def encoded(value: bytes) -> str:
|
|
return base64.b64encode(value * 32).decode()
|
|
|
|
|
|
def run_rotation(state: Path, *, tamper: bool = False):
|
|
env = {
|
|
**os.environ,
|
|
"STACKCHAIN_STATE_DIR": str(state),
|
|
"STACKCHAIN_PRIVATE_STATE_ENCRYPTION_KEYS": json.dumps(
|
|
{"old": encoded(b"o"), "next": encoded(b"n")}
|
|
),
|
|
"STACKCHAIN_PRIVATE_STATE_ACTIVE_KEY_ID": "next",
|
|
}
|
|
if tamper:
|
|
with sqlite3.connect(state / "saved-searches.sqlite3") as connection:
|
|
connection.execute(
|
|
"UPDATE saved_searches SET views = 'v1:tampered' WHERE login = 'timmy'"
|
|
)
|
|
return subprocess.run(
|
|
[sys.executable, str(SCRIPT)], env=env, text=True, capture_output=True
|
|
)
|
|
|
|
|
|
def test_rotation_command_rewraps_shared_store_rows_without_changing_metadata(tmp_path):
|
|
state = tmp_path / "state"
|
|
path = state / "saved-searches.sqlite3"
|
|
store = SavedSearchStore(path, encryption_key=b"o" * 32)
|
|
expected = store.replace(
|
|
"timmy",
|
|
0,
|
|
[{"id": "mine", "name": "My private work", "query": "assignee:timmy"}],
|
|
)
|
|
|
|
completed = run_rotation(state)
|
|
|
|
assert completed.returncode == 0, completed.stderr
|
|
report = json.loads(completed.stdout)
|
|
assert report["saved-searches"] == {
|
|
"current": 0,
|
|
"failed": 0,
|
|
"migrated": 1,
|
|
"total": 1,
|
|
}
|
|
with sqlite3.connect(path) as connection:
|
|
revision, payload = connection.execute(
|
|
"SELECT revision, views FROM saved_searches WHERE login = 'timmy'"
|
|
).fetchone()
|
|
assert revision == expected["revision"]
|
|
assert payload.startswith("v2:next:")
|
|
assert SavedSearchStore(
|
|
path, encryption_key=({"next": b"n" * 32}, "next")
|
|
).get("timmy") == expected
|
|
|
|
|
|
def test_rotation_command_fails_closed_and_never_prints_private_content(tmp_path):
|
|
state = tmp_path / "state"
|
|
secret = "private launch roadmap"
|
|
store = SavedSearchStore(state / "saved-searches.sqlite3", encryption_key=b"o" * 32)
|
|
store.replace("timmy", 0, [{"id": "secret", "name": secret, "query": "is:open"}])
|
|
|
|
completed = run_rotation(state, tamper=True)
|
|
|
|
assert completed.returncode == 1
|
|
report = json.loads(completed.stdout)
|
|
assert report["saved-searches"]["failed"] == 1
|
|
assert secret not in completed.stdout
|
|
assert "timmy" not in completed.stdout
|
|
assert completed.stderr == ""
|
|
|
|
|
|
def test_rotation_command_rewraps_completed_filed_history_without_printing_it(tmp_path):
|
|
state = tmp_path / "state"
|
|
path = state / "completed-filed-reviews.sqlite3"
|
|
private_receipt = {
|
|
"repository": "private-rotation/canary",
|
|
"number": 1252,
|
|
"updated_at": "2026-08-22T12:00:00Z",
|
|
}
|
|
CompletedFiledReviewStore(path, encryption_key=b"o" * 32).merge("timmy", [private_receipt])
|
|
|
|
completed = run_rotation(state)
|
|
|
|
assert completed.returncode == 0, completed.stderr
|
|
report = json.loads(completed.stdout)
|
|
assert report["completed-filed-reviews"] == {
|
|
"current": 0, "failed": 0, "migrated": 1, "total": 1
|
|
}
|
|
assert "private-rotation" not in completed.stdout
|
|
assert "timmy" not in completed.stdout
|
|
with sqlite3.connect(path) as connection:
|
|
payload = connection.execute(
|
|
"SELECT receipts FROM completed_filed_review_collections WHERE login = 'timmy'"
|
|
).fetchone()[0]
|
|
assert payload.startswith("v2:next:")
|
|
assert CompletedFiledReviewStore(
|
|
path, encryption_key=({"next": b"n" * 32}, "next")
|
|
).get("timmy") == {"receipts": [private_receipt]}
|