package dispatchertls import ( "cmp" "context" "crypto/tls" "errors" "fmt" "log/slog" "net" "net/http" "strings" "sync" "time" "github.com/johnandersen777/socialweb-computer-kcp/api/v1alpha1" "github.com/johnandersen777/socialweb-computer-kcp/internal/bus" "github.com/johnandersen777/socialweb-computer-kcp/internal/component" "github.com/johnandersen777/socialweb-computer-kcp/internal/contract/vault" "github.com/johnandersen777/socialweb-computer-kcp/internal/dispatcher/tlsproxy" "github.com/johnandersen777/socialweb-computer-kcp/internal/kit/hostport" "github.com/johnandersen777/socialweb-computer-kcp/internal/kit/servetls" "github.com/johnandersen777/socialweb-computer-kcp/internal/kit/strs" "github.com/johnandersen777/socialweb-computer-kcp/internal/kit/tlsreload" commonv1 "github.com/johnandersen777/socialweb-computer-kcp/api/common/v1alpha1" ) const Kind = "DispatcherTLS" const shutdownTimeout = 5 * time.Second type Directory interface { Endpoint(workspace, kind, name, endpoint string) (component.Endpoint, bool) List(workspace string) []component.Entry } type Deps struct { component.Instance Registry Directory TLS *tls.Config } type Service struct { spec v1alpha1.DispatcherTLSSpec deps Deps workspace string hostnames []string certFile string caRef *commonv1.ObjectRef upstreamURL string mu sync.Mutex log *slog.Logger http *http.Server listener net.Listener conns *connSet stopCh chan struct{} stopOnce sync.Once readiness component.ReadySignal } func New(spec v1alpha1.DispatcherTLSSpec, deps Deps) *Service { s := &Service{ spec: spec, deps: deps, workspace: cmp.Or(strings.TrimSpace(deps.Self.Workspace), "root"), } hostnames := append([]string(nil), spec.Hostnames...) if spec.TLS != nil { hostnames = strs.AppendUniqueFold(hostnames, spec.TLS.Hostnames...) s.caRef = spec.TLS.CARef if spec.TLS.CertRef != "" || spec.TLS.Enabled { s.certFile, _ = servetls.CertificateFiles(spec.TLS.CertRef, spec.Storage, hostnames) } } s.hostnames = hostnames if target, ok := upstreamEndpoint(deps.Registry, s.workspace, spec.UpstreamRef); ok { s.upstreamURL = target } return s } var _ component.Service = (*Service)(nil) func (s *Service) listenAddress() string { if s.spec.Serve == nil { return hostport.Listen("", "", 0) } return hostport.Listen("", s.spec.Serve.Host, int(s.spec.Serve.Port)) } func upstreamEndpoint(registry Directory, workspace string, ref *commonv1.ObjectRef) (string, bool) { if ref == nil || registry == nil { return "", false } kind := strings.TrimSpace(ref.Kind) kind = cmp.Or(kind, "IngressProxy") name := strings.TrimSpace(ref.Name) if name == "" { return "", false } if workspace != "" { for _, wanted := range []string{"http", "control"} { if endpoint, ok := registry.Endpoint(workspace, kind, name, wanted); ok { return hostport.EndpointURL(endpoint) } } } for _, entry := range registry.List("") { if entry.Key.Kind != kind || entry.Key.Name != name { continue } for _, endpoint := range entry.Endpoints { if endpoint.URL != "" { return endpoint.URL, true } } for _, endpoint := range entry.Endpoints { if endpoint.Address != "" { return hostport.EndpointURL(endpoint) } } } return "", false } func (s *Service) Start(ctx context.Context) error { if s.upstreamURL == "" && s.spec.UpstreamRef == nil { return errors.New("dispatchertls: the upstream is required") } upstreamURL := s.upstreamURL if target, ok := s.resolveUpstream(upstreamURL); ok { upstreamURL = target } log := s.deps.Log if log == nil { log = slog.New(slog.DiscardHandler) } tlsConfig, err := s.listenerTLS() if err != nil { return err } dial, stopTrust, err := s.upstreamDial(ctx, log) if err != nil { return err } defer stopTrust() proxy, err := tlsproxy.New(tlsproxy.Config{Hostnames: s.hostnames}, tlsproxy.Deps{ Upstream: func() (string, bool) { return s.resolveUpstream(upstreamURL) }, DialTLS: dial, Log: log, }) if err != nil { return fmt.Errorf("dispatchertls: %w", err) } hosts := proxy.Hosts() listener, err := net.Listen("tcp", s.listenAddress()) if err != nil { return fmt.Errorf("dispatchertls: listening: %w", err) } tlsListener := tls.NewListener(listener, tlsConfig) conns := newConnSet() trackedListener := &trackingListener{Listener: tlsListener, set: conns} httpServer := &http.Server{ Handler: proxy, ReadHeaderTimeout: 10 * time.Second, } stopCh := make(chan struct{}) s.mu.Lock() if s.http != nil { s.mu.Unlock() _ = listener.Close() return errors.New("dispatchertls: the service already started") } s.log = log s.http = httpServer s.listener = listener s.conns = conns s.stopCh = stopCh s.stopOnce = sync.Once{} s.mu.Unlock() log.Info("dispatcher_tls_listening", "address", listener.Addr().String(), "upstream", upstreamURL, "hostnames", hosts) errCh := make(chan error, 1) go func() { errCh <- httpServer.Serve(trackedListener) }() s.readiness.Signal() select { case <-ctx.Done(): case <-stopCh: case serveErr := <-errCh: s.finish() if serveErr != nil && !errors.Is(serveErr, http.ErrServerClosed) { return fmt.Errorf("dispatchertls: serving: %w", serveErr) } return nil } s.shutdown(ctx) serveErr := <-errCh s.finish() if serveErr != nil && !errors.Is(serveErr, http.ErrServerClosed) { return fmt.Errorf("dispatchertls: serving: %w", serveErr) } return nil } func (s *Service) Ready() component.Readiness { return &s.readiness } func (s *Service) Endpoints() []component.Endpoint { s.mu.Lock() defer s.mu.Unlock() if s.listener == nil { return nil } address := s.listener.Addr().String() return []component.Endpoint{ {Name: "https", Protocol: component.ProtocolHTTPS, URL: "https://" + address, Address: address}, } } func (s *Service) Stop(ctx context.Context) error { s.mu.Lock() stopCh := s.stopCh s.mu.Unlock() if stopCh == nil { return nil } s.stopOnce.Do(func() { close(stopCh) }) s.shutdown(ctx) return nil } func (s *Service) Address() string { s.mu.Lock() defer s.mu.Unlock() if s.listener == nil { return "" } return s.listener.Addr().String() } func (s *Service) shutdown(ctx context.Context) { s.mu.Lock() httpServer := s.http conns := s.conns s.mu.Unlock() if httpServer == nil { return } ctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), shutdownTimeout) defer cancel() if err := httpServer.Shutdown(ctx); err != nil { _ = httpServer.Close() } if conns != nil { conns.closeAll() } } func (s *Service) finish() { s.mu.Lock() s.listener = nil s.http = nil s.mu.Unlock() } func (s *Service) resolveUpstream(upstreamURL string) (string, bool) { if raw := strings.TrimSpace(upstreamURL); raw != "" { return raw, true } return upstreamEndpoint(s.deps.Registry, s.workspace, s.spec.UpstreamRef) } func (s *Service) upstreamDial(ctx context.Context, log *slog.Logger) (func(context.Context, string, string) (net.Conn, error), func(), error) { if s.caRef == nil { return nil, func() {}, nil } if s.deps.Bus == nil { return nil, nil, errors.New("dispatchertls: the CA bundle needs the bus") } trust, err := bus.Resolve[vault.Trust](s.deps.Caller(ctx), s.deps.Bus, s.deps.Self, *s.caRef, bus.Capability{Name: vault.CapabilityRead, Scope: []string{vault.TrustScope}}) if err != nil { return nil, nil, fmt.Errorf("dispatchertls: the CA bundle of %s/%s: %w", s.caRef.Kind, s.caRef.Name, err) } pool, stop := tlsreload.TrustPool(ctx, trust, log) return pool.DialTLSContext(&tls.Config{MinVersion: tls.VersionTLS12}), stop, nil } func (s *Service) listenerTLS() (*tls.Config, error) { if s.certFile != "" { return servetls.Config(s.certFile, nil, s.deps.Log) } tlsConfig := s.deps.TLS if tlsConfig == nil { return nil, errors.New("dispatchertls: a TLS config or certificate is required") } tlsConfig = tlsConfig.Clone() if len(tlsConfig.Certificates) == 0 && tlsConfig.GetCertificate == nil && tlsConfig.GetConfigForClient == nil { return nil, errors.New("dispatchertls: the TLS config has no certificate") } if len(tlsConfig.NextProtos) == 0 { tlsConfig.NextProtos = []string{"http/1.1"} } return tlsConfig, nil }