diff --git a/cmd/control-plane/main_test.go b/cmd/control-plane/main_test.go index e28f12d..7a8ed2b 100644 --- a/cmd/control-plane/main_test.go +++ b/cmd/control-plane/main_test.go @@ -4,6 +4,7 @@ import ( "bufio" "bytes" "context" + "database/sql" "encoding/json" "errors" "io" @@ -17,6 +18,8 @@ import ( "testing" "time" + "git.ipao.vip/rogee/creator-hub/internal/hub" + "git.ipao.vip/rogee/creator-hub/internal/phasea" "github.com/gofiber/fiber/v3" "github.com/sirupsen/logrus" "github.com/spf13/cobra" @@ -97,6 +100,78 @@ func TestPhaseAAccountRequestRejectsSecretsAndUnknownFields(t *testing.T) { } } +func TestPhaseAAccountHTTPWorkflowRedactsSecrets(t *testing.T) { + databaseURL := os.Getenv("CREATORHUB_POSTGRES_TEST_URL") + if databaseURL == "" { + t.Skip("set CREATORHUB_POSTGRES_TEST_URL to run PostgreSQL integration coverage") + } + ctx := context.Background() + databaseURL = isolatedControlPlaneDatabaseURL(t, databaseURL) + store, err := phasea.Open(ctx, databaseURL) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = store.Close() }) + hubStore, err := hub.Open(ctx, databaseURL) + if err != nil { + t.Fatal(err) + } + if err := hubStore.Close(); err != nil { + t.Fatal(err) + } + + app := fiber.New() + registerPhaseA(app, store, nil) + credentialKey := "creatorhub/phase-http-secret" + accountBody := `{"id":"account-http","platform":"mock","platform_account_key":"phase-http","authorization_kind":"owned","credential_reference":{"id":"credential-http","provider":"os_keyring","key":"` + credentialKey + `"}}` + request := func(method, path, body string, wantStatus int) *httptest.ResponseRecorder { + t.Helper() + response := do(app, method, path, body) + if response.Code != wantStatus { + t.Fatalf("%s %s returned %d, want %d: %s", method, path, response.Code, wantStatus, response.Body.String()) + } + if strings.Contains(response.Body.String(), credentialKey) { + t.Fatalf("%s %s leaked credential key: %s", method, path, response.Body.String()) + } + return response + } + + created := request(http.MethodPost, "/api/phase-a/accounts", accountBody, http.StatusCreated) + var account phasea.Account + if err := json.Unmarshal(created.Body.Bytes(), &account); err != nil || account.ID != "account-http" || account.RuntimeStatus != "paused" { + t.Fatalf("unexpected account response: %#v err=%v", account, err) + } + request(http.MethodPost, "/api/phase-a/accounts", accountBody, http.StatusConflict) + request(http.MethodPost, "/api/phase-a/accounts", strings.TrimSuffix(accountBody, "}")+`,"password":"`+credentialKey+`"}`, http.StatusBadRequest) + request(http.MethodGet, "/api/phase-a/accounts", "", http.StatusOK) + request(http.MethodGet, "/api/phase-a/accounts/account-http", "", http.StatusOK) + + db, err := sql.Open("pgx", databaseURL) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = db.Close() }) + if _, err := db.ExecContext(ctx, ` + INSERT INTO gateway (name, endpoint, token) VALUES ('phase-http', 'http://127.0.0.1:8081', 'phase-http-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 ('account-http', 'Phase HTTP', 'phase-http', '1', '{"seed":1}'); + INSERT INTO network_exit (id, protocol, host, port, health_status) + VALUES ('exit-http', 'socks5', '127.0.0.1', 1080, 'healthy'); + INSERT INTO environment_binding (id, account_id, browser_env_alias, network_exit_id) + VALUES ('binding-http', 'account-http', 'account-http', 'exit-http')`); err != nil { + t.Fatal(err) + } + request(http.MethodPost, "/api/phase-a/accounts/account-http/resume", "", http.StatusNoContent) + request(http.MethodPost, "/api/phase-a/accounts/account-http/pause", "", http.StatusNoContent) + request(http.MethodPost, "/api/phase-a/accounts/account-http/resume", "", http.StatusNoContent) + request(http.MethodPost, "/api/phase-a/accounts/account-http/revoke", "", http.StatusNoContent) + blocked := request(http.MethodPost, "/api/phase-a/accounts/account-http/resume", "", http.StatusConflict) + if !strings.Contains(blocked.Body.String(), `"reason_code":"account_revoked"`) { + t.Fatalf("revoked resume did not return a stable conflict reason: %s", blocked.Body.String()) + } +} + func TestPhaseAErrorRedactsInternalDetails(t *testing.T) { app := fiber.New() app.Get("/", func(c fiber.Ctx) error { diff --git a/internal/hub/environment.go b/internal/hub/environment.go index ccf2e76..77e288f 100644 --- a/internal/hub/environment.go +++ b/internal/hub/environment.go @@ -472,6 +472,27 @@ func (s *Store) GetEnvironmentContextForAccount(ctx context.Context, accountID s return s.GetEnvironmentContext(ctx, alias) } +func releaseExpiredRuntime(ctx context.Context, tx *sql.Tx, bindingID string) error { + var accountID, alias, runtimeInstanceID string + var exitID sql.NullString + var bindingVersion int64 + err := tx.QueryRowContext(ctx, ` + UPDATE runtime_instance runtime SET released_at = now() + FROM environment_binding binding + WHERE binding.id = $1 AND runtime.binding_id = binding.id + AND runtime.released_at IS NULL AND runtime.lease_until <= now() + RETURNING runtime.account_id, binding.browser_env_alias, binding.network_exit_id, + runtime.id, runtime.binding_version`, bindingID). + Scan(&accountID, &alias, &exitID, &runtimeInstanceID, &bindingVersion) + if errors.Is(err, sql.ErrNoRows) { + return nil + } + if err != nil { + return err + } + return appendRuntimeAudit(ctx, tx, "runtime_released", accountID, alias, exitID.String, runtimeInstanceID, bindingVersion) +} + func validateEnvironmentRebind(ctx context.Context, tx *sql.Tx, alias, exitID string, expectedBindingVersion int64) (string, string, error) { var accountID, bindingID string var bindingVersion int64 @@ -492,9 +513,7 @@ func validateEnvironmentRebind(ctx context.Context, tx *sql.Tx, alias, exitID st if bindingVersion != expectedBindingVersion { return "", "", ErrConflict } - if _, err := tx.ExecContext(ctx, ` - UPDATE runtime_instance SET released_at = now() - WHERE binding_id = $1 AND released_at IS NULL AND lease_until <= now()`, bindingID); err != nil { + if err := releaseExpiredRuntime(ctx, tx, bindingID); err != nil { return "", "", errors.New("expire runtime before rebind") } var allowed bool @@ -555,12 +574,16 @@ func (s *Store) RebindEnvironment(ctx context.Context, alias, exitID, runtimeID return EnvironmentContext{}, errors.New("version rebound account") } if runtimeID != "" { + runtimeInstanceID := "runtime-" + newHubID() if _, err := tx.ExecContext(ctx, ` INSERT INTO runtime_instance (id, account_id, binding_id, binding_version, runtime_id, network_id, lease_until) VALUES ($1, $2, $3, $4, $5, NULLIF($6, ''), now() + interval '1 minute')`, - "runtime-"+newHubID(), accountID, bindingID, expectedBindingVersion+1, runtimeID, networkID); err != nil { + runtimeInstanceID, accountID, bindingID, expectedBindingVersion+1, runtimeID, networkID); err != nil { return EnvironmentContext{}, publicDatabaseError(err) } + if err := appendRuntimeAudit(ctx, tx, "runtime_bound", accountID, alias, exitID, runtimeInstanceID, expectedBindingVersion+1); err != nil { + return EnvironmentContext{}, err + } } if err := commitHub(tx); err != nil { return EnvironmentContext{}, err @@ -599,9 +622,7 @@ func (s *Store) ActivateRuntime(ctx context.Context, alias, runtimeID string, bi currentBindingVersion != bindingVersion || currentExitID != exitID { return EnvironmentContext{}, ErrConflict } - if _, err := tx.ExecContext(ctx, ` - UPDATE runtime_instance SET released_at = now() - WHERE binding_id = $1 AND released_at IS NULL AND lease_until <= now()`, bindingID); err != nil { + if err := releaseExpiredRuntime(ctx, tx, bindingID); err != nil { return EnvironmentContext{}, errors.New("expire runtime before activation") } var existingInstanceID, existingRuntimeID, existingNetworkID string @@ -615,12 +636,16 @@ func (s *Store) ActivateRuntime(ctx context.Context, alias, runtimeID string, bi return EnvironmentContext{}, ErrConflict } if existingInstanceID == "" { + existingInstanceID = "runtime-" + newHubID() if _, err := tx.ExecContext(ctx, ` INSERT INTO runtime_instance (id, account_id, binding_id, binding_version, runtime_id, network_id, lease_until) VALUES ($1, $2, $3, $4, $5, NULLIF($6, ''), now() + interval '1 minute')`, - "runtime-"+newHubID(), accountID, bindingID, bindingVersion, runtimeID, networkID); err != nil { + existingInstanceID, accountID, bindingID, bindingVersion, runtimeID, networkID); err != nil { return EnvironmentContext{}, publicDatabaseError(err) } + if err := appendRuntimeAudit(ctx, tx, "runtime_bound", accountID, alias, exitID, existingInstanceID, bindingVersion); err != nil { + return EnvironmentContext{}, err + } } else if _, err := tx.ExecContext(ctx, `UPDATE runtime_instance SET lease_until = now() + interval '1 minute' WHERE id = $1`, existingInstanceID); err != nil { return EnvironmentContext{}, errors.New("renew environment runtime") } @@ -638,20 +663,33 @@ func (s *Store) ReleaseRuntime(ctx context.Context, environment EnvironmentConte if environment.RuntimeInstanceID == "" { return nil } - result, err := s.db.ExecContext(ctx, ` + tx, err := s.db.BeginTx(ctx, nil) + if err != nil { + return errors.New("begin runtime release") + } + defer tx.Rollback() + var accountID, alias, runtimeInstanceID string + var exitID sql.NullString + var bindingVersion int64 + err = tx.QueryRowContext(ctx, ` UPDATE runtime_instance runtime SET released_at = now() FROM environment_binding binding WHERE binding.browser_env_alias = $1 AND binding.id = $2 AND binding.version = $3 AND runtime.binding_id = binding.id AND runtime.binding_version = binding.version - AND runtime.id = $4 AND runtime.released_at IS NULL`, environment.Alias, environment.BindingID, - environment.BindingVersion, environment.RuntimeInstanceID) + AND runtime.id = $4 AND runtime.released_at IS NULL + RETURNING runtime.account_id, binding.browser_env_alias, binding.network_exit_id, runtime.id, runtime.binding_version`, + environment.Alias, environment.BindingID, environment.BindingVersion, environment.RuntimeInstanceID). + Scan(&accountID, &alias, &exitID, &runtimeInstanceID, &bindingVersion) + if errors.Is(err, sql.ErrNoRows) { + return ErrConflict + } if err != nil { return errors.New("release environment runtime") } - if affected, err := result.RowsAffected(); err != nil || affected != 1 { - return ErrConflict + if err := appendRuntimeAudit(ctx, tx, "runtime_released", accountID, alias, exitID.String, runtimeInstanceID, bindingVersion); err != nil { + return err } - return nil + return commitHub(tx) } func (s *Store) SetRuntimeCleanupPending(ctx context.Context, environment EnvironmentContext, pending bool) error { @@ -669,15 +707,19 @@ func (s *Store) SetRuntimeCleanupPending(ctx context.Context, environment Enviro } defer tx.Rollback() var currentPending bool + var accountID, alias string + var exitID sql.NullString var cleanupBindingVersion sql.NullInt64 var cleanupInstanceID, cleanupRuntimeID, cleanupNetworkID sql.NullString if err := tx.QueryRowContext(ctx, ` SELECT runtime_cleanup_pending, runtime_cleanup_binding_version, - runtime_cleanup_instance_id, runtime_cleanup_runtime_id, runtime_cleanup_network_id + runtime_cleanup_instance_id, runtime_cleanup_runtime_id, runtime_cleanup_network_id, + account_id, browser_env_alias, network_exit_id FROM environment_binding WHERE browser_env_alias = $1 AND id = $2 AND version = $3 FOR UPDATE`, environment.Alias, environment.BindingID, environment.BindingVersion). - Scan(¤tPending, &cleanupBindingVersion, &cleanupInstanceID, &cleanupRuntimeID, &cleanupNetworkID); errors.Is(err, sql.ErrNoRows) { + Scan(¤tPending, &cleanupBindingVersion, &cleanupInstanceID, &cleanupRuntimeID, &cleanupNetworkID, + &accountID, &alias, &exitID); errors.Is(err, sql.ErrNoRows) { return ErrConflict } else if err != nil { return publicDatabaseError(err) @@ -719,6 +761,9 @@ func (s *Store) SetRuntimeCleanupPending(ctx context.Context, environment Enviro if affected, err := result.RowsAffected(); err != nil || affected != 1 { return ErrConflict } + if err := appendRuntimeAudit(ctx, tx, "runtime_released", accountID, alias, exitID.String, runtimeInstanceID, environment.BindingVersion); err != nil { + return err + } } } if pending { @@ -741,6 +786,18 @@ func (s *Store) SetRuntimeCleanupPending(ctx context.Context, environment Enviro return commitHub(tx) } +func appendRuntimeAudit(ctx context.Context, tx *sql.Tx, eventType, accountID, alias, exitID, runtimeInstanceID string, bindingVersion int64) error { + _, err := tx.ExecContext(ctx, ` + INSERT INTO audit_event + (event_type, account_id, browser_env_alias, network_exit_id, runtime_instance_id, binding_version, actor, reason_code) + VALUES ($1, $2, $3, NULLIF($4, ''), $5, $6, 'local-user', $1)`, + eventType, accountID, alias, exitID, runtimeInstanceID, bindingVersion) + if err != nil { + return errors.New("append runtime audit event") + } + return nil +} + func (s *Store) AppendEnvironmentAction(ctx context.Context, eventType string, action EnvironmentAction) error { if (eventType != "environment_action_requested" && eventType != "environment_action_finished") || !exitIDPattern.MatchString(action.OperationID) || action.Action == "" || action.ReasonCode == "" || diff --git a/internal/hub/migration_test.go b/internal/hub/migration_test.go index d9cf3c9..9bd30cd 100644 --- a/internal/hub/migration_test.go +++ b/internal/hub/migration_test.go @@ -292,10 +292,10 @@ func isolatedDatabaseURL(t *testing.T, databaseURL string) string { return parsed.String() } -func assertDatabaseCount(t *testing.T, db *sql.DB, query string, want int) { +func assertDatabaseCount(t *testing.T, db *sql.DB, query string, want int, args ...any) { t.Helper() var got int - if err := db.QueryRow(query).Scan(&got); err != nil || got != want { + if err := db.QueryRow(query, args...).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/store_test.go b/internal/hub/store_test.go index 08c6cde..d003534 100644 --- a/internal/hub/store_test.go +++ b/internal/hub/store_test.go @@ -311,6 +311,10 @@ func TestNetworkExitBindingRuntimeAndAuditWorkflow(t *testing.T) { if err != nil || active.RuntimeInstanceID == "" { t.Fatalf("activate runtime: %#v err=%v", active, err) } + assertDatabaseCount(t, store.db, `SELECT count(*) FROM audit_event WHERE event_type = 'runtime_bound' + AND reason_code = 'runtime_bound' AND account_id = 'account-a' AND browser_env_alias = 'environment-a' + AND network_exit_id = $1 AND runtime_instance_id = $2 AND binding_version = $3 AND details = '{}'::jsonb`, + 1, active.Exit.ID, active.RuntimeInstanceID, active.BindingVersion) second, err := store.CreateNetworkExit(ctx, NetworkExit{Protocol: "http", Host: "proxy-2.example", Port: 8080}, "") if err != nil { @@ -362,12 +366,27 @@ func TestNetworkExitBindingRuntimeAndAuditWorkflow(t *testing.T) { if err != nil || rebound.Exit.ID != second.ID || rebound.BindingVersion != 2 { t.Fatalf("expired runtime must be transactionally released before rebind: %#v err=%v", rebound, err) } + assertDatabaseCount(t, store.db, `SELECT count(*) FROM audit_event WHERE event_type = 'runtime_released' + AND reason_code = 'runtime_released' AND account_id = 'account-a' AND browser_env_alias = 'environment-a' + AND network_exit_id = $1 AND runtime_instance_id = $2 AND binding_version = $3 AND details = '{}'::jsonb`, + 1, active.Exit.ID, active.RuntimeInstanceID, active.BindingVersion) if _, err := store.db.ExecContext(ctx, `UPDATE social_account SET status = 'active' WHERE id = 'account-a'`); err != nil { t.Fatal(err) } - if _, err := store.ActivateRuntime(ctx, env.Alias, "same-exit-container", rebound.BindingVersion, rebound.Exit.ID, "network-same-exit"); err != nil { - t.Fatalf("activate runtime before same-exit rebind: %v", err) + expiredBeforeActivation, err := store.ActivateRuntime(ctx, env.Alias, "expired-container", rebound.BindingVersion, rebound.Exit.ID, "network-expired") + if err != nil { + t.Fatalf("activate runtime to expire: %v", err) } + if _, err := store.db.ExecContext(ctx, `UPDATE runtime_instance SET lease_until = now() - interval '1 second' WHERE id = $1`, expiredBeforeActivation.RuntimeInstanceID); err != nil { + t.Fatal(err) + } + if _, err := store.ActivateRuntime(ctx, env.Alias, "same-exit-container", rebound.BindingVersion, rebound.Exit.ID, "network-same-exit"); err != nil { + t.Fatalf("replace expired runtime before same-exit rebind: %v", err) + } + assertDatabaseCount(t, store.db, `SELECT count(*) FROM audit_event WHERE event_type = 'runtime_released' + AND reason_code = 'runtime_released' AND account_id = 'account-a' AND browser_env_alias = 'environment-a' + AND network_exit_id = $1 AND runtime_instance_id = $2 AND binding_version = $3 AND details = '{}'::jsonb`, + 1, expiredBeforeActivation.Exit.ID, expiredBeforeActivation.RuntimeInstanceID, expiredBeforeActivation.BindingVersion) if _, err := store.db.ExecContext(ctx, `UPDATE social_account SET status = 'paused' WHERE id = 'account-a'`); err != nil { t.Fatal(err) } @@ -381,6 +400,10 @@ func TestNetworkExitBindingRuntimeAndAuditWorkflow(t *testing.T) { if err := store.ReleaseRuntime(ctx, active); err != nil { t.Fatal(err) } + assertDatabaseCount(t, store.db, `SELECT count(*) FROM audit_event WHERE event_type = 'runtime_released' + AND reason_code = 'runtime_released' AND account_id = 'account-a' AND browser_env_alias = 'environment-a' + AND network_exit_id = $1 AND runtime_instance_id = $2 AND binding_version = $3 AND details = '{}'::jsonb`, + 1, active.Exit.ID, active.RuntimeInstanceID, active.BindingVersion) rebound, err = store.RebindEnvironment(ctx, env.Alias, second.ID, "rebound-container", rebound.BindingVersion) if err != nil || rebound.BindingVersion != 3 || rebound.RuntimeID != "rebound-container" { t.Fatalf("same-exit rebind must atomically CAS the binding and runtime: %#v err=%v", rebound, err) @@ -399,6 +422,9 @@ func TestNetworkExitBindingRuntimeAndAuditWorkflow(t *testing.T) { if err := store.SetRuntimeCleanupPending(ctx, cleanup, true); err != nil { t.Fatalf("set generation cleanup pending: %v", err) } + assertDatabaseCount(t, store.db, `SELECT count(*) FROM audit_event WHERE event_type = 'runtime_released' + AND browser_env_alias = 'environment-a' AND network_exit_id = $1 AND runtime_instance_id = $2 AND binding_version = $3`, + 1, current.Exit.ID, current.RuntimeInstanceID, current.BindingVersion) wrongCleanup := cleanup wrongCleanup.RuntimeCleanupRuntimeID = "other-container" if err := store.SetRuntimeCleanupPending(ctx, wrongCleanup, false); !errors.Is(err, ErrConflict) {