From bea8efa790db1df492c29080e6f28c0e71dd4a2c Mon Sep 17 00:00:00 2001 From: k2safe Date: Fri, 18 Sep 2026 15:29:05 +0800 Subject: [PATCH] feat: make credential changes replayable --- AGENTS.md | 1 + cmd/orbit-admin/credentials.go | 223 +++++++++ cmd/orbit-admin/credentials_test.go | 86 ++++ cmd/orbit-admin/main.go | 16 +- docs/08-development-plan.md | 2 +- docs/README.md | 2 +- docs/api/README.md | 2 +- docs/api/authentication.md | 48 +- docs/reports/phase-6-credentials.md | 30 ++ internal/storage/mysql/bootstrap.go | 6 +- .../mysql/credential_integration_test.go | 195 ++++++++ internal/storage/mysql/credentials.go | 431 ++++++++++++++++++ 12 files changed, 1034 insertions(+), 8 deletions(-) create mode 100644 cmd/orbit-admin/credentials.go create mode 100644 cmd/orbit-admin/credentials_test.go create mode 100644 docs/reports/phase-6-credentials.md create mode 100644 internal/storage/mysql/credential_integration_test.go create mode 100644 internal/storage/mysql/credentials.go diff --git a/AGENTS.md b/AGENTS.md index 3402b43..edefa8c 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -84,3 +84,4 @@ - P6进程观测使用internal/observability固定指标目录，拒绝未知名称、类型和动态标签，不能引入任务/租户/业务键/凭据维度。ORBIT_METRICS_ADDR默认空（关闭），显式启用独立采集监听；业务端口不得暴露全局指标。渲染和网络I/O不持有全局记录锁，每个直方图局部快照保持桶/count/sum一致，最多2个并发抓取。 - HTTP成功升级在Hijack完成时记录握手，WS处理器存续计入独立连接指标；保持1xx临时响应、Flush、Hijack与原panic行为。DB池取本地Stats，抓取不查询业务库；后台批次success不是业务任务成功量。观测丢样或失败不得改变权威事务及恢复决定。 - 进程指标、格式验证、告警规则和标准回归证据见docs/reports/phase-6-observability.md。仍须补受理/授权/完成与流程推进延迟、持久吞吐/积压、重试/核查/恢复量、发送队列、实际采集告警及负载开销，不能把进程观测通过写成P6已完成。普通清理仍须保护有效幂等、未确认事件、未知结果和未终结流程。 +- 凭据轮换保留credential_id及订阅/Worker授权绑定，在同一事务替换摘要、固定到期时间、幂等回执和管理审计；新令牌必须重新建立会话，认证路径每次重新核对凭据状态。撤销必须让认证失效并保留原任务、订阅、交付、会话和审计证据，允许在应用或命名空间停用后执行；重复应用同一固定操作幂等返回。管理CLI先把固定操作文件写成0600，再提交；提交不明时保留文件并用原文件重放，不能重新生成秘密。日志、审计摘要、指标和业务表不得保存秘密原文或摘要。不要把轮换同一凭据ID解释成历史令牌仍有效，也不得级联删除订阅、任务或事件。 diff --git a/cmd/orbit-admin/credentials.go b/cmd/orbit-admin/credentials.go new file mode 100644 index 0000000..6f318e6 --- /dev/null +++ b/cmd/orbit-admin/credentials.go @@ -0,0 +1,223 @@ +package main + +import ( + "context" + "database/sql" + "encoding/json" + "errors" + "flag" + "io" + "os" + "path/filepath" + "strings" + "time" + + "git.yuntuo.hk/k2safe/orbit/internal/config" + "git.yuntuo.hk/k2safe/orbit/internal/storage/mysql" + core "git.yuntuo.hk/k2safe/orbit/model" +) + +func runCredential(args []string) error { + if len(args) == 0 { + return errors.New("usage: orbit-admin credential status|rotate|revoke|apply") + } + switch args[0] { + case "status": + return runCredentialStatus(args[1:]) + case "rotate": + return runCredentialChange(args[1:], "ROTATE") + case "revoke": + return runCredentialChange(args[1:], "REVOKE") + case "apply": + return runCredentialApply(args[1:]) + default: + return errors.New("usage: orbit-admin credential status|rotate|revoke|apply") + } +} + +func runCredentialStatus(args []string) error { + flags := flag.NewFlagSet("credential status", flag.ContinueOnError) + namespace, credential := "", "" + flags.StringVar(&namespace, "namespace-id", "", "credential namespace id") + flags.StringVar(&credential, "credential-id", "", "credential id") + if err := flags.Parse(args); err != nil { + return err + } + if flags.NArg() != 0 { + return errors.New("credential status does not accept positional arguments") + } + ns, id, err := credentialIDs(namespace, credential) + if err != nil { + return err + } + ctx, db, closeDB, err := adminDB(15 * time.Second) + if err != nil { + return err + } + defer closeDB() + status, err := mysql.GetCredentialStatus(ctx, db, ns, id) + if err != nil { + return errors.New("credential status failed") + } + return json.NewEncoder(os.Stdout).Encode(status) +} + +func runCredentialChange(args []string, action string) error { + flags := flag.NewFlagSet("credential "+strings.ToLower(action), flag.ContinueOnError) + namespace, credential, output, reason, validFor := "", "", "", "", "720h" + var expected uint64 + flags.StringVar(&namespace, "namespace-id", "", "credential namespace id") + flags.StringVar(&credential, "credential-id", "", "credential id") + flags.Uint64Var(&expected, "expected-revision", 0, "current credential revision required for first apply") + flags.StringVar(&reason, "reason", "", "required audit reason") + flags.StringVar(&output, "output", "", "new private operation file; an existing file is never overwritten") + flags.StringVar(&validFor, "valid-for", "720h", "new credential validity, from 1h to 8760h") + if err := flags.Parse(args); err != nil { + return err + } + if flags.NArg() != 0 || output == "" { + return errors.New("credential change requires --namespace-id, --credential-id, --expected-revision, --reason and --output") + } + ns, id, err := credentialIDs(namespace, credential) + if err != nil || expected == 0 { + return errors.New("credential change requires valid ids and a nonzero --expected-revision") + } + duration := time.Duration(0) + if action == "ROTATE" { + duration, err = time.ParseDuration(validFor) + if err != nil { + return errors.New("invalid --valid-for duration") + } + } + ctx, db, closeDB, err := adminDB(30 * time.Second) + if err != nil { + return err + } + defer closeDB() + op, err := mysql.PrepareCredentialOperation(ctx, db, mysql.CredentialChangeInput{ + NamespaceID: ns, CredentialID: id, ExpectedRevision: expected, Action: action, ValidFor: duration, Reason: reason, + }) + if err != nil { + return errors.New("credential operation could not be prepared") + } + if err := writePrivateOperation(output, op); err != nil { + return err + } + receipt, err := mysql.ApplyCredentialOperation(ctx, db, op) + if err != nil { + return errors.New("credential operation was not confirmed; the private operation file was retained for reconciliation") + } + return json.NewEncoder(os.Stdout).Encode(receipt) +} + +func runCredentialApply(args []string) error { + flags := flag.NewFlagSet("credential apply", flag.ContinueOnError) + file := "" + flags.StringVar(&file, "file", "", "existing private credential operation file") + if err := flags.Parse(args); err != nil { + return err + } + if flags.NArg() != 0 || file == "" { + return errors.New("credential apply requires --file") + } + op, err := readPrivateOperation(file) + if err != nil { + return err + } + ctx, db, closeDB, err := adminDB(30 * time.Second) + if err != nil { + return err + } + defer closeDB() + receipt, err := mysql.ApplyCredentialOperation(ctx, db, op) + if err != nil { + return errors.New("credential operation was not confirmed; retry with the same private operation file") + } + return json.NewEncoder(os.Stdout).Encode(receipt) +} + +func credentialIDs(namespace, credential string) (core.ID, core.ID, error) { + ns, id := core.ID(namespace), core.ID(credential) + if ns.Validate() != nil || id.Validate() != nil { + return "", "", errors.New("valid --namespace-id and --credential-id are required") + } + return ns, id, nil +} + +func adminDB(timeout time.Duration) (context.Context, *sql.DB, func(), error) { + ctx, cancel := context.WithTimeout(context.Background(), timeout) + cfg, err := config.Load() + if err != nil { + cancel() + return nil, nil, func() {}, err + } + db, err := mysql.Open(ctx, cfg.MySQLDSN, 2, 1) + if err != nil { + cancel() + return nil, nil, func() {}, err + } + return ctx, db, func() { db.Close(); cancel() }, nil +} + +func writePrivateOperation(path string, op mysql.CredentialOperation) error { + if path == "" || filepath.Base(path) == "." || filepath.Base(path) == string(filepath.Separator) { + return errors.New("invalid private operation path") + } + file, err := os.OpenFile(path, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0600) + if err != nil { + return errors.New("cannot create private operation file; existing files are never overwritten") + } + keep := false + defer func() { + _ = file.Close() + if !keep { + _ = os.Remove(path) + } + }() + if err := file.Chmod(0600); err != nil { + return errors.New("cannot protect private operation file") + } + if err := json.NewEncoder(file).Encode(op); err != nil { + return errors.New("cannot write private operation file") + } + if err := file.Sync(); err != nil { + return errors.New("cannot sync private operation file") + } + directory, err := os.Open(filepath.Dir(path)) + if err != nil { + return errors.New("cannot open private operation directory") + } + err = directory.Sync() + closeErr := directory.Close() + if err != nil || closeErr != nil { + return errors.New("cannot sync private operation directory") + } + keep = true + return nil +} + +func readPrivateOperation(path string) (mysql.CredentialOperation, error) { + var op mysql.CredentialOperation + info, err := os.Lstat(path) + if err != nil || !info.Mode().IsRegular() || info.Mode().Perm()&0077 != 0 || info.Size() > 64*1024 { + return op, errors.New("private operation file must be a regular 0600 file no larger than 64 KiB") + } + file, err := os.Open(path) + if err != nil { + return op, errors.New("cannot read private operation file") + } + defer file.Close() + decoder := json.NewDecoder(io.LimitReader(file, 64*1024+1)) + decoder.DisallowUnknownFields() + if err := decoder.Decode(&op); err != nil { + return op, errors.New("private operation file is invalid") + } + var extra any + if err := decoder.Decode(&extra); err != io.EOF { + return op, errors.New("private operation file must contain exactly one JSON object") + } + if err := mysql.ValidateCredentialOperation(op); err != nil { + return op, errors.New("private operation file failed validation") + } + return op, nil +} diff --git a/cmd/orbit-admin/credentials_test.go b/cmd/orbit-admin/credentials_test.go new file mode 100644 index 0000000..a1665ab --- /dev/null +++ b/cmd/orbit-admin/credentials_test.go @@ -0,0 +1,86 @@ +package main + +import ( + "os" + "path/filepath" + "testing" + "time" + + "git.yuntuo.hk/k2safe/orbit/internal/storage/mysql" + core "git.yuntuo.hk/k2safe/orbit/model" + "git.yuntuo.hk/k2safe/orbit/platform/auth" +) + +func testOperation(t *testing.T) mysql.CredentialOperation { + t.Helper() + ns, err := core.NewID() + if err != nil { + t.Fatal(err) + } + id, err := core.NewID() + if err != nil { + t.Fatal(err) + } + opID, err := core.NewID() + if err != nil { + t.Fatal(err) + } + token, _, err := auth.Issue(ns, id) + if err != nil { + t.Fatal(err) + } + expires := time.Now().UTC().Add(time.Hour) + return mysql.CredentialOperation{Version: 1, OperationID: opID, NamespaceID: ns, CredentialID: id, ExpectedRevision: 1, Action: "ROTATE", Reason: "unit test", Token: token, ExpiresAt: &expires} +} + +func TestPrivateOperationFileIsExclusiveAnd0600(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "operation.json") + op := testOperation(t) + if err := writePrivateOperation(path, op); err != nil { + t.Fatal(err) + } + info, err := os.Stat(path) + if err != nil { + t.Fatal(err) + } + if info.Mode().Perm() != 0600 { + t.Fatalf("private file mode=%v", info.Mode().Perm()) + } + read, err := readPrivateOperation(path) + if err != nil || read.OperationID != op.OperationID || read.Token != op.Token { + t.Fatalf("private operation round trip: %+v %v", read, err) + } + if err := writePrivateOperation(path, op); err == nil { + t.Fatal("existing private operation was overwritten") + } + if err := os.Chmod(path, 0640); err != nil { + t.Fatal(err) + } + if _, err := readPrivateOperation(path); err == nil { + t.Fatal("group-readable operation was accepted") + } +} + +func TestPrivateOperationFileRejectsUnknownFieldsAndSymlinks(t *testing.T) { + dir := t.TempDir() + unknown := filepath.Join(dir, "unknown.json") + if err := os.WriteFile(unknown, []byte(`{"version":1,"unexpected":true}`), 0600); err != nil { + t.Fatal(err) + } + if _, err := readPrivateOperation(unknown); err == nil { + t.Fatal("unknown operation field was accepted") + } + op := testOperation(t) + target := filepath.Join(dir, "target.json") + if err := writePrivateOperation(target, op); err != nil { + t.Fatal(err) + } + link := filepath.Join(dir, "link.json") + if err := os.Symlink(target, link); err != nil { + t.Fatal(err) + } + if _, err := readPrivateOperation(link); err == nil { + t.Fatal("symlink operation file was accepted") + } +} diff --git a/cmd/orbit-admin/main.go b/cmd/orbit-admin/main.go index c119620..1bc8e6e 100644 --- a/cmd/orbit-admin/main.go +++ b/cmd/orbit-admin/main.go @@ -22,9 +22,19 @@ func main() { } } func run() error { - if len(os.Args) < 2 || os.Args[1] != "bootstrap" { - return errors.New("usage: orbit-admin bootstrap (--application NAME | --application-id ID) --namespace NAME --environment NAME --queue NAME --output NEW_FILE") + if len(os.Args) < 2 { + return errors.New("usage: orbit-admin bootstrap|credential status|rotate|revoke|apply") } + switch os.Args[1] { + case "bootstrap": + return runBootstrap(os.Args[2:]) + case "credential": + return runCredential(os.Args[2:]) + } + return errors.New("usage: orbit-admin bootstrap|credential status|rotate|revoke|apply") +} + +func runBootstrap(args []string) error { flags := flag.NewFlagSet("bootstrap", flag.ContinueOnError) var in mysql.BootstrapInput var output string @@ -48,7 +58,7 @@ func run() error { flags.BoolVar(&in.WorkflowSignalAccess, "workflow-signal-access", false, "explicitly grant workflow signal and read actions") flags.BoolVar(&in.WorkflowRecoveryAccess, "workflow-recovery-access", false, "explicitly grant workflow recovery and read actions") flags.BoolVar(&in.WorkflowAuditAccess, "workflow-audit-access", false, "explicitly grant workflow operation audit read action") - if err := flags.Parse(os.Args[2:]); err != nil { + if err := flags.Parse(args); err != nil { return err } if flags.NArg() != 0 || output == "" { diff --git a/docs/08-development-plan.md b/docs/08-development-plan.md index 0c6b236..a892cb7 100644 --- a/docs/08-development-plan.md +++ b/docs/08-development-plan.md @@ -60,7 +60,7 @@ P4/P5定时等待与后续查询轮次的实现及回归记录见[时钟报告]( 1. 已补[首批可复现故障工具](reports/phase-6-fault-boundaries.md)：A04“派发已提交但消息尚未送达”、W01/W13事务提交结果不明，以及W16最终回执/旧PENDING结果重放/下一轮竞争。每个切点必须有实际数据库或进程证据，不能用响应错误推断事务是否已提交，也不能用普通SIGNAL竞争替代最终查询场景。 2. 已实现[基础运行观测](api/observability.md)：HTTP/WS、数据库连接池、后台批次和进程指标，以及可离线验证的告警规则；独立采集监听默认关闭，固定指标目录拒绝动态标签。仍需受理/授权/完成阶段延迟与持久吞吐、流程推进延迟、重试/核查/恢复量、待处理与到期积压、发送队列及生产采集/通知验证。 -3. 补齐凭据签发、轮换、撤销及Worker/订阅安全重绑，保持最小权限、管理审计与未知回执可恢复。 +3. 已补齐受保护管理环境中的固定凭据操作文件、轮换、撤销、幂等回执和脱敏管理审计；轮换保留授权绑定并要求新令牌重新建立会话，撤销允许在停用范围内执行且不改写历史记录。仍需补齐提交不明故障切点、多节点安全重绑演练、凭据最小权限配置核对和生产密钥托管/审计演练。 4. 实现保留/归档与受保护记录核对，再执行备份恢复、兼容升级和六类容量负载。未确认交付、有效幂等身份、当前/历史执行证据和未终结流程不能被普通清理破坏。生产HA及RPO/RTO继续按实际拓扑验证，不缩减为本机多进程检查。 ### P3 功能范围核对 diff --git a/docs/README.md b/docs/README.md index 0606bb7..d8fb967 100644 --- a/docs/README.md +++ b/docs/README.md @@ -2,7 +2,7 @@ - 文档版本：0.2 - 整理日期：2026-09-17 -- 阶段：P0–P5平台功能已通过阶段回归；[P4/P5原范围核对](reports/phase-4-5.md)和[首批故障边界](reports/phase-6-fault-boundaries.md)保留证据边界。P6生产观测与运维、P7首版交付继续推进，尚未完成全平台生产验收，数据库保持无外键。 +- 阶段：P0–P5平台功能已通过阶段回归；[P4/P5原范围核对](reports/phase-4-5.md)和[首批故障边界](reports/phase-6-fault-boundaries.md)保留证据边界。P6已补进程观测和固定操作文件凭据轮换/撤销；生产积压、阶段延迟、提交不明切点、归档备份、容量与发布交付继续推进，尚未完成全平台生产验收，数据库保持无外键。 - 适用范围：Orbit 统一平台，包含 Pulse 任务调度模块、Weave 流程编排模块与公共能力。平台面向多个业务系统复用。 ## 阅读顺序 diff --git a/docs/api/README.md b/docs/api/README.md index e39cdc1..6ee3b5e 100644 --- a/docs/api/README.md +++ b/docs/api/README.md @@ -19,7 +19,7 @@ | `/v1/application/resources` 查询及 `:configure` 管理 | P3 已实现跨命名空间应用配额 | [应用配额与 SDK 场景](application-resources.md) | | 任务 `concurrency_key` 与尝试 `concurrency_token` | P3 已实现跨队列资源键互斥 | [资源键接口与 SDK 场景](keyed-concurrency.md) | | 任务 `priority` 与节点领取池 | P3 已实现有界优先级加速、应用/队列轮转 | [优先级与公平派发](priority-and-fairness.md) | -| 凭据校验与初始化 CLI | P1 已实现 | [鉴权](authentication.md) | +| 凭据校验、初始化、固定操作文件轮换/撤销与回执重放 CLI | P1/P6 已实现；不提供公网管理路由，测试专用提交不明切点已回归 | [鉴权](authentication.md)、[凭据报告](../reports/phase-6-credentials.md) | | `/v1/worker/connect` | P2 已实现并通过阶段验收 | [Worker 契约](worker.md)、[Schema](worker-schema.md) | | 任务相对延迟 `delay_ms` | 已实现数据库计时、原键恢复与实际执行例子 | [任务接口](tasks.md)、[延迟示例](../examples/delayed-task.md) | | 任务冻结策略与 Worker `retryable` | P3 安全重试已实现 | [安全重试](retries.md) | diff --git a/docs/api/authentication.md b/docs/api/authentication.md index 00bf43c..f54cc6a 100644 --- a/docs/api/authentication.md +++ b/docs/api/authentication.md @@ -23,7 +23,53 @@ ORBIT_MYSQL_DSN 通过环境注入，命令不读取 .env。所有名称必填 资源、凭据摘要和初始化审计在一个事务中提交。交付文件先写入并同步到磁盘，成功后再提交事务。如果文件写入失败则回滚并移除文件；如果已经交付后提交无法确认，保留文件并报错，应使用其中凭据核对结果，不能直接覆盖或重复创建。 -初始文件包含真实秘密，应保管在受限目录或导入秘密管理系统。运行账户与迁移/初始化账户分开；生产只通过 HTTPS 使用凭据。凭据管理 HTTP API、轮换与细粒度运维命令在后续阶段实现，当前 bootstrap 主要用于完成独立接入与开发验证。 +初始文件包含真实秘密，应保管在受限目录或导入秘密管理系统。运行账户与迁移/初始化账户分开；生产只通过 HTTPS 使用凭据。 + +## 轮换与撤销 + +凭据轮换和撤销只允许在受保护的数据库管理环境执行，不提供公网管理接口。轮换保留 `credential_id`，因此订阅和 Worker 实例的授权绑定仍指向同一记录；新的 Worker 进程必须按现有进程身份规则重新连接，旧令牌不能再通过任何 API 或 WSS 认证。平台不会批量改写在途任务、交付记录或历史会话证据。 + +先读取脱敏状态和当前版本： + +```sh +go run ./cmd/orbit-admin credential status \ + --namespace-id \ + --credential-id +``` + +状态输出不包含令牌、秘密摘要或权限正文；`revision` 是后续变更的并发条件。 + +```sh +go run ./cmd/orbit-admin credential rotate \ + --namespace-id \ + --credential-id \ + --expected-revision 1 \ + --valid-for 720h \ + --reason "scheduled credential rotation" \ + --output .local/rotated.credentials.json +``` + +命令先读取并核对版本，生成一个固定 `operation_id`、令牌和绝对到期时间，写入新的 `0600` 私有操作文件，然后提交数据库事务。文件不得已存在；提交结果不明、网络断开或数据库返回未知错误时必须保留文件，并使用同一个文件执行 `credential apply --file`，不能重新执行 `rotate` 生成第二个秘密。`--valid-for` 限定在 1 小时到 1 年。数据库只保存秘密摘要、请求摘要和脱敏回执。 + +```sh +go run ./cmd/orbit-admin credential apply \ + --file .local/rotated.credentials.json +``` + +`apply` 会严格拒绝软链接、非 0600 文件、未知 JSON 字段、重复 JSON 对象和修改过的操作内容。重复应用已提交的文件只返回原回执，回执中的 `deduplicated` 为 `true`，不会重新轮换或新增审计。 + +发生泄露或 Worker 必须立即停用时执行撤销： + +```sh +go run ./cmd/orbit-admin credential revoke \ + --namespace-id \ + --credential-id \ + --expected-revision 2 \ + --output .local/revoke.operation.json \ + --reason "incident response" +``` + +撤销同样先写入固定操作文件，再提交事务；提交不明时用 `credential apply --file .local/revoke.operation.json` 重试。撤销只标记凭据失效，不删除订阅、任务、交付或历史记录。即使应用或命名空间已停用，管理撤销仍可执行。重复应用同一撤销文件只返回原回执，不追加重复审计。审计主体取自 MySQL 管理连接的 `CURRENT_USER()`，审计摘要不包含秘密原文或秘密摘要。 ## 凭据格式与作用域 diff --git a/docs/reports/phase-6-credentials.md b/docs/reports/phase-6-credentials.md new file mode 100644 index 0000000..95079dc --- /dev/null +++ b/docs/reports/phase-6-credentials.md @@ -0,0 +1,30 @@ +# P6：凭据操作与安全轮换 + +日期：2026-09-18。本文只记录管理凭据操作的已实现边界，不把凭据功能等同于 P6 全部生产验收。 + +## 已实现行为 + +- `credential status` 只返回名称、版本、到期/撤销时间和应用/命名空间启用状态，不返回秘密或 `secret_digest`。 +- `credential rotate` 先读取并核对 `expected_revision`，生成一次新令牌和固定绝对到期时间，写入新的 `0600` 私有操作文件，再应用事务。 +- `credential apply --file` 严格读取同一个私有操作文件。操作身份、请求摘要、凭据摘要更新、脱敏回执和管理审计在同一事务中提交；提交结果不明时重复应用同一文件只返回原回执，不重新生成秘密、不重置到期时间、不新增审计。 +- 操作文件被修改、替换为不同令牌或不同原因时返回 `IDEMPOTENCY_CONFLICT`；已有成功回执不受后续凭据轮换或撤销影响。 +- `credential revoke` 也使用固定操作文件，允许在应用或命名空间已停用时关闭凭据。撤销只改变凭据授权状态，不批量改写 Worker、订阅、任务、交付或历史会话记录；后续认证和控制操作会重新核对凭据状态。 +- 审计主体来自提交事务所在 MySQL 连接的 `CURRENT_USER()`，不是凭据 ID 或请求自报字段。审计前后摘要不包含令牌或秘密摘要。 +- 没有公网管理路由，也没有新增运行时依赖；管理 CLI 使用现有 MySQL 连接和迁移账本。 + +## 真实验证 + +以下验证使用 Go 1.27.1、darwin/arm64、Docker MySQL 8.4.10 的一次性数据库： + +```sh +GO="$(cat /tmp/orbit-toolchain-path)/go" DOCKER=/usr/local/bin/docker \ + sh /tmp/orbit-credential-test.sh +``` + +`TestIntegrationCredentialOperations` 退出 0，覆盖：初始状态、轮换后旧令牌拒绝/新令牌接受、固定操作重放、修改操作文件冲突、停用应用后撤销、重复撤销回执、审计主体和敏感字段不泄露。数据库回读确认轮换审计和撤销审计各一条，当前迁移保持零外键。 + +`TestIntegrationCredentialCommitFaults` 同样退出 0：使用测试专用 MySQL 包代理分别在 COMMIT 发出前断线和收到成功包后丢确认，独立连接回读版本，并用同一操作文件完成核对。BEFORE 情况回滚后重试首次提交，AFTER 情况直接返回原回执；两种情况都没有生成第二个令牌或第二条审计。该故障代理只存在于测试二进制，不进入生产进程。 + +随后执行标准隔离回归 `GO="$(cat /tmp/orbit-toolchain-path)/go" DOCKER=/usr/local/bin/docker sh scripts/test-mysql.sh`，退出 0。该次 root MySQL 集成组耗时 264.138 秒，包含上述两个凭据场景；流程运行器、SDK、Worker/事件进程、业务 Inbox/Outbox 和 23 个 SDK 示例均通过。另一次 CLI 专项实际构建 `orbit-admin`，在一次性 MySQL 上跑通 status、rotate、apply 首次提交、apply 重放和 revoke，且新操作文件权限为 0600。 + +以上证据仍不替代密钥托管演练、多节点重连演练、保留/归档、备份恢复和容量验收。 diff --git a/internal/storage/mysql/bootstrap.go b/internal/storage/mysql/bootstrap.go index 544deaa..4cc4a30 100644 --- a/internal/storage/mysql/bootstrap.go +++ b/internal/storage/mysql/bootstrap.go @@ -154,7 +154,11 @@ func Bootstrap(ctx context.Context, db *sql.DB, in BootstrapInput, deliver func( if err != nil { return err } - if err := appendAudit(ctx, tx, auth.Token{NamespaceID: out.NamespaceID, CredentialID: out.CredentialID}, op, out.CredentialID, "credential", "bootstrap.create", "database administrator bootstrap", nil, map[string]any{"permissions": permissions, "expires_at": out.ExpiresAt}); err != nil { + actor, err := databaseActor(ctx, tx) + if err != nil { + return err + } + if err := appendAuditActor(ctx, tx, out.NamespaceID, actor, op, "bootstrap.create", out.CredentialID, "database administrator bootstrap", nil, map[string]any{"permissions": permissions, "expires_at": out.ExpiresAt}); err != nil { return err } // 交付文件先持久化再提交；提交失败时保留文件以便核对不确定结果。 diff --git a/internal/storage/mysql/credential_integration_test.go b/internal/storage/mysql/credential_integration_test.go new file mode 100644 index 0000000..288731b --- /dev/null +++ b/internal/storage/mysql/credential_integration_test.go @@ -0,0 +1,195 @@ +package mysql + +import ( + "context" + "encoding/json" + "os" + "strings" + "testing" + "time" + + "git.yuntuo.hk/k2safe/orbit/internal/fault" + core "git.yuntuo.hk/k2safe/orbit/model" + "git.yuntuo.hk/k2safe/orbit/platform/auth" +) + +// TestIntegrationCredentialOperations verifies the private operation protocol: +// one frozen token, stable idempotency and audit identity from MySQL itself. +func TestIntegrationCredentialOperations(t *testing.T) { + dsn := os.Getenv("ORBIT_TEST_MYSQL_DSN") + if dsn == "" { + t.Skip("set ORBIT_TEST_MYSQL_DSN to a disposable database, or run make integration") + } + ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) + defer cancel() + db, err := Open(ctx, dsn, 8, 2) + if err != nil { + t.Fatal(err) + } + defer db.Close() + if err := Migrate(ctx, db); err != nil { + t.Fatal(err) + } + + var initial BootstrapResult + name := "credential-" + string(testID(t)) + err = Bootstrap(ctx, db, BootstrapInput{Application: name, Namespace: "default", Environment: "test", Queue: "credential"}, func(v BootstrapResult) error { + initial = v + return nil + }) + if err != nil { + t.Fatal(err) + } + status, err := GetCredentialStatus(ctx, db, initial.NamespaceID, initial.CredentialID) + if err != nil || status.Revision != 1 || status.RevokedAt != nil { + t.Fatalf("initial credential status: %+v %v", status, err) + } + old, err := auth.Parse(initial.Token) + if err != nil { + t.Fatal(err) + } + if err := NewTaskStore(db).AuthorizeSubmit(ctx, old); err != nil { + t.Fatalf("initial credential rejected: %v", err) + } + + op, err := PrepareCredentialOperation(ctx, db, CredentialChangeInput{ + NamespaceID: initial.NamespaceID, CredentialID: initial.CredentialID, ExpectedRevision: 1, + Action: credentialRotateAction, ValidFor: 2 * time.Hour, Reason: "scheduled rotation", + }) + if err != nil { + t.Fatal(err) + } + if op.Token == initial.Token || op.OperationID == "" || op.ExpiresAt == nil { + t.Fatal("prepare did not freeze a new token and expiry") + } + receipt, err := ApplyCredentialOperation(ctx, db, op) + if err != nil || receipt.Revision != 2 || receipt.Deduplicated { + t.Fatalf("rotation receipt: %+v %v", receipt, err) + } + newToken, err := auth.Parse(op.Token) + if err != nil || NewTaskStore(db).AuthorizeSubmit(ctx, old) == nil || NewTaskStore(db).AuthorizeSubmit(ctx, newToken) != nil { + t.Fatalf("credential authentication boundary failed: %v", err) + } + + replayed, err := ApplyCredentialOperation(ctx, db, op) + if err != nil || !replayed.Deduplicated || replayed.OperationID != receipt.OperationID || replayed.Revision != receipt.Revision { + t.Fatalf("stable operation replay: %+v %v", replayed, err) + } + changed := op + changed.Token, _, err = auth.Issue(initial.NamespaceID, initial.CredentialID) + if err != nil { + t.Fatal(err) + } + if _, err := ApplyCredentialOperation(ctx, db, changed); fault.Code(err) != "IDEMPOTENCY_CONFLICT" { + t.Fatalf("changed operation file was accepted: %v", err) + } + + // Revocation remains available for a disabled application so an incident can + // close credentials even when ordinary authentication is already unavailable. + if _, err := db.ExecContext(ctx, "UPDATE orbit_applications SET enabled=FALSE WHERE id=?", initial.ApplicationID); err != nil { + t.Fatal(err) + } + revoke, err := PrepareCredentialOperation(ctx, db, CredentialChangeInput{ + NamespaceID: initial.NamespaceID, CredentialID: initial.CredentialID, ExpectedRevision: 2, + Action: credentialRevokeAction, Reason: "incident response", + }) + if err != nil { + t.Fatal(err) + } + revoked, err := ApplyCredentialOperation(ctx, db, revoke) + if err != nil || revoked.Revision != 3 || revoked.RevokedAt == nil { + t.Fatalf("revoke receipt: %+v %v", revoked, err) + } + if _, err := ApplyCredentialOperation(ctx, db, revoke); err != nil { + t.Fatal(err) + } + if err := NewTaskStore(db).AuthorizeSubmit(ctx, newToken); fault.Code(err) != "UNAUTHENTICATED" { + t.Fatalf("revoked token remained usable: %v", err) + } + + var audits int + if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM orbit_audit_events WHERE namespace_id=? AND resource_id=? AND action IN ('credential.rotate','credential.revoke')`, initial.NamespaceID, initial.CredentialID).Scan(&audits); err != nil || audits != 2 { + t.Fatalf("credential audit count=%d err=%v", audits, err) + } + var actor, before, after string + if err := db.QueryRowContext(ctx, `SELECT actor_id,`+"`before`"+`,`+"`after`"+` FROM orbit_audit_events WHERE namespace_id=? AND operation_id=?`, initial.NamespaceID, op.OperationID).Scan(&actor, &before, &after); err != nil { + t.Fatal(err) + } + if !strings.HasPrefix(actor, "mysql:") || actor == string(initial.CredentialID) || strings.Contains(before+after, op.Token) || strings.Contains(before+after, "secret_digest") { + t.Fatalf("audit leaked or misidentified operator: actor=%q before=%s after=%s", actor, before, after) + } + var stored []byte + if err := db.QueryRowContext(ctx, `SELECT response FROM orbit_idempotency_keys WHERE namespace_id=? AND id=?`, initial.NamespaceID, op.OperationID).Scan(&stored); err != nil { + t.Fatal(err) + } + var saved CredentialChangeReceipt + if err := json.Unmarshal(stored, &saved); err != nil || saved.OperationID != op.OperationID { + t.Fatalf("stored receipt: %s %v", stored, err) + } + t.Log("credential operation file froze one token, replay returned the original receipt, tampering conflicted, and disabled-scope revocation preserved audit identity") +} + +func TestIntegrationCredentialCommitFaults(t *testing.T) { + dsn := os.Getenv("ORBIT_TEST_MYSQL_DSN") + if dsn == "" { + t.Skip("set ORBIT_TEST_MYSQL_DSN to a disposable database, or run make integration") + } + ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second) + defer cancel() + db, err := Open(ctx, dsn, 8, 2) + if err != nil { + t.Fatal(err) + } + defer db.Close() + if err := Migrate(ctx, db); err != nil { + t.Fatal(err) + } + for _, after := range []bool{false, true} { + name := "credential-commit-" + string(testID(t)) + var initial BootstrapResult + if err := Bootstrap(ctx, db, BootstrapInput{Application: name, Namespace: "default", Environment: "test", Queue: "credential"}, func(v BootstrapResult) error { initial = v; return nil }); err != nil { + t.Fatal(err) + } + op, err := PrepareCredentialOperation(ctx, db, CredentialChangeInput{NamespaceID: initial.NamespaceID, CredentialID: initial.CredentialID, ExpectedRevision: 1, Action: credentialRotateAction, ValidFor: time.Hour, Reason: "commit boundary"}) + if err != nil { + t.Fatal(err) + } + proxy, proxiedDSN := newCommitFaultProxy(t, dsn) + faultDB, err := Open(ctx, proxiedDSN, 2, 1) + if err != nil { + t.Fatal(err) + } + if after { + proxy.mode.Store(2) + } else { + proxy.mode.Store(1) + } + _, applyErr := ApplyCredentialOperation(ctx, faultDB, op) + proxy.expect(t, after, applyErr) + faultDB.Close() + status, err := GetCredentialStatus(ctx, db, initial.NamespaceID, initial.CredentialID) + if err != nil { + t.Fatal(err) + } + wantRevision := uint64(1) + if after { + wantRevision = 2 + } + if status.Revision != wantRevision { + t.Fatalf("commit %v status revision=%d want=%d", after, status.Revision, wantRevision) + } + receipt, err := ApplyCredentialOperation(ctx, db, op) + if err != nil || receipt.Revision != 2 || receipt.Deduplicated != after { + t.Fatalf("commit %v reconciliation receipt=%+v err=%v", after, receipt, err) + } + } +} + +func testID(t *testing.T) core.ID { + t.Helper() + id, err := core.NewID() + if err != nil { + t.Fatal(err) + } + return id +} diff --git a/internal/storage/mysql/credentials.go b/internal/storage/mysql/credentials.go new file mode 100644 index 0000000..e1eb7f9 --- /dev/null +++ b/internal/storage/mysql/credentials.go @@ -0,0 +1,431 @@ +package mysql + +import ( + "context" + "crypto/sha256" + "database/sql" + "encoding/json" + "fmt" + "strings" + "time" + "unicode/utf8" + + "git.yuntuo.hk/k2safe/orbit/internal/fault" + core "git.yuntuo.hk/k2safe/orbit/model" + "git.yuntuo.hk/k2safe/orbit/platform/auth" +) + +const ( + credentialOperationVersion = 1 + credentialRotateAction = "ROTATE" + credentialRevokeAction = "REVOKE" + credentialOperationName = "credential.change.v1" +) + +// CredentialStatus 是管理读取的脱敏凭据状态；永远不包含秘密或摘要。 +// Enabled 字段分别表示所属命名空间和应用当前是否允许业务认证。 +type CredentialStatus struct { + // NamespaceID 凭据所属命名空间标识。 + NamespaceID core.ID `json:"namespace_id"` + // CredentialID 凭据记录标识，轮换时保持不变。 + CredentialID core.ID `json:"credential_id"` + // Name 凭据的管理名称。 + Name string `json:"name"` + // Revision 凭据的乐观并发版本，所有变更都必须核对它。 + Revision uint64 `json:"revision"` + // ExpiresAt 当前秘密的到期时间；空值表示未设置到期时间。 + ExpiresAt *time.Time `json:"expires_at,omitempty"` + // RevokedAt 撤销时间；空值表示尚未撤销。 + RevokedAt *time.Time `json:"revoked_at,omitempty"` + // NamespaceEnabled 命名空间是否启用。 + NamespaceEnabled bool `json:"namespace_enabled"` + // ApplicationEnabled 所属应用是否启用。 + ApplicationEnabled bool `json:"application_enabled"` +} + +// CredentialChangeInput 描述一次待固定的管理变更。 +// ExpectedRevision 防止准备和提交之间的并发变更被覆盖。 +type CredentialChangeInput struct { + // NamespaceID 凭据所属命名空间。 + NamespaceID core.ID + // CredentialID 要变更的凭据。 + CredentialID core.ID + // ExpectedRevision 准备操作时观察到的版本。 + ExpectedRevision uint64 + // Action 只能是 ROTATE 或 REVOKE。 + Action string + // ValidFor 轮换后的有效时长；撤销时必须为零。 + ValidFor time.Duration + // Reason 必填审计原因。 + Reason string +} + +// CredentialOperation 是可安全重放的私有操作文件内容。 +// 令牌只存在于 ROTATE 文件中；服务端只保存其摘要和操作回执。 +type CredentialOperation struct { + // Version 操作文件格式版本，升级时不得静默改变语义。 + Version int `json:"version"` + // OperationID 稳定操作身份，同时作为审计和幂等记录的身份。 + OperationID core.ID `json:"operation_id"` + // NamespaceID 凭据所属命名空间。 + NamespaceID core.ID `json:"namespace_id"` + // CredentialID 变更目标凭据。 + CredentialID core.ID `json:"credential_id"` + // ExpectedRevision 只允许在该版本上首次应用。 + ExpectedRevision uint64 `json:"expected_revision"` + // Action 固定为 ROTATE 或 REVOKE。 + Action string `json:"action"` + // Reason 写入管理审计，不能包含控制字符或秘密。 + Reason string `json:"reason"` + // Token 轮换产生的一次性新令牌；撤销操作为空。 + Token string `json:"token,omitempty"` + // ExpiresAt 轮换后固定的绝对到期时间；重放不能重新计算。 + ExpiresAt *time.Time `json:"expires_at,omitempty"` +} + +// CredentialChangeReceipt 是服务端保存并可重复返回的脱敏结果。 +// Deduplicated 只表示本次调用命中了已有回执，不表示当前凭据仍处于该状态。 +type CredentialChangeReceipt struct { + // OperationID 稳定操作身份。 + OperationID core.ID `json:"operation_id"` + // NamespaceID 操作所属命名空间。 + NamespaceID core.ID `json:"namespace_id"` + // CredentialID 变更目标凭据。 + CredentialID core.ID `json:"credential_id"` + // Action 已应用的变更类型。 + Action string `json:"action"` + // Revision 该操作提交后凭据版本。 + Revision uint64 `json:"revision"` + // AppliedAt 操作提交事务使用的数据库时间。 + AppliedAt time.Time `json:"applied_at"` + // ExpiresAt 轮换后的固定到期时间。 + ExpiresAt *time.Time `json:"expires_at,omitempty"` + // RevokedAt 撤销提交时间。 + RevokedAt *time.Time `json:"revoked_at,omitempty"` + // Deduplicated 当前调用是否读取了既有回执。 + Deduplicated bool `json:"deduplicated"` +} + +// GetCredentialStatus 读取脱敏状态，不需要使用该凭据认证。 +func GetCredentialStatus(ctx context.Context, db *sql.DB, namespace, credential core.ID) (CredentialStatus, error) { + var out CredentialStatus + if err := validateCredentialIDs(namespace, credential); err != nil { + return out, err + } + if err := CheckSchema(ctx, db); err != nil { + return out, err + } + var expires, revoked sql.NullTime + err := db.QueryRowContext(ctx, `SELECT c.name,c.revision,c.expires_at,c.revoked_at,n.enabled,a.enabled + FROM orbit_credentials c + JOIN orbit_namespaces n ON n.id=c.namespace_id + JOIN orbit_applications a ON a.id=n.application_id + WHERE c.namespace_id=? AND c.id=?`, namespace, credential). + Scan(&out.Name, &out.Revision, &expires, &revoked, &out.NamespaceEnabled, &out.ApplicationEnabled) + if err == sql.ErrNoRows { + return out, fault.NotFound + } + if err != nil { + return out, err + } + out.NamespaceID, out.CredentialID = namespace, credential + out.ExpiresAt = nullableTime(expires) + out.RevokedAt = nullableTime(revoked) + return out, nil +} + +// PrepareCredentialOperation 只读取当前状态并生成一次秘密，不修改数据库。 +// 调用方应先把返回值写入受保护的新文件，再调用 ApplyCredentialOperation。 +func PrepareCredentialOperation(ctx context.Context, db *sql.DB, in CredentialChangeInput) (CredentialOperation, error) { + var out CredentialOperation + if err := validateCredentialChangeInput(in); err != nil { + return out, err + } + if err := CheckSchema(ctx, db); err != nil { + return out, err + } + status, err := GetCredentialStatus(ctx, db, in.NamespaceID, in.CredentialID) + if err != nil { + return out, err + } + if status.Revision != in.ExpectedRevision || status.RevokedAt != nil { + return out, fault.Conflict + } + var now time.Time + if err := db.QueryRowContext(ctx, "SELECT UTC_TIMESTAMP(6)").Scan(&now); err != nil { + return out, err + } + now = now.UTC() + if in.Action == credentialRotateAction { + secret, _, err := auth.Issue(in.NamespaceID, in.CredentialID) + if err != nil { + return out, err + } + out.Token = secret + out.ExpiresAt = timePtr(now.Add(in.ValidFor)) + } else if in.ValidFor != 0 { + return out, fault.Invalid("revoke operation cannot set validity") + } + out.Version = credentialOperationVersion + out.NamespaceID = in.NamespaceID + out.CredentialID = in.CredentialID + out.ExpectedRevision = in.ExpectedRevision + out.Action = in.Action + out.Reason = in.Reason + out.OperationID, err = core.NewID() + if err != nil { + return out, err + } + return out, ValidateCredentialOperation(out) +} + +// ValidateCredentialOperation validates the exact private operation shape. +// It is exported so the admin CLI can reject tampered files before opening a DB. +func ValidateCredentialOperation(op CredentialOperation) error { + if op.Version != credentialOperationVersion || op.OperationID.Validate() != nil || op.NamespaceID.Validate() != nil || op.CredentialID.Validate() != nil { + return fault.Invalid("credential operation identity or version is invalid") + } + if op.ExpectedRevision == 0 || (op.Action != credentialRotateAction && op.Action != credentialRevokeAction) { + return fault.Invalid("credential operation action or revision is invalid") + } + if err := validCredentialReason(op.Reason); err != nil { + return err + } + if op.Action == credentialRotateAction { + if op.ExpiresAt == nil || op.ExpiresAt.IsZero() || op.Token == "" { + return fault.Invalid("rotation operation must contain a token and fixed expiry") + } + parsed, err := auth.Parse(op.Token) + if err != nil || parsed.NamespaceID != op.NamespaceID || parsed.CredentialID != op.CredentialID { + return fault.Invalid("rotation token does not match credential identity") + } + } else if op.Token != "" || op.ExpiresAt != nil { + return fault.Invalid("revoke operation must not contain a token or expiry") + } + return nil +} + +// ApplyCredentialOperation applies one frozen operation or returns its original +// receipt. No callback or file I/O occurs inside the transaction. A commit error +// is intentionally returned unchanged; replaying the same operation reconciles it. +func ApplyCredentialOperation(ctx context.Context, db *sql.DB, op CredentialOperation) (CredentialChangeReceipt, error) { + var out CredentialChangeReceipt + if err := ValidateCredentialOperation(op); err != nil { + return out, err + } + if err := CheckSchema(ctx, db); err != nil { + return out, err + } + err := WithTx(ctx, db, func(tx *sql.Tx) error { + var name string + var revision uint64 + var expires, revoked sql.NullTime + var namespaceEnabled, applicationEnabled bool + err := tx.QueryRowContext(ctx, `SELECT c.name,c.revision,c.expires_at,c.revoked_at,n.enabled,a.enabled + FROM orbit_credentials c + JOIN orbit_namespaces n ON n.id=c.namespace_id + JOIN orbit_applications a ON a.id=n.application_id + WHERE c.namespace_id=? AND c.id=? FOR UPDATE`, op.NamespaceID, op.CredentialID). + Scan(&name, &revision, &expires, &revoked, &namespaceEnabled, &applicationEnabled) + if err == sql.ErrNoRows { + return fault.NotFound + } + if err != nil { + return err + } + identity, cached, err := reserveCredentialOperation(ctx, tx, op) + if err != nil { + return err + } + if cached != nil { + if err := json.Unmarshal(cached, &out); err != nil { + return fmt.Errorf("stored credential receipt is invalid: %w", err) + } + out.Deduplicated = true + return nil + } + if revision != op.ExpectedRevision || revoked.Valid { + return fault.Conflict + } + if revision == ^uint64(0) { + return fault.Conflict + } + now, err := dbNow(ctx, tx) + if err != nil { + return err + } + before := map[string]any{"name": name, "revision": revision, "expires_at": nullableTime(expires), "revoked_at": nullableTime(revoked), "namespace_enabled": namespaceEnabled, "application_enabled": applicationEnabled} + after := map[string]any{"name": name, "revision": revision + 1, "namespace_enabled": namespaceEnabled, "application_enabled": applicationEnabled} + if op.Action == credentialRotateAction { + parsed, err := auth.Parse(op.Token) + if err != nil { + return fault.Invalid("rotation token is invalid") + } + if !op.ExpiresAt.After(now) || op.ExpiresAt.After(now.Add(365*24*time.Hour)) { + return fault.Conflict + } + if _, err := tx.ExecContext(ctx, `UPDATE orbit_credentials + SET secret_digest=?,expires_at=?,revoked_at=NULL,revision=revision+1,updated_at=? + WHERE namespace_id=? AND id=? AND revision=?`, parsed.Digest, op.ExpiresAt.UTC(), now, op.NamespaceID, op.CredentialID, revision); err != nil { + return err + } + out.ExpiresAt = timePtr(op.ExpiresAt.UTC()) + after["expires_at"] = out.ExpiresAt + } else { + if _, err := tx.ExecContext(ctx, `UPDATE orbit_credentials + SET revoked_at=?,revision=revision+1,updated_at=? + WHERE namespace_id=? AND id=? AND revision=?`, now, now, op.NamespaceID, op.CredentialID, revision); err != nil { + return err + } + out.RevokedAt = timePtr(now) + after["revoked_at"] = now + } + out.OperationID = op.OperationID + out.NamespaceID = op.NamespaceID + out.CredentialID = op.CredentialID + out.Action = op.Action + out.Revision = revision + 1 + out.AppliedAt = now + actor, err := databaseActor(ctx, tx) + if err != nil { + return err + } + if err := appendCredentialAudit(ctx, tx, actor, op, before, after); err != nil { + return err + } + encoded, err := json.Marshal(out) + if err != nil { + return err + } + if _, err := tx.ExecContext(ctx, `UPDATE orbit_idempotency_keys SET response=?,revision=revision+1,updated_at=? WHERE namespace_id=? AND id=?`, encoded, now, op.NamespaceID, identity); err != nil { + return err + } + return nil + }) + return out, err +} + +// reserveCredentialOperation stores the frozen request digest before mutation. +// The operation ID is also the idempotency row ID, so retries never mint a new +// secret or a second audit event. +func reserveCredentialOperation(ctx context.Context, tx *sql.Tx, op CredentialOperation) (core.ID, []byte, error) { + digest, err := requestDigest(op) + if err != nil { + return "", nil, err + } + keyDigest := core.Digest(sha256.Sum256([]byte(op.OperationID))) + _, err = tx.ExecContext(ctx, `INSERT INTO orbit_idempotency_keys(namespace_id,id,operation,scope_key,key_digest,request_digest,resource_kind,resource_id,response,expires_at) + VALUES (?,?,?,?,?,?,?,?,JSON_OBJECT(),DATE_ADD(UTC_TIMESTAMP(6),INTERVAL 90 DAY)) + ON DUPLICATE KEY UPDATE id=orbit_idempotency_keys.id`, op.NamespaceID, op.OperationID, credentialOperationName, string(op.CredentialID), keyDigest, digest, "credential", op.CredentialID) + if err != nil { + return "", nil, err + } + var storedID core.ID + var storedDigest core.Digest + var response []byte + if err := tx.QueryRowContext(ctx, `SELECT id,request_digest,response FROM orbit_idempotency_keys + WHERE namespace_id=? AND operation=? AND scope_key=? AND key_digest=? FOR UPDATE`, op.NamespaceID, credentialOperationName, string(op.CredentialID), keyDigest). + Scan(&storedID, &storedDigest, &response); err != nil { + return "", nil, err + } + if subtleCompareDigest(digest, storedDigest) != 1 { + return "", nil, fault.IdempotencyConflict + } + trimmed := strings.TrimSpace(string(response)) + if trimmed != "" && trimmed != "{}" && trimmed != "null" { + return storedID, response, nil + } + return storedID, nil, nil +} + +func appendCredentialAudit(ctx context.Context, tx *sql.Tx, actor string, op CredentialOperation, before, after any) error { + return appendAuditActor(ctx, tx, op.NamespaceID, actor, op.OperationID, strings.ToLower("credential."+strings.ToLower(op.Action)), op.CredentialID, op.Reason, before, after) +} + +func appendAuditActor(ctx context.Context, tx *sql.Tx, namespace core.ID, actor string, operation core.ID, action string, resource core.ID, reason string, before, after any) error { + old, err := json.Marshal(before) + if err != nil { + return err + } + current, err := json.Marshal(after) + if err != nil { + return err + } + id, err := core.NewID() + if err != nil { + return err + } + _, err = tx.ExecContext(ctx, `INSERT INTO orbit_audit_events(namespace_id,id,operation_id,actor_id,action,resource_kind,resource_id,reason,evidence_ref,`+"`before`"+`,`+"`after`"+`) + VALUES (?,?,?,?,?,?,?,?,NULL,?,?)`, namespace, id, operation, actor, action, "credential", resource, reason, string(old), string(current)) + return err +} + +func databaseActor(ctx context.Context, tx *sql.Tx) (string, error) { + var principal string + if err := tx.QueryRowContext(ctx, "SELECT CURRENT_USER()").Scan(&principal); err != nil { + return "", err + } + principal = "mysql:" + principal + if len(principal) <= 128 { + return principal, nil + } + sum := sha256.Sum256([]byte(principal)) + return fmt.Sprintf("mysql-sha256:%x", sum[:]), nil +} + +func subtleCompareDigest(a, b core.Digest) int { + var result byte + for i := range a { + result |= a[i] ^ b[i] + } + if result == 0 { + return 1 + } + return 0 +} + +func validateCredentialIDs(namespace, credential core.ID) error { + if namespace.Validate() != nil || credential.Validate() != nil { + return fault.Invalid("credential identity is invalid") + } + return nil +} + +func validateCredentialChangeInput(in CredentialChangeInput) error { + if err := validateCredentialIDs(in.NamespaceID, in.CredentialID); err != nil { + return err + } + if in.ExpectedRevision == 0 || (in.Action != credentialRotateAction && in.Action != credentialRevokeAction) { + return fault.Invalid("credential action and expected revision are required") + } + if err := validCredentialReason(in.Reason); err != nil { + return err + } + if in.Action == credentialRotateAction && (in.ValidFor < time.Hour || in.ValidFor > 365*24*time.Hour) { + return fault.Invalid("credential validity must be between one hour and one year") + } + if in.Action == credentialRevokeAction && in.ValidFor != 0 { + return fault.Invalid("revoke operation cannot set validity") + } + return nil +} + +func validCredentialReason(reason string) error { + if strings.TrimSpace(reason) == "" || !utf8.ValidString(reason) || len([]rune(reason)) > 1024 || strings.ContainsAny(reason, "\x00\r\n") { + return fault.Invalid("credential reason is required and must not contain control newlines") + } + return nil +} + +func nullableTime(v sql.NullTime) *time.Time { + if !v.Valid { + return nil + } + t := v.Time.UTC() + return &t +} + +func timePtr(v time.Time) *time.Time { + t := v.UTC() + return &t +}