feat(523): forge-connection tokens are stored sealed under SECRET_KEY, like the backup S3 secret
CI & Build / Python lint (push) Successful in 3s
CI & Build / Plugin hooks (push) Successful in 16s
CI & Build / TypeScript typecheck (push) Successful in 56s
CI & Build / integration (push) Successful in 1m8s
CI & Build / Python tests (push) Successful in 1m54s
CI & Build / Build & push image (push) Successful in 36s
CI & Build / Python lint (push) Successful in 3s
CI & Build / Plugin hooks (push) Successful in 16s
CI & Build / TypeScript typecheck (push) Successful in 56s
CI & Build / integration (push) Successful in 1m8s
CI & Build / Python tests (push) Successful in 1m54s
CI & Build / Build & push image (push) Successful in 36s
The column becomes token_sealed (migration 0122 seals existing rows with a frozen copy of the v1 key), every reader opens it through connection_token, a row this instance cannot open is skipped with a warning, and Settings asks for that token again. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,47 @@
|
||||
"""Forge-connection tokens are stored sealed (milestone 523, #5877).
|
||||
|
||||
Migration 0122 seals the rows that existed before, with its OWN frozen copy of
|
||||
the v1 key derivation (so a later edit to services/secrets.py can never change
|
||||
what an old migration writes). That copy is only right while it agrees with
|
||||
what the app opens today — which is what this pins. If secrets.py ever moves
|
||||
to a v2 key, it must still open v1 values, and this test says so.
|
||||
"""
|
||||
import importlib.util
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import patch
|
||||
|
||||
from scribe.services import secrets
|
||||
from scribe.services.forge_connections import connection_token
|
||||
|
||||
_MIGRATION = Path(__file__).resolve().parents[1] / "alembic" / "versions" / "0122_seal_forge_tokens.py"
|
||||
|
||||
|
||||
def _migration():
|
||||
spec = importlib.util.spec_from_file_location("migration_0122", _MIGRATION)
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(module)
|
||||
return module
|
||||
|
||||
|
||||
def test_what_the_migration_seals_the_app_opens():
|
||||
sealed = _migration().seal_v1("gitea-read-token")
|
||||
assert "gitea-read-token" not in sealed
|
||||
assert secrets.unseal(sealed) == "gitea-read-token"
|
||||
|
||||
|
||||
def test_a_row_reads_through_connection_token():
|
||||
row = SimpleNamespace(token_sealed=secrets.seal("ghp_abc"))
|
||||
assert connection_token(row) == "ghp_abc"
|
||||
|
||||
|
||||
def test_a_row_sealed_under_another_key_reads_as_none():
|
||||
row = SimpleNamespace(token_sealed=secrets.seal("ghp_abc"))
|
||||
with patch.object(secrets.Config, "SECRET_KEY", "a-different-instance"):
|
||||
assert connection_token(row) is None
|
||||
|
||||
|
||||
def test_a_left_over_plain_text_token_is_not_used_as_one():
|
||||
"""A value that never went through seal is refused, not passed to the
|
||||
forge as if it were the token: a reader sees None, never the raw bytes."""
|
||||
assert connection_token(SimpleNamespace(token_sealed="plain-text-token")) is None
|
||||
@@ -169,3 +169,29 @@ async def test_env_config_serves_admins_only_and_rows_beat_it(seeded):
|
||||
assert len(selector.adapters) == 1
|
||||
assert selector.adapters[0]._token == "row-tok"
|
||||
await delete_connection(admin, row.id)
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_the_token_is_stored_sealed_and_a_key_change_drops_the_row(seeded):
|
||||
"""What a copy of the database holds is ciphertext; the reads still get
|
||||
the token; and a row this instance can no longer open serves nothing
|
||||
rather than sending ciphertext to the forge as a token."""
|
||||
from sqlalchemy import text
|
||||
|
||||
from scribe.services import secrets
|
||||
|
||||
owner = seeded["owner"]
|
||||
row = await create_connection(owner, kind="gitea", base_url=GITEA, token="plain-tok")
|
||||
await update_connection(owner, row.id, token="rotated-tok")
|
||||
async with async_session() as s:
|
||||
stored = (await s.execute(
|
||||
text("SELECT token_sealed FROM forge_connections WHERE id = :id"), {"id": row.id}
|
||||
)).scalar_one()
|
||||
assert "rotated-tok" not in stored
|
||||
assert secrets.unseal(stored) == "rotated-tok"
|
||||
|
||||
selector = await get_forges(owner, seeded["project"])
|
||||
assert [a._token for a in selector.adapters] == ["rotated-tok"]
|
||||
|
||||
with patch.object(secrets.Config, "SECRET_KEY", "a-different-instance"):
|
||||
assert not (await get_forges(owner, seeded["project"])).configured
|
||||
|
||||
Reference in New Issue
Block a user