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 <cursoragent@cursor.com>
This commit is contained in:
2026-10-12 14:49:40 +00:00
co-authored by Cursor
parent acabb85e2a
commit f13e8b1ab7
10 changed files with 701 additions and 138 deletions
+89
View File
@@ -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
}
+32 -3
View File
@@ -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 仅此一次输出;兑换后设置密码。")
+76 -70
View File
@@ -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
+11 -10
View File
@@ -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,
+130
View File
@@ -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
}
+309
View File
@@ -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
}
+4 -15
View File
@@ -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
}
+8 -27
View File
@@ -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 展示)。
+23
View File
@@ -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`,
}}},
)
}
+19 -13
View File
@@ -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
}