fix(release-worker): 启动失败带端点与脱敏原因
正式 release-worker 崩溃重启时日志只有一句 connect release target database， 主机名、库名和底层原因都不在里面。现在错误里带 host:port/db、路由标识与原因。 端点去口令不等于整条日志安全：pgx 的解析错误可能原样回显连接串。新增 internal/pgdsn.Redact（按已知口令精确替换 + 按 URL/keyword 形状兜底）， 错误保留 Unwrap 以便 errors.Is/As，技术原因不丢。cmd/worker 里只留库名的旧实现 换成同一个包，避免两份脱敏逻辑。libraryctl 用法补上 dead-letters。
This commit is contained in:
姚章浩
@@ -111,7 +111,7 @@ func main() {
|
||||
}
|
||||
|
||||
func usage() {
|
||||
fmt.Fprint(os.Stderr, `libraryctl <status|preflight|capture|list|verify|recover> [flags]
|
||||
fmt.Fprint(os.Stderr, `libraryctl <status|preflight|capture|list|verify|recover|dead-letters> [flags]
|
||||
|
||||
status probe the real source and print what is configured / blocked
|
||||
preflight read the source, run settlement + candidate validation (no writes, no stop)
|
||||
@@ -119,6 +119,8 @@ func usage() {
|
||||
list list sealed candidates and stored bytes
|
||||
verify re-verify one sealed candidate (digests, coverage, identity)
|
||||
recover restore writers from an interrupted freeze journal
|
||||
dead-letters print the source's live dead letters (read-only) with a digest the
|
||||
governance platform binds a precise release to
|
||||
|
||||
Flags (defaults target the local WKNFull test stack):
|
||||
-source-app WKNFull-app container whose env describes the source
|
||||
|
||||
@@ -0,0 +1,57 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// 正式栈的 release-worker 启动失败时只打一句 "connect release target database"，
|
||||
// 主机名、库名与底层原因都不在日志里。修好之后仍然要证明：可诊断 ≠ 泄露凭据，
|
||||
// 因为 pgx 的解析/连接错误会把连接串原样带回来。
|
||||
var diagnosisSecret = "rel" + "ease-s3cret-value"
|
||||
|
||||
func TestReleaseDatabaseErrorNamesTheEndpointWithoutCredentials(t *testing.T) {
|
||||
cause := errors.New("dial tcp 127.0.0.1:1: connect: connection refused")
|
||||
message := releaseDatabaseError("target",
|
||||
"postgres://postgres:ignored@prod-pg:25432/weknora_test?sslmode=disable", "kb-release-diagnosis", cause).Error()
|
||||
for _, want := range []string{"connect release target database", "prod-pg:25432/weknora_test",
|
||||
"kb-release-diagnosis", "connection refused"} {
|
||||
if !strings.Contains(message, want) {
|
||||
t.Fatalf("error %q does not mention %q", message, want)
|
||||
}
|
||||
}
|
||||
source := releaseDatabaseError("source", "postgres://u:p@WKNFull-postgres:5432/weknora_full", "kb-1", cause).Error()
|
||||
if !strings.Contains(source, "WKNFull-postgres:5432/weknora_full") {
|
||||
t.Fatalf("source error = %q", source)
|
||||
}
|
||||
}
|
||||
|
||||
// cause 原样回显 DSN（pgx 在解析/连接失败时确实可能这样）是最危险的一种：
|
||||
// 端点脱敏救不了它，整条错误里都藏着口令。
|
||||
func TestReleaseDatabaseErrorRedactsACauseThatEchoesTheDSN(t *testing.T) {
|
||||
cases := []string{
|
||||
"postgres://postgres:" + diagnosisSecret + "@127.0.0.1:notaport/weknora_test",
|
||||
"host=127.0.0.1 port=5432 user=postgres password=" + diagnosisSecret + " dbname=weknora_test sslmode=bogus",
|
||||
}
|
||||
for _, dsn := range cases {
|
||||
cause := fmt.Errorf("cannot parse `%s`: invalid port after host", dsn)
|
||||
message := releaseDatabaseError("target", dsn, "kb-release-diagnosis", cause).Error()
|
||||
if strings.Contains(message, diagnosisSecret) {
|
||||
t.Fatalf("error leaked the password: %q", message)
|
||||
}
|
||||
if !strings.Contains(message, "invalid port after host") {
|
||||
t.Fatalf("error lost the underlying reason: %q", message)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 保留 Unwrap：调用方仍然可以用 errors.Is/As 判断原因；只是文本被脱敏。
|
||||
func TestReleaseDatabaseErrorKeepsTheCauseChain(t *testing.T) {
|
||||
sentinel := errors.New("boom")
|
||||
err := releaseDatabaseError("target", "postgres://u:p@h:1/db", "kb-1", sentinel)
|
||||
if !errors.Is(err, sentinel) {
|
||||
t.Fatalf("cause chain broken: %v", err)
|
||||
}
|
||||
}
|
||||
@@ -3,6 +3,7 @@ package main
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"net/url"
|
||||
"path/filepath"
|
||||
"strconv"
|
||||
@@ -11,6 +12,7 @@ import (
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
|
||||
"knowledge-governance/internal/pgdsn"
|
||||
"knowledge-governance/internal/platform/weknora"
|
||||
v071 "knowledge-governance/internal/releaseadapter/v071"
|
||||
"knowledge-governance/internal/releasesnapshot"
|
||||
@@ -63,13 +65,13 @@ func buildReleaseRoute(ctx context.Context, archiveRoot string, snapshot release
|
||||
}
|
||||
sourcePool, err := openReleasePool(ctx, sourceDatabaseURL)
|
||||
if err != nil {
|
||||
return nil, nil, errors.New("connect release source database")
|
||||
return nil, nil, releaseDatabaseError("source", sourceDatabaseURL, route.KnowledgeBaseID, err)
|
||||
}
|
||||
pools := []*pgxpool.Pool{sourcePool}
|
||||
targetPool, err := openReleasePool(ctx, targetDatabaseURL)
|
||||
if err != nil {
|
||||
sourcePool.Close()
|
||||
return nil, nil, errors.New("connect release target database")
|
||||
return nil, nil, releaseDatabaseError("target", targetDatabaseURL, route.KnowledgeBaseID, err)
|
||||
}
|
||||
pools = append(pools, targetPool)
|
||||
cleanup := func(err error) (releaseworkflow.Publisher, []*pgxpool.Pool, error) {
|
||||
@@ -207,6 +209,29 @@ func readOptionalRestrictedValue(path string) (string, error) {
|
||||
return readRestrictedValue(path)
|
||||
}
|
||||
|
||||
// releaseDatabaseError 是"连不上某个库"的唯一措辞：端点 + 路由 + 底层原因。
|
||||
//
|
||||
// 为什么不用 %w 直接拼原因：pgx 的错误里可能原样带着连接串（含口令），
|
||||
// 只在端点脱敏证明不了整条日志安全。这里把原因文本交给 pgdsn.Redact 抹口令，
|
||||
// 同时保留 Unwrap，调用方仍能用 errors.Is/As 判断。技术原因（拒绝连接、
|
||||
// 认证失败、TLS 配置错误等）一个字都不丢。
|
||||
type databaseConnectError struct {
|
||||
message string
|
||||
cause error
|
||||
}
|
||||
|
||||
func (e *databaseConnectError) Error() string { return e.message }
|
||||
func (e *databaseConnectError) Unwrap() error { return e.cause }
|
||||
|
||||
func releaseDatabaseError(role, databaseURL, knowledgeBaseID string, cause error) error {
|
||||
reason := pgdsn.Redact(databaseURL, cause.Error())
|
||||
return &databaseConnectError{
|
||||
message: fmt.Sprintf("connect release %s database %s (route %s): %s",
|
||||
role, pgdsn.Endpoint(databaseURL), knowledgeBaseID, reason),
|
||||
cause: cause,
|
||||
}
|
||||
}
|
||||
|
||||
func openReleasePool(ctx context.Context, databaseURL string) (*pgxpool.Pool, error) {
|
||||
pool, err := pgxpool.New(ctx, databaseURL)
|
||||
if err != nil {
|
||||
|
||||
@@ -16,6 +16,7 @@ import (
|
||||
"knowledge-governance/internal/librarymaintenance"
|
||||
"knowledge-governance/internal/libraryrelease"
|
||||
"knowledge-governance/internal/librarysnapshot"
|
||||
"knowledge-governance/internal/pgdsn"
|
||||
postgresstore "knowledge-governance/internal/platform/postgres"
|
||||
)
|
||||
|
||||
@@ -194,7 +195,7 @@ func runLibraryRelease(ctx context.Context, store *postgresstore.Store, config a
|
||||
"目标拓扑不满足唯一入口约束，拒绝开维护窗口："+topoErr.Error(), "prebackup")
|
||||
}
|
||||
targetIdentity := fmt.Sprintf("scope=%s db=%s index=%s bucket=%s", payload.Scope,
|
||||
redactDSN(config.LibraryProductionDatabaseURL), config.LibraryProductionIndex, config.LibraryProductionS3Bucket)
|
||||
pgdsn.Endpoint(config.LibraryProductionDatabaseURL), config.LibraryProductionIndex, config.LibraryProductionS3Bucket)
|
||||
originalState, stateErr := manager.OriginalState(ctx)
|
||||
if stateErr != nil {
|
||||
return failLibraryJob(ctx, store, recorder, payload.LibraryJobID,
|
||||
|
||||
@@ -0,0 +1,81 @@
|
||||
// Package pgdsn 把连接串变成"可以写进日志/维护记录/错误信息"的端点描述。
|
||||
//
|
||||
// 为什么单独一个包：worker、release-worker 与主机侧 CLI 都要在错误里指明
|
||||
// "连不上的是哪个库"，而 pgx 的原始错误会把连接串原样带回来。把主机/端口/库名
|
||||
// 摘出来、并把原因里的口令抹掉是同一件事，只做一份，免得某处漏改又把口令打进日志。
|
||||
package pgdsn
|
||||
|
||||
import (
|
||||
"net/url"
|
||||
"regexp"
|
||||
"strconv"
|
||||
"strings"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
// Endpoint 返回 host:port/db（不含用户与口令）。解析不了就明确说"无法解析"，
|
||||
// 不把原文回显——原文里可能正好有口令。
|
||||
func Endpoint(raw string) string {
|
||||
if raw == "" {
|
||||
return "（未配置）"
|
||||
}
|
||||
config, err := pgxpool.ParseConfig(raw)
|
||||
if err != nil {
|
||||
return "（连接串无法解析）"
|
||||
}
|
||||
host := config.ConnConfig.Host
|
||||
if host == "" {
|
||||
host = "（未指定主机）"
|
||||
}
|
||||
return host + ":" + strconv.FormatUint(uint64(config.ConnConfig.Port), 10) + "/" + config.ConnConfig.Database
|
||||
}
|
||||
|
||||
// 兜底用：即使不知道原始 DSN（错误里可能夹着另一条连接串），也把这两类形状抹掉。
|
||||
var (
|
||||
userInfoPattern = regexp.MustCompile(`(://[^:/@\s]+:)[^@\s]*@`)
|
||||
keywordPassPattern = regexp.MustCompile(`(?i)(password\s*=\s*)'?[^\s'&]*'?`)
|
||||
urlPassPattern = regexp.MustCompile(`(?i)([?&]password=)[^&\s]*`)
|
||||
)
|
||||
|
||||
// Redact 把一段错误文本里可能出现的口令抹掉，保留主机、端口、库名与原因。
|
||||
//
|
||||
// 为什么必须做：只在端点上脱敏证明不了整条日志安全——pgx 解析失败时会把
|
||||
// `postgres://u:p@h:notaport/db` 这类原文带回来，keyword 串（password=…）
|
||||
// 也一样。这里先按 raw 里能取到的口令精确替换，再按形状兜底。
|
||||
func Redact(raw, text string) string {
|
||||
for _, secret := range secretsOf(raw) {
|
||||
text = strings.ReplaceAll(text, secret, "***")
|
||||
}
|
||||
text = userInfoPattern.ReplaceAllString(text, "${1}***@")
|
||||
text = keywordPassPattern.ReplaceAllString(text, "${1}***")
|
||||
text = urlPassPattern.ReplaceAllString(text, "${1}***")
|
||||
return text
|
||||
}
|
||||
|
||||
// secretsOf 从 URL 形式与 keyword 形式里取出可能的口令。
|
||||
func secretsOf(raw string) []string {
|
||||
secrets := make([]string, 0, 4)
|
||||
appendSecret := func(value string) {
|
||||
if len(value) >= 4 {
|
||||
secrets = append(secrets, value)
|
||||
}
|
||||
}
|
||||
if parsed, err := url.Parse(raw); err == nil && parsed.User != nil {
|
||||
if password, ok := parsed.User.Password(); ok {
|
||||
appendSecret(password)
|
||||
}
|
||||
}
|
||||
if index := strings.Index(raw, "?"); index >= 0 {
|
||||
if values, err := url.ParseQuery(raw[index+1:]); err == nil {
|
||||
appendSecret(values.Get("password"))
|
||||
}
|
||||
}
|
||||
for _, field := range strings.Fields(raw) {
|
||||
key, value, ok := strings.Cut(field, "=")
|
||||
if ok && strings.EqualFold(strings.TrimSpace(key), "password") {
|
||||
appendSecret(strings.Trim(value, "'"))
|
||||
}
|
||||
}
|
||||
return secrets
|
||||
}
|
||||
@@ -0,0 +1,97 @@
|
||||
package pgdsn
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
// 拼接构造，避免把"看起来像口令的字面量"直接写进源码（密钥扫描器会对它报警）。
|
||||
var testSecret = "sup3r" + "-s3cret-value"
|
||||
|
||||
// 端点描述只留 host:port/db：诊断需要主机名，口令一个字都不许出现在日志里。
|
||||
func TestEndpointKeepsTheHostAndDropsCredentials(t *testing.T) {
|
||||
endpoint := Endpoint("postgres://postgres:" + testSecret + "@prod-pg:25432/weknora_test?sslmode=disable")
|
||||
if endpoint != "prod-pg:25432/weknora_test" {
|
||||
t.Fatalf("endpoint = %q", endpoint)
|
||||
}
|
||||
if strings.Contains(endpoint, testSecret) || strings.Contains(endpoint, "postgres:") {
|
||||
t.Fatalf("endpoint leaked credentials: %q", endpoint)
|
||||
}
|
||||
}
|
||||
|
||||
func TestEndpointIsHonestAboutUnparseableValues(t *testing.T) {
|
||||
if got := Endpoint(""); got != "（未配置）" {
|
||||
t.Fatalf("empty = %q", got)
|
||||
}
|
||||
if got := Endpoint("://not-a-dsn"); !strings.Contains(got, "无法解析") {
|
||||
t.Fatalf("unparseable = %q", got)
|
||||
}
|
||||
}
|
||||
|
||||
// 真实 pgx 的解析失败：错误文本本身就是从这条连接串生成的。这里不假设
|
||||
// "pgx 会不会回显口令"（不同版本行为不同），只断言经过 Redact 之后整条消息里
|
||||
// 没有口令，且技术原因还在（不是把原因抹空）。
|
||||
func TestRedactCoversRealPGXParseErrors(t *testing.T) {
|
||||
cases := []struct{ dsn, keep string }{
|
||||
{"postgres://postgres:" + testSecret + "@127.0.0.1:notaport/weknora_test", "postgres"},
|
||||
{"postgres://postgres:" + testSecret + "@127.0.0.1:5432/weknora_test?sslmode=bogus&connect_timeout=abc", "connect_timeout"},
|
||||
{"host=127.0.0.1 port=5432 user=postgres password=" + testSecret + " dbname=weknora_test sslmode=bogus", "sslmode"},
|
||||
}
|
||||
for _, test := range cases {
|
||||
cause := parseErrorText(t, test.dsn)
|
||||
message := Redact(test.dsn, cause)
|
||||
if strings.Contains(message, testSecret) {
|
||||
t.Fatalf("redacted message still contains the password: %q", message)
|
||||
}
|
||||
if !strings.Contains(message, test.keep) {
|
||||
t.Fatalf("redaction lost diagnosability (%q missing): %q", test.keep, message)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 真正的回归点：任何一层把 DSN（或它的口令）原样回显进 cause 时，输出仍必须安全。
|
||||
// 这里用"把 DSN 直接打进 cause"来模拟这种回显，URL 与 keyword 两种形状都覆盖。
|
||||
func TestRedactCoversCausesThatEchoTheDSN(t *testing.T) {
|
||||
for _, dsn := range []string{
|
||||
"postgres://postgres:" + testSecret + "@127.0.0.1:1/weknora_test?sslmode=disable",
|
||||
"host=127.0.0.1 user=postgres password=" + testSecret + " dbname=weknora_test",
|
||||
} {
|
||||
cause := fmt.Errorf("dial failed for %s: connection refused", dsn).Error()
|
||||
if !strings.Contains(cause, testSecret) {
|
||||
t.Fatal("fixture broken: cause does not contain the secret")
|
||||
}
|
||||
message := Redact(dsn, cause)
|
||||
if strings.Contains(message, testSecret) {
|
||||
t.Fatalf("echoed DSN leaked the password: %q", message)
|
||||
}
|
||||
if !strings.Contains(message, "connection refused") {
|
||||
t.Fatalf("redaction lost the reason: %q", message)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 即使错误里夹着的是另一条连接串（调用方没把那条 DSN 传进来），按形状也要抹掉。
|
||||
func TestRedactCoversUnknownDSNsByShape(t *testing.T) {
|
||||
text := "failed: postgres://weknora:" + testSecret + "@db:5432/x?sslmode=disable password=" + testSecret + " trailing"
|
||||
redacted := Redact("", text)
|
||||
if strings.Contains(redacted, testSecret) {
|
||||
t.Fatalf("redacted = %q", redacted)
|
||||
}
|
||||
for _, keep := range []string{"db:5432/x", "failed", "trailing"} {
|
||||
if !strings.Contains(redacted, keep) {
|
||||
t.Fatalf("redaction lost diagnosability (%q missing): %q", keep, redacted)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func parseErrorText(t *testing.T, dsn string) string {
|
||||
t.Helper()
|
||||
_, err := pgxpool.ParseConfig(dsn)
|
||||
if err == nil {
|
||||
t.Fatalf("expected pgx to reject %q", dsn)
|
||||
}
|
||||
return err.Error()
|
||||
}
|
||||
Reference in New Issue
Block a user