HH-800 / HH-804: unify account schema and lifecycle (#18)

This commit is contained in:
2026-08-28 23:14:52 +08:00
parent 199f7b348a
commit 884505bb55
10 changed files with 766 additions and 122 deletions
+24 -18
View File
@@ -70,24 +70,30 @@ func TestLoadConfigRejectsInvalidDatabase(t *testing.T) {
}
}
func TestPhaseARequestRejectsUnknownFields(t *testing.T) {
app := fiber.New()
app.Post("/", func(c fiber.Ctx) error {
var input struct {
ID string `json:"id"`
}
if err := decodePhaseA(c, &input); err != nil {
return phaseAError(c, err)
}
return c.SendStatus(http.StatusNoContent)
})
response, err := app.Test(httptest.NewRequest(http.MethodPost, "/", strings.NewReader(`{"id":"account-a","password":"must-not-be-accepted"}`)))
if err != nil {
t.Fatal(err)
}
defer response.Body.Close()
if response.StatusCode != http.StatusBadRequest {
t.Fatalf("expected unknown sensitive field to be rejected, got %d", response.StatusCode)
func TestPhaseAAccountRequestRejectsSecretsAndUnknownFields(t *testing.T) {
for name, body := range map[string]string{
"raw password": `{"id":"account-a","platform":"mock","platform_account_key":"a","authorization_kind":"owned","credential_reference":{"id":"credential-a","provider":"os_keyring","key":"creatorhub/account-a"},"password":"secret"}`,
"nested secret": `{"id":"account-a","platform":"mock","platform_account_key":"a","authorization_kind":"owned","credential_reference":{"id":"credential-a","provider":"os_keyring","key":"creatorhub/account-a","secret":"secret"}}`,
"unknown field": `{"id":"account-a","platform":"mock","platform_account_key":"a","authorization_kind":"owned","credential_reference":{"id":"credential-a","provider":"os_keyring","key":"creatorhub/account-a"},"extra":true}`,
} {
t.Run(name, func(t *testing.T) {
app := fiber.New()
app.Post("/", func(c fiber.Ctx) error {
var input accountRequest
if err := decodePhaseA(c, &input); err != nil {
return phaseAError(c, err)
}
return c.SendStatus(http.StatusNoContent)
})
response, err := app.Test(httptest.NewRequest(http.MethodPost, "/", strings.NewReader(body)))
if err != nil {
t.Fatal(err)
}
defer response.Body.Close()
if response.StatusCode != http.StatusBadRequest {
t.Fatalf("expected sensitive or unknown field to be rejected, got %d", response.StatusCode)
}
})
}
}
+40 -6
View File
@@ -12,7 +12,9 @@ import (
type accountRequest struct {
ID string `json:"id"`
ProfileID string `json:"profile_id"`
Platform string `json:"platform"`
PlatformAccountKey string `json:"platform_account_key"`
AuthorizationKind string `json:"authorization_kind"`
CredentialReference struct {
ID string `json:"id"`
Provider string `json:"provider"`
@@ -26,14 +28,32 @@ func registerPhaseA(app *fiber.App, store *phasea.Store) {
if err := decodePhaseA(c, &input); err != nil {
return phaseAError(c, err)
}
err := store.CreateAccount(c.Context(), phasea.Account{
ID: input.ID, ProfileID: input.ProfileID, CredentialReferenceID: input.CredentialReference.ID,
CredentialProvider: input.CredentialReference.Provider, CredentialKey: input.CredentialReference.Key,
})
account := phasea.Account{
ID: input.ID, Platform: input.Platform, PlatformAccountKey: input.PlatformAccountKey,
AuthorizationKind: input.AuthorizationKind, CredentialKey: input.CredentialReference.Key,
CredentialReference: phasea.CredentialReference{ID: input.CredentialReference.ID, Provider: input.CredentialReference.Provider},
AuthorizationStatus: "authorized", RuntimeStatus: "paused", Version: 1,
}
if err := store.CreateAccount(c.Context(), account); err != nil {
return phaseAError(c, err)
}
return c.Status(fiber.StatusCreated).JSON(account)
})
app.Get("/api/phase-a/accounts", func(c fiber.Ctx) error {
accounts, err := store.ListAccounts(c.Context())
if err != nil {
return phaseAError(c, err)
}
return c.Status(fiber.StatusCreated).JSON(map[string]string{"id": input.ID, "profile_id": input.ProfileID})
return c.JSON(accounts)
})
app.Get("/api/phase-a/accounts/:id", func(c fiber.Ctx) error {
account, err := store.GetAccount(c.Context(), c.Params("id"))
if err != nil {
return phaseAError(c, err)
}
return c.JSON(account)
})
app.Post("/api/phase-a/accounts/:id/pause", func(c fiber.Ctx) error {
@@ -43,6 +63,20 @@ func registerPhaseA(app *fiber.App, store *phasea.Store) {
return c.SendStatus(fiber.StatusNoContent)
})
app.Post("/api/phase-a/accounts/:id/resume", func(c fiber.Ctx) error {
if err := store.ResumeAccount(c.Context(), c.Params("id")); err != nil {
return phaseAError(c, err)
}
return c.SendStatus(fiber.StatusNoContent)
})
app.Post("/api/phase-a/accounts/:id/revoke", func(c fiber.Ctx) error {
if err := store.RevokeAccount(c.Context(), c.Params("id")); err != nil {
return phaseAError(c, err)
}
return c.SendStatus(fiber.StatusNoContent)
})
app.Post("/api/phase-a/runtimes", func(c fiber.Ctx) error {
var input phasea.RuntimeBinding
if err := decodePhaseA(c, &input); err != nil {
+2 -2
View File
@@ -55,8 +55,8 @@ DOCKER_GID=$(stat -c %g /var/run/docker.sock) docker compose up --build
## 阶段 A 离线闭环
`POST /api/phase-a/accounts` 只接受 OS Keyring/Secret Manager 的引用标识,不接受秘密值;账号的 `profile_id` 全局唯一。`POST /api/phase-a/runtimes` 通过部分唯一索引保证一个账号和一个运行时都只有一条活动绑定。
`POST /api/phase-a/accounts` 接受 `{id, platform, platform_account_key, authorization_kind, credential_reference}`,只允许 OS Keyring/Secret Manager 引用,不接受秘密值;新账号默认 `paused`,`(platform, platform_account_key)` 全局唯一。`GET /api/phase-a/accounts[/:id]` 不返回引用 key;pause/revoke 会递增账号版本并将 queued 任务置为 `policy_hold`,只有具备 binding 和 healthy 出口的未撤销账号才能 resume。账号与浏览器环境通过一对一 `environment_binding` 关联,出口可复用;运行实例保留历史,并以 binding 和外部 runtime id 的部分唯一索引限制活动实例。
草稿经 `POST /api/phase-a/confirmations` 显式确认后才可投递到 `/api/phase-a/tasks`。任务由幂等键去重;`POST /api/phase-a/mock/execute` 使用 `FOR UPDATE SKIP LOCKED` 领取一分钟租约,执行前统一核对账号、草稿和确认版本。缺少确认或版本不一致会进入 `needs_confirmation`,暂停账号或 Mock 策略结果会进入 `policy_hold`,不确定结果与过期租约进入 `needs_confirmation`;这些状态都不会自动重试。`GET /api/phase-a/audit` 只导出账号、确认版本、尝试和结果等非秘密证据。
启动时控制面在事务和 advisory lock 下应用前向迁移 `internal/phasea/migrations/001_phase_a.sql`。本迁移只新建表、索引、约束和追加式审计触发器,不删除或改写现有数据;回滚需停服务后人工删除阶段 A 新表,本阶段不提供自动破坏性回滚。
启动时控制面先应用 Phase A v1,再由 Hub runner 顺序应用 v2、v3;每一步都在事务和 advisory lock 下前向执行。v3 保留旧表、列和历史记录,旧账号回填为 `platform=mock` 并暂停,仅账号 ID 与环境 alias 相同的记录自动建立 binding;其余记录等待显式绑定。本阶段不提供破坏性自动回滚。
+5 -5
View File
@@ -27,10 +27,10 @@ type Fingerprint struct {
}
var (
platforms = map[string]bool{"windows": true, "linux": true, "macos": true}
brands = map[string]bool{"Chrome": true, "Edge": true, "Opera": true, "Vivaldi": true}
spoofings = map[string]bool{"font": true, "audio": true, "canvas": true, "clientrects": true, "gpu": true}
langPattern = regexp.MustCompile(`^[A-Za-z]{1,8}(-[A-Za-z0-9]{1,8})?$`)
platforms = map[string]bool{"windows": true, "linux": true, "macos": true}
brands = map[string]bool{"Chrome": true, "Edge": true, "Opera": true, "Vivaldi": true}
spoofings = map[string]bool{"font": true, "audio": true, "canvas": true, "clientrects": true, "gpu": true}
langPattern = regexp.MustCompile(`^[A-Za-z]{1,8}(-[A-Za-z0-9]{1,8})?$`)
acceptLangPattern = regexp.MustCompile(
`^[A-Za-z]{1,8}(-[A-Za-z0-9]{1,8})?(, ?[A-Za-z]{1,8}(-[A-Za-z0-9]{1,8})*){0,7}$`)
timezonePattern = regexp.MustCompile(`^[A-Za-z0-9_+\-/]{1,64}$`)
@@ -111,7 +111,7 @@ func optionalMatch(value string, pattern *regexp.Regexp) bool {
func validProxyURL(raw string) bool {
parsed, err := url.Parse(raw)
if err != nil || parsed.Host == "" {
if err != nil || parsed.Host == "" || parsed.User != nil {
return false
}
switch parsed.Scheme {
+215
View File
@@ -0,0 +1,215 @@
package hub
import (
"context"
"database/sql"
"fmt"
"net/url"
"os"
"testing"
"time"
"git.ipao.vip/rogee/creator-hub/internal/phasea"
)
func TestUnifiedAccountMigration(t *testing.T) {
databaseURL := os.Getenv("CREATORHUB_POSTGRES_TEST_URL")
if databaseURL == "" {
t.Skip("set CREATORHUB_POSTGRES_TEST_URL to run PostgreSQL integration coverage")
}
t.Run("fresh database", func(t *testing.T) {
ctx := context.Background()
testURL := isolatedDatabaseURL(t, databaseURL)
store := openFullyMigratedHub(t, ctx, testURL)
store.Close()
db, err := sql.Open("pgx", testURL)
if err != nil {
t.Fatal(err)
}
defer db.Close()
assertDatabaseCount(t, db, `SELECT count(*) FROM schema_migration WHERE version IN (1, 2, 3)`, 3)
assertDatabaseCount(t, db, `SELECT count(*) FROM information_schema.tables WHERE table_schema = current_schema() AND table_name IN ('social_account', 'browser_env', 'network_exit', 'environment_binding')`, 4)
store = openFullyMigratedHub(t, ctx, testURL)
store.Close()
assertDatabaseCount(t, db, `SELECT count(*) FROM schema_migration WHERE version IN (1, 2, 3)`, 3)
})
t.Run("v1 and v2 data", func(t *testing.T) {
ctx := context.Background()
testURL := isolatedDatabaseURL(t, databaseURL)
phaseAStore, err := phasea.Open(ctx, testURL)
if err != nil {
t.Fatal(err)
}
phaseAStore.Close()
db, err := sql.Open("pgx", testURL)
if err != nil {
t.Fatal(err)
}
defer db.Close()
if _, err := db.Exec(migration002); err != nil {
t.Fatal(err)
}
if _, err := db.Exec(`
INSERT INTO schema_migration (version) VALUES (2);
ALTER TABLE browser_image DROP CONSTRAINT browser_image_image_ref_check;
INSERT INTO credential_reference (id, provider, reference_key) VALUES
('credential-mapped', 'os_keyring', 'creatorhub/mapped'),
('credential-unbound', 'secret_manager', 'creatorhub/unbound');
INSERT INTO social_account (id, credential_reference_id, profile_id, status) VALUES
('mapped', 'credential-mapped', 'legacy-profile-mapped', 'active'),
('unbound', 'credential-unbound', 'legacy-profile-unbound', 'active');
INSERT INTO content_draft (id, account_id, version, content) VALUES ('legacy-draft', 'mapped', 1, 'legacy');
INSERT INTO confirmation (id, account_id, account_version, draft_id, draft_version, version)
VALUES ('legacy-confirmation', 'mapped', 1, 'legacy-draft', 1, 1);
INSERT INTO operation_task
(id, idempotency_key, account_id, account_version, draft_id, draft_version, confirmation_id, confirmation_version)
VALUES ('legacy-task', 'legacy-task-key', 'mapped', 1, 'legacy-draft', 1, 'legacy-confirmation', 1);
INSERT INTO gateway (name, endpoint, token) VALUES ('legacy-gateway', 'http://127.0.0.1:8081', 'legacy-gateway-token');
INSERT INTO browser_image (version, image_ref) VALUES ('1', 'example/browser:1');
INSERT INTO browser_env (alias, name, gateway_name, image_version, fingerprint) VALUES
('mapped', 'Mapped', 'legacy-gateway', '1', '{"seed":1}'),
('orphan-env', 'Orphan', 'legacy-gateway', '1', '{"seed":2}');
INSERT INTO runtime_instance (id, account_id, runtime_id, lease_until) VALUES
('instance-mapped', 'mapped', 'runtime-mapped', now() + interval '1 hour'),
('instance-unbound', 'unbound', 'runtime-unbound', now() + interval '1 hour');
INSERT INTO audit_event (event_type, account_id) VALUES ('legacy_event', 'mapped')`); err != nil {
t.Fatal(err)
}
store, err := Open(ctx, testURL)
if err != nil {
t.Fatal(err)
}
store.Close()
assertDatabaseCount(t, db, `SELECT count(*) FROM social_account WHERE platform = 'mock' AND platform_account_key = id AND status = 'paused'`, 2)
assertDatabaseCount(t, db, `SELECT count(*) FROM social_account WHERE profile_id LIKE 'legacy-profile-%'`, 2)
assertDatabaseCount(t, db, `SELECT count(*) FROM environment_binding WHERE account_id = 'mapped' AND browser_env_alias = 'mapped' AND network_exit_id IS NULL`, 1)
assertDatabaseCount(t, db, `SELECT count(*) FROM environment_binding WHERE account_id = 'unbound'`, 0)
assertDatabaseCount(t, db, `SELECT count(*) FROM runtime_instance WHERE id = 'instance-mapped' AND binding_id = 'mapped'`, 1)
assertDatabaseCount(t, db, `SELECT count(*) FROM runtime_instance WHERE id = 'instance-unbound' AND binding_id IS NULL`, 1)
assertDatabaseCount(t, db, `SELECT count(*) FROM audit_event WHERE event_type = 'legacy_event'`, 1)
assertDatabaseCount(t, db, `SELECT count(*) FROM operation_task WHERE id = 'legacy-task' AND state = 'policy_hold'`, 1)
if _, err := db.Exec(`
INSERT INTO credential_reference (id, provider, reference_key) VALUES ('credential-duplicate', 'os_keyring', 'creatorhub/duplicate');
INSERT INTO social_account
(id, credential_reference_id, platform, platform_account_key, authorization_kind, authorization_status)
VALUES ('duplicate', 'credential-duplicate', 'mock', 'mapped', 'owned', 'authorized')`); err == nil {
t.Fatal("duplicate platform account must fail")
}
if _, err := db.Exec(`
INSERT INTO network_exit (id, protocol, host, port, health_status) VALUES ('shared-exit', 'socks5', '127.0.0.1', 1080, 'healthy');
UPDATE environment_binding SET network_exit_id = 'shared-exit' WHERE id = 'mapped';
INSERT INTO environment_binding (id, account_id, browser_env_alias, network_exit_id)
VALUES ('unbound-binding', 'unbound', 'orphan-env', 'shared-exit')`); err != nil {
t.Fatal(err)
}
assertDatabaseCount(t, db, `SELECT count(*) FROM environment_binding WHERE network_exit_id = 'shared-exit'`, 2)
if _, err := db.Exec(`
INSERT INTO browser_env (alias, name, gateway_name, image_version, fingerprint)
VALUES ('extra-env', 'Extra', 'legacy-gateway', '1', '{"seed":3}');
INSERT INTO environment_binding (id, account_id, browser_env_alias)
VALUES ('duplicate-account-binding', 'mapped', 'extra-env')`); err == nil {
t.Fatal("one account must not bind two environments")
}
if _, err := db.Exec(`
INSERT INTO credential_reference (id, provider, reference_key) VALUES ('credential-extra', 'os_keyring', 'creatorhub/extra');
INSERT INTO social_account
(id, credential_reference_id, platform, platform_account_key, authorization_kind, authorization_status)
VALUES ('extra-account', 'credential-extra', 'mock', 'extra-account', 'owned', 'authorized');
INSERT INTO environment_binding (id, account_id, browser_env_alias)
VALUES ('duplicate-env-binding', 'extra-account', 'mapped')`); err == nil {
t.Fatal("one environment must not bind two accounts")
}
if _, err := db.Exec(`
INSERT INTO runtime_instance (id, account_id, binding_id, runtime_id, lease_until)
VALUES ('instance-conflict', 'mapped', 'mapped', 'runtime-new', now() + interval '1 hour')`); err == nil {
t.Fatal("one binding must not have two active runtime instances")
}
if _, err := db.Exec(`
INSERT INTO browser_image (version, image_ref) VALUES ('2', 'example/browser:2');
UPDATE social_account SET status = 'active' WHERE id = 'mapped';
INSERT INTO content_draft (id, account_id, version, content) VALUES ('upgrade-draft', 'mapped', 1, 'test');
INSERT INTO confirmation (id, account_id, account_version, draft_id, draft_version, version)
VALUES ('upgrade-confirmation', 'mapped', 1, 'upgrade-draft', 1, 1);
INSERT INTO operation_task
(id, idempotency_key, account_id, account_version, draft_id, draft_version, confirmation_id, confirmation_version)
VALUES ('upgrade-task', 'upgrade-task-key', 'mapped', 1, 'upgrade-draft', 1, 'upgrade-confirmation', 1)`); err != nil {
t.Fatal(err)
}
store, err = Open(ctx, testURL)
if err != nil {
t.Fatal(err)
}
if err := store.UpgradeEnv(ctx, "mapped", "2"); err != nil {
t.Fatal(err)
}
store.Close()
assertDatabaseCount(t, db, `SELECT count(*) FROM browser_env WHERE alias = 'mapped' AND version = 2 AND image_version = '2'`, 1)
assertDatabaseCount(t, db, `SELECT count(*) FROM social_account WHERE id = 'mapped' AND version = 2 AND status = 'paused'`, 1)
assertDatabaseCount(t, db, `SELECT count(*) FROM operation_task WHERE id = 'upgrade-task' AND state = 'policy_hold'`, 1)
store, err = Open(ctx, testURL)
if err != nil {
t.Fatal(err)
}
store.Close()
assertDatabaseCount(t, db, `SELECT count(*) FROM schema_migration WHERE version = 3`, 1)
})
}
func openFullyMigratedHub(t *testing.T, ctx context.Context, databaseURL string) *Store {
t.Helper()
phaseAStore, err := phasea.Open(ctx, databaseURL)
if err != nil {
t.Fatal(err)
}
phaseAStore.Close()
store, err := Open(ctx, databaseURL)
if err != nil {
t.Fatal(err)
}
return store
}
func isolatedDatabaseURL(t *testing.T, databaseURL string) string {
t.Helper()
admin, err := sql.Open("pgx", databaseURL)
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { admin.Close() })
schema := fmt.Sprintf("creatorhub_hh804_%d", time.Now().UnixNano())
if _, err := admin.Exec("CREATE SCHEMA " + schema); err != nil {
t.Fatal(err)
}
t.Cleanup(func() {
if _, err := admin.Exec("DROP SCHEMA " + schema + " CASCADE"); err != nil {
t.Errorf("drop test schema: %v", err)
}
})
parsed, err := url.Parse(databaseURL)
if err != nil {
t.Fatal(err)
}
query := parsed.Query()
query.Set("search_path", schema)
parsed.RawQuery = query.Encode()
return parsed.String()
}
func assertDatabaseCount(t *testing.T, db *sql.DB, query string, want int) {
t.Helper()
var got int
if err := db.QueryRow(query).Scan(&got); err != nil || got != want {
t.Fatalf("count mismatch: got=%d want=%d err=%v query=%s", got, want, err, query)
}
}
@@ -0,0 +1,97 @@
ALTER TABLE social_account
ADD COLUMN platform text,
ADD COLUMN platform_account_key text,
ADD COLUMN authorization_kind text,
ADD COLUMN authorization_status text,
ADD COLUMN revoked_at timestamptz;
UPDATE social_account
SET platform = 'mock',
platform_account_key = id,
authorization_kind = 'owned',
authorization_status = 'authorized',
status = 'paused',
paused_at = COALESCE(paused_at, now());
UPDATE operation_task
SET state = 'policy_hold', updated_at = now()
WHERE state = 'queued';
ALTER TABLE social_account
ALTER COLUMN platform SET NOT NULL,
ALTER COLUMN platform_account_key SET NOT NULL,
ALTER COLUMN authorization_kind SET NOT NULL,
ALTER COLUMN authorization_status SET NOT NULL,
ALTER COLUMN status SET DEFAULT 'paused',
ALTER COLUMN profile_id DROP NOT NULL,
ADD CONSTRAINT social_account_platform_check
CHECK (platform ~ '^[a-z0-9][a-z0-9-]{0,31}$'),
ADD CONSTRAINT social_account_platform_key_check
CHECK (platform_account_key ~ '^[A-Za-z0-9][A-Za-z0-9._:@/-]{0,127}$'),
ADD CONSTRAINT social_account_authorization_kind_check
CHECK (authorization_kind IN ('owned', 'authorized')),
ADD CONSTRAINT social_account_authorization_status_check
CHECK (authorization_status IN ('authorized', 'revoked')),
ADD CONSTRAINT social_account_platform_key_unique
UNIQUE (platform, platform_account_key);
ALTER TABLE browser_env
ADD COLUMN version bigint NOT NULL DEFAULT 1 CHECK (version > 0);
ALTER TABLE browser_image
DROP CONSTRAINT IF EXISTS browser_image_image_ref_check,
ADD CONSTRAINT browser_image_image_ref_check
CHECK (length(image_ref) <= 301 AND image_ref ~ '^[A-Za-z0-9][A-Za-z0-9._:/@-]*$');
CREATE TABLE network_exit (
id text PRIMARY KEY CHECK (id ~ '^[A-Za-z0-9][A-Za-z0-9._/-]{0,127}$'),
protocol text NOT NULL CHECK (protocol IN ('http', 'https', 'socks4', 'socks5')),
host text NOT NULL CHECK (host <> '' AND host !~ '[[:space:]@]' AND length(host) <= 253),
port integer NOT NULL CHECK (port BETWEEN 1 AND 65535),
credential_reference_id text REFERENCES credential_reference(id),
expected_public_ip inet,
expected_region text NOT NULL DEFAULT '' CHECK (length(expected_region) <= 64),
observed_public_ip inet,
observed_region text NOT NULL DEFAULT '' CHECK (length(observed_region) <= 64),
health_status text NOT NULL DEFAULT 'unchecked'
CHECK (health_status IN ('unchecked', 'healthy', 'unhealthy', 'disabled')),
version bigint NOT NULL DEFAULT 1 CHECK (version > 0),
last_checked_at timestamptz,
created_at timestamptz NOT NULL DEFAULT now(),
updated_at timestamptz NOT NULL DEFAULT now()
);
CREATE TABLE environment_binding (
id text PRIMARY KEY CHECK (id ~ '^[A-Za-z0-9][A-Za-z0-9._/-]{0,127}$'),
account_id text NOT NULL UNIQUE REFERENCES social_account(id),
browser_env_alias text NOT NULL UNIQUE REFERENCES browser_env(alias),
network_exit_id text REFERENCES network_exit(id),
version bigint NOT NULL DEFAULT 1 CHECK (version > 0),
created_at timestamptz NOT NULL DEFAULT now(),
updated_at timestamptz NOT NULL DEFAULT now()
);
INSERT INTO environment_binding (id, account_id, browser_env_alias)
SELECT account.id, account.id, environment.alias
FROM social_account account
JOIN browser_env environment ON environment.alias = account.id;
ALTER TABLE runtime_instance
ADD COLUMN binding_id text REFERENCES environment_binding(id);
UPDATE runtime_instance runtime
SET binding_id = binding.id
FROM environment_binding binding
WHERE binding.account_id = runtime.account_id;
CREATE UNIQUE INDEX one_active_runtime_per_binding
ON runtime_instance (binding_id)
WHERE binding_id IS NOT NULL AND released_at IS NULL;
ALTER TABLE audit_event
ADD COLUMN browser_env_alias text REFERENCES browser_env(alias),
ADD COLUMN network_exit_id text REFERENCES network_exit(id),
ADD COLUMN runtime_instance_id text REFERENCES runtime_instance(id),
ADD COLUMN binding_version bigint CHECK (binding_version > 0),
ADD COLUMN actor text,
ADD COLUMN reason_code text;
+45 -12
View File
@@ -22,6 +22,9 @@ import (
//go:embed migrations/002_hub.sql
var migration002 string
//go:embed migrations/003_unified_accounts.sql
var migration003 string
var (
ErrConflict = errors.New("resource conflicts with existing state")
ErrInvalid = errors.New("invalid hub input")
@@ -101,16 +104,22 @@ func (s *Store) migrate(ctx context.Context) error {
if _, err := tx.ExecContext(ctx, `CREATE TABLE IF NOT EXISTS schema_migration (version integer PRIMARY KEY, applied_at timestamptz NOT NULL DEFAULT now())`); err != nil {
return errors.New("create schema migration table")
}
var applied bool
if err := tx.QueryRowContext(ctx, `SELECT EXISTS (SELECT 1 FROM schema_migration WHERE version = 2)`).Scan(&applied); err != nil {
return errors.New("read hub schema migration state")
}
if !applied {
if _, err := tx.ExecContext(ctx, migration002); err != nil {
return fmt.Errorf("apply hub schema migration 2: %w", err)
for _, migration := range []struct {
version int
sql string
}{{2, migration002}, {3, migration003}} {
var applied bool
if err := tx.QueryRowContext(ctx, `SELECT EXISTS (SELECT 1 FROM schema_migration WHERE version = $1)`, migration.version).Scan(&applied); err != nil {
return errors.New("read hub schema migration state")
}
if _, err := tx.ExecContext(ctx, `INSERT INTO schema_migration (version) VALUES (2)`); err != nil {
return errors.New("record hub schema migration 2")
if applied {
continue
}
if _, err := tx.ExecContext(ctx, migration.sql); err != nil {
return fmt.Errorf("apply hub schema migration %d: %w", migration.version, err)
}
if _, err := tx.ExecContext(ctx, `INSERT INTO schema_migration (version) VALUES ($1)`, migration.version); err != nil {
return fmt.Errorf("record hub schema migration %d", migration.version)
}
}
if err := tx.Commit(); err != nil {
@@ -327,14 +336,31 @@ func (s *Store) UpgradeEnv(ctx context.Context, alias, version string) error {
if !aliasPattern.MatchString(alias) || !imageVersionPattern.MatchString(version) {
return ErrInvalid
}
tx, err := s.db.BeginTx(ctx, nil)
if err != nil {
return errors.New("begin environment upgrade")
}
defer tx.Rollback()
var updated string
if err := s.db.QueryRowContext(ctx, `
UPDATE browser_env SET image_version = $2
if err := tx.QueryRowContext(ctx, `
UPDATE browser_env SET image_version = $2, version = version + 1
WHERE alias = $1 AND EXISTS (SELECT 1 FROM browser_image WHERE version = $2 AND enabled)
RETURNING alias`, alias, version).Scan(&updated); err != nil {
return rowError(err)
}
return nil
if _, err := tx.ExecContext(ctx, `
WITH changed AS (
UPDATE social_account account
SET status = 'paused', paused_at = now(), version = account.version + 1, updated_at = now()
FROM environment_binding binding
WHERE binding.browser_env_alias = $1 AND binding.account_id = account.id
RETURNING account.id
)
UPDATE operation_task task SET state = 'policy_hold', updated_at = now()
FROM changed WHERE task.account_id = changed.id AND task.state = 'queued'`, alias); err != nil {
return errors.New("invalidate upgraded environment account")
}
return commitHub(tx)
}
func (s *Store) DeleteEnv(ctx context.Context, alias string) error {
@@ -385,6 +411,13 @@ func newToken() string {
return hex.EncodeToString(value[:])
}
func commitHub(tx *sql.Tx) error {
if err := tx.Commit(); err != nil {
return errors.New("commit hub transaction")
}
return nil
}
func rowError(err error) error {
if errors.Is(err, sql.ErrNoRows) {
return ErrNotFound
+20 -21
View File
@@ -58,20 +58,21 @@ func TestFingerprintArgsFollowUpstreamCommandLineContract(t *testing.T) {
func TestFingerprintValidateRejectsUnsupportedValues(t *testing.T) {
invalid := map[string]Fingerprint{
"seed zero": {Seed: 0},
"seed overflow": {Seed: 2147483648},
"platform": {Seed: 1, Platform: "android"},
"brand": {Seed: 1, Brand: "Firefox"},
"platform version": {Seed: 1, PlatformVersion: "bad value"},
"brand version": {Seed: 1, BrandVersion: strings.Repeat("x", 33)},
"concurrency": {Seed: 1, HardwareConcurrency: 129},
"lang": {Seed: 1, Lang: "zh CN"},
"accept lang": {Seed: 1, AcceptLang: "zh-CN;drop"},
"timezone": {Seed: 1, Timezone: "Asia/Shanghai\n"},
"proxy scheme": {Seed: 1, ProxyServer: "ftp://proxy:21"},
"proxy host": {Seed: 1, ProxyServer: "http://"},
"spoofing unknown": {Seed: 1, DisableSpoofing: "webrtc"},
"spoofing repeated": {Seed: 1, DisableSpoofing: "font,font"},
"seed zero": {Seed: 0},
"seed overflow": {Seed: 2147483648},
"platform": {Seed: 1, Platform: "android"},
"brand": {Seed: 1, Brand: "Firefox"},
"platform version": {Seed: 1, PlatformVersion: "bad value"},
"brand version": {Seed: 1, BrandVersion: strings.Repeat("x", 33)},
"concurrency": {Seed: 1, HardwareConcurrency: 129},
"lang": {Seed: 1, Lang: "zh CN"},
"accept lang": {Seed: 1, AcceptLang: "zh-CN;drop"},
"timezone": {Seed: 1, Timezone: "Asia/Shanghai\n"},
"proxy scheme": {Seed: 1, ProxyServer: "ftp://proxy:21"},
"proxy host": {Seed: 1, ProxyServer: "http://"},
"proxy userinfo": {Seed: 1, ProxyServer: "socks5://user:password@proxy:1080"},
"spoofing unknown": {Seed: 1, DisableSpoofing: "webrtc"},
"spoofing repeated": {Seed: 1, DisableSpoofing: "font,font"},
}
for name, fingerprint := range invalid {
t.Run(name, func(t *testing.T) {
@@ -119,13 +120,11 @@ func TestHubWorkflow(t *testing.T) {
if databaseURL == "" {
t.Skip("set CREATORHUB_POSTGRES_TEST_URL to run PostgreSQL integration coverage")
}
databaseURL = isolatedDatabaseURL(t, databaseURL)
ctx := context.Background()
store, err := Open(ctx, databaseURL)
if err != nil {
t.Fatal(err)
}
store := openFullyMigratedHub(t, ctx, databaseURL)
t.Cleanup(func() { _ = store.Close() })
if _, err := store.db.ExecContext(ctx, `TRUNCATE browser_env, browser_image, gateway`); err != nil {
if _, err := store.db.ExecContext(ctx, `TRUNCATE browser_env, browser_image, gateway CASCADE`); err != nil {
t.Fatal(err)
}
@@ -144,7 +143,7 @@ func TestHubWorkflow(t *testing.T) {
t.Fatalf("expected duplicate gateway conflict, got %v", err)
}
if err := store.CreateImage(ctx, Image{Version: "148.0.7778.215", ImageRef: "git.ipao.vip/rogee/fingerprint-chromium:148", Note: "主力版本"}); err != nil {
if err := store.CreateImage(ctx, Image{Version: "148.0.7778.215", ImageRef: "git.ipao.vip/rogee/fingerprint-chromium:148", Note: "主力版本", Enabled: true}); err != nil {
t.Fatal(err)
}
if err := store.CreateImage(ctx, Image{Version: "144.0.7559.132", ImageRef: "git.ipao.vip/rogee/fingerprint-chromium:144", Enabled: false}); err != nil {
@@ -188,7 +187,7 @@ func TestHubWorkflow(t *testing.T) {
if ref, err := store.ImageRef(ctx, "144.0.7559.132"); err != nil || !strings.HasSuffix(ref, ":144") {
t.Fatalf("enabled version must resolve: %v %q", err, ref)
}
if _, err := store.ImageRef(ctx, "missing"); !errors.Is(err, ErrNotFound) {
if _, err := store.ImageRef(ctx, "999"); !errors.Is(err, ErrNotFound) {
t.Fatalf("expected missing version, got %v", err)
}
+183 -33
View File
@@ -26,17 +26,27 @@ var (
ErrNotFound = errors.New("resource not found")
idPattern = regexp.MustCompile(`^[a-z0-9][a-z0-9-]{0,31}$`)
refPattern = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._/-]{0,127}$`)
platformKeyPattern = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._:@/-]{0,127}$`)
credentialKeyPattern = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._-]{0,63}/[A-Za-z0-9][A-Za-z0-9._/-]{0,126}$`)
)
type Store struct{ db *sql.DB }
type Account struct {
ID string `json:"id"`
CredentialReferenceID string `json:"credential_reference_id"`
CredentialProvider string `json:"credential_provider"`
CredentialKey string `json:"-"`
ProfileID string `json:"profile_id"`
ID string `json:"id"`
Platform string `json:"platform"`
PlatformAccountKey string `json:"platform_account_key"`
AuthorizationKind string `json:"authorization_kind"`
AuthorizationStatus string `json:"authorization_status"`
RuntimeStatus string `json:"runtime_status"`
CredentialReference CredentialReference `json:"credential_reference"`
CredentialKey string `json:"-"`
Version int64 `json:"version"`
}
type CredentialReference struct {
ID string `json:"id"`
Provider string `json:"provider"`
}
type RuntimeBinding struct {
@@ -92,6 +102,12 @@ type AuditEvent struct {
ConfirmationVersion int64 `json:"confirmation_version,omitempty"`
AttemptID string `json:"attempt_id,omitempty"`
TaskID string `json:"task_id,omitempty"`
BrowserEnvAlias string `json:"browser_env_alias,omitempty"`
NetworkExitID string `json:"network_exit_id,omitempty"`
RuntimeInstanceID string `json:"runtime_instance_id,omitempty"`
BindingVersion int64 `json:"binding_version,omitempty"`
Actor string `json:"actor,omitempty"`
ReasonCode string `json:"reason_code,omitempty"`
Details json.RawMessage `json:"details"`
CreatedAt time.Time `json:"created_at"`
}
@@ -149,9 +165,11 @@ func (s *Store) migrate(ctx context.Context) error {
}
func (s *Store) CreateAccount(ctx context.Context, account Account) error {
if !idPattern.MatchString(account.ID) || !refPattern.MatchString(account.CredentialReferenceID) ||
!credentialKeyPattern.MatchString(account.CredentialKey) || !refPattern.MatchString(account.ProfileID) ||
(account.CredentialProvider != "os_keyring" && account.CredentialProvider != "secret_manager") {
if !idPattern.MatchString(account.ID) || !idPattern.MatchString(account.Platform) ||
!platformKeyPattern.MatchString(account.PlatformAccountKey) ||
(account.AuthorizationKind != "owned" && account.AuthorizationKind != "authorized") ||
!refPattern.MatchString(account.CredentialReference.ID) || !credentialKeyPattern.MatchString(account.CredentialKey) ||
(account.CredentialReference.Provider != "os_keyring" && account.CredentialReference.Provider != "secret_manager") {
return ErrInvalid
}
tx, err := s.db.BeginTx(ctx, nil)
@@ -160,19 +178,68 @@ func (s *Store) CreateAccount(ctx context.Context, account Account) error {
}
defer tx.Rollback()
if _, err := tx.ExecContext(ctx, `INSERT INTO credential_reference (id, provider, reference_key) VALUES ($1, $2, $3)`,
account.CredentialReferenceID, account.CredentialProvider, account.CredentialKey); err != nil {
account.CredentialReference.ID, account.CredentialReference.Provider, account.CredentialKey); err != nil {
return publicDatabaseError(err)
}
if _, err := tx.ExecContext(ctx, `INSERT INTO social_account (id, credential_reference_id, profile_id) VALUES ($1, $2, $3)`,
account.ID, account.CredentialReferenceID, account.ProfileID); err != nil {
if _, err := tx.ExecContext(ctx, `
INSERT INTO social_account
(id, credential_reference_id, platform, platform_account_key, authorization_kind, authorization_status, status)
VALUES ($1, $2, $3, $4, $5, 'authorized', 'paused')`, account.ID, account.CredentialReference.ID,
account.Platform, account.PlatformAccountKey, account.AuthorizationKind); err != nil {
return publicDatabaseError(err)
}
if err := appendAudit(ctx, tx, "account_created", account.ID, "", 0, "", "", map[string]string{"profile_id": account.ProfileID}); err != nil {
if err := appendAudit(ctx, tx, "account_created", account.ID, "", 0, "", "", map[string]string{"platform": account.Platform}); err != nil {
return err
}
return commit(tx)
}
func (s *Store) ListAccounts(ctx context.Context) ([]Account, error) {
rows, err := s.db.QueryContext(ctx, `
SELECT account.id, account.platform, account.platform_account_key, account.authorization_kind,
account.authorization_status, account.status, reference.id, reference.provider, account.version
FROM social_account account
JOIN credential_reference reference ON reference.id = account.credential_reference_id
ORDER BY account.created_at, account.id`)
if err != nil {
return nil, errors.New("read accounts")
}
defer rows.Close()
accounts := []Account{}
for rows.Next() {
account, err := scanAccount(rows)
if err != nil {
return nil, err
}
accounts = append(accounts, account)
}
return accounts, rows.Err()
}
func (s *Store) GetAccount(ctx context.Context, id string) (Account, error) {
if !idPattern.MatchString(id) {
return Account{}, ErrInvalid
}
return scanAccount(s.db.QueryRowContext(ctx, `
SELECT account.id, account.platform, account.platform_account_key, account.authorization_kind,
account.authorization_status, account.status, reference.id, reference.provider, account.version
FROM social_account account
JOIN credential_reference reference ON reference.id = account.credential_reference_id
WHERE account.id = $1`, id))
}
type accountScanner interface{ Scan(...any) error }
func scanAccount(row accountScanner) (Account, error) {
var account Account
if err := row.Scan(&account.ID, &account.Platform, &account.PlatformAccountKey, &account.AuthorizationKind,
&account.AuthorizationStatus, &account.RuntimeStatus, &account.CredentialReference.ID,
&account.CredentialReference.Provider, &account.Version); err != nil {
return Account{}, rowError(err)
}
return account, nil
}
func (s *Store) BindRuntime(ctx context.Context, binding RuntimeBinding) error {
if !refPattern.MatchString(binding.ID) || !idPattern.MatchString(binding.AccountID) ||
!refPattern.MatchString(binding.RuntimeID) || !binding.LeaseUntil.After(time.Now()) {
@@ -188,8 +255,11 @@ func (s *Store) BindRuntime(ctx context.Context, binding RuntimeBinding) error {
}
var accountID string
if err := tx.QueryRowContext(ctx, `
INSERT INTO runtime_instance (id, account_id, runtime_id, lease_until)
SELECT $1, id, $3, $4 FROM social_account WHERE id = $2 AND status = 'active'
INSERT INTO runtime_instance (id, account_id, binding_id, runtime_id, lease_until)
SELECT $1, account.id, binding.id, $3, $4
FROM social_account account
JOIN environment_binding binding ON binding.account_id = account.id
WHERE account.id = $2 AND account.status = 'active'
RETURNING account_id`, binding.ID, binding.AccountID, binding.RuntimeID, binding.LeaseUntil).Scan(&accountID); err != nil {
if errors.Is(err, sql.ErrNoRows) {
return ErrConflict
@@ -320,38 +390,109 @@ func (s *Store) Enqueue(ctx context.Context, task Task) (Task, bool, error) {
}
func (s *Store) PauseAccount(ctx context.Context, accountID string) error {
return s.disableAccount(ctx, accountID, false)
}
func (s *Store) RevokeAccount(ctx context.Context, accountID string) error {
return s.disableAccount(ctx, accountID, true)
}
func (s *Store) disableAccount(ctx context.Context, accountID string, revoke bool) error {
if !idPattern.MatchString(accountID) {
return ErrInvalid
}
tx, err := s.db.BeginTx(ctx, nil)
if err != nil {
return errors.New("begin pause transaction")
return errors.New("begin account state transaction")
}
defer tx.Rollback()
var version int64
var status string
if err := tx.QueryRowContext(ctx, `SELECT version, status FROM social_account WHERE id = $1 FOR UPDATE`, accountID).Scan(&version, &status); err != nil {
var authorizationStatus, runtimeStatus string
if err := tx.QueryRowContext(ctx, `SELECT version, authorization_status, status FROM social_account WHERE id = $1 FOR UPDATE`, accountID).
Scan(&version, &authorizationStatus, &runtimeStatus); err != nil {
return rowError(err)
}
if status == "paused" {
unchanged := (revoke && authorizationStatus == "revoked") || (!revoke && runtimeStatus == "paused")
if !unchanged {
if err := tx.QueryRowContext(ctx, `
UPDATE social_account
SET authorization_status = CASE WHEN $2 THEN 'revoked' ELSE authorization_status END,
status = 'paused', paused_at = now(), revoked_at = CASE WHEN $2 THEN now() ELSE revoked_at END,
version = version + 1, updated_at = now()
WHERE id = $1 RETURNING version`, accountID, revoke).Scan(&version); err != nil {
return errors.New("change account state")
}
}
held, err := holdQueuedTasks(ctx, tx, accountID)
if err != nil {
return err
}
if unchanged {
return commit(tx)
}
if err := tx.QueryRowContext(ctx, `
UPDATE social_account SET status = 'paused', paused_at = now(), version = version + 1, updated_at = now()
WHERE id = $1 RETURNING version`, accountID).Scan(&version); err != nil {
return errors.New("pause account")
eventType := "account_paused"
if revoke {
eventType = "account_revoked"
}
if err := appendAudit(ctx, tx, eventType, accountID, "", 0, "", "", map[string]any{"account_version": version, "tasks_held": held}); err != nil {
return err
}
return commit(tx)
}
func (s *Store) ResumeAccount(ctx context.Context, accountID string) error {
if !idPattern.MatchString(accountID) {
return ErrInvalid
}
tx, err := s.db.BeginTx(ctx, nil)
if err != nil {
return errors.New("begin resume transaction")
}
defer tx.Rollback()
var authorizationStatus, runtimeStatus string
if err := tx.QueryRowContext(ctx, `SELECT authorization_status, status FROM social_account WHERE id = $1 FOR UPDATE`, accountID).
Scan(&authorizationStatus, &runtimeStatus); err != nil {
return rowError(err)
}
if authorizationStatus == "revoked" {
return ErrConflict
}
var ready bool
if err := tx.QueryRowContext(ctx, `
SELECT EXISTS (
SELECT 1 FROM environment_binding binding
JOIN network_exit network ON network.id = binding.network_exit_id
WHERE binding.account_id = $1 AND network.health_status = 'healthy'
)`, accountID).Scan(&ready); err != nil {
return errors.New("validate account binding")
}
if !ready {
return ErrConflict
}
if runtimeStatus == "active" {
return commit(tx)
}
var version int64
if err := tx.QueryRowContext(ctx, `
UPDATE social_account SET status = 'active', paused_at = NULL, version = version + 1, updated_at = now()
WHERE id = $1 RETURNING version`, accountID).Scan(&version); err != nil {
return errors.New("resume account")
}
if err := appendAudit(ctx, tx, "account_resumed", accountID, "", 0, "", "", map[string]any{"account_version": version}); err != nil {
return err
}
return commit(tx)
}
func holdQueuedTasks(ctx context.Context, tx *sql.Tx, accountID string) (int64, error) {
result, err := tx.ExecContext(ctx, `
UPDATE operation_task SET state = 'policy_hold', updated_at = now()
WHERE account_id = $1 AND state = 'queued'`, accountID)
if err != nil {
return errors.New("hold queued account tasks")
return 0, errors.New("hold queued account tasks")
}
held, _ := result.RowsAffected()
if err := appendAudit(ctx, tx, "account_paused", accountID, "", 0, "", "", map[string]any{"account_version": version, "tasks_held": held}); err != nil {
return err
}
return commit(tx)
return held, nil
}
func (s *Store) CancelTask(ctx context.Context, taskID string) error {
@@ -587,7 +728,9 @@ func quarantineInvalid(ctx context.Context, tx *sql.Tx) error {
func (s *Store) Audit(ctx context.Context) ([]AuditEvent, error) {
rows, err := s.db.QueryContext(ctx, `
SELECT id, event_type, account_id, confirmation_id, confirmation_version, attempt_id, task_id, details, created_at
SELECT id, event_type, account_id, confirmation_id, confirmation_version, attempt_id, task_id,
browser_env_alias, network_exit_id, runtime_instance_id, binding_version, actor, reason_code,
details, created_at
FROM audit_event ORDER BY id`)
if err != nil {
return nil, errors.New("read audit events")
@@ -596,13 +739,19 @@ func (s *Store) Audit(ctx context.Context) ([]AuditEvent, error) {
var events []AuditEvent
for rows.Next() {
var event AuditEvent
var accountID, confirmationID, attemptID, taskID sql.NullString
var confirmationVersion sql.NullInt64
if err := rows.Scan(&event.ID, &event.EventType, &accountID, &confirmationID, &confirmationVersion, &attemptID, &taskID, &event.Details, &event.CreatedAt); err != nil {
var accountID, confirmationID, attemptID, taskID, browserEnvAlias, networkExitID sql.NullString
var runtimeInstanceID, actor, reasonCode sql.NullString
var confirmationVersion, bindingVersion sql.NullInt64
if err := rows.Scan(&event.ID, &event.EventType, &accountID, &confirmationID, &confirmationVersion, &attemptID, &taskID,
&browserEnvAlias, &networkExitID, &runtimeInstanceID, &bindingVersion, &actor, &reasonCode,
&event.Details, &event.CreatedAt); err != nil {
return nil, errors.New("decode audit event")
}
event.AccountID, event.ConfirmationID, event.ConfirmationVersion = accountID.String, confirmationID.String, confirmationVersion.Int64
event.AttemptID, event.TaskID = attemptID.String, taskID.String
event.BrowserEnvAlias, event.NetworkExitID = browserEnvAlias.String, networkExitID.String
event.RuntimeInstanceID, event.BindingVersion = runtimeInstanceID.String, bindingVersion.Int64
event.Actor, event.ReasonCode = actor.String, reasonCode.String
events = append(events, event)
}
return events, rows.Err()
@@ -617,8 +766,9 @@ func appendAudit(ctx context.Context, tx *sql.Tx, eventType, accountID, confirma
return errors.New("encode audit details")
}
_, err = tx.ExecContext(ctx, `
INSERT INTO audit_event (event_type, account_id, confirmation_id, confirmation_version, attempt_id, task_id, details)
VALUES ($1, NULLIF($2, ''), NULLIF($3, ''), NULLIF($4, 0), NULLIF($5, ''), NULLIF($6, ''), $7)`,
INSERT INTO audit_event
(event_type, account_id, confirmation_id, confirmation_version, attempt_id, task_id, actor, reason_code, details)
VALUES ($1, NULLIF($2, ''), NULLIF($3, ''), NULLIF($4, 0), NULLIF($5, ''), NULLIF($6, ''), 'local-user', $1, $7)`,
eventType, accountID, confirmationID, confirmationVersion, attemptID, taskID, encoded)
if err != nil {
return errors.New("append audit event")
+135 -25
View File
@@ -6,6 +6,7 @@ import (
"errors"
"fmt"
"os"
"path/filepath"
"strings"
"sync"
"sync/atomic"
@@ -15,12 +16,28 @@ import (
func TestValidationRejectsInvalidInputsBeforePersistence(t *testing.T) {
store := &Store{}
if err := store.CreateAccount(context.Background(), Account{ID: "INVALID"}); !errors.Is(err, ErrInvalid) {
t.Fatalf("expected invalid account, got %v", err)
valid := Account{
ID: "account-a", Platform: "mock", PlatformAccountKey: "platform-a", AuthorizationKind: "owned",
CredentialReference: CredentialReference{ID: "credential-a", Provider: "os_keyring"}, CredentialKey: "creatorhub/account-a",
}
for name, mutate := range map[string]func(*Account){
"id": func(account *Account) { account.ID = "INVALID" },
"platform": func(account *Account) { account.Platform = "bad platform" },
"platform account key": func(account *Account) { account.PlatformAccountKey = "secret value" },
"authorization kind": func(account *Account) { account.AuthorizationKind = "borrowed" },
"credential provider": func(account *Account) { account.CredentialReference.Provider = "plain_text" },
} {
t.Run(name, func(t *testing.T) {
account := valid
mutate(&account)
if err := store.CreateAccount(context.Background(), account); !errors.Is(err, ErrInvalid) {
t.Fatalf("expected invalid account, got %v", err)
}
})
}
if err := store.CreateAccount(context.Background(), Account{
ID: "account-a", CredentialReferenceID: "credential-a", CredentialProvider: "os_keyring",
CredentialKey: "raw-value", ProfileID: "profile-a",
ID: "account-a", Platform: "mock", PlatformAccountKey: "platform-a", AuthorizationKind: "owned",
CredentialReference: CredentialReference{ID: "credential-a", Provider: "os_keyring"}, CredentialKey: "raw-value",
}); !errors.Is(err, ErrInvalid) {
t.Fatalf("expected non-reference credential value to be rejected, got %v", err)
}
@@ -43,26 +60,75 @@ func TestPhaseAOfflineWorkflow(t *testing.T) {
t.Fatal(err)
}
t.Cleanup(func() { _ = store.Close() })
applyHubMigrationsForPhaseATest(t, store)
if _, err := store.db.ExecContext(ctx, `
TRUNCATE audit_event, execution_attempt, operation_task, confirmation, content_draft,
runtime_instance, social_account, credential_reference RESTART IDENTITY CASCADE`); err != nil {
runtime_instance, environment_binding, network_exit, social_account, credential_reference,
browser_env, browser_image, gateway RESTART IDENTITY CASCADE`); err != nil {
t.Fatal(err)
}
accounts := []Account{
{ID: "account-a", CredentialReferenceID: "credential-a", CredentialProvider: "os_keyring", CredentialKey: "creatorhub/account-a", ProfileID: "profile-a"},
{ID: "account-b", CredentialReferenceID: "credential-b", CredentialProvider: "os_keyring", CredentialKey: "creatorhub/account-b", ProfileID: "profile-b"},
{ID: "account-a", Platform: "mock", PlatformAccountKey: "platform-a", AuthorizationKind: "owned", CredentialReference: CredentialReference{ID: "credential-a", Provider: "os_keyring"}, CredentialKey: "creatorhub/account-a"},
{ID: "account-b", Platform: "mock", PlatformAccountKey: "platform-b", AuthorizationKind: "authorized", CredentialReference: CredentialReference{ID: "credential-b", Provider: "os_keyring"}, CredentialKey: "creatorhub/account-b"},
}
for _, account := range accounts {
if err := store.CreateAccount(ctx, account); err != nil {
t.Fatal(err)
}
}
createdAccount, err := store.GetAccount(ctx, "account-a")
if err != nil || createdAccount.RuntimeStatus != "paused" || createdAccount.Version != 1 {
t.Fatalf("new account must start paused: %#v %v", createdAccount, err)
}
if err := store.ResumeAccount(ctx, "account-a"); !errors.Is(err, ErrConflict) {
t.Fatalf("account without a healthy binding must not resume: %v", err)
}
if err := store.CreateAccount(ctx, Account{
ID: "account-c", CredentialReferenceID: "credential-c", CredentialProvider: "os_keyring",
CredentialKey: "creatorhub/account-c", ProfileID: "profile-a",
ID: "account-c", Platform: "mock", PlatformAccountKey: "platform-a", AuthorizationKind: "owned",
CredentialReference: CredentialReference{ID: "credential-c", Provider: "os_keyring"}, CredentialKey: "creatorhub/account-c",
}); !errors.Is(err, ErrConflict) {
t.Fatalf("expected profiles to be unique, got %v", err)
t.Fatalf("expected platform accounts to be unique, got %v", err)
}
if _, err := store.db.ExecContext(ctx, `
INSERT INTO gateway (name, endpoint, token) VALUES ('test-gateway', 'http://127.0.0.1:8081', 'phase-a-test-gateway-token')`); err != nil {
t.Fatal(err)
}
if _, err := store.db.ExecContext(ctx, `
INSERT INTO browser_image (version, image_ref) VALUES ('1', 'example/browser:1')`); err != nil {
t.Fatal(err)
}
if _, err := store.db.ExecContext(ctx, `
INSERT INTO browser_env (alias, name, gateway_name, image_version, fingerprint) VALUES
('account-a', 'Account A', 'test-gateway', '1', '{"seed":1}'),
('account-b', 'Account B', 'test-gateway', '1', '{"seed":2}');
INSERT INTO network_exit (id, protocol, host, port, health_status)
VALUES ('exit-shared', 'socks5', '127.0.0.1', 1080, 'healthy');
INSERT INTO environment_binding (id, account_id, browser_env_alias, network_exit_id) VALUES
('binding-a', 'account-a', 'account-a', 'exit-shared'),
('binding-b', 'account-b', 'account-b', 'exit-shared')`); err != nil {
t.Fatal(err)
}
for _, id := range []string{"account-a", "account-b"} {
if err := store.ResumeAccount(ctx, id); err != nil {
t.Fatalf("resume %s: %v", id, err)
}
}
accountA, err := store.GetAccount(ctx, "account-a")
if err != nil || accountA.RuntimeStatus != "active" || accountA.Version != 2 {
t.Fatalf("unexpected resumed account: %#v %v", accountA, err)
}
exportedAccount, _ := json.Marshal(accountA)
if strings.Contains(string(exportedAccount), "creatorhub/account-a") || strings.Contains(string(exportedAccount), "reference_key") {
t.Fatalf("account response leaked credential key: %s", exportedAccount)
}
listedAccounts, err := store.ListAccounts(ctx)
if err != nil || len(listedAccounts) != 2 {
t.Fatalf("unexpected account list: %#v %v", listedAccounts, err)
}
accountB, err := store.GetAccount(ctx, "account-b")
if err != nil {
t.Fatal(err)
}
lease := time.Now().Add(time.Minute)
@@ -79,9 +145,9 @@ func TestPhaseAOfflineWorkflow(t *testing.T) {
t.Fatal(err)
}
createApprovedDraft(t, store, "account-a", "draft-a", "confirmation-a")
createApprovedDraft(t, store, "account-a", accountA.Version, "draft-a", "confirmation-a")
for index := range 20 {
task := approvedTask(index, "account-a", "draft-a", "confirmation-a")
task := approvedTask(index, "account-a", accountA.Version, "draft-a", "confirmation-a")
if _, inserted, err := store.Enqueue(ctx, task); err != nil || !inserted {
t.Fatalf("enqueue task %d: inserted=%v err=%v", index, inserted, err)
}
@@ -122,14 +188,14 @@ func TestPhaseAOfflineWorkflow(t *testing.T) {
assertCount(t, store, `SELECT count(*) FROM operation_task WHERE state = 'succeeded'`, 20)
assertCount(t, store, `SELECT count(*) FROM execution_attempt`, 20)
createApprovedDraft(t, store, "account-b", "draft-b", "confirmation-b")
unconfirmed := approvedTask(30, "account-b", "draft-b", "")
createApprovedDraft(t, store, "account-b", accountB.Version, "draft-b", "confirmation-b")
unconfirmed := approvedTask(30, "account-b", accountB.Version, "draft-b", "")
unconfirmed.ConfirmationVersion = 0
if _, _, err := store.Enqueue(ctx, unconfirmed); err != nil {
t.Fatal(err)
}
mismatch := approvedTask(31, "account-b", "draft-b", "confirmation-b")
mismatch.AccountVersion = 2
mismatch := approvedTask(31, "account-b", accountB.Version, "draft-b", "confirmation-b")
mismatch.AccountVersion++
if _, _, err := store.Enqueue(ctx, mismatch); err != nil {
t.Fatal(err)
}
@@ -139,7 +205,7 @@ func TestPhaseAOfflineWorkflow(t *testing.T) {
assertCount(t, store, `SELECT count(*) FROM operation_task WHERE id IN ('task-30', 'task-31') AND state = 'needs_confirmation'`, 2)
assertCount(t, store, `SELECT count(*) FROM execution_attempt WHERE task_id IN ('task-30', 'task-31')`, 0)
uncertain := approvedTask(32, "account-b", "draft-b", "confirmation-b")
uncertain := approvedTask(32, "account-b", accountB.Version, "draft-b", "confirmation-b")
if _, _, err := store.Enqueue(ctx, uncertain); err != nil {
t.Fatal(err)
}
@@ -151,7 +217,7 @@ func TestPhaseAOfflineWorkflow(t *testing.T) {
}
assertCount(t, store, `SELECT count(*) FROM execution_attempt WHERE task_id = 'task-32'`, 1)
policy := approvedTask(33, "account-b", "draft-b", "confirmation-b")
policy := approvedTask(33, "account-b", accountB.Version, "draft-b", "confirmation-b")
if _, _, err := store.Enqueue(ctx, policy); err != nil {
t.Fatal(err)
}
@@ -162,7 +228,7 @@ func TestPhaseAOfflineWorkflow(t *testing.T) {
t.Fatalf("policy task was retried: execution=%+v err=%v", execution, err)
}
paused := approvedTask(34, "account-b", "draft-b", "confirmation-b")
paused := approvedTask(34, "account-b", accountB.Version, "draft-b", "confirmation-b")
if _, _, err := store.Enqueue(ctx, paused); err != nil {
t.Fatal(err)
}
@@ -177,12 +243,30 @@ func TestPhaseAOfflineWorkflow(t *testing.T) {
t.Fatalf("paused account task was claimed: execution=%+v err=%v", execution, err)
}
assertCount(t, store, `SELECT count(*) FROM operation_task WHERE id = 'task-34' AND state = 'policy_hold'`, 1)
pausedAccount, err := store.GetAccount(ctx, "account-b")
if err != nil || pausedAccount.Version != accountB.Version+1 {
t.Fatalf("idempotent pause must increment once: %#v %v", pausedAccount, err)
}
var pausedAt time.Time
if err := store.db.QueryRowContext(ctx, `SELECT paused_at FROM social_account WHERE id = 'account-b'`).Scan(&pausedAt); err != nil || pausedAt.Sub(pauseStarted) >= time.Minute {
t.Fatalf("account was not paused within one minute: paused_at=%v err=%v", pausedAt, err)
}
cancelled := approvedTask(35, "account-a", "draft-a", "confirmation-a")
if err := store.RevokeAccount(ctx, "account-b"); err != nil {
t.Fatal(err)
}
if err := store.RevokeAccount(ctx, "account-b"); err != nil {
t.Fatalf("revoke should be idempotent: %v", err)
}
revokedAccount, err := store.GetAccount(ctx, "account-b")
if err != nil || revokedAccount.AuthorizationStatus != "revoked" || revokedAccount.Version != accountB.Version+2 {
t.Fatalf("unexpected revoked account: %#v %v", revokedAccount, err)
}
if err := store.ResumeAccount(ctx, "account-b"); !errors.Is(err, ErrConflict) {
t.Fatalf("revoked account must not resume: %v", err)
}
cancelled := approvedTask(35, "account-a", accountA.Version, "draft-a", "confirmation-a")
if _, _, err := store.Enqueue(ctx, cancelled); err != nil {
t.Fatal(err)
}
@@ -191,7 +275,7 @@ func TestPhaseAOfflineWorkflow(t *testing.T) {
}
assertCount(t, store, `SELECT count(*) FROM operation_task WHERE id = 'task-35' AND state = 'cancelled'`, 1)
expired := approvedTask(36, "account-a", "draft-a", "confirmation-a")
expired := approvedTask(36, "account-a", accountA.Version, "draft-a", "confirmation-a")
if _, _, err := store.Enqueue(ctx, expired); err != nil {
t.Fatal(err)
}
@@ -233,27 +317,53 @@ func TestPhaseAOfflineWorkflow(t *testing.T) {
}
}
func createApprovedDraft(t *testing.T, store *Store, accountID, draftID, confirmationID string) {
func createApprovedDraft(t *testing.T, store *Store, accountID string, accountVersion int64, draftID, confirmationID string) {
t.Helper()
ctx := context.Background()
if err := store.CreateDraft(ctx, Draft{ID: draftID, AccountID: accountID, Version: 1, Content: "offline mock content"}); err != nil {
t.Fatal(err)
}
if err := store.Confirm(ctx, Confirmation{
ID: confirmationID, AccountID: accountID, AccountVersion: 1, DraftID: draftID, DraftVersion: 1, Version: 1,
ID: confirmationID, AccountID: accountID, AccountVersion: accountVersion, DraftID: draftID, DraftVersion: 1, Version: 1,
}); err != nil {
t.Fatal(err)
}
}
func approvedTask(index int, accountID, draftID, confirmationID string) Task {
func approvedTask(index int, accountID string, accountVersion int64, draftID, confirmationID string) Task {
return Task{
ID: fmt.Sprintf("task-%02d", index), IdempotencyKey: fmt.Sprintf("idempotency-%02d", index),
AccountID: accountID, AccountVersion: 1, DraftID: draftID, DraftVersion: 1,
AccountID: accountID, AccountVersion: accountVersion, DraftID: draftID, DraftVersion: 1,
ConfirmationID: confirmationID, ConfirmationVersion: 1,
}
}
func applyHubMigrationsForPhaseATest(t *testing.T, store *Store) {
t.Helper()
for _, migrationFile := range []struct {
version int
name string
}{{2, "002_hub.sql"}, {3, "003_unified_accounts.sql"}} {
var applied bool
if err := store.db.QueryRow(`SELECT EXISTS (SELECT 1 FROM schema_migration WHERE version = $1)`, migrationFile.version).Scan(&applied); err != nil {
t.Fatal(err)
}
if applied {
continue
}
migration, err := os.ReadFile(filepath.Join("..", "hub", "migrations", migrationFile.name))
if err != nil {
t.Fatal(err)
}
if _, err := store.db.Exec(string(migration)); err != nil {
t.Fatalf("apply test migration %d: %v", migrationFile.version, err)
}
if _, err := store.db.Exec(`INSERT INTO schema_migration (version) VALUES ($1)`, migrationFile.version); err != nil {
t.Fatal(err)
}
}
}
func assertCount(t *testing.T, store *Store, query string, expected int) {
t.Helper()
var actual int