diff --git a/cmd/control-plane/main_test.go b/cmd/control-plane/main_test.go index 5b92227..e28f12d 100644 --- a/cmd/control-plane/main_test.go +++ b/cmd/control-plane/main_test.go @@ -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) + } + }) } } diff --git a/cmd/control-plane/phasea.go b/cmd/control-plane/phasea.go index 8817238..5faf6c4 100644 --- a/cmd/control-plane/phasea.go +++ b/cmd/control-plane/phasea.go @@ -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 { diff --git a/docs/architecture/container-control.md b/docs/architecture/container-control.md index 0ab20b9..29db0f5 100644 --- a/docs/architecture/container-control.md +++ b/docs/architecture/container-control.md @@ -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;其余记录等待显式绑定。本阶段不提供破坏性自动回滚。 diff --git a/internal/hub/fingerprint.go b/internal/hub/fingerprint.go index bd52b7e..e1f3f70 100644 --- a/internal/hub/fingerprint.go +++ b/internal/hub/fingerprint.go @@ -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 { diff --git a/internal/hub/migration_test.go b/internal/hub/migration_test.go new file mode 100644 index 0000000..2eb1073 --- /dev/null +++ b/internal/hub/migration_test.go @@ -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) + } +} diff --git a/internal/hub/migrations/003_unified_accounts.sql b/internal/hub/migrations/003_unified_accounts.sql new file mode 100644 index 0000000..0572241 --- /dev/null +++ b/internal/hub/migrations/003_unified_accounts.sql @@ -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; diff --git a/internal/hub/store.go b/internal/hub/store.go index f08e4eb..e521a10 100644 --- a/internal/hub/store.go +++ b/internal/hub/store.go @@ -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 diff --git a/internal/hub/store_test.go b/internal/hub/store_test.go index fcdcc8f..cb7cdb5 100644 --- a/internal/hub/store_test.go +++ b/internal/hub/store_test.go @@ -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) } diff --git a/internal/phasea/store.go b/internal/phasea/store.go index 79fff72..194428a 100644 --- a/internal/phasea/store.go +++ b/internal/phasea/store.go @@ -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") diff --git a/internal/phasea/store_test.go b/internal/phasea/store_test.go index bc721e7..68b820d 100644 --- a/internal/phasea/store_test.go +++ b/internal/phasea/store_test.go @@ -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