easyai-ai-gateway/apps/api/internal/securityevents/connection_manager.go

package securityevents

import (
	"bytes"
	"context"
	"crypto/rand"
	"crypto/tls"
	"encoding/base64"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net"
	"net/http"
	"net/netip"
	"net/url"
	"strings"
	"sync"
	"time"

	"github.com/easyai/easyai-ai-gateway/apps/api/internal/auth"
	"github.com/easyai/easyai-ai-gateway/apps/api/internal/store"
	"github.com/google/uuid"
)

const (
	sessionRevokedEvent = "https://schemas.openid.net/secevent/caep/event-type/session-revoked"
	pushDeliveryMethod  = "urn:ietf:rfc:8935"
)

type ConnectionRepository interface {
	Repository
	StateRepository
	CreateSecurityEventConnection(context.Context, store.CreateSecurityEventConnectionInput) (store.SecurityEventConnection, error)
	SecurityEventConnection(context.Context) (store.SecurityEventConnection, error)
	BindSecurityEventConnection(context.Context, string, string, string, string, int64) (store.SecurityEventConnection, error)
	SetSecurityEventManagementCredential(context.Context, string, string, string, int64) (store.SecurityEventConnection, error)
	TransitionSecurityEventConnectionLifecycle(context.Context, string, string, int64, string, *string) (store.SecurityEventConnection, error)
	SetSecurityEventNextCredential(context.Context, string, string, int64) (store.SecurityEventConnection, error)
	PromoteSecurityEventCredential(context.Context, string, string) (store.SecurityEventConnection, error)
	DiscardPreparedSecurityEventConnection(context.Context, string, string, int64) error
	FinalizeRetiringSecurityEventConnection(context.Context, string, int64) error
	SecurityEventMetrics(context.Context, string, string) (string, *time.Time, error)
	QueueIdentitySecretCleanup(context.Context, string, time.Time) error
	RemovePendingIdentitySecretCleanup(context.Context, string) (bool, error)
}

type ConnectionManagerConfig struct {
	AppEnv                    string
	OIDCEnabled               bool
	OIDCIssuer                string
	OIDCTenantID              string
	ManagementClientID        string
	ManagementClientSecret    string
	ExpectedTransmitterIssuer string
	ExpectedAudience          string
	ExpectedOwnerKey          string
	StrictRevisionBinding     bool
	PublicBaseURL             string
	HeartbeatInterval         time.Duration
	StaleAfter                time.Duration
	ClockSkew                 time.Duration
	BootstrapDuration         time.Duration
	CredentialOverlapDuration time.Duration
	RetirementDuration        time.Duration
	HTTPClient                *http.Client
}

type ConnectionView struct {
	ConnectionID       string     `json:"connectionId"`
	TransmitterIssuer  string     `json:"transmitterIssuer"`
	ReceiverEndpoint   string     `json:"receiverEndpoint"`
	Audience           *string    `json:"audience,omitempty"`
	StreamID           *string    `json:"streamId,omitempty"`
	LifecycleStatus    string     `json:"lifecycleStatus"`
	HealthMode         string     `json:"healthMode"`
	ManagementClientID string     `json:"managementClientId"`
	LastVerificationAt *time.Time `json:"lastVerificationAt,omitempty"`
	LastErrorCategory  *string    `json:"lastErrorCategory,omitempty"`
	Version            int64      `json:"version"`
}

type connectionRuntime struct {
	handler *Handler
	service *Service
	cancel  context.CancelFunc
}

type connectionRetirementAuthorization struct {
	connectionID  string
	credentialRef string
	ownerKey      string
}

type ConnectionManager struct {
	ctx             context.Context
	repository      ConnectionRepository
	secrets         SecretStore
	config          ConnectionManagerConfig
	metrics         *Metrics
	mutex           sync.RWMutex
	operation       sync.Mutex
	runtime         *connectionRuntime
	retirement      *connectionRetirementAuthorization
	bindingMismatch bool
}

type transmitterConfiguration struct {
	SpecVersion              string   `json:"spec_version"`
	Issuer                   string   `json:"issuer"`
	JWKSURI                  string   `json:"jwks_uri"`
	ConfigurationEndpoint    string   `json:"configuration_endpoint"`
	StatusEndpoint           string   `json:"status_endpoint"`
	VerificationEndpoint     string   `json:"verification_endpoint"`
	DeliveryMethodsSupported []string `json:"delivery_methods_supported"`
}

type streamConfiguration struct {
	StreamID        string   `json:"stream_id"`
	Issuer          string   `json:"iss"`
	Audience        string   `json:"aud"`
	EventsRequested []string `json:"events_requested"`
	Delivery        struct {
		Method      string `json:"method"`
		EndpointURL string `json:"endpoint_url"`
	} `json:"delivery"`
	Description *string `json:"description,omitempty"`
}

type remoteHTTPError struct{ status int }

func (e remoteHTTPError) Error() string {
	return fmt.Sprintf("SSF endpoint returned HTTP %d", e.status)
}

type safeConnectionError struct {
	category string
	cause    error
}

func (err safeConnectionError) Error() string {
	return "security event operation failed: " + err.category
}
func (err safeConnectionError) Unwrap() error             { return err.cause }
func (err safeConnectionError) SafeErrorCategory() string { return err.category }

func NewConnectionManager(ctx context.Context, repository ConnectionRepository, secrets SecretStore, config ConnectionManagerConfig, metrics *Metrics) (*ConnectionManager, error) {
	if repository == nil || secrets == nil || !config.OIDCEnabled || config.OIDCIssuer == "" || config.OIDCTenantID == "" {
		return nil, errors.New("security event connection prerequisites are incomplete")
	}
	if config.HeartbeatInterval <= 0 {
		config.HeartbeatInterval = time.Minute
	}
	if config.StaleAfter < 2*config.HeartbeatInterval {
		config.StaleAfter = 3 * time.Minute
	}
	if config.ClockSkew == 0 {
		config.ClockSkew = time.Minute
	}
	if config.BootstrapDuration == 0 {
		config.BootstrapDuration = 6 * time.Minute
	}
	if config.CredentialOverlapDuration == 0 {
		config.CredentialOverlapDuration = 3 * time.Minute
	}
	if config.RetirementDuration == 0 {
		config.RetirementDuration = defaultSecurityEventRetirementDuration
	}
	manager := &ConnectionManager{ctx: ctx, repository: repository, secrets: secrets, config: config, metrics: metrics}
	connection, err := repository.SecurityEventConnection(ctx)
	if errors.Is(err, store.ErrSecurityEventConnectionNotFound) {
		if config.StrictRevisionBinding {
			return nil, safeConnectionError{category: "connection_binding_missing", cause: err}
		}
		return manager, nil
	}
	if err != nil {
		return nil, fmt.Errorf("load security event connection: %w", err)
	}
	if config.StrictRevisionBinding {
		if err := manager.validateConnectionBinding(connection); err != nil {
			return nil, err
		}
	} else if hasRevisionBindingExpectation(config) {
		if err := manager.validateConnectionBinding(connection); err != 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.
			if connection.LifecycleStatus == "retiring" {
				go manager.finishRetirement(connection)
			}
			manager.bindingMismatch = true
			return manager, nil
		}
	}
	if connection.LifecycleStatus == "retiring" {
		if connection.StreamID != nil && connection.Audience != nil {
			_ = manager.activate(ctx, connection)
		}
		go manager.finishRetirement(connection)
		return manager, nil
	}
	if connection.LifecycleStatus == "disconnect_pending" {
		if connection.StreamID != nil && connection.Audience != nil {
			_ = manager.activate(ctx, connection)
		}
		go manager.retryDisconnect(connection.ConnectionID)
		return manager, nil
	}
	if connection.StreamID != nil && connection.Audience != nil {
		if err := manager.activate(ctx, connection); err != nil {
			category := "credential_unavailable"
			_, _ = repository.TransitionSecurityEventConnectionLifecycle(
				ctx, connection.ConnectionID, connection.LifecycleStatus, connection.Version, "degraded", &category,
			)
		}
	}
	if connection.LifecycleStatus == "bootstrap" {
		go manager.finishBootstrap(connection.ConnectionID)
	}
	return manager, nil
}

func hasRevisionBindingExpectation(config ConnectionManagerConfig) bool {
	return strings.TrimSpace(config.ExpectedTransmitterIssuer) != "" ||
		strings.TrimSpace(config.ExpectedAudience) != "" ||
		strings.TrimSpace(config.ExpectedOwnerKey) != ""
}

// ValidateConfiguredConnectionBinding verifies that the singleton persisted
// SSF connection belongs to the identity Revision represented by this manager.
// Recovery managers call this explicitly only when they are about to become an
// Active Runtime; cleanup and retirement remain able to inspect old bindings.
func (m *ConnectionManager) ValidateConfiguredConnectionBinding(ctx context.Context) error {
	connection, err := m.repository.SecurityEventConnection(ctx)
	if errors.Is(err, store.ErrSecurityEventConnectionNotFound) {
		return safeConnectionError{category: "connection_binding_missing", cause: err}
	}
	if err != nil {
		return safeConnectionError{category: "connection_binding_unavailable", cause: err}
	}
	return m.validateConnectionBinding(connection)
}

func (m *ConnectionManager) validateConnectionBinding(connection store.SecurityEventConnection) error {
	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 safeConnectionError{category: "connection_binding_invalid", cause: errors.New("security event Revision binding is incomplete")}
	}
	if connection.TransmitterIssuer != expectedIssuer || connection.Audience == nil || *connection.Audience != expectedAudience ||
		connection.ManagementClientID != expectedClientID || connection.IdempotencyKey != expectedOwner {
		return safeConnectionError{category: "connection_binding_mismatch", cause: store.ErrSecurityEventConnectionConflict}
	}
	return nil
}

func (m *ConnectionManager) Get(ctx context.Context) (ConnectionView, error) {
	connection, err := m.repository.SecurityEventConnection(ctx)
	if err != nil {
		return ConnectionView{}, err
	}
	return m.view(connection), nil
}

func (m *ConnectionManager) IntrospectionCredential(ctx context.Context) (string, []byte, error) {
	if err := m.requireOperationalBinding(); err != nil {
		return "", nil, err
	}
	connection, err := m.repository.SecurityEventConnection(ctx)
	if errors.Is(err, store.ErrSecurityEventConnectionNotFound) {
		if m.config.ManagementClientID != "" && m.config.ManagementClientSecret != "" {
			return m.config.ManagementClientID, []byte(m.config.ManagementClientSecret), nil
		}
		return "", nil, errors.New("managed machine credential is unavailable")
	}
	if err != nil {
		return "", nil, err
	}
	return m.managementCredential(ctx, connection)
}

func (m *ConnectionManager) Connect(ctx context.Context, issuer, managementClientID string, managementSecret []byte, idempotencyKey string) (ConnectionView, error) {
	m.operation.Lock()
	defer m.operation.Unlock()
	issuer = strings.TrimRight(strings.TrimSpace(issuer), "/")
	managementClientID = strings.TrimSpace(managementClientID)
	if idempotencyKey == "" {
		return ConnectionView{}, errors.New("idempotency key is required")
	}
	existing, err := m.repository.SecurityEventConnection(ctx)
	if err != nil && !errors.Is(err, store.ErrSecurityEventConnectionNotFound) {
		return ConnectionView{}, err
	}
	credentialsProvided := managementClientID != "" || len(managementSecret) > 0
	if (managementClientID == "") != (len(managementSecret) == 0) {
		return ConnectionView{}, errors.New("machine client id and secret must be provided together")
	}
	if err == nil && (existing.LifecycleStatus == "retiring" ||
		existing.LifecycleStatus == "disconnect_pending" && existing.IdempotencyKey == idempotencyKey) {
		return ConnectionView{}, safeConnectionError{category: "retirement_pending", cause: store.ErrSecurityEventConnectionConflict}
	}
	if !credentialsProvided {
		if err == nil {
			managementClientID, managementSecret, _ = m.managementCredential(ctx, existing)
		} else {
			managementClientID = strings.TrimSpace(m.config.ManagementClientID)
			managementSecret = []byte(m.config.ManagementClientSecret)
		}
	}
	defer clear(managementSecret)
	if err := m.validatePrerequisites(issuer, managementClientID, managementSecret, errors.Is(err, store.ErrSecurityEventConnectionNotFound)); err != nil {
		return ConnectionView{}, err
	}
	if err == nil {
		if m.bindingMismatch && existing.IdempotencyKey == idempotencyKey {
			return ConnectionView{}, safeConnectionError{category: "credential_handoff_unsafe", cause: store.ErrSecurityEventConnectionConflict}
		}
		if existing.TransmitterIssuer != issuer {
			return ConnectionView{}, safeConnectionError{category: "credential_handoff_unsafe", cause: store.ErrSecurityEventConnectionConflict}
		}
		if credentialsProvided {
			if existing.IdempotencyKey != idempotencyKey {
				if existing.ManagementClientID == "" || existing.ManagementClientID != managementClientID {
					return ConnectionView{}, safeConnectionError{category: "credential_handoff_unsafe", cause: store.ErrSecurityEventConnectionConflict}
				}
				if existing.StreamID == nil {
					m.retirement = &connectionRetirementAuthorization{
						connectionID: existing.ConnectionID, credentialRef: valueOrEmpty(existing.ManagementCredentialRef), ownerKey: idempotencyKey,
					}
					return ConnectionView{}, safeConnectionError{category: "connection_conflict", cause: store.ErrSecurityEventConnectionConflict}
				}
				if m.retirement == nil || m.retirement.connectionID != existing.ConnectionID ||
					m.retirement.credentialRef != valueOrEmpty(existing.ManagementCredentialRef) ||
					m.retirement.ownerKey != idempotencyKey {
					if err := m.proveManagementCredentialHandoff(ctx, existing, managementClientID, managementSecret); err != nil {
						return ConnectionView{}, safeConnectionError{category: "credential_handoff_unsafe", cause: err}
					}
					existing, err = m.persistManagementCredential(ctx, existing, managementClientID, managementSecret)
					if err != nil {
						return ConnectionView{}, err
					}
					m.retirement = &connectionRetirementAuthorization{
						connectionID: existing.ConnectionID, credentialRef: valueOrEmpty(existing.ManagementCredentialRef), ownerKey: idempotencyKey,
					}
				}
				if existing.LifecycleStatus == "disconnect_pending" {
					go m.retryDisconnect(existing.ConnectionID)
					return ConnectionView{}, safeConnectionError{category: "retirement_pending", cause: store.ErrSecurityEventConnectionConflict}
				}
				return ConnectionView{}, safeConnectionError{category: "connection_conflict", cause: store.ErrSecurityEventConnectionConflict}
			}
			if existing.ManagementClientID != "" && existing.ManagementClientID != managementClientID {
				return ConnectionView{}, store.ErrSecurityEventConnectionConflict
			}
			existing, err = m.persistManagementCredential(ctx, existing, managementClientID, managementSecret)
			if err != nil {
				return ConnectionView{}, err
			}
		}
		if existing.StreamID != nil && !resumableConnectionLifecycle(existing.LifecycleStatus) {
			if existing.Audience != nil {
				if err := m.activate(ctx, existing); err != nil {
					return ConnectionView{}, err
				}
			}
			return m.view(existing), nil
		}
		return m.resumeConnecting(ctx, existing)
	}
	connectionID := uuid.NewString()
	reference := "ssf-push-" + connectionID
	managementReference := "ssf-management-" + connectionID
	if err := m.stageIdentitySecret(ctx, managementReference, managementSecret); err != nil {
		return ConnectionView{}, err
	}
	secret, err := randomPushBearer()
	if err != nil {
		_ = m.retireIdentitySecret(ctx, managementReference)
		return ConnectionView{}, err
	}
	defer clear(secret)
	if err := m.stageIdentitySecret(ctx, reference, secret); err != nil {
		_ = m.retireIdentitySecret(ctx, managementReference)
		return ConnectionView{}, err
	}
	endpoint := strings.TrimRight(m.config.PublicBaseURL, "/") + "/api/v1/security-events/ssf"
	connection, err := m.repository.CreateSecurityEventConnection(ctx, store.CreateSecurityEventConnectionInput{
		ConnectionID: connectionID, TransmitterIssuer: issuer, EndpointURL: endpoint,
		CredentialRef: reference, ManagementClientID: managementClientID,
		ManagementCredentialRef: managementReference, IdempotencyKey: idempotencyKey,
	})
	if err != nil {
		// Commit outcome can be ambiguous: rollback keeps the staging rows,
		// while commit atomically removes them. Requeueing here could delete
		// credentials already referenced by the committed connection.
		return ConnectionView{}, err
	}
	return m.resumeConnecting(ctx, connection)
}

func (m *ConnectionManager) persistManagementCredential(ctx context.Context, connection store.SecurityEventConnection, clientID string, secret []byte) (store.SecurityEventConnection, error) {
	reference := "ssf-management-" + connection.ConnectionID + "-" + uuid.NewString()
	if err := m.stageIdentitySecret(ctx, reference, secret); err != nil {
		return store.SecurityEventConnection{}, err
	}
	updated, err := m.repository.SetSecurityEventManagementCredential(ctx, connection.ConnectionID, clientID, reference, connection.Version)
	if err != nil {
		// The staging row is the source of truth when commit outcome is unknown.
		return store.SecurityEventConnection{}, err
	}
	return updated, nil
}

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 {
		return ConnectionView{}, err
	}
	defer clear(secret)
	configuration, client, err := m.discover(ctx, connection.TransmitterIssuer)
	if err != nil {
		return ConnectionView{}, m.connectionError(ctx, connection, "discovery_failed", err)
	}
	token, err := m.managementToken(ctx, client, "ssf.stream.read ssf.stream.verify ssf.stream.manage")
	if err != nil {
		return ConnectionView{}, m.connectionError(ctx, connection, "management_token_failed", err)
	}
	description := "easyai-gateway:" + connection.ConnectionID
	stream, err := m.findStream(ctx, client, configuration, token, description, connection.EndpointURL)
	if errors.Is(err, store.ErrSecurityEventConnectionNotFound) {
		stream, err = m.createStream(ctx, client, configuration, token, connection.EndpointURL, string(secret), description)
	}
	if err != nil {
		return ConnectionView{}, m.connectionError(ctx, connection, "stream_create_failed", err)
	}
	if err := validateCreatedStream(stream, connection.TransmitterIssuer, connection.EndpointURL); err != nil {
		return ConnectionView{}, m.connectionError(ctx, connection, "stream_response_invalid", err)
	}
	connection, err = m.repository.BindSecurityEventConnection(ctx, connection.ConnectionID, stream.Audience, stream.StreamID, "verifying", connection.Version)
	if err != nil {
		return ConnectionView{}, err
	}
	started := time.Now().UTC()
	if err := m.activate(ctx, connection); err != nil {
		return ConnectionView{}, m.connectionError(ctx, connection, "receiver_activation_failed", err)
	}
	go m.finishAutomaticVerification(connection.ConnectionID, configuration, client, token, started, false)
	return m.view(connection), nil
}

func (m *ConnectionManager) Verify(ctx context.Context) (ConnectionView, error) {
	if err := m.requireOperationalBinding(); err != nil {
		return ConnectionView{}, err
	}
	connection, err := m.repository.SecurityEventConnection(ctx)
	if err != nil {
		return ConnectionView{}, err
	}
	m.mutex.RLock()
	runtime := m.runtime
	m.mutex.RUnlock()
	if runtime == nil || runtime.service == nil {
		return ConnectionView{}, errors.New("security event receiver is unavailable")
	}
	started := time.Now().UTC()
	runtime.service.RequestVerification(ctx)
	if connection.LifecycleStatus == "verifying" || connection.LifecycleStatus == "rotating" {
		configuration, client, discoverErr := m.discover(ctx, connection.TransmitterIssuer)
		if discoverErr != nil {
			return ConnectionView{}, discoverErr
		}
		token, tokenErr := m.managementToken(ctx, client, "ssf.stream.read ssf.stream.verify ssf.stream.manage")
		if tokenErr != nil {
			return ConnectionView{}, tokenErr
		}
		go m.finishAutomaticVerification(connection.ConnectionID, configuration, client, token, started, connection.LifecycleStatus == "rotating")
	}
	return m.view(connection), nil
}

func (m *ConnectionManager) RotateCredential(ctx context.Context) (ConnectionView, error) {
	m.operation.Lock()
	defer m.operation.Unlock()
	if err := m.requireOperationalBinding(); err != nil {
		return ConnectionView{}, err
	}
	connection, err := m.repository.SecurityEventConnection(ctx)
	if err != nil {
		return ConnectionView{}, err
	}
	if connection.StreamID == nil || connection.Audience == nil {
		return ConnectionView{}, store.ErrSecurityEventConnectionConflict
	}
	var nextReference string
	var next []byte
	if connection.NextCredentialRef == nil {
		nextReference = "ssf-push-" + connection.ConnectionID + "-next-" + uuid.NewString()
		next, err = randomPushBearer()
		if err != nil {
			return ConnectionView{}, err
		}
		if err := m.stageIdentitySecret(ctx, nextReference, next); err != nil {
			clear(next)
			return ConnectionView{}, err
		}
		connection, err = m.repository.SetSecurityEventNextCredential(ctx, connection.ConnectionID, nextReference, connection.Version)
		if err != nil {
			clear(next)
			// Never recreate cleanup intent after an ambiguous adoption commit.
			return ConnectionView{}, err
		}
	} else {
		nextReference = *connection.NextCredentialRef
		next, err = m.secrets.Get(ctx, nextReference)
		if err != nil {
			return ConnectionView{}, err
		}
	}
	defer clear(next)
	configuration, client, err := m.discover(ctx, connection.TransmitterIssuer)
	if err != nil {
		return ConnectionView{}, err
	}
	token, err := m.managementToken(ctx, client, "ssf.stream.read ssf.stream.verify ssf.stream.manage")
	if err != nil {
		return ConnectionView{}, err
	}
	if err := m.setStreamStatus(ctx, client, configuration, token, *connection.StreamID, "paused", "credential_rotation"); err != nil {
		return ConnectionView{}, err
	}
	if err := m.patchStreamCredential(ctx, client, configuration, token, *connection.StreamID, connection.EndpointURL, string(next)); err != nil {
		return ConnectionView{}, err
	}
	started := time.Now().UTC()
	if err := m.activate(ctx, connection); err != nil {
		return ConnectionView{}, err
	}
	go m.finishAutomaticVerification(connection.ConnectionID, configuration, client, token, started, true)
	return m.view(connection), nil
}

func (m *ConnectionManager) Disconnect(ctx context.Context) (ConnectionView, error) {
	m.operation.Lock()
	defer m.operation.Unlock()
	if err := m.requireOperationalBinding(); err != nil {
		return ConnectionView{}, err
	}
	connection, err := m.repository.SecurityEventConnection(ctx)
	if err != nil {
		return ConnectionView{}, err
	}
	return m.disconnect(ctx, connection)
}

// RetireConflictingConnection is a pairing-scoped recovery operation. It may
// retire only a connection owned by another workflow; a stale request becomes
// a no-op once the current pairing owns the singleton connection.
func (m *ConnectionManager) RetireConflictingConnection(ctx context.Context, currentOwnerKey string) error {
	m.operation.Lock()
	defer m.operation.Unlock()
	connection, err := m.repository.SecurityEventConnection(ctx)
	if errors.Is(err, store.ErrSecurityEventConnectionNotFound) {
		return nil
	}
	if err != nil {
		return err
	}
	if strings.TrimSpace(currentOwnerKey) == "" {
		return errors.New("current security event owner is required")
	}
	if connection.IdempotencyKey == currentOwnerKey {
		return nil
	}
	if connection.LifecycleStatus == "retiring" {
		go m.finishRetirement(connection)
		return nil
	}
	if m.retirement == nil || m.retirement.connectionID != connection.ConnectionID ||
		m.retirement.credentialRef != valueOrEmpty(connection.ManagementCredentialRef) ||
		m.retirement.ownerKey != currentOwnerKey {
		return safeConnectionError{category: "credential_handoff_unsafe", cause: store.ErrSecurityEventConnectionConflict}
	}
	if connection.LifecycleStatus == "disconnect_pending" {
		go m.retryDisconnect(connection.ConnectionID)
		return nil
	}
	_, err = m.disconnect(ctx, connection)
	return err
}

func (m *ConnectionManager) disconnect(ctx context.Context, connection store.SecurityEventConnection) (ConnectionView, error) {
	if connection.LifecycleStatus == "retiring" || connection.LifecycleStatus == "disconnect_pending" {
		return m.view(connection), nil
	}
	if connection.StreamID != nil {
		configuration, client, discoverErr := m.discover(ctx, connection.TransmitterIssuer)
		if discoverErr == nil {
			token, tokenErr := m.managementToken(ctx, client, "ssf.stream.read ssf.stream.verify ssf.stream.manage")
			if tokenErr == nil {
				discoverErr = m.deleteStream(ctx, client, configuration, token, *connection.StreamID)
				if isRemoteHTTPStatus(discoverErr, http.StatusNotFound) {
					discoverErr = nil
				}
			}
		}
		if discoverErr != nil {
			category := "remote_disconnect_failed"
			transitioned, transitionErr := m.repository.TransitionSecurityEventConnectionLifecycle(
				ctx, connection.ConnectionID, connection.LifecycleStatus, connection.Version, "disconnect_pending", &category,
			)
			if transitionErr != nil {
				return ConnectionView{}, transitionErr
			}
			connection = transitioned
			go m.retryDisconnect(connection.ConnectionID)
			return m.view(connection), nil
		}
	}
	connection, err := m.repository.TransitionSecurityEventConnectionLifecycle(
		ctx, connection.ConnectionID, connection.LifecycleStatus, connection.Version, "retiring", nil,
	)
	if err != nil {
		return ConnectionView{}, err
	}
	go m.finishRetirement(connection)
	return m.view(connection), nil
}

// DiscardPreparedConnection removes only the connection created by one identity
// pairing. Bound streams continue through the normal retirement window so
// revocation watermarks and introspection fallback are never bypassed.
func (m *ConnectionManager) DiscardPreparedConnection(ctx context.Context, ownerKey string) error {
	m.operation.Lock()
	defer m.operation.Unlock()
	connection, err := m.repository.SecurityEventConnection(ctx)
	if errors.Is(err, store.ErrSecurityEventConnectionNotFound) {
		return nil
	}
	if err != nil {
		return safeConnectionError{category: "connection_cleanup_failed", cause: err}
	}
	if strings.TrimSpace(ownerKey) == "" || connection.IdempotencyKey != ownerKey {
		return nil
	}
	if connection.StreamID != nil {
		if connection.LifecycleStatus != "retiring" && connection.LifecycleStatus != "disconnect_pending" {
			if _, err := m.disconnect(ctx, connection); err != nil {
				return safeConnectionError{category: "connection_cleanup_failed", cause: err}
			}
		}
		return safeConnectionError{category: "retirement_pending", cause: errors.New("security event retirement is pending")}
	}
	err = m.repository.DiscardPreparedSecurityEventConnection(
		ctx, connection.ConnectionID, ownerKey, connection.Version,
	)
	if err != nil && !errors.Is(err, store.ErrSecurityEventConnectionNotFound) {
		return safeConnectionError{category: "connection_cleanup_failed", cause: err}
	}
	m.stopRuntime()
	return nil
}

func connectionSecretReferences(connection store.SecurityEventConnection) []string {
	references := []string{connection.CredentialRef}
	if connection.ManagementCredentialRef != nil {
		references = append(references, *connection.ManagementCredentialRef)
	}
	if connection.NextCredentialRef != nil {
		references = append(references, *connection.NextCredentialRef)
	}
	return references
}

func (m *ConnectionManager) retryDisconnect(connectionID string) {
	delay := time.Second
	for {
		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 != "disconnect_pending" || connection.StreamID == nil {
			return
		}
		configuration, client, err := m.discover(m.ctx, connection.TransmitterIssuer)
		if err == nil {
			var token string
			token, err = m.managementToken(m.ctx, client, "ssf.stream.read ssf.stream.verify ssf.stream.manage")
			if err == nil {
				err = m.deleteStream(m.ctx, client, configuration, token, *connection.StreamID)
				if isRemoteHTTPStatus(err, http.StatusNotFound) {
					err = nil
				}
			}
		}
		if err == nil {
			connection, err = m.repository.TransitionSecurityEventConnectionLifecycle(
				m.ctx, connectionID, "disconnect_pending", connection.Version, "retiring", nil,
			)
			if err == nil {
				go m.finishRetirement(connection)
			}
			return
		}
		if delay < time.Minute {
			delay *= 2
		}
	}
}

func (m *ConnectionManager) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	m.mutex.RLock()
	runtime := m.runtime
	m.mutex.RUnlock()
	if runtime == nil || runtime.handler == nil {
		w.Header().Set("Cache-Control", "no-store")
		if _, err := m.repository.SecurityEventConnection(r.Context()); errors.Is(err, store.ErrSecurityEventConnectionNotFound) {
			http.NotFound(w, r)
			return
		}
		w.Header().Set("Retry-After", "1")
		w.WriteHeader(http.StatusServiceUnavailable)
		return
	}
	runtime.handler.ServeHTTP(w, r)
}

func (m *ConnectionManager) Evaluate(ctx context.Context, identity auth.OIDCSecurityEventIdentity) (auth.OIDCSecurityEventEvaluation, error) {
	connection, err := m.repository.SecurityEventConnection(ctx)
	if errors.Is(err, store.ErrSecurityEventConnectionNotFound) {
		return auth.OIDCSecurityEventEvaluation{}, nil
	}
	if err != nil {
		return auth.OIDCSecurityEventEvaluation{Enabled: true, RequireIntrospection: true}, err
	}
	m.mutex.RLock()
	runtime := m.runtime
	m.mutex.RUnlock()
	if runtime == nil || runtime.service == nil {
		if connection.Audience == nil {
			return auth.OIDCSecurityEventEvaluation{Enabled: true, RequireIntrospection: true}, nil
		}
		result, evaluateErr := m.repository.EvaluateOIDCSecurityEvent(
			ctx, connection.TransmitterIssuer, *connection.Audience, identity.Issuer, identity.TenantID, identity.Subject,
			identity.IssuedAt, time.Now().UTC(), m.config.StaleAfter,
		)
		if result.Revoked && m.metrics != nil {
			m.metrics.ObserveWatermarkRejection()
		}
		return auth.OIDCSecurityEventEvaluation{Enabled: true, Revoked: result.Revoked, RequireIntrospection: true}, evaluateErr
	}
	evaluation, err := runtime.service.Evaluate(ctx, identity)
	evaluation.Enabled = true
	if connection.LifecycleStatus != "enabled" && connection.LifecycleStatus != "bootstrap" {
		evaluation.RequireIntrospection = true
	}
	return evaluation, err
}

func (m *ConnectionManager) activate(ctx context.Context, connection store.SecurityEventConnection) error {
	if connection.StreamID == nil || connection.Audience == nil {
		return errors.New("security event connection binding is incomplete")
	}
	current, err := m.secrets.Get(ctx, connection.CredentialRef)
	if err != nil {
		return err
	}
	defer clear(current)
	managementClientID, managementSecret, err := m.managementCredential(ctx, connection)
	if err != nil {
		return err
	}
	defer clear(managementSecret)
	var next []byte
	if connection.NextCredentialRef != nil {
		next, err = m.secrets.Get(ctx, *connection.NextCredentialRef)
		if err != nil {
			return err
		}
		defer clear(next)
	}
	issuerURL, err := validatedIssuerURL(connection.TransmitterIssuer, m.config.AppEnv)
	if err != nil {
		return err
	}
	safeClient := m.config.HTTPClient
	if safeClient == nil {
		safeClient = safeHTTPClient(issuerURL, m.config.AppEnv)
	}
	verifier, err := NewVerifier(VerifierConfig{
		TransmitterIssuer: connection.TransmitterIssuer, Audience: *connection.Audience,
		SubjectIssuer: m.config.OIDCIssuer, TenantID: m.config.OIDCTenantID, StreamID: *connection.StreamID,
		ClockSkew: m.config.ClockSkew, HTTPClient: safeClient,
	})
	if err != nil {
		return err
	}
	verifier.SetMetrics(m.metrics)
	handler, err := NewHandler(verifier, m.repository, connection.TransmitterIssuer, *connection.Audience, *connection.StreamID, string(current), string(next))
	if err != nil {
		return err
	}
	handler.SetMetrics(m.metrics)
	service, err := NewService(m.repository, ServiceConfig{
		Issuer: connection.TransmitterIssuer, Audience: *connection.Audience, StreamID: *connection.StreamID,
		ManagementTokenURL: strings.TrimRight(m.config.OIDCIssuer, "/") + "/token",
		ManagementClientID: managementClientID, ManagementSecret: string(managementSecret),
		HeartbeatInterval: m.config.HeartbeatInterval, StaleAfter: m.config.StaleAfter,
		HTTPClient: safeClient,
	})
	if err != nil {
		return err
	}
	service.SetMetrics(m.metrics)
	runtimeContext, cancel := context.WithCancel(m.ctx)
	if err := service.Initialize(runtimeContext); err != nil {
		cancel()
		return err
	}
	m.mutex.Lock()
	previous := m.runtime
	m.runtime = &connectionRuntime{handler: handler, service: service, cancel: cancel}
	m.mutex.Unlock()
	if previous != nil && previous.cancel != nil {
		previous.cancel()
	}
	go service.Run(runtimeContext)
	return nil
}

func (m *ConnectionManager) stopRuntime() {
	m.mutex.Lock()
	previous := m.runtime
	m.runtime = nil
	m.mutex.Unlock()
	if previous != nil && previous.cancel != nil {
		previous.cancel()
	}
}

func (m *ConnectionManager) finishAutomaticVerification(connectionID string, configuration transmitterConfiguration, client *http.Client, token string, started time.Time, rotating bool) {
	ctx, cancel := context.WithTimeout(m.ctx, 90*time.Second)
	defer cancel()
	ticker := time.NewTicker(250 * time.Millisecond)
	defer ticker.Stop()
	expectedLifecycle := "verifying"
	if rotating {
		expectedLifecycle = "rotating"
	}
	for {
		connection, err := m.repository.SecurityEventConnection(ctx)
		if err != nil || connection.ConnectionID != connectionID || connection.StreamID == nil || connection.LifecycleStatus != expectedLifecycle {
			return
		}
		_, verifiedAt, _ := m.repository.SecurityEventMetrics(ctx, connection.TransmitterIssuer, valueOrEmpty(connection.Audience))
		if verifiedAt != nil && !verifiedAt.Before(started) {
			if err := m.setStreamStatus(ctx, client, configuration, token, *connection.StreamID, "enabled", "verification_succeeded"); err != nil {
				category := "stream_enable_failed"
				_, _ = m.repository.TransitionSecurityEventConnectionLifecycle(
					ctx, connectionID, expectedLifecycle, connection.Version, expectedLifecycle, &category,
				)
				return
			}
			if rotating && connection.NextCredentialRef != nil {
				nextReference := *connection.NextCredentialRef
				select {
				case <-m.ctx.Done():
					return
				case <-time.After(m.config.CredentialOverlapDuration):
				}
				current, currentErr := m.repository.SecurityEventConnection(m.ctx)
				if currentErr != nil || current.ConnectionID != connectionID || current.LifecycleStatus != "rotating" ||
					current.NextCredentialRef == nil || *current.NextCredentialRef != nextReference {
					return
				}
				connection, err = m.repository.PromoteSecurityEventCredential(m.ctx, connectionID, nextReference)
				if err == nil {
					_ = m.activate(m.ctx, connection)
					go m.finishBootstrap(connectionID)
				}
				return
			}
			connection, err = m.repository.TransitionSecurityEventConnectionLifecycle(
				ctx, connectionID, expectedLifecycle, connection.Version, "bootstrap", nil,
			)
			if err == nil {
				go m.finishBootstrap(connectionID)
			}
			return
		}
		select {
		case <-ctx.Done():
			category := "verification_timeout"
			_, _ = m.repository.TransitionSecurityEventConnectionLifecycle(
				m.ctx, connectionID, expectedLifecycle, connection.Version, expectedLifecycle, &category,
			)
			return
		case <-ticker.C:
		}
	}
}

func (m *ConnectionManager) finishBootstrap(connectionID string) {
	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" {
			_, _ = m.repository.TransitionSecurityEventConnectionLifecycle(
				m.ctx, connectionID, "bootstrap", connection.Version, "enabled", nil,
			)
			return
		}
		category := "bootstrap_verification_stale"
		_, _ = m.repository.TransitionSecurityEventConnectionLifecycle(
			m.ctx, connectionID, "bootstrap", connection.Version, "degraded", &category,
		)
	}
}

func (m *ConnectionManager) finishRetirement(retirement store.SecurityEventConnection) {
	retryDelay := time.Second
	for {
		connection, err := m.repository.SecurityEventConnection(m.ctx)
		if errors.Is(err, store.ErrSecurityEventConnectionNotFound) {
			return
		}
		if err != nil {
			if !waitForConnectionRetry(m.ctx, retryDelay) {
				return
			}
			retryDelay = nextConnectionRetryDelay(retryDelay)
			continue
		}
		if connection.ConnectionID != retirement.ConnectionID || connection.LifecycleStatus != "retiring" {
			return
		}
		if remaining := m.config.RetirementDuration - time.Since(connection.UpdatedAt); remaining > 0 {
			select {
			case <-m.ctx.Done():
				return
			case <-time.After(remaining):
			}
			continue
		}
		m.stopRuntime()
		err = m.repository.FinalizeRetiringSecurityEventConnection(m.ctx, connection.ConnectionID, connection.Version)
		if err == nil || errors.Is(err, store.ErrSecurityEventConnectionNotFound) {
			return
		}
		if !waitForConnectionRetry(m.ctx, retryDelay) {
			return
		}
		retryDelay = nextConnectionRetryDelay(retryDelay)
	}
}

func waitForConnectionRetry(ctx context.Context, delay time.Duration) bool {
	select {
	case <-ctx.Done():
		return false
	case <-time.After(delay):
		return true
	}
}

func nextConnectionRetryDelay(delay time.Duration) time.Duration {
	if delay < time.Minute {
		return delay * 2
	}
	return delay
}

const identitySecretStagingTTL = 10 * time.Minute

func (m *ConnectionManager) stageIdentitySecret(ctx context.Context, reference string, value []byte) error {
	if err := m.repository.QueueIdentitySecretCleanup(ctx, reference, time.Now().UTC().Add(identitySecretStagingTTL)); err != nil {
		return err
	}
	if err := m.secrets.Put(ctx, reference, value); err != nil {
		deleteErr := m.secrets.Delete(ctx, reference)
		if deleteErr == nil || errors.Is(deleteErr, ErrSecretNotFound) {
			_, _ = m.repository.RemovePendingIdentitySecretCleanup(ctx, reference)
		}
		return err
	}
	return nil
}

func (m *ConnectionManager) retireIdentitySecret(ctx context.Context, reference string) error {
	return m.repository.QueueIdentitySecretCleanup(ctx, reference, time.Now().UTC())
}

func (m *ConnectionManager) validatePrerequisites(issuer, managementClientID string, managementSecret []byte, requirePublicBaseURL bool) error {
	if !m.config.OIDCEnabled || strings.TrimSpace(managementClientID) == "" || len(managementSecret) < 16 {
		return errors.New("OIDC and RFC 7662 machine client must be configured")
	}
	if _, err := uuid.Parse(m.config.OIDCTenantID); err != nil {
		return errors.New("OIDC tenant id is invalid")
	}
	if _, err := validatedIssuerURL(issuer, m.config.AppEnv); err != nil {
		return err
	}
	if !requirePublicBaseURL {
		return nil
	}
	endpoint, err := url.Parse(strings.TrimRight(m.config.PublicBaseURL, "/") + "/api/v1/security-events/ssf")
	if err != nil || endpoint.Host == "" || endpoint.User != nil || endpoint.RawQuery != "" || endpoint.Fragment != "" {
		return errors.New("AI_GATEWAY_PUBLIC_BASE_URL is invalid")
	}
	if endpoint.Scheme != "https" && !(isLocalEnvironment(m.config.AppEnv) && endpoint.Scheme == "http" && isLocalHostname(endpoint.Hostname())) {
		return errors.New("AI_GATEWAY_PUBLIC_BASE_URL must use HTTPS")
	}
	return nil
}

func (m *ConnectionManager) discover(ctx context.Context, issuer string) (transmitterConfiguration, *http.Client, error) {
	issuerURL, err := validatedIssuerURL(issuer, m.config.AppEnv)
	if err != nil {
		return transmitterConfiguration{}, nil, err
	}
	client := m.config.HTTPClient
	if client == nil {
		client = safeHTTPClient(issuerURL, m.config.AppEnv)
	}
	discoveryURL := *issuerURL
	discoveryURL.Path = "/.well-known/ssf-configuration" + strings.TrimRight(issuerURL.EscapedPath(), "/")
	var configuration transmitterConfiguration
	if err := requestJSON(ctx, client, http.MethodGet, discoveryURL.String(), "", nil, &configuration); err != nil {
		return transmitterConfiguration{}, nil, err
	}
	if configuration.SpecVersion != "1_0" || strings.TrimRight(configuration.Issuer, "/") != strings.TrimRight(issuer, "/") ||
		!contains(configuration.DeliveryMethodsSupported, pushDeliveryMethod) {
		return transmitterConfiguration{}, nil, errors.New("SSF discovery metadata is incompatible")
	}
	for _, endpoint := range []string{configuration.JWKSURI, configuration.ConfigurationEndpoint, configuration.StatusEndpoint, configuration.VerificationEndpoint} {
		parsed, parseErr := url.Parse(endpoint)
		if parseErr != nil || parsed.Scheme != issuerURL.Scheme || !strings.EqualFold(parsed.Host, issuerURL.Host) || parsed.User != nil || parsed.Fragment != "" {
			return transmitterConfiguration{}, nil, errors.New("SSF discovery endpoint is not trusted")
		}
	}
	return configuration, client, nil
}

func (m *ConnectionManager) managementToken(ctx context.Context, client *http.Client, scopes string) (string, error) {
	connection, err := m.repository.SecurityEventConnection(ctx)
	if err != nil {
		return "", err
	}
	managementClientID, managementSecret, err := m.managementCredential(ctx, connection)
	if err != nil {
		return "", err
	}
	defer clear(managementSecret)
	return m.requestManagementToken(ctx, client, scopes, managementClientID, managementSecret)
}

func (m *ConnectionManager) requestManagementToken(ctx context.Context, client *http.Client, scopes, managementClientID string, managementSecret []byte) (string, error) {
	form := url.Values{"grant_type": {"client_credentials"}, "scope": {scopes}}
	request, err := http.NewRequestWithContext(ctx, http.MethodPost, strings.TrimRight(m.config.OIDCIssuer, "/")+"/token", strings.NewReader(form.Encode()))
	if err != nil {
		return "", err
	}
	request.SetBasicAuth(managementClientID, string(managementSecret))
	request.Header.Set("Content-Type", "application/x-www-form-urlencoded")
	response, err := client.Do(request)
	if err != nil {
		return "", err
	}
	defer response.Body.Close()
	if response.StatusCode != http.StatusOK {
		_, _ = io.Copy(io.Discard, io.LimitReader(response.Body, 64*1024))
		return "", fmt.Errorf("machine token request returned HTTP %d", response.StatusCode)
	}
	var payload struct {
		AccessToken string `json:"access_token"`
		TokenType   string `json:"token_type"`
	}
	if err := decodeLimitedJSON(response.Body, &payload); err != nil || payload.AccessToken == "" || !strings.EqualFold(payload.TokenType, "Bearer") {
		return "", errors.New("machine token response is invalid")
	}
	return payload.AccessToken, nil
}

func (m *ConnectionManager) proveManagementCredentialHandoff(ctx context.Context, connection store.SecurityEventConnection, managementClientID string, managementSecret []byte) error {
	if connection.StreamID == nil || connection.Audience == nil {
		return store.ErrSecurityEventConnectionConflict
	}
	configuration, client, err := m.discover(ctx, connection.TransmitterIssuer)
	if err != nil {
		return err
	}
	token, err := m.requestManagementToken(ctx, client, "ssf.stream.read ssf.stream.manage", managementClientID, managementSecret)
	if err != nil {
		return err
	}
	var streams []streamConfiguration
	if err := requestJSON(ctx, client, http.MethodGet, configuration.ConfigurationEndpoint, token, nil, &streams); err != nil {
		return err
	}
	for _, stream := range streams {
		if stream.StreamID != *connection.StreamID {
			continue
		}
		if err := validateCreatedStream(stream, connection.TransmitterIssuer, connection.EndpointURL); err != nil || stream.Audience != *connection.Audience {
			return store.ErrSecurityEventConnectionConflict
		}
		return nil
	}
	// An authenticated empty list proves that the new credential belongs to a
	// token issuer trusted by the old transmitter. Disconnect treats the absent
	// Stream as an idempotent remote deletion.
	return nil
}

func (m *ConnectionManager) requireOperationalBinding() error {
	if m.bindingMismatch {
		return safeConnectionError{category: "credential_handoff_unsafe", cause: store.ErrSecurityEventConnectionConflict}
	}
	return nil
}

func (m *ConnectionManager) managementCredential(ctx context.Context, connection store.SecurityEventConnection) (string, []byte, error) {
	if connection.ManagementCredentialRef != nil && connection.ManagementClientID != "" {
		secret, err := m.secrets.Get(ctx, *connection.ManagementCredentialRef)
		if err != nil {
			return "", nil, err
		}
		if len(secret) < 16 {
			clear(secret)
			return "", nil, errors.New("managed machine credential is invalid")
		}
		return connection.ManagementClientID, secret, nil
	}
	if m.config.ManagementClientID != "" && m.config.ManagementClientSecret != "" {
		return m.config.ManagementClientID, []byte(m.config.ManagementClientSecret), nil
	}
	return "", nil, errors.New("managed machine credential is unavailable")
}

func (m *ConnectionManager) findStream(ctx context.Context, client *http.Client, configuration transmitterConfiguration, token, description, endpoint string) (streamConfiguration, error) {
	var streams []streamConfiguration
	if err := requestJSON(ctx, client, http.MethodGet, configuration.ConfigurationEndpoint, token, nil, &streams); err != nil {
		return streamConfiguration{}, err
	}
	for _, stream := range streams {
		if stream.Description != nil && *stream.Description == description && stream.Delivery.EndpointURL == endpoint {
			return stream, nil
		}
	}
	return streamConfiguration{}, store.ErrSecurityEventConnectionNotFound
}

func (m *ConnectionManager) createStream(ctx context.Context, client *http.Client, configuration transmitterConfiguration, token, endpoint, secret, description string) (streamConfiguration, error) {
	body := map[string]any{
		"delivery":         map[string]any{"method": pushDeliveryMethod, "endpoint_url": endpoint, "authorization_header": "Bearer " + secret},
		"events_requested": []string{sessionRevokedEvent}, "description": description,
	}
	var stream streamConfiguration
	err := requestJSON(ctx, client, http.MethodPost, configuration.ConfigurationEndpoint, token, body, &stream)
	return stream, err
}

func (m *ConnectionManager) patchStreamCredential(ctx context.Context, client *http.Client, configuration transmitterConfiguration, token, streamID, endpoint, secret string) error {
	body := map[string]any{"stream_id": streamID, "delivery": map[string]any{
		"method": pushDeliveryMethod, "endpoint_url": endpoint, "authorization_header": "Bearer " + secret,
	}}
	return requestJSON(ctx, client, http.MethodPatch, configuration.ConfigurationEndpoint, token, body, &streamConfiguration{})
}

func (m *ConnectionManager) setStreamStatus(ctx context.Context, client *http.Client, configuration transmitterConfiguration, token, streamID, status, reason string) error {
	body := map[string]any{"stream_id": streamID, "status": status, "reason": reason}
	return requestJSON(ctx, client, http.MethodPost, configuration.StatusEndpoint, token, body, &map[string]any{})
}

func (m *ConnectionManager) deleteStream(ctx context.Context, client *http.Client, configuration transmitterConfiguration, token, streamID string) error {
	endpoint, _ := url.Parse(configuration.ConfigurationEndpoint)
	query := endpoint.Query()
	query.Set("stream_id", streamID)
	endpoint.RawQuery = query.Encode()
	return requestJSON(ctx, client, http.MethodDelete, endpoint.String(), token, nil, nil)
}

func (m *ConnectionManager) connectionError(ctx context.Context, connection store.SecurityEventConnection, category string, cause error) error {
	_, _ = m.repository.TransitionSecurityEventConnectionLifecycle(
		ctx, connection.ConnectionID, connection.LifecycleStatus, connection.Version, "error", &category,
	)
	return safeConnectionError{category: category, cause: cause}
}

func (m *ConnectionManager) view(connection store.SecurityEventConnection) ConnectionView {
	return ConnectionView{
		ConnectionID: connection.ConnectionID, TransmitterIssuer: connection.TransmitterIssuer,
		ReceiverEndpoint: connection.EndpointURL, Audience: connection.Audience, StreamID: connection.StreamID,
		LifecycleStatus: connection.LifecycleStatus, HealthMode: connection.HealthMode,
		ManagementClientID: managementClientIDForView(connection.ManagementClientID, m.config.ManagementClientID), LastVerificationAt: connection.LastVerificationAt,
		LastErrorCategory: connection.LastErrorCategory, Version: connection.Version,
	}
}

func managementClientIDForView(managed, fallback string) string {
	if strings.TrimSpace(managed) != "" {
		return managed
	}
	return fallback
}

func validateCreatedStream(stream streamConfiguration, issuer, endpoint string) error {
	if _, err := uuid.Parse(stream.StreamID); err != nil || stream.Issuer != issuer || stream.Audience == "" ||
		stream.Delivery.Method != pushDeliveryMethod || stream.Delivery.EndpointURL != endpoint ||
		len(stream.EventsRequested) != 1 || stream.EventsRequested[0] != sessionRevokedEvent {
		return errors.New("created SSF stream does not match the requested receiver")
	}
	return nil
}

func randomPushBearer() ([]byte, error) {
	raw := make([]byte, 32)
	if _, err := rand.Read(raw); err != nil {
		return nil, err
	}
	encoded := make([]byte, base64.RawURLEncoding.EncodedLen(len(raw)))
	base64.RawURLEncoding.Encode(encoded, raw)
	clear(raw)
	return encoded, nil
}

func requestJSON(ctx context.Context, client *http.Client, method, endpoint, token string, body, output any) error {
	var reader io.Reader
	if body != nil {
		payload, err := json.Marshal(body)
		if err != nil {
			return err
		}
		reader = bytes.NewReader(payload)
	}
	request, err := http.NewRequestWithContext(ctx, method, endpoint, reader)
	if err != nil {
		return err
	}
	request.Header.Set("Accept", "application/json")
	if token != "" {
		request.Header.Set("Authorization", "Bearer "+token)
	}
	if body != nil {
		request.Header.Set("Content-Type", "application/json")
	}
	response, err := client.Do(request)
	if err != nil {
		return err
	}
	defer response.Body.Close()
	if response.StatusCode < 200 || response.StatusCode >= 300 {
		_, _ = io.Copy(io.Discard, io.LimitReader(response.Body, 64*1024))
		return remoteHTTPError{status: response.StatusCode}
	}
	if output == nil || response.StatusCode == http.StatusNoContent {
		_, _ = io.Copy(io.Discard, io.LimitReader(response.Body, 64*1024))
		return nil
	}
	return decodeLimitedJSON(response.Body, output)
}

func isRemoteHTTPStatus(err error, status int) bool {
	var remoteError remoteHTTPError
	return errors.As(err, &remoteError) && remoteError.status == status
}

func decodeLimitedJSON(reader io.Reader, output any) error {
	payload, err := io.ReadAll(io.LimitReader(reader, 64*1024+1))
	if err != nil || len(payload) > 64*1024 {
		return errors.New("security event response is too large")
	}
	if len(payload) == 0 {
		return nil
	}
	return json.Unmarshal(payload, output)
}

func validatedIssuerURL(raw, appEnv string) (*url.URL, error) {
	parsed, err := url.Parse(strings.TrimSpace(raw))
	if err != nil || parsed.Host == "" || parsed.User != nil || parsed.RawQuery != "" || parsed.Fragment != "" {
		return nil, errors.New("transmitter issuer is invalid")
	}
	if parsed.Scheme != "https" && !(isLocalEnvironment(appEnv) && parsed.Scheme == "http" && isLocalHostname(parsed.Hostname())) {
		return nil, errors.New("transmitter issuer must use HTTPS")
	}
	return parsed, nil
}

func safeHTTPClient(issuer *url.URL, appEnv string) *http.Client {
	transport := http.DefaultTransport.(*http.Transport).Clone()
	transport.Proxy = nil
	transport.DisableCompression = true
	transport.TLSClientConfig = &tls.Config{MinVersion: tls.VersionTLS12}
	transport.DialContext = func(ctx context.Context, network, address string) (net.Conn, error) {
		host, port, err := net.SplitHostPort(address)
		if err != nil {
			return nil, err
		}
		addresses, err := net.DefaultResolver.LookupIPAddr(ctx, host)
		if err != nil || len(addresses) == 0 {
			return nil, errors.New("transmitter DNS resolution failed")
		}
		allowLocal := isLocalEnvironment(appEnv) && isLocalHostname(issuer.Hostname())
		for _, address := range addresses {
			if blockedAddress(address.IP) && !allowLocal {
				return nil, errors.New("transmitter resolved to a blocked network")
			}
		}
		dialer := &net.Dialer{Timeout: 5 * time.Second}
		return dialResolvedAddresses(ctx, network, port, addresses, dialer)
	}
	return &http.Client{Timeout: 10 * time.Second, Transport: transport, CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }}
}

func dialResolvedAddresses(ctx context.Context, network, port string, addresses []net.IPAddr, dialer *net.Dialer) (net.Conn, error) {
	var attempts []error
	for _, address := range addresses {
		connection, err := dialer.DialContext(ctx, network, net.JoinHostPort(address.IP.String(), port))
		if err == nil {
			return connection, nil
		}
		attempts = append(attempts, err)
		if ctx.Err() != nil {
			break
		}
	}
	return nil, fmt.Errorf("transmitter connection failed for every resolved address: %w", errors.Join(attempts...))
}

func blockedAddress(ip net.IP) bool {
	if ip.IsLoopback() || ip.IsPrivate() || ip.IsLinkLocalUnicast() || ip.IsLinkLocalMulticast() ||
		ip.IsUnspecified() || ip.IsMulticast() {
		return true
	}
	address, ok := netip.AddrFromSlice(ip)
	if !ok {
		return true
	}
	address = address.Unmap()
	for _, prefix := range blockedNetworkPrefixes {
		if prefix.Contains(address) {
			return true
		}
	}
	return false
}

var blockedNetworkPrefixes = []netip.Prefix{
	netip.MustParsePrefix("0.0.0.0/8"), netip.MustParsePrefix("100.64.0.0/10"),
	netip.MustParsePrefix("192.0.0.0/24"), netip.MustParsePrefix("192.0.2.0/24"),
	netip.MustParsePrefix("198.18.0.0/15"), netip.MustParsePrefix("198.51.100.0/24"),
	netip.MustParsePrefix("203.0.113.0/24"), netip.MustParsePrefix("240.0.0.0/4"),
	netip.MustParsePrefix("100::/64"), netip.MustParsePrefix("2001:db8::/32"),
}

func isLocalHostname(value string) bool {
	value = strings.ToLower(strings.TrimSpace(value))
	if value == "localhost" {
		return true
	}
	ip := net.ParseIP(value)
	return ip != nil && ip.IsLoopback()
}

func isLocalEnvironment(value string) bool {
	switch strings.ToLower(strings.TrimSpace(value)) {
	case "development", "dev", "local", "test":
		return true
	default:
		return false
	}
}

func contains(values []string, expected string) bool {
	for _, value := range values {
		if value == expected {
			return true
		}
	}
	return false
}

func valueOrEmpty(value *string) string {
	if value == nil {
		return ""
	}
	return *value
}