fix(release): rewrite nested bucket in target resource locators
模拟发布暴露的第 8 个契约缺陷:PG 通道改写资源行时只替换了外层 storage://<backend-id> 前缀,内层 oss://<bucket>/<key> 照搬源桶。 WeKnora v0.7.1 官方 GetKnowledgeFile 从路径内层桶名直接 GetObject, 导致目标端所有文档预览以生产凭据读源桶失败(403),前端表现为预览空白。 - PostgresImporterConfig 新增 TargetBucket;nested locator 导入时改写 内层桶名为目标桶,location_hash 随改写后的路径重算 - 未配置目标桶时保持旧行为透传,不做猜测 - 4 个单测锁定:桶改写+hash重算、非嵌套 locator 不变、无配置透传、非法桶名拒绝 - 端到端复验:重封 canary 批次重新发布 4/4 verified,目标端资源行 全部指向 dobot-weknora-test,location_hash 与路径逐行比对一致
This commit is contained in:
@@ -215,7 +215,7 @@ func main() {
|
||||
fmt.Println(" target PG connect FAIL", err)
|
||||
return
|
||||
}
|
||||
pgImporter, err := v071.NewPostgresImporter(targetPool, v071.PostgresImporterConfig{TargetStorageBackendID: route.Target.Object.StorageBackendID, TargetProvider: route.Target.Object.Provider})
|
||||
pgImporter, err := v071.NewPostgresImporter(targetPool, v071.PostgresImporterConfig{TargetStorageBackendID: route.Target.Object.StorageBackendID, TargetProvider: route.Target.Object.Provider, TargetBucket: route.Target.Object.Bucket})
|
||||
if err != nil {
|
||||
fmt.Println(" importer config FAIL", err)
|
||||
return
|
||||
|
||||
@@ -129,7 +129,7 @@ func buildReleaseRoute(ctx context.Context, archiveRoot string, route releaseRou
|
||||
if err != nil {
|
||||
return cleanup(errors.New("create source PostgreSQL exporter"))
|
||||
}
|
||||
postgresImporter, err := v071.NewPostgresImporter(targetPool, v071.PostgresImporterConfig{TargetStorageBackendID: route.Target.Object.StorageBackendID, TargetProvider: route.Target.Object.Provider})
|
||||
postgresImporter, err := v071.NewPostgresImporter(targetPool, v071.PostgresImporterConfig{TargetStorageBackendID: route.Target.Object.StorageBackendID, TargetProvider: route.Target.Object.Provider, TargetBucket: route.Target.Object.Bucket})
|
||||
if err != nil {
|
||||
return cleanup(errors.New("create target PostgreSQL importer"))
|
||||
}
|
||||
|
||||
@@ -20,6 +20,11 @@ type postgresTransactionalDB interface {
|
||||
type PostgresImporterConfig struct {
|
||||
TargetStorageBackendID string
|
||||
TargetProvider string
|
||||
// TargetBucket rewrites the bucket embedded in nested provider locators
|
||||
// (oss://<bucket>/<key>). WeKnora resolves GetFile from the path's inner
|
||||
// bucket, so a source-env bucket leaking into the target row makes every
|
||||
// preview/download read the wrong bucket with target credentials.
|
||||
TargetBucket string
|
||||
}
|
||||
|
||||
type PostgresImporter struct {
|
||||
@@ -37,10 +42,14 @@ func NewPostgresImporter(db postgresTransactionalDB, options ...PostgresImporter
|
||||
}
|
||||
config.TargetStorageBackendID = strings.TrimSpace(config.TargetStorageBackendID)
|
||||
config.TargetProvider = strings.TrimSpace(config.TargetProvider)
|
||||
config.TargetBucket = strings.TrimSpace(config.TargetBucket)
|
||||
if (config.TargetStorageBackendID == "") != (config.TargetProvider == "") ||
|
||||
(config.TargetStorageBackendID != "" && (!validSearchIdentifier(config.TargetStorageBackendID) || !validObjectProvider(config.TargetProvider))) {
|
||||
return nil, errors.New("invalid target storage backend configuration")
|
||||
}
|
||||
if config.TargetBucket != "" && !validSearchIdentifier(config.TargetBucket) {
|
||||
return nil, errors.New("invalid target storage bucket configuration")
|
||||
}
|
||||
return &PostgresImporter{db: db, config: config}, nil
|
||||
}
|
||||
|
||||
@@ -286,6 +295,17 @@ func (importer *PostgresImporter) prepareResourceRow(raw json.RawMessage, tenant
|
||||
} else if !validObjectLocator(physicalPath) {
|
||||
return nil, resourceIdentityPair{}, errors.New("invalid source storage locator")
|
||||
}
|
||||
// WeKnora's official file service parses the bucket out of the nested
|
||||
// provider locator (oss://<bucket>/<key>) and uses it directly for
|
||||
// GetObject. Carrying the source bucket across environments makes every
|
||||
// target-side preview read the wrong bucket, so rewrite it to the target
|
||||
// bucket whenever the locator embeds one and a target bucket is configured.
|
||||
if inner := strings.Index(providerPath, "://"); inner >= 0 && importer.config.TargetBucket != "" {
|
||||
rest := providerPath[inner+3:]
|
||||
if boundary := strings.Index(rest, "/"); boundary > 0 && rest[:boundary] != importer.config.TargetBucket {
|
||||
providerPath = providerPath[:inner+3] + importer.config.TargetBucket + rest[boundary:]
|
||||
}
|
||||
}
|
||||
targetPath := "storage://" + importer.config.TargetStorageBackendID + "/" + providerPath
|
||||
digest := sha256.Sum256([]byte(targetPath))
|
||||
row["storage_backend_id"] = importer.config.TargetStorageBackendID
|
||||
|
||||
@@ -0,0 +1,132 @@
|
||||
package v071
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"testing"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
)
|
||||
|
||||
// nilPGDB is a non-nil postgresTransactionalDB whose Begin is never invoked by
|
||||
// prepareResourceRow; it exists only to satisfy NewPostgresImporter's nil check.
|
||||
type nilPGDB struct{}
|
||||
|
||||
func (nilPGDB) Begin(context.Context) (pgx.Tx, error) { panic("unused") }
|
||||
|
||||
func resourceRowJSON(t *testing.T, fields map[string]any) json.RawMessage {
|
||||
t.Helper()
|
||||
encoded, err := json.Marshal(fields)
|
||||
if err != nil {
|
||||
t.Fatalf("encode resource row: %v", err)
|
||||
}
|
||||
return encoded
|
||||
}
|
||||
|
||||
func rewriteResource(t *testing.T, config PostgresImporterConfig, row json.RawMessage) map[string]any {
|
||||
t.Helper()
|
||||
importer, err := NewPostgresImporter(nilPGDB{}, config)
|
||||
if err != nil {
|
||||
t.Fatalf("new importer: %v", err)
|
||||
}
|
||||
rewritten, pair, err := importer.prepareResourceRow(row, 10000)
|
||||
if err != nil {
|
||||
t.Fatalf("prepare resource row: %v", err)
|
||||
}
|
||||
var out map[string]any
|
||||
if err := json.Unmarshal(rewritten, &out); err != nil {
|
||||
t.Fatalf("decode rewritten row: %v", err)
|
||||
}
|
||||
if pair.id != out["id"] {
|
||||
t.Fatalf("pair identity mismatch: %q vs %v", pair.id, out["id"])
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// WeKnora's official file service resolves GetFile from the bucket embedded in
|
||||
// the nested oss:// locator, not from the storage backend row. If the release
|
||||
// import carried the source bucket into the target row, every preview on the
|
||||
// target would read the source bucket with target credentials (403). This test
|
||||
// locks the rewrite to the configured target bucket and recomputes
|
||||
// location_hash exactly like the official resourceLocationHash.
|
||||
func TestPostgresResourceRewriteReplacesNestedSourceBucket(t *testing.T) {
|
||||
backendID := "6129cfd1-a26c-41dd-88cb-e6a800daa3bb"
|
||||
row := resourceRowJSON(t, map[string]any{
|
||||
"id": "res-1",
|
||||
"handle": "ABCDEFGHIJKLMNOPQRSTUV",
|
||||
"tenant_id": 10000,
|
||||
"provider": "oss",
|
||||
"storage_backend_id": "source-backend",
|
||||
"physical_path": "storage://source-backend/oss://yzh-weknora-test/weknora/10000/knowledge-1/source.md",
|
||||
"location_hash": "stale",
|
||||
})
|
||||
out := rewriteResource(t, PostgresImporterConfig{
|
||||
TargetStorageBackendID: backendID, TargetProvider: "oss", TargetBucket: "dobot-weknora-test",
|
||||
}, row)
|
||||
want := "storage://" + backendID + "/oss://dobot-weknora-test/weknora/10000/knowledge-1/source.md"
|
||||
if out["physical_path"] != want {
|
||||
t.Fatalf("physical_path=%v want %v", out["physical_path"], want)
|
||||
}
|
||||
if out["storage_backend_id"] != backendID || out["provider"] != "oss" {
|
||||
t.Fatalf("backend/provider not rewritten: %v %v", out["storage_backend_id"], out["provider"])
|
||||
}
|
||||
digest := sha256.Sum256([]byte(want))
|
||||
if out["location_hash"] != hex.EncodeToString(digest[:]) {
|
||||
t.Fatalf("location_hash not recomputed for rewritten path: %v", out["location_hash"])
|
||||
}
|
||||
}
|
||||
|
||||
// Locators without an embedded bucket (plain provider paths used by some
|
||||
// community deployments) must keep their key unchanged while still receiving
|
||||
// the backend rewrite.
|
||||
func TestPostgresResourceRewriteKeepsNonNestedLocator(t *testing.T) {
|
||||
backendID := "target-backend"
|
||||
row := resourceRowJSON(t, map[string]any{
|
||||
"id": "res-2",
|
||||
"handle": "ABCDEFGHIJKLMNOPQRSTUV",
|
||||
"tenant_id": 10000,
|
||||
"provider": "minio",
|
||||
"storage_backend_id": "source-backend",
|
||||
"physical_path": "storage://source-backend/10000/knowledge-2/source.txt",
|
||||
"location_hash": "stale",
|
||||
})
|
||||
out := rewriteResource(t, PostgresImporterConfig{
|
||||
TargetStorageBackendID: backendID, TargetProvider: "minio", TargetBucket: "target-bucket",
|
||||
}, row)
|
||||
want := "storage://" + backendID + "/10000/knowledge-2/source.txt"
|
||||
if out["physical_path"] != want {
|
||||
t.Fatalf("physical_path=%v want %v", out["physical_path"], want)
|
||||
}
|
||||
}
|
||||
|
||||
// Without a configured target bucket the rewrite must stay conservative:
|
||||
// the nested bucket passes through untouched (legacy behaviour) instead of
|
||||
// guessing a bucket name.
|
||||
func TestPostgresResourceRewriteWithoutTargetBucketKeepsLocator(t *testing.T) {
|
||||
row := resourceRowJSON(t, map[string]any{
|
||||
"id": "res-3",
|
||||
"handle": "ABCDEFGHIJKLMNOPQRSTUV",
|
||||
"tenant_id": 10000,
|
||||
"provider": "oss",
|
||||
"storage_backend_id": "source-backend",
|
||||
"physical_path": "storage://source-backend/oss://legacy-bucket/weknora/10000/knowledge-3/source.md",
|
||||
"location_hash": "stale",
|
||||
})
|
||||
out := rewriteResource(t, PostgresImporterConfig{
|
||||
TargetStorageBackendID: "target-backend", TargetProvider: "oss",
|
||||
}, row)
|
||||
want := "storage://target-backend/oss://legacy-bucket/weknora/10000/knowledge-3/source.md"
|
||||
if out["physical_path"] != want {
|
||||
t.Fatalf("physical_path=%v want %v", out["physical_path"], want)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPostgresImporterRejectsInvalidTargetBucket(t *testing.T) {
|
||||
if _, err := NewPostgresImporter(nilPGDB{}, PostgresImporterConfig{
|
||||
TargetStorageBackendID: "backend", TargetProvider: "oss", TargetBucket: "bad bucket!",
|
||||
}); err == nil {
|
||||
t.Fatal("expected invalid target bucket rejection")
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user