iop/apps/control-plane/internal/credentialstore/schema.go
toki 4c8441e6c9 feat(credential): Provider Credential Slot 라우팅을 구현한다
사용자별 credential 저장, lease, projection, runtime 전달과 OpenAI-compatible 계약 및 검증 근거를 함께 반영한다.
2026-08-02 09:10:11 +09:00

513 lines
19 KiB
Go

package credentialstore
import (
"context"
"database/sql"
"fmt"
"strings"
)
// The dialect-specific schemas are idempotent DDL for principal and token tables.
//
// Design notes:
// - principals.alias is unique so duplicate issuance is rejected at the
// database layer even under concurrency.
// - tokens.token_ref is unique and is the only column that can join a
// verifier (Edge) to a principal. The raw token is never persisted.
// - tokens.digest is the SHA-256 hex encoding of the raw token; it is the
// single source of truth for authentication.
// - tokens.status transitions are enforced by the application layer; the
// schema only constrains not-null and length.
// - tokens.revision is an integer monotonic counter used for compare-and-swap
// disable/revoke. Revoked rows are never re-enabled by any method.
// - created_at / updated_at / revoked_at are UTC timestamps.
const sqlitePrincipalsSchema = `
CREATE TABLE IF NOT EXISTS principals (
id TEXT PRIMARY KEY,
alias TEXT NOT NULL UNIQUE,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL
)
`
const sqliteTokensSchema = `
CREATE TABLE IF NOT EXISTS tokens (
id TEXT PRIMARY KEY,
principal_id TEXT NOT NULL REFERENCES principals(id) ON DELETE CASCADE,
token_ref TEXT NOT NULL UNIQUE,
digest TEXT NOT NULL,
status TEXT NOT NULL CHECK(status IN ('active','disabled','revoked')),
revision INTEGER NOT NULL DEFAULT 0,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
revoked_at TEXT,
UNIQUE(principal_id, token_ref)
)
`
const postgresPrincipalsSchema = `
CREATE TABLE IF NOT EXISTS principals (
id TEXT PRIMARY KEY,
alias TEXT NOT NULL UNIQUE,
created_at TIMESTAMPTZ NOT NULL,
updated_at TIMESTAMPTZ NOT NULL
)
`
const postgresTokensSchema = `
CREATE TABLE IF NOT EXISTS tokens (
id TEXT PRIMARY KEY,
principal_id TEXT NOT NULL REFERENCES principals(id) ON DELETE CASCADE,
token_ref TEXT NOT NULL UNIQUE,
digest TEXT NOT NULL,
status TEXT NOT NULL CHECK(status IN ('active','disabled','revoked')),
revision BIGINT NOT NULL DEFAULT 0,
created_at TIMESTAMPTZ NOT NULL,
updated_at TIMESTAMPTZ NOT NULL,
revoked_at TIMESTAMPTZ,
UNIQUE(principal_id, token_ref)
)
`
const sqliteCredentialSlotsSchema = `
CREATE TABLE IF NOT EXISTS credential_slots (
id TEXT PRIMARY KEY,
principal_id TEXT NOT NULL REFERENCES principals(id) ON DELETE CASCADE,
vendor TEXT NOT NULL,
credential_kind TEXT NOT NULL,
alias TEXT,
status TEXT NOT NULL CHECK(status IN ('draft','active','disabled','revoked')),
revision INTEGER NOT NULL DEFAULT 0,
algorithm TEXT NOT NULL,
key_id TEXT NOT NULL,
key_version INTEGER NOT NULL,
nonce BLOB NOT NULL,
ciphertext BLOB NOT NULL,
aad BLOB,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
revoked_at TEXT,
UNIQUE(principal_id, id)
)
`
const sqliteSlotRevisionsSchema = `
CREATE TABLE IF NOT EXISTS credential_slot_revisions (
id TEXT PRIMARY KEY,
slot_id TEXT NOT NULL REFERENCES credential_slots(id) ON DELETE CASCADE,
revision INTEGER NOT NULL,
algorithm TEXT NOT NULL,
key_id TEXT NOT NULL,
key_version INTEGER NOT NULL,
nonce BLOB NOT NULL,
ciphertext BLOB NOT NULL,
aad BLOB,
created_at TEXT NOT NULL,
UNIQUE(slot_id, revision)
)
`
const sqliteRoutesSchema = `
CREATE TABLE IF NOT EXISTS routes (
id TEXT PRIMARY KEY,
principal_id TEXT NOT NULL REFERENCES principals(id) ON DELETE CASCADE,
slot_id TEXT NOT NULL REFERENCES credential_slots(id) ON DELETE CASCADE,
alias TEXT,
profile_id TEXT NOT NULL,
upstream_model TEXT NOT NULL,
resource_selector TEXT NOT NULL DEFAULT 'default',
status TEXT NOT NULL CHECK(status IN ('active','disabled','revoked')),
revision INTEGER NOT NULL DEFAULT 0,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
revoked_at TEXT,
FOREIGN KEY(principal_id, slot_id) REFERENCES credential_slots(principal_id, id) ON DELETE CASCADE
)
`
const postgresCredentialSlotsSchema = `
CREATE TABLE IF NOT EXISTS credential_slots (
id TEXT PRIMARY KEY,
principal_id TEXT NOT NULL REFERENCES principals(id) ON DELETE CASCADE,
vendor TEXT NOT NULL,
credential_kind TEXT NOT NULL,
alias TEXT,
status TEXT NOT NULL CHECK(status IN ('draft','active','disabled','revoked')),
revision BIGINT NOT NULL DEFAULT 0,
algorithm TEXT NOT NULL,
key_id TEXT NOT NULL,
key_version BIGINT NOT NULL,
nonce BYTEA NOT NULL,
ciphertext BYTEA NOT NULL,
aad BYTEA,
created_at TIMESTAMPTZ NOT NULL,
updated_at TIMESTAMPTZ NOT NULL,
revoked_at TIMESTAMPTZ,
UNIQUE(principal_id, id)
)
`
const postgresSlotRevisionsSchema = `
CREATE TABLE IF NOT EXISTS credential_slot_revisions (
id TEXT PRIMARY KEY,
slot_id TEXT NOT NULL REFERENCES credential_slots(id) ON DELETE CASCADE,
revision BIGINT NOT NULL,
algorithm TEXT NOT NULL,
key_id TEXT NOT NULL,
key_version BIGINT NOT NULL,
nonce BYTEA NOT NULL,
ciphertext BYTEA NOT NULL,
aad BYTEA,
created_at TIMESTAMPTZ NOT NULL,
UNIQUE(slot_id, revision)
)
`
const postgresRoutesSchema = `
CREATE TABLE IF NOT EXISTS routes (
id TEXT PRIMARY KEY,
principal_id TEXT NOT NULL REFERENCES principals(id) ON DELETE CASCADE,
slot_id TEXT NOT NULL REFERENCES credential_slots(id) ON DELETE CASCADE,
alias TEXT,
profile_id TEXT NOT NULL,
upstream_model TEXT NOT NULL,
resource_selector TEXT NOT NULL DEFAULT 'default',
status TEXT NOT NULL CHECK(status IN ('active','disabled','revoked')),
revision BIGINT NOT NULL DEFAULT 0,
created_at TIMESTAMPTZ NOT NULL,
updated_at TIMESTAMPTZ NOT NULL,
revoked_at TIMESTAMPTZ,
FOREIGN KEY(principal_id, slot_id) REFERENCES credential_slots(principal_id, id) ON DELETE CASCADE
)
`
const sqliteProjectionStateSchema = `
CREATE TABLE IF NOT EXISTS principal_projection_state (
singleton INTEGER PRIMARY KEY CHECK(singleton = 1),
generation INTEGER NOT NULL CHECK(generation >= 0)
)
`
const postgresProjectionStateSchema = `
CREATE TABLE IF NOT EXISTS principal_projection_state (
singleton SMALLINT PRIMARY KEY CHECK(singleton = 1),
generation BIGINT NOT NULL CHECK(generation >= 0)
)
`
// migrate runs the principal/token/slot/route DDL and any post-migration adjustments.
// It is safe to call repeatedly; every statement is idempotent.
func migrate(ctx context.Context, db *sql.DB, dialect string) error {
for _, statement := range schemaStatements(dialect) {
if _, err := db.ExecContext(ctx, statement); err != nil {
return fmt.Errorf("schema: %w", err)
}
}
if err := migrateAddRevokedAt(ctx, db, dialect); err != nil {
return fmt.Errorf("migrate add revoked_at: %w", err)
}
if err := migrateCredentialSlotContract(ctx, db, dialect); err != nil {
return fmt.Errorf("migrate credential slot contract: %w", err)
}
if err := validateMigratedCredentialBindings(ctx, db, dialect); err != nil {
return fmt.Errorf("validate migrated credential bindings: %w", err)
}
if _, err := db.ExecContext(ctx, `CREATE UNIQUE INDEX IF NOT EXISTS idx_tokens_principal_token ON tokens(principal_id, token_ref)`); err != nil {
return fmt.Errorf("migrate token_ref unique: %w", err)
}
if _, err := db.ExecContext(ctx, `CREATE UNIQUE INDEX IF NOT EXISTS idx_routes_principal_alias ON routes(principal_id, alias) WHERE alias IS NOT NULL AND alias != ''`); err != nil {
return fmt.Errorf("migrate route alias unique: %w", err)
}
if _, err := db.ExecContext(ctx, `CREATE UNIQUE INDEX IF NOT EXISTS idx_credential_slots_principal_alias ON credential_slots(principal_id, alias) WHERE alias IS NOT NULL AND alias != ''`); err != nil {
return fmt.Errorf("migrate slot alias unique: %w", err)
}
if _, err := db.ExecContext(ctx, `INSERT INTO principal_projection_state (singleton, generation) VALUES (1, 0) ON CONFLICT(singleton) DO NOTHING`); err != nil {
return fmt.Errorf("migrate projection generation: %w", err)
}
return nil
}
func schemaStatements(dialect string) []string {
principals, tokens := sqlitePrincipalsSchema, sqliteTokensSchema
slots, revisions, routes := sqliteCredentialSlotsSchema, sqliteSlotRevisionsSchema, sqliteRoutesSchema
projectionState := sqliteProjectionStateSchema
if dialect == dialectPostgres {
principals, tokens = postgresPrincipalsSchema, postgresTokensSchema
slots, revisions, routes = postgresCredentialSlotsSchema, postgresSlotRevisionsSchema, postgresRoutesSchema
projectionState = postgresProjectionStateSchema
}
return []string{
principals,
tokens,
slots,
revisions,
routes,
projectionState,
`CREATE INDEX IF NOT EXISTS idx_tokens_digest ON tokens(digest)`,
`CREATE INDEX IF NOT EXISTS idx_tokens_principal_status ON tokens(principal_id, status)`,
`CREATE INDEX IF NOT EXISTS idx_credential_slots_principal_status ON credential_slots(principal_id, status)`,
`CREATE INDEX IF NOT EXISTS idx_routes_principal_status ON routes(principal_id, status)`,
`CREATE INDEX IF NOT EXISTS idx_routes_slot_id ON routes(slot_id)`,
}
}
// migrateAddRevokedAt adds the revoked_at column if missing. The original
// schema did not include it; this migration backfills it for existing rows.
func migrateAddRevokedAt(ctx context.Context, db *sql.DB, dialect string) error {
if dialect == dialectPostgres {
_, err := db.ExecContext(ctx, `ALTER TABLE tokens ADD COLUMN IF NOT EXISTS revoked_at TIMESTAMPTZ`)
return err
}
if hasColumn(ctx, db, dialect, "tokens", "revoked_at") {
return nil
}
_, err := db.ExecContext(ctx, `ALTER TABLE tokens ADD COLUMN revoked_at TEXT`)
return err
}
// migrateCredentialSlotContract upgrades the first credential catalog schema
// to the S04 contract. Existing rows receive the deterministic bearer kind,
// while new rows can remain draft and omit an alias. SQLite needs a table
// rebuild because it cannot relax NOT NULL or replace a CHECK constraint.
func migrateCredentialSlotContract(ctx context.Context, db *sql.DB, dialect string) error {
if dialect == dialectPostgres {
hadCredentialKind := hasColumn(ctx, db, dialect, "credential_slots", "credential_kind")
if !hadCredentialKind {
if _, err := db.ExecContext(ctx, `ALTER TABLE credential_slots ADD COLUMN credential_kind TEXT NOT NULL DEFAULT 'bearer'`); err != nil {
return err
}
if _, err := db.ExecContext(ctx, `UPDATE credential_slots
SET credential_kind = CASE
WHEN lower(trim(vendor)) IN ('anthropic', 'mimo_messages', 'seulgi_messages', 'seulgivibe_claude') THEN 'api_key'
ELSE 'bearer'
END`); err != nil {
return err
}
}
if _, err := db.ExecContext(ctx, `ALTER TABLE credential_slots ALTER COLUMN alias DROP NOT NULL`); err != nil {
return err
}
_, err := db.ExecContext(ctx, `
DO $$
BEGIN
IF EXISTS (SELECT 1 FROM pg_constraint WHERE conname = 'credential_slots_status_check') THEN
ALTER TABLE credential_slots DROP CONSTRAINT credential_slots_status_check;
END IF;
IF NOT EXISTS (SELECT 1 FROM pg_constraint WHERE conname = 'credential_slots_status_check') THEN
ALTER TABLE credential_slots ADD CONSTRAINT credential_slots_status_check
CHECK (status IN ('draft','active','disabled','revoked'));
END IF;
END $$`)
return err
}
if hasColumn(ctx, db, dialect, "credential_slots", "credential_kind") && sqliteSlotSchemaIsCurrent(ctx, db) {
return nil
}
return rebuildSQLiteCredentialCatalog(ctx, db)
}
func sqliteSlotSchemaIsCurrent(ctx context.Context, db *sql.DB) bool {
var sqlText string
if err := db.QueryRowContext(ctx, `SELECT sql FROM sqlite_master WHERE type='table' AND name='credential_slots'`).Scan(&sqlText); err != nil {
return false
}
return strings.Contains(strings.ToLower(sqlText), "'draft'")
}
func rebuildSQLiteCredentialCatalog(ctx context.Context, db *sql.DB) (err error) {
conn, err := db.Conn(ctx)
if err != nil {
return err
}
defer conn.Close()
if _, err := conn.ExecContext(ctx, `PRAGMA foreign_keys = OFF`); err != nil {
return err
}
defer func() {
_, restoreErr := conn.ExecContext(ctx, `PRAGMA foreign_keys = ON`)
if err == nil && restoreErr != nil {
err = restoreErr
}
}()
tx, err := conn.BeginTx(ctx, nil)
if err != nil {
return err
}
defer func() {
if err != nil {
_ = tx.Rollback()
}
}()
for _, statement := range []string{
`ALTER TABLE routes RENAME TO routes_legacy_credential_catalog`,
`ALTER TABLE credential_slot_revisions RENAME TO credential_slot_revisions_legacy_credential_catalog`,
`ALTER TABLE credential_slots RENAME TO credential_slots_legacy_credential_catalog`,
sqliteCredentialSlotsSchema,
sqliteSlotRevisionsSchema,
sqliteRoutesSchema,
`INSERT INTO credential_slots (id, principal_id, vendor, credential_kind, alias, status, revision, algorithm, key_id, key_version, nonce, ciphertext, aad, created_at, updated_at, revoked_at)
SELECT id, principal_id, vendor,
CASE WHEN lower(trim(vendor)) IN ('anthropic', 'mimo_messages', 'seulgi_messages', 'seulgivibe_claude') THEN 'api_key' ELSE 'bearer' END,
alias, status, revision, algorithm, key_id, key_version, nonce, ciphertext, aad, created_at, updated_at, revoked_at
FROM credential_slots_legacy_credential_catalog`,
`INSERT INTO credential_slot_revisions (id, slot_id, revision, algorithm, key_id, key_version, nonce, ciphertext, aad, created_at)
SELECT id, slot_id, revision, algorithm, key_id, key_version, nonce, ciphertext, aad, created_at
FROM credential_slot_revisions_legacy_credential_catalog`,
`INSERT INTO routes (id, principal_id, slot_id, alias, profile_id, upstream_model, resource_selector, status, revision, created_at, updated_at, revoked_at)
SELECT id, principal_id, slot_id, alias, profile_id, upstream_model, resource_selector, status, revision, created_at, updated_at, revoked_at
FROM routes_legacy_credential_catalog`,
`DROP TABLE routes_legacy_credential_catalog`,
`DROP TABLE credential_slot_revisions_legacy_credential_catalog`,
`DROP TABLE credential_slots_legacy_credential_catalog`,
} {
if _, err = tx.ExecContext(ctx, statement); err != nil {
return err
}
}
err = tx.Commit()
return err
}
// validateMigratedCredentialBindings fences legacy active state using the
// same closed resolver used by route mutations. It deliberately reports only
// typed, redacted compatibility failures: database rows can contain encrypted
// credential material but no migration error may expose it.
func validateMigratedCredentialBindings(ctx context.Context, db *sql.DB, dialect string) error {
activeRoutes, err := db.QueryContext(ctx, bindQuery(dialect, `
SELECT r.principal_id, r.slot_id, r.profile_id,
s.id, s.principal_id, s.vendor, s.credential_kind, s.alias, s.status
FROM routes r
LEFT JOIN credential_slots s ON s.id = r.slot_id
WHERE r.status = ?
`), StatusActive)
if err != nil {
return err
}
defer activeRoutes.Close()
for activeRoutes.Next() {
var routePrincipalID string
var routeSlotID string
var profileID string
var joinedSlotID, joinedPrincipalID, joinedVendor, joinedKind, slotAlias, joinedStatus sql.NullString
if err := activeRoutes.Scan(
&routePrincipalID, &routeSlotID, &profileID,
&joinedSlotID, &joinedPrincipalID, &joinedVendor, &joinedKind, &slotAlias, &joinedStatus,
); err != nil {
return err
}
if !joinedSlotID.Valid {
return fmt.Errorf("%w: active migrated route has no slot", ErrIncompatibleProfile)
}
slot := CredentialSlotRecord{
ID: joinedSlotID.String,
PrincipalID: joinedPrincipalID.String,
Vendor: joinedVendor.String,
CredentialKind: joinedKind.String,
Status: joinedStatus.String,
}
if slotAlias.Valid {
slot.Alias = slotAlias.String
}
if routePrincipalID != slot.PrincipalID {
return fmt.Errorf("%w: active migrated route has mismatched ownership", ErrIncompatibleProfile)
}
switch slot.Status {
case StatusActive, StatusDisabled, StatusRevoked:
// Disabled and revoked slots retain their bindings so a disabled slot
// can be re-enabled without recreating routes. They are not serving-ready,
// but their retained active bindings must still be compatible below.
default:
return fmt.Errorf("%w: active migrated route has an unavailable slot", ErrIncompatibleProfile)
}
if _, err := resolveSlotProfile(slot, profileID); err != nil {
return fmt.Errorf("active migrated route compatibility: %w", err)
}
}
if err := activeRoutes.Err(); err != nil {
return err
}
activeSlots, err := db.QueryContext(ctx, bindQuery(dialect, `
SELECT id, principal_id, vendor, credential_kind, alias, status
FROM credential_slots
WHERE status = ?
`), StatusActive)
if err != nil {
return err
}
defer activeSlots.Close()
for activeSlots.Next() {
var slot CredentialSlotRecord
var slotAlias sql.NullString
if err := activeSlots.Scan(&slot.ID, &slot.PrincipalID, &slot.Vendor, &slot.CredentialKind, &slotAlias, &slot.Status); err != nil {
return err
}
if slotAlias.Valid {
slot.Alias = slotAlias.String
}
if err := migratedSlotHasCompatibleBinding(ctx, db, dialect, slot); err != nil {
return err
}
}
return activeSlots.Err()
}
func migratedSlotHasCompatibleBinding(ctx context.Context, db *sql.DB, dialect string, slot CredentialSlotRecord) error {
rows, err := db.QueryContext(ctx, bindQuery(dialect, `
SELECT profile_id
FROM routes
WHERE principal_id = ? AND slot_id = ? AND status <> ?
`), slot.PrincipalID, slot.ID, StatusRevoked)
if err != nil {
return err
}
defer rows.Close()
for rows.Next() {
var profileID string
if err := rows.Scan(&profileID); err != nil {
return err
}
if _, err := resolveSlotProfile(slot, profileID); err == nil {
return nil
}
}
if err := rows.Err(); err != nil {
return err
}
return fmt.Errorf("%w: active migrated slot has no compatible binding", ErrIncompatibleProfile)
}
// hasColumn reports whether table has a column named col.
// Uses PRAGMA table_info for SQLite compatibility and information_schema for PostgreSQL.
func hasColumn(ctx context.Context, db *sql.DB, dialect, table, col string) bool {
if dialect == dialectPostgres {
var name string
err := db.QueryRowContext(ctx,
`SELECT column_name FROM information_schema.columns WHERE table_name=$1 AND column_name=$2`, table, col,
).Scan(&name)
return err == nil
}
rows, err := db.QueryContext(ctx, `PRAGMA table_info(`+table+`)`)
if err == nil {
defer rows.Close()
for rows.Next() {
var cid int
var name, dtype string
var notnull, pk int
var dfltValue interface{}
if err := rows.Scan(&cid, &name, &dtype, &notnull, &dfltValue, &pk); err != nil {
return false
}
if name == col {
return true
}
}
return false
}
return false
}