fix(ssf): 允许同一配对恢复未完成连接

首次创建 SSF Stream 失败后，恢复管理器会因缺少尚未生成的 Audience 而将同一 Revision 误判为凭据交接冲突。现在仅对同一 owner、Transmitter、机器客户端且尚未绑定 Stream 的可恢复状态放行重试，跨 Revision 与不完整绑定配置仍保持安全拒绝。\n\n验证：gofmt、go vet ./...、env -u AI_GATEWAY_TEST_DATABASE_URL go test ./... -count=1。
2026-07-28 04:48:15 +08:00
parent 9e8722dc9f
commit 1b48a91af0
2 changed files with 114 additions and 0 deletions
@@ -183,6 +183,9 @@ func NewConnectionManager(ctx context.Context, repository ConnectionRepository,
 		}
 	} else if hasRevisionBindingExpectation(config) {
 		if err := manager.validateConnectionBinding(connection); err != nil {
+			if manager.recoverableIncompleteConnectionBinding(connection) {
+				return manager, nil
+			}
 			// A recovery manager may inspect a foreign singleton connection, but
 			// it must not activate it: activate() would combine the old persisted
 			// Secret with the new Revision's token endpoint.
@@ -257,6 +260,22 @@ func (m *ConnectionManager) validateConnectionBinding(connection store.SecurityE
 	return nil
 }

+func (m *ConnectionManager) recoverableIncompleteConnectionBinding(connection store.SecurityEventConnection) bool {
+	expectedIssuer := strings.TrimRight(strings.TrimSpace(m.config.ExpectedTransmitterIssuer), "/")
+	expectedAudience := strings.TrimSpace(m.config.ExpectedAudience)
+	expectedClientID := strings.TrimSpace(m.config.ManagementClientID)
+	expectedOwner := strings.TrimSpace(m.config.ExpectedOwnerKey)
+	if expectedIssuer == "" || expectedAudience == "" || expectedClientID == "" || expectedOwner == "" {
+		return false
+	}
+	return connection.StreamID == nil &&
+		connection.Audience == nil &&
+		resumableConnectionLifecycle(connection.LifecycleStatus) &&
+		strings.TrimRight(strings.TrimSpace(connection.TransmitterIssuer), "/") == expectedIssuer &&
+		strings.TrimSpace(connection.ManagementClientID) == expectedClientID &&
+		strings.TrimSpace(connection.IdempotencyKey) == expectedOwner
+}
+
 func (m *ConnectionManager) Get(ctx context.Context) (ConnectionView, error) {
 	connection, err := m.repository.SecurityEventConnection(ctx)
 	if err != nil {
@@ -409,6 +409,101 @@ func TestConnectionManagerRecoveryCanLoadDifferentPersistedConnection(t *testing
 	}
 }

+func TestConnectionManagerRecoveryResumesSameOwnerIncompleteConnection(t *testing.T) {
+	var requests atomic.Int32
+	transmitter := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		requests.Add(1)
+		http.Error(w, "expected retry probe", http.StatusServiceUnavailable)
+	}))
+	defer transmitter.Close()
+
+	const (
+		managementClient = "gateway-machine"
+		managementSecret = "new-machine-secret-value-0000000000"
+		managementRef    = "ssf-management-current"
+		pushRef          = "ssf-push-current"
+		ownerKey         = "identity-pairing-ssf-current-revision"
+	)
+	managementReference := managementRef
+	repository := &memoryConnectionRepository{connection: &store.SecurityEventConnection{
+		ConnectionID:            uuid.NewString(),
+		TransmitterIssuer:       transmitter.URL + "/ssf",
+		EndpointURL:             transmitter.URL + "/api/v1/security-events/ssf",
+		CredentialRef:           pushRef,
+		ManagementClientID:      managementClient,
+		ManagementCredentialRef: &managementReference,
+		LifecycleStatus:         "error",
+		IdempotencyKey:          ownerKey,
+		Version:                 2,
+	}}
+	secrets := &memorySecretStore{values: map[string][]byte{
+		managementRef: []byte(managementSecret),
+		pushRef:       []byte("push-secret-value-0000000000000000"),
+	}}
+	manager, err := NewConnectionManager(context.Background(), repository, secrets, ConnectionManagerConfig{
+		AppEnv: "test", OIDCEnabled: true, OIDCIssuer: transmitter.URL + "/issuer", OIDCTenantID: uuid.NewString(),
+		ManagementClientID: managementClient, ExpectedTransmitterIssuer: transmitter.URL + "/ssf",
+		ExpectedAudience: "urn:easyai:ssf:receiver:current", ExpectedOwnerKey: ownerKey,
+		PublicBaseURL: transmitter.URL, HTTPClient: transmitter.Client(),
+	}, &Metrics{})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	_, err = manager.Connect(
+		context.Background(),
+		transmitter.URL+"/ssf",
+		managementClient,
+		[]byte(managementSecret),
+		ownerKey,
+	)
+	var categorized interface{ SafeErrorCategory() string }
+	if !errors.As(err, &categorized) || categorized.SafeErrorCategory() != "discovery_failed" {
+		t.Fatalf("same-owner incomplete retry error=%v category=%v", err, categorized)
+	}
+	if requests.Load() != 1 {
+		t.Fatalf("same-owner incomplete retry sent %d discovery requests", requests.Load())
+	}
+}
+
+func TestConnectionManagerRecoveryRejectsIncompleteRevisionExpectation(t *testing.T) {
+	const (
+		managementClient = "gateway-machine"
+		managementSecret = "new-machine-secret-value-0000000000"
+		ownerKey         = "identity-pairing-ssf-current-revision"
+	)
+	repository := &memoryConnectionRepository{connection: &store.SecurityEventConnection{
+		ConnectionID:       uuid.NewString(),
+		TransmitterIssuer:  "https://auth.example/ssf",
+		EndpointURL:        "https://gateway.example/api/v1/security-events/ssf",
+		CredentialRef:      "ssf-push-current",
+		ManagementClientID: managementClient,
+		LifecycleStatus:    "error",
+		IdempotencyKey:     ownerKey,
+		Version:            2,
+	}}
+	manager, err := NewConnectionManager(context.Background(), repository, &memorySecretStore{}, ConnectionManagerConfig{
+		OIDCEnabled: true, OIDCIssuer: "https://auth.example/issuer", OIDCTenantID: uuid.NewString(),
+		ManagementClientID: managementClient, ExpectedTransmitterIssuer: "https://auth.example/ssf",
+		ExpectedOwnerKey: ownerKey,
+	}, &Metrics{})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	_, err = manager.Connect(
+		context.Background(),
+		"https://auth.example/ssf",
+		managementClient,
+		[]byte(managementSecret),
+		ownerKey,
+	)
+	var categorized interface{ SafeErrorCategory() string }
+	if !errors.As(err, &categorized) || categorized.SafeErrorCategory() != "credential_handoff_unsafe" {
+		t.Fatalf("incomplete Revision expectation error=%v category=%v", err, categorized)
+	}
+}
+
 func TestForeignRevisionRecoveryNeverUsesPersistedCredentialOnGenericOperations(t *testing.T) {
 	var requests atomic.Int32
 	endpoint := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {