Files
FabledGrimoire/alembic/versions/0122_seal_forge_tokens.py
T
bvandeusenandClaude Opus 5.5 79655fd19f
CI & Build / Python lint (push) Successful in 2s
CI & Build / Plugin hooks (push) Successful in 13s
CI & Build / TypeScript typecheck (push) Successful in 59s
CI & Build / integration (push) Successful in 1m32s
CI & Build / Python tests (push) Successful in 2m33s
CI & Build / Build & push image (push) Successful in 26s
docs(6207): the remaining migrations state their rationale in terms of the data
Record, milestone, phase and plan citations go from every migration header;
the "end before start" fix (0043) and the suppressed-count column (0095)
state the condition they prevent rather than the incident. 0053 and 0055
pointed at design specs under docs/superpowers that are not in the repo;
those pointers are removed.

Kept: alembic's generated "Create Date:" headers (template metadata, not
rationale), 0121's ARM_BORN justification beside the constant, and 0051's
illustrative "update task 42" format example.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-21 20:49:32 -04:00

83 lines
2.7 KiB
Python

"""Seal forge-connection tokens at rest
Revision ID: 0122
Revises: 0121
Create Date: 2026-10-15
A forge token was stored as plain text, so a copy of the database (or of a
backup tool's snapshot of it) was enough to read every user's forge
credential. Tokens are now sealed the way the backup destination's S3 secret
is (services/secrets.py): Fernet under a key derived from SECRET_KEY, which
lives in the deployment's environment rather than in the data.
The column is renamed `token` -> `token_sealed` in the same step. The name
says what the bytes are, and every reader has to change with it, so a reader
missed here fails on a missing attribute instead of quietly sending
ciphertext to the forge as a token.
THE KEY DERIVATION IS FROZEN HERE, not imported. This revision must seal with
exactly v1's key forever; importing services.secrets would let a later edit
there change what an old migration writes. The unit lane checks that what
this produces is what `secrets.unseal` opens today.
Cost: one row per (user, forge host) — a handful per instance.
Downgrade opens each token back into plain text. A row sealed under a
different SECRET_KEY cannot be opened and comes back empty; its owner enters
the token again.
"""
import base64
import hashlib
import sqlalchemy as sa
from alembic import op
revision = "0122"
down_revision = "0121"
branch_labels = None
depends_on = None
_LABEL_V1 = b"scribe.secrets.v1\x00"
def _fernet():
from cryptography.fernet import Fernet
from scribe.config import Config
digest = hashlib.sha256(_LABEL_V1 + Config.SECRET_KEY.encode()).digest()
return Fernet(base64.urlsafe_b64encode(digest))
def seal_v1(plain: str) -> str:
return _fernet().encrypt(plain.encode()).decode()
def upgrade() -> None:
op.alter_column("forge_connections", "token", new_column_name="token_sealed")
bind = op.get_bind()
rows = bind.execute(sa.text("SELECT id, token_sealed FROM forge_connections")).all()
for row_id, plain in rows:
bind.execute(
sa.text("UPDATE forge_connections SET token_sealed = :v WHERE id = :id"),
{"v": seal_v1(plain or ""), "id": row_id},
)
def downgrade() -> None:
from cryptography.fernet import InvalidToken
bind = op.get_bind()
fernet = _fernet()
rows = bind.execute(sa.text("SELECT id, token_sealed FROM forge_connections")).all()
for row_id, sealed in rows:
try:
plain = fernet.decrypt((sealed or "").encode()).decode()
except (InvalidToken, ValueError):
plain = ""
bind.execute(
sa.text("UPDATE forge_connections SET token_sealed = :v WHERE id = :id"),
{"v": plain, "id": row_id},
)
op.alter_column("forge_connections", "token_sealed", new_column_name="token")