From 8e33d1d33e57bcdc11805c802127e6d9ae4f24d7 Mon Sep 17 00:00:00 2001 From: chengcheng Date: Thu, 16 Jul 2026 12:30:14 +0800 Subject: [PATCH] =?UTF-8?q?fix(ssf):=20=E6=94=AF=E6=8C=81=E6=96=AD?= =?UTF-8?q?=E5=BC=80=E5=90=8E=E5=AE=89=E5=85=A8=E9=87=8D=E8=BF=9E?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 重建 Stream 时仅重置流级 Verification 状态，保留历史收据和撤销水位；允许显式恢复失败连接，并在进程重启后按持久化时间继续 bootstrap。\n\n已通过 pnpm test、pnpm lint、pnpm build、go vet、PostgreSQL 集成测试以及本地真实断开重连和停机重试验收。 --- .../securityevents/connection_manager.go | 31 +++++++++--- .../securityevents/connection_manager_test.go | 47 +++++++++++++++++++ apps/api/internal/store/security_events.go | 37 +++++++++++---- .../store/security_events_integration_test.go | 27 +++++++++++ 4 files changed, 127 insertions(+), 15 deletions(-) diff --git a/apps/api/internal/securityevents/connection_manager.go b/apps/api/internal/securityevents/connection_manager.go index d0e0448..451a73c 100644 --- a/apps/api/internal/securityevents/connection_manager.go +++ b/apps/api/internal/securityevents/connection_manager.go @@ -167,6 +167,9 @@ func NewConnectionManager(ctx context.Context, repository ConnectionRepository, _, _ = repository.UpdateSecurityEventConnectionLifecycle(ctx, connection.ConnectionID, "degraded", &category) } } + if connection.LifecycleStatus == "bootstrap" { + go manager.finishBootstrap(connection.ConnectionID) + } return manager, nil } @@ -193,7 +196,7 @@ func (m *ConnectionManager) Connect(ctx context.Context, issuer, idempotencyKey if existing.TransmitterIssuer != issuer { return ConnectionView{}, store.ErrSecurityEventConnectionConflict } - if existing.StreamID != nil { + if existing.StreamID != nil && !resumableConnectionLifecycle(existing.LifecycleStatus) { return m.view(existing), nil } return m.resumeConnecting(ctx, existing) @@ -223,6 +226,15 @@ func (m *ConnectionManager) Connect(ctx context.Context, issuer, idempotencyKey return m.resumeConnecting(ctx, connection) } +func resumableConnectionLifecycle(status string) bool { + switch status { + case "connecting", "verifying", "degraded", "error": + return true + default: + return false + } +} + func (m *ConnectionManager) resumeConnecting(ctx context.Context, connection store.SecurityEventConnection) (ConnectionView, error) { secret, err := m.secrets.Get(ctx, connection.CredentialRef) if err != nil { @@ -594,12 +606,19 @@ func (m *ConnectionManager) finishAutomaticVerification(connectionID string, con } func (m *ConnectionManager) finishBootstrap(connectionID string) { - select { - case <-m.ctx.Done(): - return - case <-time.After(m.config.BootstrapDuration): - } connection, err := m.repository.SecurityEventConnection(m.ctx) + if err != nil || connection.ConnectionID != connectionID || connection.LifecycleStatus != "bootstrap" { + return + } + delay := m.config.BootstrapDuration - time.Since(connection.UpdatedAt) + if delay > 0 { + select { + case <-m.ctx.Done(): + return + case <-time.After(delay): + } + } + connection, err = m.repository.SecurityEventConnection(m.ctx) if err == nil && connection.ConnectionID == connectionID && connection.LifecycleStatus == "bootstrap" { mode, _, metricsErr := m.repository.SecurityEventMetrics(m.ctx, connection.TransmitterIssuer, valueOrEmpty(connection.Audience)) if metricsErr == nil && mode == "push_healthy" { diff --git a/apps/api/internal/securityevents/connection_manager_test.go b/apps/api/internal/securityevents/connection_manager_test.go index 4fbd96a..2647b80 100644 --- a/apps/api/internal/securityevents/connection_manager_test.go +++ b/apps/api/internal/securityevents/connection_manager_test.go @@ -130,12 +130,59 @@ func TestConnectionManagerBootstrapOnlyBecomesEnabledWhenPushIsHealthy(t *testin } } +func TestConnectionManagerResumesBootstrapAfterRestart(t *testing.T) { + repository := &memoryConnectionRepository{mode: "push_healthy", connection: &store.SecurityEventConnection{ + ConnectionID: uuid.NewString(), TransmitterIssuer: "https://auth.example/ssf", + LifecycleStatus: "bootstrap", Version: 1, UpdatedAt: time.Now().Add(-time.Second), + }} + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + if _, err := NewConnectionManager(ctx, repository, &memorySecretStore{}, ConnectionManagerConfig{ + OIDCEnabled: true, OIDCIssuer: "https://auth.example/issuer", OIDCTenantID: uuid.NewString(), + BootstrapDuration: 10 * time.Millisecond, + }, &Metrics{}); err != nil { + t.Fatal(err) + } + deadline := time.Now().Add(time.Second) + for time.Now().Before(deadline) { + connection, err := repository.SecurityEventConnection(ctx) + if err == nil && connection.LifecycleStatus == "enabled" { + return + } + time.Sleep(10 * time.Millisecond) + } + t.Fatal("bootstrap lifecycle did not resume after process restart") +} + func TestRemoteNotFoundIsRecognizedForIdempotentDisconnect(t *testing.T) { if !isRemoteHTTPStatus(remoteHTTPError{status: http.StatusNotFound}, http.StatusNotFound) { t.Fatal("remote 404 was not recognized as an already-deleted Stream") } } +func TestConnectionLifecycleCanResumeOnlyIncompleteOrFailedConnections(t *testing.T) { + for _, test := range []struct { + status string + want bool + }{ + {status: "connecting", want: true}, + {status: "verifying", want: true}, + {status: "degraded", want: true}, + {status: "error", want: true}, + {status: "bootstrap"}, + {status: "enabled"}, + {status: "rotating"}, + {status: "disconnect_pending"}, + {status: "retiring"}, + } { + t.Run(test.status, func(t *testing.T) { + if got := resumableConnectionLifecycle(test.status); got != test.want { + t.Fatalf("resumableConnectionLifecycle(%q)=%t want %t", test.status, got, test.want) + } + }) + } +} + func TestTransmitterSSRFProtectionBlocksReservedNetworks(t *testing.T) { for _, address := range []string{ "127.0.0.1", "10.0.0.1", "169.254.169.254", "100.64.0.1", "192.0.2.1", "198.51.100.1", "203.0.113.1", "2001:db8::1", diff --git a/apps/api/internal/store/security_events.go b/apps/api/internal/store/security_events.go index 4ccac66..df04a03 100644 --- a/apps/api/internal/store/security_events.go +++ b/apps/api/internal/store/security_events.go @@ -108,19 +108,38 @@ WHERE session.gateway_user_id = gateway_user.id } func (s *Store) EnsureSecurityEventStreamState(ctx context.Context, issuer, audience, streamID string) error { - tag, err := s.pool.Exec(ctx, ` -INSERT INTO gateway_security_event_stream_state(issuer,audience,stream_id,mode) -VALUES($1,$2,$3::uuid,'bootstrap') -ON CONFLICT(issuer,audience) DO UPDATE -SET stream_id=EXCLUDED.stream_id,updated_at=now() -WHERE gateway_security_event_stream_state.stream_id=EXCLUDED.stream_id`, issuer, audience, streamID) + tx, err := s.pool.Begin(ctx) if err != nil { return err } - if tag.RowsAffected() == 0 { - return errors.New("security event stream id cannot be changed") + defer func() { _ = tx.Rollback(ctx) }() + if _, err := tx.Exec(ctx, ` +INSERT INTO gateway_security_event_stream_state(issuer,audience,stream_id,mode) +VALUES($1,$2,$3::uuid,'bootstrap') +ON CONFLICT(issuer,audience) DO NOTHING`, issuer, audience, streamID); err != nil { + return err } - return nil + var currentStreamID string + if err := tx.QueryRow(ctx, `SELECT stream_id::text FROM gateway_security_event_stream_state + WHERE issuer=$1 AND audience=$2 FOR UPDATE`, issuer, audience).Scan(¤tStreamID); err != nil { + return err + } + if currentStreamID != streamID { + if _, err := tx.Exec(ctx, `DELETE FROM gateway_security_event_verification_challenges + WHERE issuer=$1 AND audience=$2`, issuer, audience); err != nil { + return err + } + if _, err := tx.Exec(ctx, ` +UPDATE gateway_security_event_stream_state +SET stream_id=$3::uuid,mode='bootstrap',stream_status='unknown', + last_verification_at=NULL,bootstrap_until=NULL,consecutive_verifications=0, + pending_state_hash=NULL,pending_state_created_at=NULL,fallback_since=NULL, + last_error_category=NULL,created_at=now(),updated_at=now() +WHERE issuer=$1 AND audience=$2`, issuer, audience, streamID); err != nil { + return err + } + } + return tx.Commit(ctx) } func (s *Store) BeginSecurityEventVerification(ctx context.Context, issuer, audience string, stateHash []byte, now time.Time) error { diff --git a/apps/api/internal/store/security_events_integration_test.go b/apps/api/internal/store/security_events_integration_test.go index b84ed96..cef4e2b 100644 --- a/apps/api/internal/store/security_events_integration_test.go +++ b/apps/api/internal/store/security_events_integration_test.go @@ -155,4 +155,31 @@ func TestSecurityEventWatermarkVerificationAndFallbackLifecycle(t *testing.T) { if lifecycle != "enabled" || lastError != nil { t.Fatalf("recovered connection lifecycle=%q lastError=%v", lifecycle, lastError) } + replacementStreamID := uuid.NewString() + if err := db.EnsureSecurityEventStreamState(ctx, issuer, audience, replacementStreamID); err != nil { + t.Fatalf("rebind retained receiver state: %v", err) + } + var reboundStreamID, reboundMode, reboundStatus string + var reboundVerification *time.Time + if err := db.pool.QueryRow(ctx, `SELECT stream_id::text,mode,stream_status,last_verification_at + FROM gateway_security_event_stream_state WHERE issuer=$1 AND audience=$2`, issuer, audience). + Scan(&reboundStreamID, &reboundMode, &reboundStatus, &reboundVerification); err != nil { + t.Fatal(err) + } + if reboundStreamID != replacementStreamID || reboundMode != "bootstrap" || reboundStatus != "unknown" || reboundVerification != nil { + t.Fatalf("rebound state stream=%q mode=%q status=%q verification=%v", + reboundStreamID, reboundMode, reboundStatus, reboundVerification) + } + var retainedReceipts, retainedWatermarks int + if err := db.pool.QueryRow(ctx, `SELECT count(*) FROM gateway_security_event_receipts + WHERE issuer=$1`, issuer).Scan(&retainedReceipts); err != nil { + t.Fatal(err) + } + if err := db.pool.QueryRow(ctx, `SELECT count(*) FROM gateway_oidc_revocation_watermarks + WHERE issuer=$1 AND tenant_id=$2 AND subject=$3`, subjectIssuer, tenantID, subject).Scan(&retainedWatermarks); err != nil { + t.Fatal(err) + } + if retainedReceipts == 0 || retainedWatermarks != 1 { + t.Fatalf("rebind discarded security history receipts=%d watermarks=%d", retainedReceipts, retainedWatermarks) + } }