server

package
v0.8.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 29, 2026 License: Apache-2.0 Imports: 44 Imported by: 0

Documentation

Index

Constants

View Source 
const MaxStreamLogsPods = 5

MaxStreamLogsPods is the maximum number of pod log streams per StreamLogs request.

View Source 
const (

	// tunnelProxyPort is the dedicated port for tunnel proxy (no auth).
	TunnelProxyPort = 8081
)

Variables

This section is empty.

Functions

func AuthorizeAction added in v0.6.0 

func AuthorizeAction(ctx context.Context, k8sClient kubernetes.Interface, auditLogger *AuditLogger, verb, namespace, resource string) error

AuthorizeAction performs a Kubernetes SubjectAccessReview for the authenticated user against a Diverge resource. The user's identity is extracted from the request context (set by the auth middleware).

func AuthorizePodLogs added in v0.6.0 

func AuthorizePodLogs(ctx context.Context, k8sClient kubernetes.Interface, auditLogger *AuditLogger, namespace string) error

AuthorizePodLogs performs a SubjectAccessReview for pods/log access. StreamLogs requires both environment read AND pod log read permissions.

func CRDEnvToProto added in v0.7.0 

func CRDEnvToProto(crd *v1alpha1.Environment) (*pb.Environment, error)

CRDEnvToProto maps a CRD Environment to the protobuf type.

func CRDPgToProto added in v0.7.0 

func CRDPgToProto(crd *v1alpha1.PreviewGroup) (*pb.PreviewGroup, error)

CRDPgToProto maps a CRD PreviewGroup to the protobuf type.

func GetBroadcasterMetrics added in v0.5.0 

func GetBroadcasterMetrics() streaming.BroadcasterMetrics

func ListenAndServeTunnelProxy added in v0.7.0 

func ListenAndServeTunnelProxy(tm *TunnelManager, port int, logger *slog.Logger) error

ListenAndServeTunnelProxy starts the tunnel proxy server on the given port.

func NewAuthInterceptor 

func NewAuthInterceptor(authAttempts *prometheus.CounterVec) connect.Interceptor

func NewAuthMetrics added in v0.6.0 

func NewAuthMetrics() *auth.AuthMetrics

NewAuthMetrics returns the auth metrics wired to the auth middleware's expected types.

func NewAuthService 

func NewAuthService(k8s kubernetes.Interface, logger *slog.Logger, auditLogger *AuditLogger) divergev1alpha1connect.AuthServiceHandler

func NewClusterService 

func NewClusterService(c client.Client, k8s kubernetes.Interface, logger *slog.Logger, auditLogger *AuditLogger, version string) divergev1alpha1connect.ClusterServiceHandler

NewClusterService creates a ClusterService with the given version string. Pass the build-injected version (e.g. from ldflags) for accurate reporting.

func NewEnvironmentService 

func NewEnvironmentService(c client.Client, k8s kubernetes.Interface, informerMgr *streaming.InformerManager, logStreamer *streaming.LogStreamer, limiter *StreamLimiter, logger *slog.Logger, audit *AuditLogger) divergev1alpha1connect.EnvironmentServiceHandler

func NewMetricsInterceptor added in v0.5.0 

func NewMetricsInterceptor() connect.Interceptor

func NewPreviewGroupService 

func NewPreviewGroupService(c client.Client, k8s kubernetes.Interface, informerMgr *streaming.InformerManager, limiter *StreamLimiter, logger *slog.Logger, audit *AuditLogger) divergev1alpha1connect.PreviewGroupServiceHandler

func NewTunnelProxyServer added in v0.7.0 

func NewTunnelProxyServer(tm *TunnelManager, port int, logger *slog.Logger) *http.Server

NewTunnelProxyServer creates a dedicated HTTP server for the tunnel proxy. P0 #3: This runs on a SEPARATE port from the RPC server. P0 #6: NO auth middleware — cluster-internal traffic only. Host-header validation prevents open proxy abuse.

func ProtoEnvToCRD added in v0.7.0 

func ProtoEnvToCRD(proto *pb.Environment) (*v1alpha1.Environment, error)

ProtoEnvToCRD maps a protobuf Environment to the CRD type.

func ProtoPgToCRD added in v0.7.0 

func ProtoPgToCRD(proto *pb.PreviewGroup) (*v1alpha1.PreviewGroup, error)

ProtoPgToCRD maps a protobuf PreviewGroup to the CRD type.

func ProtocolTelemetryMiddleware added in v0.7.0 

func ProtocolTelemetryMiddleware(next http.Handler) http.Handler

ProtocolTelemetryMiddleware records the ConnectRPC wire protocol and client SDK for every request. Designed for <1μs overhead per request.

func SanitizeK8sError added in v0.6.0 

func SanitizeK8sError(logger *slog.Logger, err error) error

SanitizeK8sError maps Kubernetes API errors to safe Connect error codes. The raw K8s error is logged server-side but never returned to the client. Returns nil when err is nil.

func SetStreamLimiterMax added in v0.7.0 

func SetStreamLimiterMax(max int)

SetStreamLimiterMax sets the static capacity gauge (called once at startup).

func ValidateDNS1123Label added in v0.6.0 

func ValidateDNS1123Label(value, field string) error

ValidateDNS1123Label checks that a string is a valid DNS-1123 label.

func ValidateNamespaceMatch added in v0.6.0 

func ValidateNamespaceMatch(requestNS, resourceNS string) error

ValidateNamespaceMatch ensures that the namespace in the request wrapper matches the namespace in the resource, preventing RBAC bypass where a user authorized for namespace "foo" creates a resource in namespace "bar". Both requestNS and resourceNS must be non-empty for comparison. Callers must ensure requestNS is defaulted before calling this function.

Types

type AuditLogger added in v0.6.0 

type AuditLogger struct {
	// contains filtered or unexported fields
}

AuditLogger emits structured audit events for authentication, authorization, and resource mutations. All events are JSON-formatted via slog for compatibility with standard log aggregators.

func NewAuditLogger added in v0.6.0 

func NewAuditLogger(logger *slog.Logger) *AuditLogger

NewAuditLogger creates an audit logger wrapping the given structured logger.

func (*AuditLogger) LogAuth added in v0.6.0 

func (a *AuditLogger) LogAuth(ctx context.Context, event string, user *auth.UserInfo, r *http.Request, attrs ...slog.Attr)

LogAuth logs authentication events (success, failure, cache hit).

func (*AuditLogger) LogAuthz added in v0.6.0 

func (a *AuditLogger) LogAuthz(ctx context.Context, event string, user *auth.UserInfo, verb, resource, namespace string)

LogAuthz logs authorization events (denied, error).

func (*AuditLogger) LogMutation added in v0.6.0 

func (a *AuditLogger) LogMutation(ctx context.Context, event string, resourceType, name, namespace string)

LogMutation logs resource mutation events (create, update, delete).

type AuthService 

type AuthService struct {
	// contains filtered or unexported fields
}

func (*AuthService) GetCurrentUser 

func (s *AuthService) GetCurrentUser(ctx context.Context, req *connect.Request[pb.GetCurrentUserRequest]) (*connect.Response[pb.GetCurrentUserResponse], error)

func (*AuthService) ListPermissions 

func (s *AuthService) ListPermissions(ctx context.Context, req *connect.Request[pb.ListPermissionsRequest]) (*connect.Response[pb.ListPermissionsResponse], error)

type ClusterService 

type ClusterService struct {
	// contains filtered or unexported fields
}

func (*ClusterService) GetClusterInfo 

func (s *ClusterService) GetClusterInfo(ctx context.Context, req *connect.Request[pb.GetClusterInfoRequest]) (*connect.Response[pb.GetClusterInfoResponse], error)

type EnvironmentService 

type EnvironmentService struct {
	// contains filtered or unexported fields
}

func (*EnvironmentService) CreateEnvironment 

func (s *EnvironmentService) CreateEnvironment(ctx context.Context, req *connect.Request[pb.CreateEnvironmentRequest]) (*connect.Response[pb.CreateEnvironmentResponse], error)

func (*EnvironmentService) DeleteEnvironment 

func (s *EnvironmentService) DeleteEnvironment(ctx context.Context, req *connect.Request[pb.DeleteEnvironmentRequest]) (*connect.Response[pb.DeleteEnvironmentResponse], error)

func (*EnvironmentService) ExtendTTL 

func (s *EnvironmentService) ExtendTTL(ctx context.Context, req *connect.Request[pb.ExtendTTLRequest]) (*connect.Response[pb.ExtendTTLResponse], error)

func (*EnvironmentService) GetEnvironment 

func (s *EnvironmentService) GetEnvironment(ctx context.Context, req *connect.Request[pb.GetEnvironmentRequest]) (*connect.Response[pb.GetEnvironmentResponse], error)

func (*EnvironmentService) ListEnvironments 

func (s *EnvironmentService) ListEnvironments(ctx context.Context, req *connect.Request[pb.ListEnvironmentsRequest]) (*connect.Response[pb.ListEnvironmentsResponse], error)

func (*EnvironmentService) ListHookJobs added in v0.7.0 

func (s *EnvironmentService) ListHookJobs(ctx context.Context, req *connect.Request[pb.ListHookJobsRequest]) (*connect.Response[pb.ListHookJobsResponse], error)

ListHookJobs returns hook Jobs for an environment, sorted newest-first, with each Job's K8s status mapped to a proto phase (Pending/Running/Succeeded/Failed).

func (*EnvironmentService) RetryHook added in v0.7.0 

func (s *EnvironmentService) RetryHook(ctx context.Context, req *connect.Request[pb.RetryHookRequest]) (*connect.Response[pb.RetryHookResponse], error)

RetryHook deletes the newest failed Job matching the hook type, then annotates the Environment CR with a retry marker so the controller reconciles a new run. It does NOT accept arbitrary image/command — only re-triggers from CRD spec.

func (*EnvironmentService) StreamLogs 

func (s *EnvironmentService) StreamLogs(ctx context.Context, req *connect.Request[pb.StreamLogsRequest], stream *connect.ServerStream[pb.StreamLogsResponse]) error

func (*EnvironmentService) UpdateEnvironment 

func (s *EnvironmentService) UpdateEnvironment(ctx context.Context, req *connect.Request[pb.UpdateEnvironmentRequest]) (*connect.Response[pb.UpdateEnvironmentResponse], error)

func (*EnvironmentService) WatchEnvironments 

func (s *EnvironmentService) WatchEnvironments(ctx context.Context, req *connect.Request[pb.WatchEnvironmentsRequest], stream *connect.ServerStream[pb.WatchEnvironmentsResponse]) error

type OIDCHandler added in v0.7.0 

type OIDCHandler struct {
	// contains filtered or unexported fields
}

OIDCHandler implements the OIDC authorization code flow endpoints.

func NewOIDCHandler added in v0.7.0 

func NewOIDCHandler(cfg OIDCHandlerConfig) (*OIDCHandler, error)

NewOIDCHandler creates a new OIDC handler. It performs OIDC discovery on startup.

func (*OIDCHandler) HandleCallback added in v0.7.0 

func (h *OIDCHandler) HandleCallback(w http.ResponseWriter, r *http.Request)

HandleCallback processes the OIDC provider's authorization code callback.

func (*OIDCHandler) HandleConfig added in v0.7.0 

func (h *OIDCHandler) HandleConfig(w http.ResponseWriter, r *http.Request)

HandleConfig returns the public auth configuration for the frontend.

func (*OIDCHandler) HandleLogin added in v0.7.0 

func (h *OIDCHandler) HandleLogin(w http.ResponseWriter, r *http.Request)

HandleLogin initiates the OIDC authorization code flow.

func (*OIDCHandler) HandleLogout added in v0.7.0 

func (h *OIDCHandler) HandleLogout(w http.ResponseWriter, r *http.Request)

HandleLogout clears the session cookie and redirects to the login page.

func (*OIDCHandler) RegisterRoutes added in v0.7.0 

func (h *OIDCHandler) RegisterRoutes(mux *http.ServeMux)

RegisterRoutes registers the OIDC auth endpoints on the given mux.

type OIDCHandlerConfig added in v0.7.0 

type OIDCHandlerConfig struct {
	// IssuerURL is the OIDC provider's issuer URL.
	IssuerURL string
	// ClientID is the OIDC client ID.
	ClientID string
	// ClientSecret is the OIDC client secret.
	ClientSecret string
	// RedirectURL is the callback URL (e.g. https://diverge.example.com/auth/callback).
	RedirectURL string
	// Scopes are the OIDC scopes to request.
	Scopes []string
	// ProviderName is the display name shown on the login button (e.g. "Okta", "Google").
	ProviderName string
	// SessionManager mints and verifies session JWTs.
	SessionManager *auth.SessionManager
	// SessionMaxAge is the cookie max age. Defaults to 24 hours.
	SessionMaxAge time.Duration
	// SecureCookies sets the Secure flag on cookies. Should be true in production.
	SecureCookies bool
	// UsernameClaim is the JWT claim used for the username.
	UsernameClaim string
	// GroupsClaim is the JWT claim used for group membership.
	GroupsClaim string
	// Logger for request logging.
	Logger *slog.Logger
}

OIDCHandlerConfig configures the OIDC authentication HTTP handlers.

type PreviewGroupService 

type PreviewGroupService struct {
	// contains filtered or unexported fields
}

func (*PreviewGroupService) CreatePreviewGroup 

func (s *PreviewGroupService) CreatePreviewGroup(ctx context.Context, req *connect.Request[pb.CreatePreviewGroupRequest]) (*connect.Response[pb.CreatePreviewGroupResponse], error)

func (*PreviewGroupService) DeletePreviewGroup 

func (s *PreviewGroupService) DeletePreviewGroup(ctx context.Context, req *connect.Request[pb.DeletePreviewGroupRequest]) (*connect.Response[pb.DeletePreviewGroupResponse], error)

func (*PreviewGroupService) GetPreviewGroup 

func (s *PreviewGroupService) GetPreviewGroup(ctx context.Context, req *connect.Request[pb.GetPreviewGroupRequest]) (*connect.Response[pb.GetPreviewGroupResponse], error)

func (*PreviewGroupService) ListPreviewGroups 

func (s *PreviewGroupService) ListPreviewGroups(ctx context.Context, req *connect.Request[pb.ListPreviewGroupsRequest]) (*connect.Response[pb.ListPreviewGroupsResponse], error)

func (*PreviewGroupService) UpdatePreviewGroup added in v0.6.0 

func (s *PreviewGroupService) UpdatePreviewGroup(ctx context.Context, req *connect.Request[pb.UpdatePreviewGroupRequest]) (*connect.Response[pb.UpdatePreviewGroupResponse], error)

func (*PreviewGroupService) WatchPreviewGroups 

func (s *PreviewGroupService) WatchPreviewGroups(ctx context.Context, req *connect.Request[pb.WatchPreviewGroupsRequest], stream *connect.ServerStream[pb.WatchPreviewGroupsResponse]) error

type ServeMuxConfig added in v0.6.0 

type ServeMuxConfig struct {
	Client           client.Client
	K8sClient        kubernetes.Interface
	InformerMgr      *streaming.InformerManager
	LogStreamer      *streaming.LogStreamer
	StreamLimiter    *StreamLimiter
	Logger           *slog.Logger
	AuditLogger      *AuditLogger
	Version          string
	DashboardEnabled bool
}

ServeMuxConfig holds all dependencies for the ConnectRPC server mux.

type StreamLimiter added in v0.7.0 

type StreamLimiter struct {
	// contains filtered or unexported fields
}

StreamLimiter enforces per-user and global stream concurrency limits. It replaces the previous global channel semaphore to prevent a single user from exhausting all stream slots (noisy-neighbor DoS).

func NewStreamLimiter added in v0.7.0 

func NewStreamLimiter(maxGlobal, maxPerUser int, metrics ...StreamLimiterMetrics) *StreamLimiter

NewStreamLimiter creates a limiter with the given global and per-user caps. Panics if maxGlobal <= 0 or maxPerUser <= 0 or maxPerUser > maxGlobal.

func (*StreamLimiter) Acquire added in v0.7.0 

func (l *StreamLimiter) Acquire(ctx context.Context) (func(), error)

Acquire reserves a stream slot for the authenticated user in ctx. Returns a release function that MUST be called when the stream ends. The release function is always non-nil and safe to call multiple times.

Errors:

  • CodeUnauthenticated: missing or anonymous user identity
  • CodeResourceExhausted: per-user or global limit reached

func (*StreamLimiter) ActiveStreams added in v0.7.0 

func (l *StreamLimiter) ActiveStreams() int

ActiveStreams returns the current global stream count (for testing/metrics).

func (*StreamLimiter) ActiveStreamsForUser added in v0.7.0 

func (l *StreamLimiter) ActiveStreamsForUser(username string) int

ActiveStreamsForUser returns the active stream count for a user (for testing/metrics).

type StreamLimiterMetrics added in v0.7.0 

type StreamLimiterMetrics struct {
	IncActive func()
	DecActive func()
	Rejected  func(reason string) // "per_user" or "global"
}

StreamLimiterMetrics provides decoupled metric callbacks. Matches the BroadcasterMetrics pattern to avoid coupling to Prometheus.

func GetStreamLimiterMetrics added in v0.7.0 

func GetStreamLimiterMetrics() StreamLimiterMetrics

GetStreamLimiterMetrics returns callbacks wired to Prometheus collectors.

type TunnelGC added in v0.7.0 

type TunnelGC struct {
	// contains filtered or unexported fields
}

TunnelGC periodically garbage-collects expired tunnel K8s resources. Resources are annotated with diverge.dev/tunnel-expires (RFC3339). If the annotation is past, the resource is deleted.

func NewTunnelGC added in v0.7.0 

func NewTunnelGC(k8s kubernetes.Interface, logger *slog.Logger) *TunnelGC

func (*TunnelGC) Run added in v0.7.0 

func (gc *TunnelGC) Run(ctx context.Context, namespaces []string)

Run starts the GC loop. Blocks until ctx is cancelled.

type TunnelLease added in v0.7.0 

type TunnelLease struct {
	// contains filtered or unexported fields
}

TunnelLease provides distributed fencing for tunnel ownership using K8s Leases. When multiple server replicas exist, only one can own a tunnel at a time.

func NewTunnelLease added in v0.7.0 

func NewTunnelLease(k8s kubernetes.Interface, logger *slog.Logger, podName string) *TunnelLease

func (*TunnelLease) Acquire added in v0.7.0 

func (tl *TunnelLease) Acquire(ctx context.Context, namespace, previewID, tunnelID string) (previousHolder string, err error)

Acquire attempts to acquire or steal the lease for a tunnel. Returns the previous holder (empty if none) and any error.

func (*TunnelLease) Release added in v0.7.0 

func (tl *TunnelLease) Release(ctx context.Context, namespace, previewID, tunnelID string)

Release gives up the lease.

func (*TunnelLease) Renew added in v0.7.0 

func (tl *TunnelLease) Renew(ctx context.Context, namespace, previewID, tunnelID string) bool

Renew extends the lease. Returns false if we no longer hold it.

type TunnelManager added in v0.7.0 

type TunnelManager struct {
	// contains filtered or unexported fields
}

func NewServeMux 

func NewServeMux(cfg ServeMuxConfig) (*http.ServeMux, *TunnelManager)

NewServeMux creates the ConnectRPC service mux with all handlers registered. Auth is NOT applied here — it is applied at the net/http middleware layer.

func NewTunnelManager added in v0.7.0 

func NewTunnelManager(crdClient client.Client, k8s kubernetes.Interface, logger *slog.Logger, audit *AuditLogger) *TunnelManager

func (*TunnelManager) ForwardRequest added in v0.7.0 

func (tm *TunnelManager) ForwardRequest(ctx context.Context, previewID string, req *pb.TunnelHTTPRequest) (*pb.TunnelHTTPResponse, error)

ForwardRequest sends an HTTP request through the tunnel to the CLI. P0 #4: Supports chunked streaming — for large bodies, sends chunks via TunnelRequestChunk.

func (*TunnelManager) HasTunnel added in v0.7.0 

func (tm *TunnelManager) HasTunnel(previewID string) bool

func (*TunnelManager) NewTunnelProxyHandler added in v0.7.0 

func (tm *TunnelManager) NewTunnelProxyHandler() http.Handler

NewTunnelProxyHandler creates the HTTP handler for the tunnel proxy. P0 #3: This handler is served on a DEDICATED port (8081), NOT the RPC mux. P0 #6: No auth middleware — cluster-internal traffic only. Validates Host header to prevent open proxy abuse.

func (*TunnelManager) StartGC added in v0.7.0 

func (tm *TunnelManager) StartGC(ctx context.Context, namespaces []string)

StartGC starts the background garbage collector for tunnel resources.

func (*TunnelManager) Tunnel added in v0.7.0 

func (tm *TunnelManager) Tunnel(ctx context.Context, stream *connect.BidiStream[pb.TunnelServiceTunnelRequest, pb.TunnelServiceTunnelResponse]) error

Source Files

View all Source files

Directories

Path Synopsis
Package dashboard provides embedded static assets for the Diverge web dashboard.
 Package dashboard provides embedded static assets for the Diverge web dashboard.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.