From f13e8b1ab7fb0a5b6ecdce644f947d3f266b722c Mon Sep 17 00:00:00 2001 From: weichen Date: Mon, 12 Oct 2026 14:49:40 +0000 Subject: [PATCH 1/5] Commit account changes and audits in one transaction Why: creating an account and its activation key, and changing a password, were separate commits, so a crash could leave a user without a key or a new password with a still-valid login key. Evidence: SELF go test ./internal/store including TestConcurrentCreateOneWinner, TestPasswordChangeIsAtomic, TestOpenExistingDoesNotMigrateOrCreate, and TestBootstrapRefusesExistingBusiness. PostgreSQL cases are NOT_RUN locally. Impact: strict insert returns ErrConflict instead of overwriting. Account CLI opens an existing database without migrating. --dsn and --password still work and warn. capabilities --json does not touch the database. Refs: docs/svc-deploy/u7-account-tx-audit.md Co-authored-by: Cursor --- cmd/agent-system/account_io.go | 89 +++++++ cmd/agent-system/main.go | 35 ++- internal/app/app.go | 146 ++++++----- internal/server/accounts.go | 21 +- internal/store/account_lifecycle_test.go | 130 ++++++++++ internal/store/account_tx.go | 309 +++++++++++++++++++++++ internal/store/auth.go | 19 +- internal/store/login_keys.go | 35 +-- internal/store/migrate.go | 23 ++ internal/store/store.go | 32 ++- 10 files changed, 701 insertions(+), 138 deletions(-) create mode 100644 cmd/agent-system/account_io.go create mode 100644 internal/store/account_lifecycle_test.go create mode 100644 internal/store/account_tx.go diff --git a/cmd/agent-system/account_io.go b/cmd/agent-system/account_io.go new file mode 100644 index 0000000..65db6da --- /dev/null +++ b/cmd/agent-system/account_io.go @@ -0,0 +1,89 @@ +package main + +import ( + "encoding/json" + "flag" + "fmt" + "io" + "os" +) + +func runCapabilities(args []string) int { + fs := flag.NewFlagSet("capabilities", flag.ExitOnError) + asJSON := fs.Bool("json", false, "以 JSON 打印能力，不打开数据库") + _ = fs.Parse(args) + caps := map[string]any{ + "accountTransaction": true, + "accountAudit": true, + "accountOpsOpen": true, + "passwordStdin": true, + "dsnFile": true, + "outFile": true, + } + if !*asJSON { + fmt.Fprintln(os.Stderr, "用法：agent-system capabilities --json") + return 2 + } + enc := json.NewEncoder(os.Stdout) + enc.SetIndent("", " ") + if err := enc.Encode(caps); err != nil { + fmt.Fprintln(os.Stderr, "capabilities:", err) + return 1 + } + return 0 +} + +func resolveAccountDSN(flagDSN, dsnFile string) (string, error) { + if flagDSN != "" { + fmt.Fprintln(os.Stderr, "warning: --dsn is deprecated; use AGENT_SYSTEM_DSN or --dsn-file") + } + if dsnFile != "" { + info, err := os.Stat(dsnFile) + if err != nil { + return "", err + } + if info.Mode().Perm() != 0o600 { + return "", fmt.Errorf("dsn file %s must be mode 0600", dsnFile) + } + raw, err := os.ReadFile(dsnFile) + if err != nil { + return "", err + } + return string(trimNewline(raw)), nil + } + if env := os.Getenv("AGENT_SYSTEM_DSN"); env != "" { + return env, nil + } + return flagDSN, nil +} + +func resolveAccountPassword(flagPassword string, fromStdin bool) (string, error) { + if flagPassword != "" { + fmt.Fprintln(os.Stderr, "warning: --password is deprecated; use --password-stdin") + } + if !fromStdin { + return flagPassword, nil + } + raw, err := io.ReadAll(os.Stdin) + if err != nil { + return "", err + } + return string(trimNewline(raw)), nil +} + +func writeExclusiveSecret(path, body string) error { + f, err := os.OpenFile(path, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0o600) + if err != nil { + return err + } + defer f.Close() + _, err = io.WriteString(f, body+"\n") + return err +} + +func trimNewline(raw []byte) []byte { + for len(raw) > 0 && (raw[len(raw)-1] == '\n' || raw[len(raw)-1] == '\r') { + raw = raw[:len(raw)-1] + } + return raw +} diff --git a/cmd/agent-system/main.go b/cmd/agent-system/main.go index 391d27a..4dc04e1 100644 --- a/cmd/agent-system/main.go +++ b/cmd/agent-system/main.go @@ -35,6 +35,8 @@ func main() { os.Exit(runUser(os.Args[2:])) case "identity": os.Exit(runIdentity(os.Args[2:])) + case "capabilities": + os.Exit(runCapabilities(os.Args[2:])) } } fs := flag.NewFlagSet("serve", flag.ExitOnError) @@ -236,10 +238,14 @@ func runUser(args []string) int { sub := args[0] fs := flag.NewFlagSet("user "+sub, flag.ExitOnError) data := fs.String("data", "./data", "数据目录") - dsn := fs.String("dsn", "", "PostgreSQL 档 DSN（空 = 嵌入式档）") + dsn := fs.String("dsn", "", "PostgreSQL 档 DSN（空 = 嵌入式档；已弃用，优先 AGENT_SYSTEM_DSN 或 --dsn-file）") + dsnFile := fs.String("dsn-file", "", "0600 权限的 DSN 文件") owner := fs.String("owner", "", "用户名（owner）") name := fs.String("name", "", "显示名") - password := fs.String("password", "", "密码（create 空=激活 Key；加 --generate-password 则随机）") + password := fs.String("password", "", "密码（已弃用；请用 --password-stdin）") + passwordStdin := fs.Bool("password-stdin", false, "从 stdin 读取密码") + outFile := fs.String("out-file", "", "把明文凭据独占写入该文件（0600）") + instanceID := fs.String("instance-id", "", "严格核验库内服务身份") admin := fs.Bool("admin", false, "管理员") role := fs.String("role", "user", "角色 user|admin") openExisting := fs.Bool("open-existing", false, "只打开既有存储") @@ -249,6 +255,21 @@ func runUser(args []string) int { revokeAll := fs.Bool("revoke-all-credentials", false, "签发时撤销全部长期 API Key 与会话") keyID := fs.String("id", "", "revoke-login-key 凭据编号") fs.Parse(args[1:]) + resolvedDSN, err := resolveAccountDSN(*dsn, *dsnFile) + if err != nil { + fmt.Fprintln(os.Stderr, "user:", err) + return 1 + } + resolvedPassword, err := resolveAccountPassword(*password, *passwordStdin) + if err != nil { + fmt.Fprintln(os.Stderr, "user:", err) + return 1 + } + if *instanceID != "" { + _ = os.Setenv("AGENT_SYSTEM_INSTANCE_ID", *instanceID) + } + *dsn = resolvedDSN + *password = resolvedPassword switch sub { case "create": if *owner == "" { @@ -264,11 +285,19 @@ func runUser(args []string) int { fmt.Fprintf(os.Stderr, "user create: 未知角色 %q\n", *role) return 2 } - pw, err := app.CreateUserAccount(*dsn, *data, *owner, *name, *password, isAdmin, *openExisting, *generatePassword) + pw, opID, err := app.CreateUserAccount(*dsn, *data, *owner, *name, *password, isAdmin, *openExisting, *generatePassword) if err != nil { fmt.Fprintln(os.Stderr, "user create:", err) return 1 } + if *outFile != "" { + if err := writeExclusiveSecret(*outFile, pw); err != nil { + fmt.Fprintln(os.Stderr, "user create:", err) + return 1 + } + fmt.Fprintf(os.Stderr, "operation_id=%s\n", opID) + return 0 + } fmt.Println(pw) if *password == "" && !*generatePassword { fmt.Fprintln(os.Stderr, "一次性激活 Key 仅此一次输出；兑换后设置密码。") diff --git a/internal/app/app.go b/internal/app/app.go index 2244da4..143ddfe 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -724,8 +724,8 @@ func Run(opt Options) error { // CreateUserAccount CLI 建号。 // password 空且 generatePassword=false：无初始密码，同事务签发 activate Key，返回明文 Key。 // generatePassword=true 或显式 password：兼容旧路径（随机/给定密码 + must_change）。 -func CreateUserAccount(storageDSN, dataDir, owner, name, password string, isAdmin, openExisting, generatePassword bool) (string, error) { - var out string +func CreateUserAccount(storageDSN, dataDir, owner, name, password string, isAdmin, openExisting, generatePassword bool) (string, string, error) { + var out, opID string err := withStoreMode(storageDSN, dataDir, openExisting, func(st *store.Store) error { if password != "" || generatePassword { pw := password @@ -736,23 +736,29 @@ func CreateUserAccount(storageDSN, dataDir, owner, name, password string, isAdmi if err != nil { return err } - if err := st.CreateAccount(owner, name, hash, isAdmin); err != nil { + outc, err := st.CreateActivatedAccount(store.AccountWrite{ + Owner: owner, Name: name, PasswordHash: hash, IsAdmin: isAdmin, MustChange: true, + Actor: cliActor(), InstanceID: accountInstanceID(), Action: "account.create", + }) + if err != nil { return err } out = pw + opID = outc.OperationID return nil } - if err := st.CreateAccount(owner, name, "", isAdmin); err != nil { - return err - } - plain, _, err := st.IssueLoginKey(owner, store.LoginKeyPurposeActivate, "cli", 0) + outc, err := st.CreateActivatedAccount(store.AccountWrite{ + Owner: owner, Name: name, IsAdmin: isAdmin, Purpose: store.LoginKeyPurposeActivate, CreatedBy: "cli", + Actor: cliActor(), InstanceID: accountInstanceID(), Action: "account.create", + }) if err != nil { return err } - out = plain + out = outc.Plain + opID = outc.OperationID return nil }) - return out, err + return out, opID, err } // IssueUserLoginKey CLI 签发激活/恢复 Key。 @@ -760,10 +766,12 @@ func IssueUserLoginKey(storageDSN, dataDir, owner, purpose string, openExisting, var plain string err := withStoreMode(storageDSN, dataDir, openExisting, func(st *store.Store) error { if revokeAllCreds { - if err := st.RevokeAllAPIKeys(owner); err != nil { - return err - } - st.DeleteUserSessions(owner, "") + outc, err := st.RevokeAllAndIssue(store.AccountWrite{ + Owner: owner, Purpose: purpose, CreatedBy: "cli", TTL: ttl, + Actor: cliActor(), InstanceID: accountInstanceID(), + }) + plain = outc.Plain + return err } p, _, err := st.IssueLoginKey(owner, purpose, "cli", ttl) plain = p @@ -783,18 +791,10 @@ func RevokeUserLoginKey(storageDSN, dataDir, id string, openExisting bool) error func BootstrapAdmin(storageDSN, dataDir, owner string, openExisting bool) (string, error) { var plain string err := withStoreMode(storageDSN, dataDir, openExisting, func(st *store.Store) error { - users, err := st.ListUsersAdmin() - if err != nil { - return err - } - if len(users) > 0 { - return fmt.Errorf("bootstrap-admin: users already exist (%d); refuse takeover", len(users)) - } - if err := st.CreateAccount(owner, owner, "", true); err != nil { - return err - } - p, _, err := st.IssueLoginKey(owner, store.LoginKeyPurposeActivate, "cli-bootstrap", 0) - plain = p + outc, err := st.BootstrapFirstAdmin(store.AccountWrite{ + Owner: owner, Name: owner, Actor: cliActor(), InstanceID: accountInstanceID(), + }) + plain = outc.Plain return err }) return plain, err @@ -810,11 +810,11 @@ func ResetUserPassword(storageDSN, dataDir, owner, password string, openExisting return "", err } if err := withStoreMode(storageDSN, dataDir, openExisting, func(st *store.Store) error { - if err := st.SetUserPassword(owner, hash); err != nil { - return err - } - st.DeleteUserSessions(owner, "") - return nil + _, err := st.ApplyPasswordChange(store.AccountWrite{ + Owner: owner, PasswordHash: hash, MustChange: true, + Actor: cliActor(), InstanceID: accountInstanceID(), Action: "account.reset_password", + }) + return err }); err != nil { return "", err } @@ -857,58 +857,56 @@ func withStoreMode(storageDSN, dataDir string, openExisting bool, fn func(*store return err } defer st.Close() + if id := strings.TrimSpace(os.Getenv("AGENT_SYSTEM_INSTANCE_ID")); id != "" { + dbref := strings.TrimSpace(os.Getenv("AGENT_SYSTEM_DBREF")) + if dbref == "" { + if storageDSN != "" { + dbref = "pg" + } else { + dbref = "sqlite" + } + } + if err := st.VerifySystemIdentity(id, dbref); err != nil { + return err + } + } return fn(st) } // CreateAPIKey 是 apikey 子命令的实现：在指定存储档上为 owner 创建密钥（明文仅打印一次）。 func CreateAPIKey(storageDSN, dataDir, owner, name string) (string, error) { - var st *store.Store - var err error - if storageDSN != "" { - st, err = store.OpenPostgres(storageDSN, filepath.Join(dataDir, "artifacts")) - } else { - st, err = store.OpenSQLite(filepath.Join(dataDir, "agent-system.db")) - } - if err != nil { - return "", err - } - defer st.Close() - return st.CreateAPIKey(owner, name) + var plain string + err := withStoreMode(storageDSN, dataDir, true, func(st *store.Store) error { + p, e := st.CreateAPIKey(owner, name) + plain = p + return e + }) + return plain, err } // ListAPIKeys 列出租户密钥（摘要视图）。 func ListAPIKeys(storageDSN, dataDir, owner string) ([]store.APIKeyInfo, error) { - var st *store.Store - var err error - if storageDSN != "" { - st, err = store.OpenPostgres(storageDSN, filepath.Join(dataDir, "artifacts")) - } else { - st, err = store.OpenSQLite(filepath.Join(dataDir, "agent-system.db")) - } - if err != nil { - return nil, err - } - defer st.Close() - if _, err := st.GetUser(owner); err != nil { - return nil, fmt.Errorf("owner %q 不存在（先 apikey create 创建）", owner) - } - return st.ListAPIKeys(owner) + var out []store.APIKeyInfo + err := withStoreMode(storageDSN, dataDir, true, func(st *store.Store) error { + if _, err := st.GetUser(owner); err != nil { + return fmt.Errorf("owner %q 不存在（先 apikey create 创建）", owner) + } + var e error + out, e = st.ListAPIKeys(owner) + return e + }) + return out, err } // RevokeAPIKey 撤销指定名称密钥。 func RevokeAPIKey(storageDSN, dataDir, owner, name string) (bool, error) { - var st *store.Store - var err error - if storageDSN != "" { - st, err = store.OpenPostgres(storageDSN, filepath.Join(dataDir, "artifacts")) - } else { - st, err = store.OpenSQLite(filepath.Join(dataDir, "agent-system.db")) - } - if err != nil { - return false, err - } - defer st.Close() - return st.RevokeAPIKeyByName(owner, name) + var ok bool + err := withStoreMode(storageDSN, dataDir, true, func(st *store.Store) error { + var e error + ok, e = st.RevokeAPIKeyByName(owner, name) + return e + }) + return ok, err } // EnsureSystemIdentity CLI：显式建立库内服务身份（svc init / I2→I3 引导）。 @@ -920,11 +918,19 @@ func EnsureSystemIdentity(storageDSN, dataDir, instanceID, dbref string) error { // VerifySystemIdentity CLI：校验库内服务身份（svc start 前；失败不插入）。 func VerifySystemIdentity(storageDSN, dataDir, instanceID, dbref string) error { - return withStore(storageDSN, dataDir, func(st *store.Store) error { + return withStoreMode(storageDSN, dataDir, true, func(st *store.Store) error { return st.VerifySystemIdentity(instanceID, dbref) }) } +func cliActor() string { + return fmt.Sprintf("uid:%d", os.Getuid()) +} + +func accountInstanceID() string { + return strings.TrimSpace(os.Getenv("AGENT_SYSTEM_INSTANCE_ID")) +} + // cachedInt 带短 TTL 的整型缓存读（限流值热读取免打库）。 func cachedInt(ttl time.Duration, fn func() int) func() int { var mu sync.Mutex diff --git a/internal/server/accounts.go b/internal/server/accounts.go index a753161..16dc240 100644 --- a/internal/server/accounts.go +++ b/internal/server/accounts.go @@ -194,7 +194,9 @@ func (s *Server) changePassword(w http.ResponseWriter, r *http.Request) { return } if forceSet { - if err := s.st.CompleteSetPassword(owner, hash); err != nil { + if _, err := s.st.ApplyPasswordChange(store.AccountWrite{ + Owner: owner, PasswordHash: hash, Actor: owner, Action: "account.set_password", + }); err != nil { writeStoreErr(w, err) return } @@ -205,15 +207,12 @@ func (s *Server) changePassword(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusOK, out) return } - if err := s.st.SetUserPassword(owner, hash); err != nil { + if _, err := s.st.ApplyPasswordChange(store.AccountWrite{ + Owner: owner, PasswordHash: hash, ExceptToken: tok, Actor: owner, Action: "account.change_password", + }); err != nil { writeStoreErr(w, err) return } - if err := s.st.ClearMustChangePassword(owner); err != nil { - writeStoreErr(w, err) - return - } - s.st.DeleteUserSessions(owner, tok) st, _ := s.st.GetUserAuthState(owner) writeJSON(w, http.StatusOK, authStateJSON(st)) } @@ -289,7 +288,8 @@ func (s *Server) createAdminUser(w http.ResponseWriter, r *http.Request) { // resetAdminUserPassword：管理员重置密码 → must_change=1，吊销该用户全部会话。 func (s *Server) resetAdminUserPassword(w http.ResponseWriter, r *http.Request) { - if _, ok := s.requireAdmin(w, r); !ok { + actor, ok := s.requireAdmin(w, r) + if !ok { return } owner := r.PathValue("owner") @@ -309,11 +309,12 @@ func (s *Server) resetAdminUserPassword(w http.ResponseWriter, r *http.Request) http.Error(w, err.Error(), http.StatusInternalServerError) return } - if err := s.st.SetUserPassword(owner, hash); err != nil { + if _, err := s.st.ApplyPasswordChange(store.AccountWrite{ + Owner: owner, PasswordHash: hash, MustChange: true, Actor: actor, Action: "account.reset_password", + }); err != nil { writeStoreErr(w, err) return } - s.st.DeleteUserSessions(owner, "") st, _ := s.st.GetUserAuthState(owner) writeJSON(w, http.StatusOK, map[string]any{ "owner": st.Owner, "mustChangePassword": st.MustChangePassword, diff --git a/internal/store/account_lifecycle_test.go b/internal/store/account_lifecycle_test.go new file mode 100644 index 0000000..40f3c0f --- /dev/null +++ b/internal/store/account_lifecycle_test.go @@ -0,0 +1,130 @@ +package store + +import ( + "errors" + "fmt" + "os" + "path/filepath" + "sync" + "testing" +) + +func TestConcurrentCreateOneWinner(t *testing.T) { + st := openLife(t) + const n = 8 + var wg sync.WaitGroup + wins := make(chan error, n) + for i := 0; i < n; i++ { + wg.Add(1) + go func() { + defer wg.Done() + _, err := st.CreateActivatedAccount(AccountWrite{ + Owner: "same", Name: "Same", Purpose: LoginKeyPurposeActivate, CreatedBy: "test", + Action: "account.create", + }) + wins <- err + }() + } + wg.Wait() + close(wins) + okN, conflict := 0, 0 + for err := range wins { + switch { + case err == nil: + okN++ + case errors.Is(err, ErrConflict): + conflict++ + default: + t.Fatal(err) + } + } + if okN != 1 || conflict != n-1 { + t.Fatalf("wins=%d conflicts=%d", okN, conflict) + } +} + +func TestPasswordChangeIsAtomic(t *testing.T) { + st := openLife(t) + hash, err := HashPassword("old-password") + if err != nil { + t.Fatal(err) + } + if _, err := st.CreateActivatedAccount(AccountWrite{Owner: "ada", Name: "Ada", PasswordHash: hash, Action: "account.create"}); err != nil { + t.Fatal(err) + } + plain, _, err := st.IssueLoginKey("ada", LoginKeyPurposeActivate, "test", 0) + if err != nil { + t.Fatal(err) + } + st.SetAccountFault(fmt.Errorf("injected")) + next, err := HashPassword("new-password") + if err != nil { + t.Fatal(err) + } + if _, err := st.ApplyPasswordChange(AccountWrite{Owner: "ada", PasswordHash: next, Action: "account.password"}); err == nil { + t.Fatal("faulted change returned success") + } + st.SetAccountFault(nil) + if _, ok := st.VerifyLogin("ada", "old-password"); !ok { + t.Fatal("old password did not survive the failed change") + } + if _, err := st.ApplyPasswordChange(AccountWrite{Owner: "ada", PasswordHash: next, Action: "account.password"}); err != nil { + t.Fatal(err) + } + if _, _, err := st.RedeemLoginKey(plain); err == nil { + t.Fatal("old key redeemed after password change") + } + if _, ok := st.VerifyLogin("ada", "new-password"); !ok { + t.Fatal("new password missing") + } + n, err := st.CountAudit("ada", "rejected") + if err != nil || n == 0 { + t.Fatalf("rejected audit n=%d err=%v", n, err) + } +} + +func TestOpenExistingDoesNotMigrateOrCreate(t *testing.T) { + missing := filepath.Join(t.TempDir(), "absent", "agent-system.db") + if _, err := OpenExistingSQLite(missing); err == nil { + t.Fatal("missing file opened") + } + if _, err := os.Stat(filepath.Dir(missing)); !os.IsNotExist(err) { + t.Fatalf("parent created: %v", err) + } + path := filepath.Join(t.TempDir(), "agent-system.db") + st, err := OpenSQLite(path) + if err != nil { + t.Fatal(err) + } + if err := st.Close(); err != nil { + t.Fatal(err) + } + again, err := OpenExistingSQLite(path) + if err != nil { + t.Fatal(err) + } + again.Close() +} + +func TestBootstrapRefusesExistingBusiness(t *testing.T) { + st := openLife(t) + if err := st.EnsureSystemIdentity("inst", "sqlite"); err != nil { + t.Fatal(err) + } + if _, err := st.CreateProject("owner", "p"); err != nil { + t.Fatal(err) + } + if _, err := st.BootstrapFirstAdmin(AccountWrite{Owner: "root", Name: "root"}); err == nil { + t.Fatal("bootstrap accepted a database that already has a project") + } +} + +func openLife(t *testing.T) *Store { + t.Helper() + st, err := OpenSQLite(filepath.Join(t.TempDir(), "agent-system.db")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { st.Close() }) + return st +} diff --git a/internal/store/account_tx.go b/internal/store/account_tx.go new file mode 100644 index 0000000..a5fd4c9 --- /dev/null +++ b/internal/store/account_tx.go @@ -0,0 +1,309 @@ +package store + +import ( + "crypto/rand" + "database/sql" + "encoding/hex" + "fmt" + "time" +) + +// AccountWrite 是一次账户变更的调用参数。成功审计与变更同事务。 +type AccountWrite struct { + Owner string + Name string + PasswordHash string + IsAdmin bool + MustChange bool + ExceptToken string + Purpose string + CreatedBy string + TTL time.Duration + RevokeAllKeys bool + Actor string + InstanceID string + Action string +} + +// AccountOutcome 是成功提交后可以对外展示的标识。不含密码、哈希或明文。 +type AccountOutcome struct { + Plain string + CredentialID string + OperationID string +} + +// CreateActivatedAccount 严格插入账户。passwordHash 为空且 Purpose 非空时，同一事务签发登录 Key。 +func (s *Store) CreateActivatedAccount(w AccountWrite) (AccountOutcome, error) { + if w.Action == "" { + w.Action = "account.create" + } + var out AccountOutcome + err := s.accountTx(w, func(tx *sql.Tx, opID string) error { + if err := insertAccount(s, tx, w); err != nil { + return err + } + if w.PasswordHash == "" && w.Purpose != "" { + plain, id, err := insertLoginKey(s, tx, w) + if err != nil { + return err + } + out.Plain, out.CredentialID = plain, id + } + out.OperationID = opID + return nil + }) + return out, err +} + +// ApplyPasswordChange 在同一事务里写哈希、更新强制改密标志、作废未使用登录 Key，并处理会话。 +func (s *Store) ApplyPasswordChange(w AccountWrite) (AccountOutcome, error) { + if w.Action == "" { + w.Action = "account.password" + } + var out AccountOutcome + err := s.accountTx(w, func(tx *sql.Tx, opID string) error { + res, err := tx.Exec(s.q(`UPDATE users SET password_hash=?, password_set_at=?, must_change_password=? WHERE owner=?`), + w.PasswordHash, nowISO(), w.MustChange, w.Owner) + if err != nil { + return err + } + if n, _ := res.RowsAffected(); n == 0 { + return ErrNotFound + } + if _, err := tx.Exec(s.q(`UPDATE login_keys SET status=? WHERE owner=? AND status=?`), + LoginKeyStatusRevoked, w.Owner, LoginKeyStatusUnused); err != nil { + return err + } + if w.ExceptToken != "" { + _, err = tx.Exec(s.q(`DELETE FROM sessions WHERE owner=? AND token_hash<>?`), w.Owner, hashKey(w.ExceptToken)) + } else { + _, err = tx.Exec(s.q(`DELETE FROM sessions WHERE owner=?`), w.Owner) + } + if err != nil { + return err + } + out.OperationID = opID + return nil + }) + return out, err +} + +// BootstrapFirstAdmin 在立即事务（SQLite）或身份行锁（PostgreSQL）内复核空库后建首位管理员。 +func (s *Store) BootstrapFirstAdmin(w AccountWrite) (AccountOutcome, error) { + w.IsAdmin = true + w.PasswordHash = "" + w.Purpose = LoginKeyPurposeActivate + if w.CreatedBy == "" { + w.CreatedBy = "cli-bootstrap" + } + if w.Action == "" { + w.Action = "account.bootstrap" + } + var out AccountOutcome + err := s.accountTx(w, func(tx *sql.Tx, opID string) error { + if err := lockIdentity(s, tx); err != nil { + return err + } + if err := requireEmptyBusiness(s, tx); err != nil { + return err + } + if err := insertAccount(s, tx, w); err != nil { + return err + } + plain, id, err := insertLoginKey(s, tx, w) + if err != nil { + return err + } + out.Plain, out.CredentialID, out.OperationID = plain, id, opID + return nil + }) + return out, err +} + +// RevokeAllAndIssue 把吊销长期密钥、删除会话、作废未使用登录 Key 和签发新 Key 放进一次提交。 +func (s *Store) RevokeAllAndIssue(w AccountWrite) (AccountOutcome, error) { + if w.Action == "" { + w.Action = "account.revoke_credentials" + } + var out AccountOutcome + err := s.accountTx(w, func(tx *sql.Tx, opID string) error { + if _, err := tx.Exec(s.q(`UPDATE api_keys SET revoked=1 WHERE owner=? AND revoked=0`), w.Owner); err != nil { + return err + } + if _, err := tx.Exec(s.q(`DELETE FROM sessions WHERE owner=?`), w.Owner); err != nil { + return err + } + if _, err := tx.Exec(s.q(`UPDATE login_keys SET status=? WHERE owner=? AND status=?`), + LoginKeyStatusRevoked, w.Owner, LoginKeyStatusUnused); err != nil { + return err + } + plain, id, err := insertLoginKey(s, tx, w) + if err != nil { + return err + } + out.Plain, out.CredentialID, out.OperationID = plain, id, opID + return nil + }) + return out, err +} + +func (s *Store) accountTx(w AccountWrite, fn func(*sql.Tx, string) error) error { + opID := newID("op") + tx, err := s.db.Begin() + if err != nil { + return err + } + committed := false + defer func() { + if !committed { + _ = tx.Rollback() + } + }() + if err := fn(tx, opID); err != nil { + _ = tx.Rollback() + committed = true + s.recordRejected(w, opID) + return err + } + if s.accountFault != nil { + err = s.accountFault + _ = tx.Rollback() + committed = true + s.recordRejected(w, opID) + return err + } + if err := insertAudit(s, tx, w, opID, "success"); err != nil { + return err + } + if err := tx.Commit(); err != nil { + s.recordRejected(w, opID) + return err + } + committed = true + return nil +} + +func (s *Store) recordRejected(w AccountWrite, opID string) { + _, _ = s.db.Exec(s.q(`INSERT INTO audit_events(time,owner,action,actor,target,entry,instance_id,operation_id,outcome) VALUES(?,?,?,?,?,?,?,?,?)`), + nowISO(), w.Owner, w.Action, w.Actor, w.Owner, "account", w.InstanceID, opID, "rejected") +} + +func insertAudit(s *Store, tx *sql.Tx, w AccountWrite, opID, outcome string) error { + _, err := tx.Exec(s.q(`INSERT INTO audit_events(time,owner,action,actor,target,entry,instance_id,operation_id,outcome) VALUES(?,?,?,?,?,?,?,?,?)`), + nowISO(), w.Owner, w.Action, w.Actor, w.Owner, "account", w.InstanceID, opID, outcome) + return err +} + +func insertAccount(s *Store, tx *sql.Tx, w AccountWrite) error { + mustChange := w.PasswordHash != "" || w.MustChange + setAt := "" + if w.PasswordHash != "" { + setAt = nowISO() + } + _, err := tx.Exec(s.q(`INSERT INTO users(id,owner,name,created_at,password_hash,password_set_at,must_change_password,is_admin) VALUES(?,?,?,?,?,?,?,?)`), + newID("usr"), w.Owner, w.Name, nowISO(), w.PasswordHash, setAt, mustChange, w.IsAdmin) + if isUniqueViolation(err) { + return ErrConflict + } + return err +} + +func insertLoginKey(s *Store, tx *sql.Tx, w AccountWrite) (string, string, error) { + purpose := w.Purpose + if purpose != LoginKeyPurposeActivate && purpose != LoginKeyPurposeRecover { + return "", "", fmt.Errorf("login key: invalid purpose %q", purpose) + } + ttl := w.TTL + if ttl <= 0 { + ttl = defaultLoginKeyTTL + } + if ttl < minLoginKeyTTL { + ttl = minLoginKeyTTL + } + if ttl > maxLoginKeyTTL { + ttl = maxLoginKeyTTL + } + raw := make([]byte, 32) + if _, err := rand.Read(raw); err != nil { + return "", "", err + } + idRaw := make([]byte, 8) + if _, err := rand.Read(idRaw); err != nil { + return "", "", err + } + plain := "alk_" + hex.EncodeToString(raw) + id := "lk_" + hex.EncodeToString(idRaw) + now := time.Now().UTC() + if _, err := tx.Exec(s.q(`UPDATE login_keys SET status=? WHERE owner=? AND purpose=? AND status=?`), + LoginKeyStatusRevoked, w.Owner, purpose, LoginKeyStatusUnused); err != nil { + return "", "", err + } + _, err := tx.Exec(s.q(`INSERT INTO login_keys(id,owner,purpose,key_hash,status,expires_at,created_at,created_by,instance_id) VALUES(?,?,?,?,?,?,?,?,?)`), + id, w.Owner, purpose, hashKey(plain), LoginKeyStatusUnused, now.Add(ttl).Format(time.RFC3339Nano), now.Format(time.RFC3339Nano), w.CreatedBy, w.InstanceID) + return plain, id, err +} + +func lockIdentity(s *Store, tx *sql.Tx) error { + q := `SELECT instance_id FROM agent_system_identity` + if s.dialect.name() == "postgres" { + q += ` FOR UPDATE` + } + rows, err := tx.Query(q) + if err != nil { + return err + } + defer rows.Close() + n := 0 + for rows.Next() { + var id string + if err := rows.Scan(&id); err != nil { + return err + } + n++ + } + if err := rows.Err(); err != nil { + return err + } + if n == 0 { + return fmt.Errorf("%w: agent_system_identity 无行", ErrSystemIdentityMissing) + } + return nil +} + +func requireEmptyBusiness(s *Store, tx *sql.Tx) error { + for _, table := range []string{"users", "projects", "conversations", "messages", "runs", "api_keys", "sessions", "login_keys"} { + var n int + if err := tx.QueryRow(`SELECT COUNT(1) FROM ` + table).Scan(&n); err != nil { + return err + } + if n > 0 { + return fmt.Errorf("bootstrap: %s is not empty", table) + } + } + return nil +} + +func (s *Store) requireCurrentSchema() error { + applied, err := s.MigrationStatus() + if err != nil || len(applied) == 0 { + return fmt.Errorf("schema 版本不可读；请运行 agent-system migrate，或先启动一次服务") + } + latest := migrations[len(migrations)-1].version + got := applied[len(applied)-1] + if got != latest { + return fmt.Errorf("schema v%d 与本二进制 v%d 不一致；请运行 agent-system migrate，或先启动一次服务", got, latest) + } + return nil +} + +// SetAccountFault 让下一次账户事务在提交前失败，用于半态测试。 +func (s *Store) SetAccountFault(err error) { + s.accountFault = err +} + +// CountAudit 返回指定 outcome 的账户审计条数。 +func (s *Store) CountAudit(owner, outcome string) (int, error) { + var n int + err := s.db.QueryRow(s.q(`SELECT COUNT(1) FROM audit_events WHERE owner=? AND entry=? AND outcome=?`), owner, "account", outcome).Scan(&n) + return n, err +} diff --git a/internal/store/auth.go b/internal/store/auth.go index 0161cb7..ce53a9e 100644 --- a/internal/store/auth.go +++ b/internal/store/auth.go @@ -221,21 +221,10 @@ func VerifyPassword(hash, password string) bool { // CreateAccount 管理员建号：owner 唯一。 // passwordHash 非空时 must_change=1；空哈希表示未设密（走激活 Key），must_change=0。 func (s *Store) CreateAccount(owner, name, passwordHash string, isAdmin bool) error { - if _, err := s.GetUser(owner); err == nil { - return ErrConflict - } - if err := s.EnsureUser(owner, name); err != nil { - return err - } - mustChange := passwordHash != "" - setAt := "" - if passwordHash != "" { - setAt = nowISO() - } - // 布尔列一律绑 Go bool 参数、不写 1/0 字面量：SQLite 两侧都容忍， - // PostgreSQL 对 boolean 列收到 integer 会 42804（2026-10-05 user create 半失败根因）。 - _, err := s.db.Exec(s.q(`UPDATE users SET name=?, password_hash=?, password_set_at=?, must_change_password=?, is_admin=? WHERE owner=?`), - name, passwordHash, setAt, mustChange, isAdmin, owner) + _, err := s.CreateActivatedAccount(AccountWrite{ + Owner: owner, Name: name, PasswordHash: passwordHash, IsAdmin: isAdmin, + Action: "account.create", + }) return err } diff --git a/internal/store/login_keys.go b/internal/store/login_keys.go index 3a3f4ad..032439f 100644 --- a/internal/store/login_keys.go +++ b/internal/store/login_keys.go @@ -23,10 +23,10 @@ const ( SessionScopeFull = "full" SessionScopeSetPasswordOnly = "set_password_only" - defaultLoginKeyTTL = 30 * time.Minute - minLoginKeyTTL = 5 * time.Minute - maxLoginKeyTTL = 120 * time.Minute - restrictedSessionTTL = 10 * time.Minute + defaultLoginKeyTTL = 30 * time.Minute + minLoginKeyTTL = 5 * time.Minute + maxLoginKeyTTL = 120 * time.Minute + restrictedSessionTTL = 10 * time.Minute ) // LoginKeyMeta 对外可见元数据（无明文）。 @@ -79,8 +79,8 @@ func (s *Store) IssueLoginKey(owner, purpose, createdBy string, ttl time.Duratio LoginKeyStatusRevoked, owner, purpose, LoginKeyStatusUnused); err != nil { return "", meta, err } - if _, err := tx.Exec(s.q(`INSERT INTO login_keys(id,owner,purpose,key_hash,status,expires_at,created_at,created_by) VALUES(?,?,?,?,?,?,?,?)`), - id, owner, purpose, hashKey(plain), LoginKeyStatusUnused, exp.Format(time.RFC3339Nano), now.Format(time.RFC3339Nano), createdBy); err != nil { + if _, err := tx.Exec(s.q(`INSERT INTO login_keys(id,owner,purpose,key_hash,status,expires_at,created_at,created_by,instance_id) VALUES(?,?,?,?,?,?,?,?,?)`), + id, owner, purpose, hashKey(plain), LoginKeyStatusUnused, exp.Format(time.RFC3339Nano), now.Format(time.RFC3339Nano), createdBy, ""); err != nil { return "", meta, err } if err := tx.Commit(); err != nil { @@ -158,27 +158,8 @@ func (s *Store) RedeemLoginKey(plain string) (token string, owner string, err er // CompleteSetPassword 设密事务：写 hash、清 must_change、作废未用登录 Key、删除全部会话。 func (s *Store) CompleteSetPassword(owner, passwordHash string) error { - tx, err := s.db.Begin() - if err != nil { - return err - } - defer tx.Rollback() - res, err := tx.Exec(s.q(`UPDATE users SET password_hash=?, password_set_at=?, must_change_password=? WHERE owner=?`), - passwordHash, nowISO(), false, owner) - if err != nil { - return err - } - if n, _ := res.RowsAffected(); n == 0 { - return ErrNotFound - } - if _, err := tx.Exec(s.q(`UPDATE login_keys SET status=? WHERE owner=? AND status=?`), - LoginKeyStatusRevoked, owner, LoginKeyStatusUnused); err != nil { - return err - } - if _, err := tx.Exec(s.q(`DELETE FROM sessions WHERE owner=?`), owner); err != nil { - return err - } - return tx.Commit() + _, err := s.ApplyPasswordChange(AccountWrite{Owner: owner, PasswordHash: passwordHash, Action: "account.set_password"}) + return err } // CountUnusedLoginKeys 未使用登录 Key 计数（list 展示）。 diff --git a/internal/store/migrate.go b/internal/store/migrate.go index 6dbac23..31869f9 100644 --- a/internal/store/migrate.go +++ b/internal/store/migrate.go @@ -778,6 +778,29 @@ func init() { }, "postgres": { `DROP TABLE IF EXISTS agent_system_identity`, }}}, + migration{version: 28, name: "account_audit", + up: map[string][]string{"sqlite": { + `ALTER TABLE audit_events ADD COLUMN entry TEXT NOT NULL DEFAULT ''`, + `ALTER TABLE audit_events ADD COLUMN instance_id TEXT NOT NULL DEFAULT ''`, + `ALTER TABLE audit_events ADD COLUMN operation_id TEXT NOT NULL DEFAULT ''`, + `ALTER TABLE audit_events ADD COLUMN outcome TEXT NOT NULL DEFAULT ''`, + }, "postgres": { + `ALTER TABLE audit_events ADD COLUMN entry TEXT NOT NULL DEFAULT ''`, + `ALTER TABLE audit_events ADD COLUMN instance_id TEXT NOT NULL DEFAULT ''`, + `ALTER TABLE audit_events ADD COLUMN operation_id TEXT NOT NULL DEFAULT ''`, + `ALTER TABLE audit_events ADD COLUMN outcome TEXT NOT NULL DEFAULT ''`, + }}, + down: map[string][]string{"sqlite": { + `ALTER TABLE audit_events DROP COLUMN outcome`, + `ALTER TABLE audit_events DROP COLUMN operation_id`, + `ALTER TABLE audit_events DROP COLUMN instance_id`, + `ALTER TABLE audit_events DROP COLUMN entry`, + }, "postgres": { + `ALTER TABLE audit_events DROP COLUMN outcome`, + `ALTER TABLE audit_events DROP COLUMN operation_id`, + `ALTER TABLE audit_events DROP COLUMN instance_id`, + `ALTER TABLE audit_events DROP COLUMN entry`, + }}}, ) } diff --git a/internal/store/store.go b/internal/store/store.go index ba64e01..9595115 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -77,6 +77,7 @@ type Store struct { migrateMu sync.Mutex migrateFault error // 非空时 migratePass 在「DDL 已执行、版本未登记」处失败（原子性测试） + accountFault error // 非空时账户事务在提交前失败（半态测试） vecPGOnce sync.Once // pgvector 列探测（U6-3：一次性缓存） vecPG bool @@ -87,11 +88,11 @@ func OpenSQLite(path string) (*Store, error) { if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { return nil, err } - return openSQLiteFile(path, true) + return openSQLiteFile(path, true, true) } -// OpenExistingSQLite 打开既有嵌入式库：数据文件不存在则失败，不创建空库或父目录。 -// 仍执行兼容迁移（requireCompatible）；用于账户运维，避免恢复操作顺带建库。 +// OpenExistingSQLite 打开既有嵌入式库：数据文件不存在则失败，不创建空库或父目录，也不迁移。 +// schema 与本二进制不一致时拒绝，并提示使用 migrate 或先启动服务。 func OpenExistingSQLite(path string) (*Store, error) { if _, err := os.Stat(path); err != nil { if os.IsNotExist(err) { @@ -99,10 +100,10 @@ func OpenExistingSQLite(path string) (*Store, error) { } return nil, err } - return openSQLiteFile(path, false) + return openSQLiteFile(path, false, false) } -func openSQLiteFile(path string, createArtifacts bool) (*Store, error) { +func openSQLiteFile(path string, createArtifacts, migrateSchema bool) (*Store, error) { // _txlock=immediate：写事务在 BEGIN 即取库写锁（而非首个写语句升级）， // 跨进程迁移/写路径依赖 busy_timeout 排队而不是中途 SQLITE_BUSY（KTD8）。 // busy_timeout 30s：迁移竞争测试在慢速 CI 盘上 4 个迁移者串行全程可超 5s @@ -114,7 +115,7 @@ func openSQLiteFile(path string, createArtifacts bool) (*Store, error) { } // SQLite 单写者：串行化连接，语句级原子 + 版本化迁移事务。 db.SetMaxOpenConns(1) - return finishOpen(db, sqliteDialect{}, filepath.Join(filepath.Dir(path), "artifacts"), nil, PoolOptions{}, createArtifacts) + return finishOpen(db, sqliteDialect{}, filepath.Join(filepath.Dir(path), "artifacts"), nil, PoolOptions{}, createArtifacts, migrateSchema) } // OpenPostgres 打开 PostgreSQL 档存储并执行迁移（默认连接预算）。 @@ -132,15 +133,15 @@ func OpenPostgresWithBlobs(dsn, artifactsDir string, blobs blobstore.Store) (*St // OpenPostgresWithOptions 按 KTD12 连接预算打开 PG 档；QueryTimeoutSec>0 时 // 以 statement_timeout 运行参数附着到 DSN（对 URL 与键值两种形式均生效）。 func OpenPostgresWithOptions(dsn, artifactsDir string, pool PoolOptions, blobs blobstore.Store) (*Store, error) { - return openPostgresWithOptions(dsn, artifactsDir, pool, blobs, true) + return openPostgresWithOptions(dsn, artifactsDir, pool, blobs, true, true) } -// OpenExistingPostgres 打开既有 PG 库并执行兼容迁移；不创建 artifacts 目录。 +// OpenExistingPostgres 打开既有 PG 库，不创建 artifacts 目录，也不迁移。 func OpenExistingPostgres(dsn, artifactsDir string) (*Store, error) { - return openPostgresWithOptions(dsn, artifactsDir, PoolOptions{}, nil, false) + return openPostgresWithOptions(dsn, artifactsDir, PoolOptions{}, nil, false, false) } -func openPostgresWithOptions(dsn, artifactsDir string, pool PoolOptions, blobs blobstore.Store, createArtifacts bool) (*Store, error) { +func openPostgresWithOptions(dsn, artifactsDir string, pool PoolOptions, blobs blobstore.Store, createArtifacts, migrateSchema bool) (*Store, error) { pool = pool.effective() if pool.QueryTimeoutSec > 0 { dsn = withPGRuntimeParam(dsn, "statement_timeout", strconv.Itoa(pool.QueryTimeoutSec*1000)) @@ -155,7 +156,7 @@ func openPostgresWithOptions(dsn, artifactsDir string, pool PoolOptions, blobs b if pool.ConnMaxLifetimeSec > 0 { db.SetConnMaxLifetime(time.Duration(pool.ConnMaxLifetimeSec) * time.Second) } - return finishOpen(db, postgresDialect{}, artifactsDir, blobs, pool, createArtifacts) + return finishOpen(db, postgresDialect{}, artifactsDir, blobs, pool, createArtifacts, migrateSchema) } // withPGRuntimeParam 向 DSN 追加运行时参数（libpq 语义：URL 查询串或键值对）。 @@ -181,7 +182,7 @@ func AppendPGRuntimeParam(dsn, key, value string) string { return dsn + sep + kv } -func finishOpen(db *sql.DB, d dialect, artifactsDir string, blobs blobstore.Store, pool PoolOptions, createArtifacts bool) (*Store, error) { +func finishOpen(db *sql.DB, d dialect, artifactsDir string, blobs blobstore.Store, pool PoolOptions, createArtifacts, migrateSchema bool) (*Store, error) { if blobs == nil { var err error if createArtifacts { @@ -195,7 +196,12 @@ func finishOpen(db *sql.DB, d dialect, artifactsDir string, blobs blobstore.Stor } } s := &Store{db: db, dialect: d, dataDir: artifactsDir, blobs: blobs, pool: pool} - if err := s.migrate(); err != nil { + if migrateSchema { + if err := s.migrate(); err != nil { + db.Close() + return nil, err + } + } else if err := s.requireCurrentSchema(); err != nil { db.Close() return nil, err } From 545a75b54e52f8fb4dd38474870c05d3436f2e3d Mon Sep 17 00:00:00 2001 From: weichen Date: Mon, 12 Oct 2026 14:52:21 +0000 Subject: [PATCH 2/5] Prove a password still verifies after the database is reopened Why: a commit that only checks the password in the same process does not show the hash was durable. Evidence: SELF go test TestPasswordSurvivesReopen. Impact: test only. Refs: docs/svc-deploy/u7-account-tx-audit.md Co-authored-by: Cursor --- internal/store/account_lifecycle_test.go | 26 ++++++++++++++++++++++++ 1 file changed, 26 insertions(+) diff --git a/internal/store/account_lifecycle_test.go b/internal/store/account_lifecycle_test.go index 40f3c0f..0004d14 100644 --- a/internal/store/account_lifecycle_test.go +++ b/internal/store/account_lifecycle_test.go @@ -119,6 +119,32 @@ func TestBootstrapRefusesExistingBusiness(t *testing.T) { } } +func TestPasswordSurvivesReopen(t *testing.T) { + path := filepath.Join(t.TempDir(), "agent-system.db") + st, err := OpenSQLite(path) + if err != nil { + t.Fatal(err) + } + hash, err := HashPassword("keep-password") + if err != nil { + t.Fatal(err) + } + if _, err := st.CreateActivatedAccount(AccountWrite{Owner: "ada", Name: "Ada", PasswordHash: hash}); err != nil { + t.Fatal(err) + } + if err := st.Close(); err != nil { + t.Fatal(err) + } + st, err = OpenSQLite(path) + if err != nil { + t.Fatal(err) + } + defer st.Close() + if _, ok := st.VerifyLogin("ada", "keep-password"); !ok { + t.Fatal("password did not survive reopen") + } +} + func openLife(t *testing.T) *Store { t.Helper() st, err := OpenSQLite(filepath.Join(t.TempDir(), "agent-system.db")) From 571e2bd28af6c21fc13b5fbc55ec5ded1c15b0a8 Mon Sep 17 00:00:00 2001 From: weichen Date: Wed, 14 Oct 2026 03:42:57 +0000 Subject: [PATCH 3/5] Verify every maintenance entry refuses an absent database Why: The safe open requirement covers user, apikey and identity verify, including default CLI paths. Evidence: All three paths use private missing directories; previous full Go and PostgreSQL receipts cover the implementation, with a focused recheck before push. Impact: Test coverage only; no runtime behavior or migration changes. Refs: Backend PR #185; Dev PR #486 --- internal/app/account_ops_existing_test.go | 32 ++++++++++++++++++----- 1 file changed, 26 insertions(+), 6 deletions(-) diff --git a/internal/app/account_ops_existing_test.go b/internal/app/account_ops_existing_test.go index 00fba27..9580649 100644 --- a/internal/app/account_ops_existing_test.go +++ b/internal/app/account_ops_existing_test.go @@ -9,12 +9,32 @@ import ( func TestAccountMaintenanceDefaultsToNonCreatingOpen(t *testing.T) { t.Setenv("AGENT_SYSTEM_DSN", "") - dir := filepath.Join(t.TempDir(), "absent") - if _, _, err := CreateUserAccount("", dir, "ada", "Ada", "", false, false, false); err == nil { - t.Error("ordinary maintenance created a missing database") - } - if _, err := os.Stat(dir); !os.IsNotExist(err) { - t.Error("failed maintenance created a directory") + for _, tc := range []struct { + name string + run func(string) error + }{ + {"user", func(dir string) error { + _, _, err := CreateUserAccount("", dir, "ada", "Ada", "", false, false, false) + return err + }}, + {"apikey", func(dir string) error { + _, err := CreateAPIKey("", dir, "ada", "test") + return err + }}, + {"identity verify", func(dir string) error { + _, err := VerifySystemIdentity("", dir, "fixture", "sqlite") + return err + }}, + } { + t.Run(tc.name, func(t *testing.T) { + dir := filepath.Join(t.TempDir(), "absent") + if err := tc.run(dir); err == nil { + t.Error("maintenance created a missing database") + } + if _, err := os.Stat(dir); !os.IsNotExist(err) { + t.Error("failed maintenance created a directory") + } + }) } } func TestAccountMaintenanceCannotImplicitlyMigrate(t *testing.T) { From 76e7cc7c18b68a0847306e3a273de16c6455b614 Mon Sep 17 00:00:00 2001 From: weichen Date: Wed, 14 Oct 2026 03:47:11 +0000 Subject: [PATCH 4/5] Deliver API keys through exclusive private files Why: API-key issuance shares the account maintenance no-create and credential-output contract; the old CLI printed its key and lacked --out-file. Evidence: The new subprocess test failed before the flag existed, then passed for private output and pre-mutation collision refusal; current go vet and all Go tests pass. Impact: apikey create accepts protected --dsn-file and --out-file, preserves stdout default, and never overwrites an existing destination. Refs: Backend PR #185; Dev PR #486; U7 KTD14 --- cmd/agent-system/account_io_test.go | 68 +++++++++++++++++++++++++++++ cmd/agent-system/main.go | 23 +++++++++- 2 files changed, 90 insertions(+), 1 deletion(-) diff --git a/cmd/agent-system/account_io_test.go b/cmd/agent-system/account_io_test.go index 4d91a02..1096b8c 100644 --- a/cmd/agent-system/account_io_test.go +++ b/cmd/agent-system/account_io_test.go @@ -2,8 +2,10 @@ package main import ( "agent-system-backend/internal/store" + "bytes" "io" "os" + "os/exec" "path/filepath" "testing" ) @@ -88,3 +90,69 @@ func TestExistingOutputAndConflictingFormatsRefuseBeforeMutation(t *testing.T) { }) } } + +func TestAPIKeyCreateUsesExclusivePrivateOutput(t *testing.T) { + if dir := os.Getenv("U7_APIKEY_CHILD_DIR"); dir != "" { + rc := runAPIKey([]string{"create", "--data", dir, "--owner", "ada", "--name", "fixture", "--out-file", os.Getenv("U7_APIKEY_CHILD_OUT")}) + if rc != 0 { + t.Fatal("apikey create rejected its private output destination") + } + return + } + for _, existing := range []bool{false, true} { + name := "new" + if existing { + name = "existing" + } + t.Run(name, func(t *testing.T) { + dir, st := cliStore(t) + if err := st.CreateAccount("ada", "Ada", "", false); err != nil { + t.Fatal(err) + } + out := filepath.Join(t.TempDir(), "credential") + if existing { + if err := os.WriteFile(out, []byte("preserve"), 0600); err != nil { + t.Fatal(err) + } + } + cmd := exec.Command(os.Args[0], "-test.run=^TestAPIKeyCreateUsesExclusivePrivateOutput$") + cmd.Env = append(os.Environ(), "U7_APIKEY_CHILD_DIR="+dir, "U7_APIKEY_CHILD_OUT="+out) + var stdout, stderr bytes.Buffer + cmd.Stdout = &stdout + cmd.Stderr = &stderr + err := cmd.Run() + if existing { + if err == nil { + t.Fatal("existing output was overwritten or mutation succeeded") + } + raw, readErr := os.ReadFile(out) + if readErr != nil || string(raw) != "preserve" { + t.Fatal("existing output was modified") + } + keys, listErr := st.ListAPIKeys("ada") + if listErr != nil || len(keys) != 0 { + t.Fatal("mutation happened before output preflight") + } + return + } + if err != nil { + t.Fatal("new private output was rejected") + } + info, statErr := os.Stat(out) + if statErr != nil || info.Mode().Perm() != 0600 { + t.Fatal("credential output missing or not private") + } + raw, readErr := os.ReadFile(out) + if readErr != nil || len(raw) == 0 { + t.Fatal("credential output empty") + } + if bytes.Contains(stdout.Bytes(), bytes.TrimSpace(raw)) { + t.Fatal("credential was written to stdout") + } + keys, listErr := st.ListAPIKeys("ada") + if listErr != nil || len(keys) != 1 { + t.Fatal("successful issue was not committed") + } + }) + } +} diff --git a/cmd/agent-system/main.go b/cmd/agent-system/main.go index 4b88321..66a22a9 100644 --- a/cmd/agent-system/main.go +++ b/cmd/agent-system/main.go @@ -196,22 +196,43 @@ func runAPIKey(args []string) int { fs := flag.NewFlagSet("apikey "+sub, flag.ExitOnError) data := fs.String("data", "./data", "数据目录") dsn := fs.String("dsn", "", "PostgreSQL 档 DSN（空 = 嵌入式档）") + dsnFile := fs.String("dsn-file", "", "0600权限的PostgreSQL DSN文件") owner := fs.String("owner", "", "租户名（owner）") name := fs.String("name", "default", "密钥名称") + outFile := fs.String("out-file", "", "把明文密钥独占写入该文件（0600）") fs.Parse(args[1:]) if *owner == "" { fmt.Fprintln(os.Stderr, "apikey: --owner 必填") return 2 } + if sub != "create" && *outFile != "" { + fmt.Fprintln(os.Stderr, "apikey: 此子命令不生成密钥") + return 2 + } + resolvedDSN, err := resolveAccountDSN(*dsn, *dsnFile) + if err != nil { + fmt.Fprintln(os.Stderr, "apikey:", err) + return 1 + } + *dsn = resolvedDSN switch sub { case "create": + delivery, err := reserveCredentialOutput(*outFile, "") + if err != nil { + fmt.Fprintln(os.Stderr, "apikey create:", err) + return 1 + } + defer delivery.abort() key, err := app.CreateAPIKey(*dsn, *data, *owner, *name) if err != nil { fmt.Fprintln(os.Stderr, "apikey create:", err) return 1 } // 明文仅此一次输出；库中只存 SHA-256 摘要。 - fmt.Println(key) + if err := delivery.deliver("api-key", *owner, key, nil); err != nil { + fmt.Fprintln(os.Stderr, "apikey create:", err) + return 1 + } return 0 case "list": keys, err := app.ListAPIKeys(*dsn, *data, *owner) From a4e33f192fbf5aa7ac6196b8ea1fcabc8832bd1d Mon Sep 17 00:00:00 2001 From: weichen Date: Wed, 14 Oct 2026 04:08:14 +0000 Subject: [PATCH 5/5] Serialize account credential and session changes Why: Concurrent password changes, recovery key reissue, and session admission could accept retired credentials or omit rejected audits. Evidence: Local Go vet and full tests passed; real PostgreSQL store suite, deterministic interleavings, and two-process contention passed with SELF receipts. Impact: Account writes and session admission serialize per owner; stale sessions fail closed; checked rejections carry an audit row. Refs: Agent-System-Dev #486; Agent-System-Backend #185; merge queue Issue #518 --- internal/app/app.go | 2 +- internal/store/account_process_crash_test.go | 16 +- .../account_security_integration_test.go | 84 ++++++ .../store/account_serialization_pg_test.go | 167 +++++++++++ internal/store/account_two_process_pg_test.go | 59 +++- internal/store/account_tx.go | 50 ++-- internal/store/auth.go | 36 ++- internal/store/dialect.go | 8 + internal/store/login_keys.go | 261 ++++++++++-------- 9 files changed, 529 insertions(+), 154 deletions(-) create mode 100644 internal/store/account_serialization_pg_test.go diff --git a/internal/app/app.go b/internal/app/app.go index 0effc9b..fe050c6 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -959,7 +959,7 @@ func VerifySystemIdentity(storageDSN, dataDir, instanceID, dbref string) (int, e } func cliActor() string { - return fmt.Sprintf("uid:%d", os.Getuid()) + return fmt.Sprintf("%d", os.Getuid()) } func accountInstanceID() string { diff --git a/internal/store/account_process_crash_test.go b/internal/store/account_process_crash_test.go index 4ec21c5..61fb4fe 100644 --- a/internal/store/account_process_crash_test.go +++ b/internal/store/account_process_crash_test.go @@ -2,6 +2,7 @@ package store import ( "bufio" + "errors" "fmt" "os" "os/exec" @@ -27,9 +28,8 @@ func TestAccountPasswordSurvivesProcessTermination(t *testing.T) { if err != nil { t.Fatal(err) } - if err := st.Close(); err != nil { - t.Fatal(err) - } + // Keep the database connection and process alive at the signal point; + // a graceful close would only test reopen, not abrupt termination. fmt.Println("U7_COMMITTED") select {} } @@ -82,7 +82,15 @@ func TestAccountPasswordSurvivesProcessTermination(t *testing.T) { if err := child.Process.Signal(tc.signal); err != nil { t.Fatal(err) } - _ = child.Wait() + waitErr := child.Wait() + var exitErr *exec.ExitError + if !errors.As(waitErr, &exitErr) { + t.Fatalf("child was not terminated by %s: %v", tc.name, waitErr) + } + status, ok := exitErr.Sys().(syscall.WaitStatus) + if !ok || !status.Signaled() || status.Signal() != tc.signal { + t.Fatalf("child ended without requested %s: %v", tc.name, waitErr) + } reopened, err := OpenExistingSQLite(path) if err != nil { t.Fatal(err) diff --git a/internal/store/account_security_integration_test.go b/internal/store/account_security_integration_test.go index 1dd18e6..8359ad5 100644 --- a/internal/store/account_security_integration_test.go +++ b/internal/store/account_security_integration_test.go @@ -101,3 +101,87 @@ func TestRevokeAllRefusesMissingOwnerWithoutCredential(t *testing.T) { t.Fatalf("orphan credential rows=%d err=%v", n, err) } } + +func TestActivePasswordChangeRevokesUnusedRecoveryKey(t *testing.T) { + s := activeAccount(t) + key, _, err := s.IssueLoginKey("ada", LoginKeyPurposeRecover, "admin", 0) + if err != nil { + t.Fatal(err) + } + session, err := s.CreateSession("ada") + if err != nil { + t.Fatal(err) + } + hash, err := HashPassword("replacement-password") + if err != nil { + t.Fatal(err) + } + if err := s.ChangeUserPasswordActive("ada", "old-password", hash, session); err != nil { + t.Fatal(err) + } + if _, _, err := s.RedeemLoginKey(key); err == nil { + t.Fatal("unused recovery key issued before active password change still redeems") + } +} + +func TestRejectedActiveChangeIsAudited(t *testing.T) { + s := activeAccount(t) + h, err := HashPassword("next-password") + if err != nil { + t.Fatal(err) + } + err = s.ChangeUserPasswordActive("ada", "incorrect-current-password", h, "") + if !errors.Is(err, ErrBadCredentials) { + t.Fatalf("expected bad credentials, got %v", err) + } + count, err := s.CountAudit("ada", "rejected") + if err != nil { + t.Fatal(err) + } + if count != 1 { + t.Fatalf("rejected checked account mutation has %d rejected audit rows, want1", count) + } +} + +func TestDelayedOldVersionSessionIsRejectedAtUse(t *testing.T) { + s := activeAccount(t) + _, ver, ok := s.VerifyLoginWithVersion("ada", "old-password") + if !ok { + t.Fatal("healthy verification missing") + } + h, err := HashPassword("new-password") + if err != nil { + t.Fatal(err) + } + if _, err := s.ApplyPasswordChange(AccountWrite{Owner: "ada", PasswordHash: h}); err != nil { + t.Fatal(err) + } + // Represents an insert racing after the replacement's earlier DELETE. + token := "sess_private_stale_fixture" + _, err = s.db.Exec(s.q(`INSERT INTO sessions(token_hash,owner,created_at,expires_at,last_seen_at,scope,auth_version) VALUES(?,?,?,?,?,?,?)`), + hashKey(token), "ada", nowISO(), time.Now().Add(time.Hour).UTC().Format(time.RFC3339Nano), nowISO(), SessionScopeFull, ver) + if err != nil { + t.Fatal(err) + } + if _, _, ok := s.SessionInfo(token); ok { + t.Fatal("session carrying the retired auth_version became authorized") + } +} + +func TestActivePasswordChangePreservesCurrentVersionedSession(t *testing.T) { + s := activeAccount(t) + token, err := s.CreateSession("ada") + if err != nil { + t.Fatal(err) + } + h, err := HashPassword("new-password") + if err != nil { + t.Fatal(err) + } + if err := s.ChangeUserPasswordActive("ada", "old-password", h, token); err != nil { + t.Fatal(err) + } + if owner, scope, ok := s.SessionInfo(token); !ok || owner != "ada" || scope != SessionScopeFull { + t.Fatal("active change did not retain its current full session") + } +} diff --git a/internal/store/account_serialization_pg_test.go b/internal/store/account_serialization_pg_test.go new file mode 100644 index 0000000..436c086 --- /dev/null +++ b/internal/store/account_serialization_pg_test.go @@ -0,0 +1,167 @@ +package store + +import ( + "context" + "testing" + "time" +) + +// The trigger holds a transaction after its authorization read. A second +// connection must wait for that owner's lock before replacing credentials. +func holdPGAccountWrite(t *testing.T, s *Store, table, event, suffix string) func() { + t.Helper() + lock := "u7_account_" + suffix + _, err := s.db.Exec(`CREATE FUNCTION u7_pause_account() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN PERFORM pg_advisory_xact_lock(hashtext(current_schema() || ':` + lock + `')); RETURN NEW; END $$`) + if err != nil { + t.Fatal(err) + } + _, err = s.db.Exec(`CREATE TRIGGER u7_pause_account BEFORE ` + event + ` ON ` + table + ` FOR EACH ROW EXECUTE FUNCTION u7_pause_account()`) + if err != nil { + t.Fatal(err) + } + conn, err := s.db.Conn(context.Background()) + if err != nil { + t.Fatal(err) + } + if _, err := conn.ExecContext(context.Background(), `SELECT pg_advisory_lock(hashtext(current_schema() || ':`+lock+`'))`); err != nil { + conn.Close() + t.Fatal(err) + } + released := false + release := func() { + if !released { + released = true + _, _ = conn.ExecContext(context.Background(), `SELECT pg_advisory_unlock(hashtext(current_schema() || ':`+lock+`'))`) + _ = conn.Close() + } + } + t.Cleanup(release) + return release +} + +func waitForPGAccountWrite(t *testing.T, s *Store, suffix string) { + t.Helper() + lock := "u7_account_" + suffix + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) { + var n int + err := s.db.QueryRow(`SELECT count(*) FROM pg_locks WHERE locktype='advisory' AND NOT granted + AND objid=(hashtext(current_schema() || ':` + lock + `')::bigint & 4294967295)::oid`).Scan(&n) + if err != nil { + t.Fatal(err) + } + if n > 0 { + return + } + time.Sleep(10 * time.Millisecond) + } + t.Fatal("account write did not reach the deterministic PostgreSQL pause") +} + +func awaitPGAccountResult[T any](t *testing.T, done <-chan T) T { + t.Helper() + select { + case result := <-done: + return result + case <-time.After(5 * time.Second): + t.Fatal("serialized account transaction did not finish") + var zero T + return zero + } +} + +func TestPGVerifiedSessionAndPasswordChangeSerialize(t *testing.T) { + s := activeAccount(t) + if s.Dialect() != "postgres" { + t.Skip("real PostgreSQL required") + } + _, version, ok := s.VerifyLoginWithVersion("ada", "old-password") + if !ok { + t.Fatal("initial login failed") + } + release := holdPGAccountWrite(t, s, "sessions", "INSERT", "session") + type sessionResult struct { + token string + err error + } + sessionDone := make(chan sessionResult, 1) + go func() { + token, err := s.CreateSessionVerified("ada", SessionScopeFull, time.Hour, version) + sessionDone <- sessionResult{token, err} + }() + waitForPGAccountWrite(t, s, "session") + hash, err := HashPassword("replacement-password") + if err != nil { + t.Fatal(err) + } + changeDone := make(chan error, 1) + go func() { + _, err := s.ApplyPasswordChange(AccountWrite{Owner: "ada", PasswordHash: hash, Actor: "admin"}) + changeDone <- err + }() + select { + case err := <-changeDone: + t.Fatalf("replacement committed ahead of the in-flight verified session: %v", err) + case <-time.After(100 * time.Millisecond): + } + release() + session := awaitPGAccountResult(t, sessionDone) + if session.err != nil { + t.Fatal(session.err) + } + if err := awaitPGAccountResult(t, changeDone); err != nil { + t.Fatal(err) + } + if _, _, ok := s.SessionInfo(session.token); ok { + t.Fatal("old verified session survived later password replacement") + } +} + +func TestPGRestrictedPasswordSetAndKeyReissueSerialize(t *testing.T) { + s := openTestStore(t) + if s.Dialect() != "postgres" { + t.Skip("real PostgreSQL required") + } + if err := s.CreateAccount("ada", "Ada", "", false); err != nil { + t.Fatal(err) + } + key, _, err := s.IssueLoginKey("ada", LoginKeyPurposeRecover, "admin", 0) + if err != nil { + t.Fatal(err) + } + token, _, err := s.RedeemLoginKey(key) + if err != nil { + t.Fatal(err) + } + hash, err := HashPassword("replacement-password") + if err != nil { + t.Fatal(err) + } + release := holdPGAccountWrite(t, s, "users", "UPDATE", "password") + setDone := make(chan error, 1) + go func() { + _, err := s.CompleteSetPasswordChecked(token, hash) + setDone <- err + }() + waitForPGAccountWrite(t, s, "password") + issueDone := make(chan error, 1) + go func() { + _, _, err := s.IssueLoginKey("ada", LoginKeyPurposeRecover, "admin", 0) + issueDone <- err + }() + select { + case err := <-issueDone: + t.Fatalf("key replacement committed ahead of the in-flight password set: %v", err) + case <-time.After(100 * time.Millisecond): + } + release() + if err := awaitPGAccountResult(t, setDone); err != nil { + t.Fatal(err) + } + if err := awaitPGAccountResult(t, issueDone); err != nil { + t.Fatal(err) + } + if _, _, ok := s.SessionInfo(token); ok { + t.Fatal("old restricted session survived password set and key reissue") + } +} diff --git a/internal/store/account_two_process_pg_test.go b/internal/store/account_two_process_pg_test.go index b281e9e..fea8054 100644 --- a/internal/store/account_two_process_pg_test.go +++ b/internal/store/account_two_process_pg_test.go @@ -2,6 +2,7 @@ package store import ( "bytes" + "context" "database/sql" "errors" "fmt" @@ -24,6 +25,9 @@ func TestAccountTwoProcessPostgresAdmission(t *testing.T) { t.Fatal(err) } defer st.Close() + if err := os.WriteFile(os.Getenv("U7_ACCOUNT_CHILD_READY"), []byte("ready"), 0o600); err != nil { + t.Fatal(err) + } for i := 0; i < 500; i++ { if _, err := os.Stat(barrier); err == nil { break @@ -96,14 +100,38 @@ func TestAccountTwoProcessPostgresAdmission(t *testing.T) { t.Fatal(err) } st.Close() + // Hold the shared admission point so both independent processes + // are demonstrably contending before either may commit. + lockKey := "bootstrap:first-admin" + if tc.mode == "create" { + lockKey = "account:same" + } + lockConn, err := admin.Conn(context.Background()) + if err != nil { + t.Fatal(err) + } + defer lockConn.Close() + if _, err := lockConn.ExecContext(context.Background(), `SELECT pg_advisory_lock(hashtext($1))`, lockKey); err != nil { + t.Fatal(err) + } + released := false + release := func() { + if !released { + released = true + _, _ = lockConn.ExecContext(context.Background(), `SELECT pg_advisory_unlock(hashtext($1))`, lockKey) + } + } + defer release() barrier := filepath.Join(t.TempDir(), "go") + readyDir := t.TempDir() var children [2]*exec.Cmd var logs [2]bytes.Buffer for i := range children { + readyPath := filepath.Join(readyDir, fmt.Sprintf("ready-%d", i)) cmd := exec.Command(os.Args[0], "-test.run=^TestAccountTwoProcessPostgresAdmission$", "-test.v") cmd.Env = append(os.Environ(), "U7_ACCOUNT_CHILD_MODE="+tc.mode, "U7_ACCOUNT_CHILD_DSN="+dsn, "U7_ACCOUNT_CHILD_OWNER="+tc.owners[i], "U7_ACCOUNT_CHILD_ARTIFACTS="+artifacts, - "U7_ACCOUNT_CHILD_BARRIER="+barrier) + "U7_ACCOUNT_CHILD_BARRIER="+barrier, "U7_ACCOUNT_CHILD_READY="+readyPath) cmd.Stdout = &logs[i] cmd.Stderr = &logs[i] if err := cmd.Start(); err != nil { @@ -111,9 +139,38 @@ func TestAccountTwoProcessPostgresAdmission(t *testing.T) { } children[i] = cmd } + deadline := time.Now().Add(10 * time.Second) + for i := range children { + readyPath := filepath.Join(readyDir, fmt.Sprintf("ready-%d", i)) + for { + if _, err := os.Stat(readyPath); err == nil { + break + } + if time.Now().After(deadline) { + t.Fatalf("child %d never opened the shared database: %s", i, logs[i].String()) + } + time.Sleep(5 * time.Millisecond) + } + } if err := os.WriteFile(barrier, []byte("go"), 0o600); err != nil { t.Fatal(err) } + for { + var waiters int + err := admin.QueryRow(`SELECT count(*) FROM pg_locks WHERE locktype='advisory' AND NOT granted + AND objid=(hashtext($1)::bigint & 4294967295)::oid`, lockKey).Scan(&waiters) + if err != nil { + t.Fatal(err) + } + if waiters >= 2 { + break + } + if time.Now().After(deadline) { + t.Fatalf("only %d of 2 child processes reached the shared admission lock: %s | %s", waiters, logs[0].String(), logs[1].String()) + } + time.Sleep(5 * time.Millisecond) + } + release() wins, losses := 0, 0 for i, cmd := range children { if err := cmd.Wait(); err != nil { diff --git a/internal/store/account_tx.go b/internal/store/account_tx.go index 561625b..88f72b1 100644 --- a/internal/store/account_tx.go +++ b/internal/store/account_tx.go @@ -104,6 +104,12 @@ func (s *Store) ApplyPasswordChange(w AccountWrite) (AccountOutcome, error) { if err != nil { return err } + if w.ExceptToken != "" { + if _, err := tx.Exec(s.q(`UPDATE sessions SET auth_version=(SELECT auth_version FROM users WHERE owner=?) WHERE owner=? AND token_hash=?`), + w.Owner, w.Owner, hashKey(w.ExceptToken)); err != nil { + return err + } + } out.OperationID = opID return nil }) @@ -204,6 +210,16 @@ func (s *Store) accountTx(w AccountWrite, fn func(*sql.Tx, string) error) error _ = tx.Rollback() } }() + if w.Owner != "" { + if err := s.dialect.lockAccount(tx, w.Owner); err != nil { + return err + } + } + if w.InstanceID == "" { + if err := tx.QueryRow(`SELECT instance_id FROM agent_system_identity`).Scan(&w.InstanceID); err != nil && !errors.Is(err, sql.ErrNoRows) { + return err + } + } if err := fn(tx, opID); err != nil { _ = tx.Rollback() committed = true @@ -218,10 +234,14 @@ func (s *Store) accountTx(w AccountWrite, fn func(*sql.Tx, string) error) error return err } if err := insertAudit(s, tx, w, opID, "success"); err != nil { + _ = tx.Rollback() + committed = true + s.recordRejected(w, opID) return err } if err := tx.Commit(); err != nil { - s.recordRejected(w, opID) + // Commit may have reached the database even if the client lost its + // response. Recording a rejection here would assert an unknown outcome. return err } committed = true @@ -350,31 +370,3 @@ func (s *Store) CountAudit(owner, outcome string) (int, error) { err := s.db.QueryRow(s.q(`SELECT COUNT(1) FROM audit_events WHERE owner=? AND entry=? AND outcome=?`), owner, "account", outcome).Scan(&n) return n, err } - -// commitAccountMutation extends existing checked transactions without moving -// their authorization checks outside the transaction or replacing account_events. -func (s *Store) commitAccountMutation(tx *sql.Tx, w AccountWrite) error { - opID := newID("op") - if w.InstanceID == "" { - err := tx.QueryRow(`SELECT instance_id FROM agent_system_identity`).Scan(&w.InstanceID) - if err != nil && !errors.Is(err, sql.ErrNoRows) { - return err - } - } - if s.accountFault != nil { - err := s.accountFault - _ = tx.Rollback() - s.recordRejected(w, opID) - return err - } - if err := insertAudit(s, tx, w, opID, "success"); err != nil { - _ = tx.Rollback() - s.recordRejected(w, opID) - return err - } - if err := tx.Commit(); err != nil { - s.recordRejected(w, opID) - return err - } - return nil -} diff --git a/internal/store/auth.go b/internal/store/auth.go index 379c354..bb87030 100644 --- a/internal/store/auth.go +++ b/internal/store/auth.go @@ -313,17 +313,32 @@ func (s *Store) CreateSessionScoped(owner, scope string, ttl time.Duration) (str if ttl <= 0 { ttl = sessionTTL } - ver := 0 - _ = s.db.QueryRow(s.q(`SELECT auth_version FROM users WHERE owner=?`), owner).Scan(&ver) raw := make([]byte, 32) if _, err := rand.Read(raw); err != nil { return "", err } token := "sess_" + hex.EncodeToString(raw) exp := time.Now().Add(ttl) - _, err := s.db.Exec(s.q(`INSERT INTO sessions(token_hash,owner,created_at,expires_at,last_seen_at,scope,auth_version) VALUES(?,?,?,?,?,?,?)`), + tx, err := s.db.Begin() + if err != nil { + return "", err + } + defer tx.Rollback() + if err := s.dialect.lockAccount(tx, owner); err != nil { + return "", err + } + var ver int + if err := tx.QueryRow(s.q(`SELECT auth_version FROM users WHERE owner=?`), owner).Scan(&ver); errors.Is(err, sql.ErrNoRows) { + return "", ErrNotFound + } else if err != nil { + return "", err + } + _, err = tx.Exec(s.q(`INSERT INTO sessions(token_hash,owner,created_at,expires_at,last_seen_at,scope,auth_version) VALUES(?,?,?,?,?,?,?)`), hashKey(token), owner, nowISO(), exp.UTC().Format(time.RFC3339Nano), nowISO(), scope, ver) - return token, err + if err != nil { + return "", err + } + return token, tx.Commit() } // CreateSessionVerified 建会话并绑定期望认证版本：事务内重读账户版本，与密码验证 @@ -346,6 +361,9 @@ func (s *Store) CreateSessionVerified(owner, scope string, ttl time.Duration, ex return "", err } defer tx.Rollback() + if err := s.dialect.lockAccount(tx, owner); err != nil { + return "", err + } var ver int err = tx.QueryRow(s.q(`SELECT auth_version FROM users WHERE owner=?`), owner).Scan(&ver) if errors.Is(err, sql.ErrNoRows) { @@ -376,11 +394,17 @@ func (s *Store) SessionInfo(token string) (owner, scope string, ok bool) { return "", "", false } var exp string - err := s.db.QueryRow(s.q(`SELECT owner,expires_at,COALESCE(scope,'full') FROM sessions WHERE token_hash=?`), hashKey(token)). - Scan(&owner, &exp, &scope) + var sessionVersion, accountVersion int + err := s.db.QueryRow(s.q(`SELECT s.owner,s.expires_at,COALESCE(s.scope,'full'),s.auth_version,u.auth_version + FROM sessions s JOIN users u ON u.owner=s.owner WHERE s.token_hash=?`), hashKey(token)). + Scan(&owner, &exp, &scope, &sessionVersion, &accountVersion) if err != nil { return "", "", false } + if sessionVersion != accountVersion { + _, _ = s.db.Exec(s.q(`DELETE FROM sessions WHERE token_hash=? AND auth_version=?`), hashKey(token), sessionVersion) + return "", "", false + } if expT, err := time.Parse(time.RFC3339Nano, exp); err != nil || time.Now().After(expT) { s.db.Exec(s.q(`DELETE FROM sessions WHERE token_hash=?`), hashKey(token)) return "", "", false diff --git a/internal/store/dialect.go b/internal/store/dialect.go index 9bf2c84..5c54cea 100644 --- a/internal/store/dialect.go +++ b/internal/store/dialect.go @@ -20,6 +20,8 @@ type dialect interface { // lockBootstrap 锁定首次管理员引导的「核验无用户→建号→激活 Key」串行段 //（KTD12：防两个进程并发引导出双管理员）。 lockBootstrap(tx *sql.Tx) error + // lockAccount serializes credential and session admission for one owner. + lockAccount(tx *sql.Tx, owner string) error } // ---- SQLite（嵌入式档）---- @@ -32,6 +34,7 @@ func (sqliteDialect) lockConversation(*sql.Tx, string) error { return nil } func (sqliteDialect) lockScope(*sql.Tx, string, string) error { return nil } func (sqliteDialect) lockAdmission(*sql.Tx) error { return nil } func (sqliteDialect) lockBootstrap(*sql.Tx) error { return nil } +func (sqliteDialect) lockAccount(*sql.Tx, string) error { return nil } // ---- PostgreSQL（多用户/多实例档）---- @@ -72,3 +75,8 @@ func (postgresDialect) lockBootstrap(tx *sql.Tx) error { _, err := tx.Exec(`SELECT pg_advisory_xact_lock(hashtext($1))`, "bootstrap:first-admin") return err } + +func (postgresDialect) lockAccount(tx *sql.Tx, owner string) error { + _, err := tx.Exec(`SELECT pg_advisory_xact_lock(hashtext($1))`, "account:"+owner) + return err +} diff --git a/internal/store/login_keys.go b/internal/store/login_keys.go index 2413883..7ac1a73 100644 --- a/internal/store/login_keys.go +++ b/internal/store/login_keys.go @@ -112,6 +112,15 @@ func (s *Store) IssueLoginKey(owner, purpose, createdBy string, ttl time.Duratio return "", meta, err } defer tx.Rollback() + if err := s.dialect.lockAccount(tx, owner); err != nil { + return "", meta, err + } + var exists int + if err := tx.QueryRow(s.q(`SELECT 1 FROM users WHERE owner=?`), owner).Scan(&exists); errors.Is(err, sql.ErrNoRows) { + return "", meta, ErrNotFound + } else if err != nil { + return "", meta, err + } meta, err = s.insertLoginKeyTx(tx, owner, purpose, createdBy, plain, id, now, exp) if err != nil { return "", meta, err @@ -204,11 +213,20 @@ func (s *Store) RevokeLoginKey(id string) error { // RevokeLoginKeyCascade 撤销 Key 并作废该用户全部 set_password_only 会话（管理员 // HTTP 入口）。Key 已消费/过期时仍返回属主并作废会话；编号不存在才报 NotFound。 func (s *Store) RevokeLoginKeyCascade(id, actor string) (owner string, err error) { + if err := s.db.QueryRow(s.q(`SELECT owner FROM login_keys WHERE id=?`), id).Scan(&owner); errors.Is(err, sql.ErrNoRows) { + return "", ErrNotFound + } else if err != nil { + return "", err + } tx, err := s.db.Begin() if err != nil { return "", err } defer tx.Rollback() + if err := s.dialect.lockAccount(tx, owner); err != nil { + return "", err + } + lockedOwner := owner err = tx.QueryRow(s.q(`SELECT owner FROM login_keys WHERE id=?`), id).Scan(&owner) if errors.Is(err, sql.ErrNoRows) { return "", ErrNotFound @@ -216,6 +234,9 @@ func (s *Store) RevokeLoginKeyCascade(id, actor string) (owner string, err error if err != nil { return "", err } + if owner != lockedOwner { + return "", ErrConflict + } if _, err := tx.Exec(s.q(`UPDATE login_keys SET status=? WHERE id=? AND status=?`), LoginKeyStatusRevoked, id, LoginKeyStatusUnused); err != nil { return "", err @@ -279,11 +300,20 @@ func (s *Store) RedeemLoginKey(plain string) (token string, owner string, err er if plain == "" || len(plain) < 5 || plain[:4] != "alk_" { return "", "", ErrNotFound } + var lockOwner string + if err := s.db.QueryRow(s.q(`SELECT owner FROM login_keys WHERE key_hash=?`), hashKey(plain)).Scan(&lockOwner); errors.Is(err, sql.ErrNoRows) { + return "", "", ErrNotFound + } else if err != nil { + return "", "", err + } tx, err := s.db.Begin() if err != nil { return "", "", err } defer tx.Rollback() + if err := s.dialect.lockAccount(tx, lockOwner); err != nil { + return "", "", err + } var id, o, status, exp string err = tx.QueryRow(s.q(`SELECT id,owner,status,expires_at FROM login_keys WHERE key_hash=?`), hashKey(plain)). Scan(&id, &o, &status, &exp) @@ -293,6 +323,9 @@ func (s *Store) RedeemLoginKey(plain string) (token string, owner string, err er if err != nil { return "", "", err } + if o != lockOwner { + return "", "", ErrConflict + } expT, err := time.Parse(time.RFC3339Nano, exp) if err != nil || time.Now().After(expT) { _, _ = tx.Exec(s.q(`UPDATE login_keys SET status=? WHERE id=? AND status=?`), LoginKeyStatusExpired, id, LoginKeyStatusUnused) @@ -339,137 +372,139 @@ func (s *Store) RedeemLoginKey(plain string) (token string, owner string, err er // 写 hash、清 must_change、bump auth_version、作废未用登录 Key、删除全部会话、 // 写成功审计。会话或版本失效时返回 ErrConflict，不做任何写入。 func (s *Store) CompleteSetPasswordChecked(token, passwordHash string) (owner string, err error) { - tx, err := s.db.Begin() + // Discover only the lock key here. Authorization is checked again after + // acquiring the owner's transaction lock, alongside key reissue/revocation. + if err := s.db.QueryRow(s.q(`SELECT owner FROM sessions WHERE token_hash=?`), hashKey(token)).Scan(&owner); errors.Is(err, sql.ErrNoRows) { + return "", ErrNotFound + } else if err != nil { + return "", err + } + lockOwner := owner + err = s.accountTx(AccountWrite{Owner: owner, Actor: owner, Action: "account.set_password"}, func(tx *sql.Tx, _ string) error { + var scope, sessExp, actualOwner string + var sessVer, userVer int + var mustChange bool + var hash sql.NullString + err := tx.QueryRow(s.q(`SELECT s.owner,s.scope,s.expires_at,s.auth_version,u.auth_version,u.must_change_password,u.password_hash + FROM sessions s JOIN users u ON u.owner=s.owner WHERE s.token_hash=?`), hashKey(token)). + Scan(&actualOwner, &scope, &sessExp, &sessVer, &userVer, &mustChange, &hash) + if errors.Is(err, sql.ErrNoRows) { + return ErrNotFound + } + if err != nil { + return err + } + if actualOwner != lockOwner { + return ErrConflict + } + if expT, perr := time.Parse(time.RFC3339Nano, sessExp); perr != nil || time.Now().After(expT) { + return ErrNotFound + } + if scope != SessionScopeSetPasswordOnly && !mustChange && hash.Valid && hash.String != "" { + return ErrConflict + } + if sessVer != userVer { + return ErrConflict + } + res, err := tx.Exec(s.q(`UPDATE users SET password_hash=?, password_set_at=?, must_change_password=?, auth_version=auth_version+1 WHERE owner=?`), + passwordHash, nowISO(), false, lockOwner) + if err != nil { + return err + } + if n, _ := res.RowsAffected(); n != 1 { + return ErrNotFound + } + if _, err := tx.Exec(s.q(`UPDATE login_keys SET status=? WHERE owner=? AND status=?`), + LoginKeyStatusRevoked, lockOwner, LoginKeyStatusUnused); err != nil { + return err + } + if _, err := tx.Exec(s.q(`DELETE FROM sessions WHERE owner=?`), lockOwner); err != nil { + return err + } + entrance := "must-change" + if scope == SessionScopeSetPasswordOnly { + entrance = "recovery" + } + return s.insertAccountEvent(tx, lockOwner, entrance, lockOwner, "password_set", entrance) + }) if err != nil { return "", err } - defer tx.Rollback() - var scope, sessExp string - var sessVer, userVer int - var mustChange bool - var hash sql.NullString - err = tx.QueryRow(s.q(`SELECT s.owner,s.scope,s.expires_at,s.auth_version,u.auth_version,u.must_change_password,u.password_hash - FROM sessions s JOIN users u ON u.owner=s.owner WHERE s.token_hash=?`), hashKey(token)). - Scan(&owner, &scope, &sessExp, &sessVer, &userVer, &mustChange, &hash) - if errors.Is(err, sql.ErrNoRows) { - return "", ErrNotFound - } - if err != nil { - return "", err - } - if expT, perr := time.Parse(time.RFC3339Nano, sessExp); perr != nil || time.Now().After(expT) { - return "", ErrNotFound - } - // 受限会话、待改密、未设密三者之一才允许免旧密码设密；其余按主动改密路径走。 - if scope != SessionScopeSetPasswordOnly && !mustChange && (hash.Valid && hash.String != "") { - return "", ErrConflict - } - // 会话创建后账户凭据已变（并发改密/重置）：本次授权过期，拒绝继续。 - if sessVer != userVer { - return "", ErrConflict - } - res, err := tx.Exec(s.q(`UPDATE users SET password_hash=?, password_set_at=?, must_change_password=?, auth_version=auth_version+1 WHERE owner=?`), - passwordHash, nowISO(), false, owner) - if err != nil { - return "", err - } - if n, _ := res.RowsAffected(); n == 0 { - return "", ErrNotFound - } - if _, err := tx.Exec(s.q(`UPDATE login_keys SET status=? WHERE owner=? AND status=?`), - LoginKeyStatusRevoked, owner, LoginKeyStatusUnused); err != nil { - return "", err - } - if _, err := tx.Exec(s.q(`DELETE FROM sessions WHERE owner=?`), owner); err != nil { - return "", err - } - entrance := "must-change" - if scope == SessionScopeSetPasswordOnly { - entrance = "recovery" - } - if err := s.insertAccountEvent(tx, owner, entrance, owner, "password_set", entrance); err != nil { - return "", err - } - return owner, s.commitAccountMutation(tx, AccountWrite{Owner: owner, Actor: owner, Action: "account.set_password"}) + return lockOwner, nil } // ChangeUserPasswordActive 主动改密单事务：校验当前密码 → 更新（带认证版本乐观 // 检查）→ bump auth_version → 吊销其他会话（保留当前）→ 审计。中途失败无部分成功。 func (s *Store) ChangeUserPasswordActive(owner, current, passwordHash, keepToken string) error { - tx, err := s.db.Begin() - if err != nil { - return err - } - defer tx.Rollback() - var hash string - var ver int - var mustChange bool - err = tx.QueryRow(s.q(`SELECT password_hash,auth_version,must_change_password FROM users WHERE owner=?`), owner). - Scan(&hash, &ver, &mustChange) - if errors.Is(err, sql.ErrNoRows) { - return ErrNotFound - } - if err != nil { - return err - } - if mustChange || hash == "" { - // 强制态必须走受限设密路径，不在此处用旧密码换新密码。 - return ErrConflict - } - if !VerifyPassword(hash, current) { - return ErrBadCredentials - } - res, err := tx.Exec(s.q(`UPDATE users SET password_hash=?, password_set_at=?, must_change_password=?, auth_version=auth_version+1 WHERE owner=? AND auth_version=?`), - passwordHash, nowISO(), false, owner, ver) - if err != nil { - return err - } - if n, _ := res.RowsAffected(); n == 0 { - return ErrConflict // 并发改密：版本已前进 - } - if keepToken != "" { - if _, err := tx.Exec(s.q(`DELETE FROM sessions WHERE owner=? AND token_hash<>?`), owner, hashKey(keepToken)); err != nil { + return s.accountTx(AccountWrite{Owner: owner, Actor: owner, Action: "account.change_password"}, func(tx *sql.Tx, _ string) error { + var hash string + var ver int + var mustChange bool + err := tx.QueryRow(s.q(`SELECT password_hash,auth_version,must_change_password FROM users WHERE owner=?`), owner). + Scan(&hash, &ver, &mustChange) + if errors.Is(err, sql.ErrNoRows) { + return ErrNotFound + } + if err != nil { return err } - } else if _, err := tx.Exec(s.q(`DELETE FROM sessions WHERE owner=?`), owner); err != nil { - return err - } - if err := s.insertAccountEvent(tx, owner, "settings", owner, "password_change", ""); err != nil { - return err - } - return s.commitAccountMutation(tx, AccountWrite{Owner: owner, Actor: owner, Action: "account.change_password"}) + if mustChange || hash == "" { + return ErrConflict + } + if !VerifyPassword(hash, current) { + return ErrBadCredentials + } + res, err := tx.Exec(s.q(`UPDATE users SET password_hash=?, password_set_at=?, must_change_password=?, auth_version=auth_version+1 WHERE owner=? AND auth_version=?`), + passwordHash, nowISO(), false, owner, ver) + if err != nil { + return err + } + if n, _ := res.RowsAffected(); n != 1 { + return ErrConflict + } + if _, err := tx.Exec(s.q(`UPDATE login_keys SET status=? WHERE owner=? AND status=?`), + LoginKeyStatusRevoked, owner, LoginKeyStatusUnused); err != nil { + return err + } + if keepToken != "" { + if _, err := tx.Exec(s.q(`DELETE FROM sessions WHERE owner=? AND token_hash<>?`), owner, hashKey(keepToken)); err != nil { + return err + } + if _, err := tx.Exec(s.q(`UPDATE sessions SET auth_version=? WHERE owner=? AND token_hash=?`), + ver+1, owner, hashKey(keepToken)); err != nil { + return err + } + } else if _, err := tx.Exec(s.q(`DELETE FROM sessions WHERE owner=?`), owner); err != nil { + return err + } + return s.insertAccountEvent(tx, owner, "settings", owner, "password_change", "") + }) } // AdminResetUserPassword 管理员/CLI 重置密码单事务：写 hash、must_change=true、 // bump auth_version、作废未用登录 Key、吊销全部会话、审计（actor 为操作者）。 func (s *Store) AdminResetUserPassword(owner, passwordHash, actor, entrance string) error { - tx, err := s.db.Begin() - if err != nil { - return err - } - defer tx.Rollback() - res, err := tx.Exec(s.q(`UPDATE users SET password_hash=?, password_set_at=?, must_change_password=?, auth_version=auth_version+1 WHERE owner=?`), - passwordHash, nowISO(), true, owner) - if err != nil { - return err - } - if n, _ := res.RowsAffected(); n == 0 { - return ErrNotFound - } - if _, err := tx.Exec(s.q(`UPDATE login_keys SET status=? WHERE owner=? AND status=?`), - LoginKeyStatusRevoked, owner, LoginKeyStatusUnused); err != nil { - return err - } - if _, err := tx.Exec(s.q(`DELETE FROM sessions WHERE owner=?`), owner); err != nil { - return err - } if entrance == "" { entrance = entranceOf(actor) } - if err := s.insertAccountEvent(tx, actor, entrance, owner, "admin_password_reset", ""); err != nil { - return err - } - return s.commitAccountMutation(tx, AccountWrite{Owner: owner, Actor: actor, Action: "account.reset_password"}) + return s.accountTx(AccountWrite{Owner: owner, Actor: actor, Action: "account.reset_password"}, func(tx *sql.Tx, _ string) error { + res, err := tx.Exec(s.q(`UPDATE users SET password_hash=?, password_set_at=?, must_change_password=?, auth_version=auth_version+1 WHERE owner=?`), + passwordHash, nowISO(), true, owner) + if err != nil { + return err + } + if n, _ := res.RowsAffected(); n != 1 { + return ErrNotFound + } + if _, err := tx.Exec(s.q(`UPDATE login_keys SET status=? WHERE owner=? AND status=?`), + LoginKeyStatusRevoked, owner, LoginKeyStatusUnused); err != nil { + return err + } + if _, err := tx.Exec(s.q(`DELETE FROM sessions WHERE owner=?`), owner); err != nil { + return err + } + return s.insertAccountEvent(tx, actor, entrance, owner, "admin_password_reset", "") + }) } // CountUnusedLoginKeys 未使用登录 Key 计数（list 展示）。