304 lines
9.4 KiB
Go
304 lines
9.4 KiB
Go
package gateway
|
|
|
|
import (
|
|
"context"
|
|
"crypto/tls"
|
|
"errors"
|
|
"linkport/internal/cluster"
|
|
"linkport/internal/model"
|
|
"linkport/internal/outbound"
|
|
"linkport/internal/security"
|
|
"linkport/internal/store"
|
|
"net"
|
|
"net/netip"
|
|
"sync"
|
|
"time"
|
|
)
|
|
|
|
type Status struct {
|
|
ManagementBoundAddress string `json:"managementBoundAddress,omitempty"`
|
|
PersistentTargets []PersistentTargetStatus `json:"persistentTargets"`
|
|
BoundAddress string `json:"boundAddress,omitempty"`
|
|
ClientDestinations []string `json:"clientDestinations"`
|
|
ClientDestinationScope string `json:"clientDestinationScope,omitempty"`
|
|
Running bool `json:"running"`
|
|
Connections int `json:"connections"`
|
|
Rejected int64 `json:"rejected"`
|
|
Error string `json:"error,omitempty"`
|
|
}
|
|
type instance struct {
|
|
managementListener net.Listener
|
|
persistent *persistentTargets
|
|
config model.Proxy // Immutable snapshot replaced under Manager.mu.
|
|
ctx context.Context
|
|
listener net.Listener
|
|
cancel context.CancelFunc
|
|
connections map[net.Conn]struct{}
|
|
rejected int64
|
|
stopping bool
|
|
wg sync.WaitGroup
|
|
}
|
|
|
|
type preparedStart struct {
|
|
managementListener net.Listener
|
|
config model.Proxy
|
|
listener net.Listener
|
|
handler model.Handler
|
|
}
|
|
|
|
type Manager struct {
|
|
slowRequests []SlowRequest
|
|
mu sync.Mutex
|
|
instances map[string]*instance
|
|
snapshots map[string]model.Proxy
|
|
failures map[string]string
|
|
handlers map[string]model.Handler
|
|
tls *tls.Config
|
|
store *store.Store
|
|
metrics map[string]*Metrics
|
|
health map[string]HealthResult
|
|
healthBusy map[string]bool
|
|
healthMu sync.Mutex
|
|
healthCancel context.CancelFunc
|
|
healthDone chan struct{}
|
|
healthInterval int
|
|
connections map[string]*connectionState
|
|
shutdownHooks []func()
|
|
}
|
|
|
|
func New(s *store.Store, t *tls.Config, h map[string]model.Handler) *Manager {
|
|
return &Manager{instances: map[string]*instance{}, snapshots: map[string]model.Proxy{}, failures: map[string]string{}, handlers: h, tls: t, store: s, metrics: map[string]*Metrics{}, health: map[string]HealthResult{}, healthBusy: map[string]bool{}, connections: map[string]*connectionState{}}
|
|
}
|
|
func (m *Manager) Start(p model.Proxy) error {
|
|
m.mu.Lock()
|
|
defer m.mu.Unlock()
|
|
prepared, err := m.prepareStartLocked(p)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
m.publishStartLocked(prepared)
|
|
return nil
|
|
}
|
|
|
|
func (m *Manager) prepareStartLocked(p model.Proxy) (preparedStart, error) {
|
|
if !p.HasDestination() {
|
|
return preparedStart{}, errors.New("select a target before starting the gateway")
|
|
}
|
|
if err := p.ValidateTargetSelection(); err != nil {
|
|
return preparedStart{}, err
|
|
}
|
|
if _, ok := m.instances[p.ID]; ok {
|
|
return preparedStart{}, errors.New("proxy already running or stopping")
|
|
}
|
|
h := m.handlers[p.Type]
|
|
if h == nil {
|
|
return preparedStart{}, errors.New("unsupported protocol")
|
|
}
|
|
if err := p.ValidateGatewayHostKeys(); err != nil {
|
|
return preparedStart{}, err
|
|
}
|
|
address, err := listenAddress(p.ListenHost, p.ListenPort)
|
|
if err != nil {
|
|
m.failures[p.ID] = err.Error()
|
|
return preparedStart{}, err
|
|
}
|
|
l, err := net.Listen("tcp", address)
|
|
if err != nil {
|
|
m.failures[p.ID] = listenStartError(address, err).Error()
|
|
metric := m.metric(p.ID)
|
|
metric.RecentError, _ = listenFailure(err)
|
|
metric.RecentErrorAt = time.Now().UTC()
|
|
return preparedStart{}, errors.New(m.failures[p.ID])
|
|
}
|
|
var management net.Listener
|
|
if p.Type == "rabbitmq" && p.ManagementListenPort != 0 {
|
|
if m.handlers["http"] == nil {
|
|
l.Close()
|
|
return preparedStart{}, errors.New("HTTP management handler unavailable")
|
|
}
|
|
managementAddress, e := listenAddress(p.ListenHost, p.ManagementListenPort)
|
|
if e == nil {
|
|
management, e = net.Listen("tcp", managementAddress)
|
|
}
|
|
if e != nil {
|
|
l.Close()
|
|
m.failures[p.ID] = listenStartError(managementAddress, e).Error()
|
|
return preparedStart{}, errors.New(m.failures[p.ID])
|
|
}
|
|
}
|
|
return preparedStart{config: p, listener: l, managementListener: management, handler: h}, nil
|
|
}
|
|
|
|
func (m *Manager) publishStartLocked(prepared preparedStart) {
|
|
p, l, h := prepared.config, prepared.listener, prepared.handler
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
inst := &instance{config: p, ctx: ctx, listener: l, cancel: cancel, connections: map[net.Conn]struct{}{}}
|
|
inst.managementListener = prepared.managementListener
|
|
inst.persistent = newPersistentTargets(ctx, m.handlers)
|
|
inst.persistent.Reconcile(p)
|
|
m.instances[p.ID] = inst
|
|
m.snapshots[p.ID] = p
|
|
delete(m.failures, p.ID)
|
|
inst.wg.Add(1)
|
|
go m.accept(ctx, inst, l, h)
|
|
if inst.managementListener != nil {
|
|
inst.wg.Add(1)
|
|
go m.accept(ctx, inst, inst.managementListener, m.handlers["http"])
|
|
}
|
|
}
|
|
func (m *Manager) accept(ctx context.Context, i *instance, listener net.Listener, h model.Handler) {
|
|
defer i.wg.Done()
|
|
for {
|
|
c, err := listener.Accept()
|
|
if err != nil {
|
|
return
|
|
}
|
|
source, _, _ := net.SplitHostPort(c.RemoteAddr().String())
|
|
var connectionCtx context.Context
|
|
var connection *connectionState
|
|
m.mu.Lock()
|
|
p := i.config
|
|
p.TargetDialer = i.persistent.Dialer
|
|
p.PreparedTarget = i.persistent.Borrow
|
|
if listener == i.managementListener {
|
|
// Management HTTP needs its own port, including for SSH/Kubernetes tunnels.
|
|
p.TargetDialer = outbound.Dialer
|
|
p.PreparedTarget = nil
|
|
}
|
|
metric := m.metric(p.ID)
|
|
metric.ConnectionAttempts++
|
|
allowed := !i.stopping && Allowed(p.AllowCIDRs, source) && len(i.connections) < p.MaxConnections
|
|
if allowed {
|
|
metric.TotalConnections++
|
|
i.connections[c] = struct{}{}
|
|
i.wg.Add(1)
|
|
var connectionCancel context.CancelFunc
|
|
connectionCtx, connectionCancel = context.WithCancel(ctx)
|
|
connection = &connectionState{ConnectionInfo: ConnectionInfo{ID: security.Token(), GatewayID: p.ID, Protocol: p.Type, Source: source, StartedAt: time.Now().UTC(), TargetID: p.TargetID, TargetHost: p.TargetHost, TargetPort: p.TargetPort, TargetAccountID: p.TargetAccountID}, conn: c, cancel: connectionCancel}
|
|
m.connections[connection.ID] = connection
|
|
} else {
|
|
metric.ConnectionRejected++
|
|
}
|
|
m.mu.Unlock()
|
|
if !allowed {
|
|
c.Close()
|
|
m.store.Event(p.ID, source, "connection_rejected", "source_or_limit")
|
|
continue
|
|
}
|
|
session := &connectionSession{
|
|
manager: m, instance: i, config: p, connection: connection,
|
|
metric: metric, ctx: connectionCtx, gatewayCtx: ctx, auditProxy: p,
|
|
}
|
|
go session.serve(h)
|
|
}
|
|
}
|
|
func (m *Manager) Stop(id string) {
|
|
m.mu.Lock()
|
|
i := m.instances[id]
|
|
if i == nil {
|
|
delete(m.failures, id)
|
|
m.mu.Unlock()
|
|
return
|
|
}
|
|
i.stopLocked()
|
|
m.mu.Unlock()
|
|
i.waitStopped()
|
|
m.mu.Lock()
|
|
if m.instances[id] == i {
|
|
m.snapshots[id] = i.config
|
|
delete(m.instances, id)
|
|
}
|
|
delete(m.failures, id)
|
|
m.mu.Unlock()
|
|
}
|
|
func (m *Manager) Shutdown() {
|
|
m.shutdown(30 * time.Second)
|
|
}
|
|
func (m *Manager) shutdown(grace time.Duration) {
|
|
m.mu.Lock()
|
|
hooks := m.shutdownHooks
|
|
m.shutdownHooks = nil
|
|
m.mu.Unlock()
|
|
for _, hook := range hooks {
|
|
hook()
|
|
}
|
|
m.setHealthInterval(0, false)
|
|
m.mu.Lock()
|
|
ids := make([]string, 0, len(m.instances))
|
|
instances := make([]*instance, 0, len(m.instances))
|
|
for id, inst := range m.instances {
|
|
ids = append(ids, id)
|
|
instances = append(instances, inst)
|
|
inst.stopping = true
|
|
inst.listener.Close()
|
|
if inst.managementListener != nil {
|
|
inst.managementListener.Close()
|
|
}
|
|
}
|
|
m.mu.Unlock()
|
|
var wg sync.WaitGroup
|
|
for _, inst := range instances {
|
|
wg.Add(1)
|
|
go func(inst *instance) { defer wg.Done(); inst.wg.Wait() }(inst)
|
|
}
|
|
done := make(chan struct{})
|
|
go func() { wg.Wait(); close(done) }()
|
|
select {
|
|
case <-done:
|
|
case <-time.After(grace):
|
|
}
|
|
for _, id := range ids {
|
|
m.Stop(id)
|
|
}
|
|
}
|
|
func (m *Manager) Status(id string) Status {
|
|
m.mu.Lock()
|
|
defer m.mu.Unlock()
|
|
if i := m.instances[id]; i != nil {
|
|
bound := i.listener.Addr().String()
|
|
management := ""
|
|
if i.managementListener != nil {
|
|
management = i.managementListener.Addr().String()
|
|
}
|
|
return Status{ManagementBoundAddress: management, Running: true, Connections: len(i.connections), Rejected: i.rejected, BoundAddress: bound, ClientDestinations: clientDestinations(bound), ClientDestinationScope: clientDestinationScope(bound), PersistentTargets: i.persistent.Status()}
|
|
}
|
|
return Status{Error: m.failures[id], ClientDestinations: []string{}}
|
|
}
|
|
func (m *Manager) Test(ctx context.Context, p model.Proxy) (model.TestResult, error) {
|
|
// An explicit connection test must observe a fresh setup, never consume or
|
|
// create background resources belonging to a running gateway.
|
|
p.PreparedTarget, p.TargetDialer = nil, nil
|
|
h := m.handlers[p.Type]
|
|
if h == nil {
|
|
return model.TestResult{}, errors.New("unsupported protocol")
|
|
}
|
|
dial := outbound.Dialer(p)
|
|
resolved, info, err := cluster.Resolve(ctx, p, dial)
|
|
if err != nil {
|
|
return model.TestResult{Cluster: info}, err
|
|
}
|
|
result, err := h.Test(ctx, resolved, dial)
|
|
result.Cluster = info
|
|
return result, err
|
|
}
|
|
func Allowed(cidrs []string, source string) bool {
|
|
ip, err := netip.ParseAddr(source)
|
|
if err != nil {
|
|
return false
|
|
}
|
|
ip = ip.Unmap()
|
|
if len(cidrs) == 0 {
|
|
return true
|
|
}
|
|
for _, s := range cidrs {
|
|
if prefix, err := netip.ParsePrefix(s); err == nil {
|
|
if prefix.Contains(ip) {
|
|
return true
|
|
}
|
|
} else if addr, err := netip.ParseAddr(s); err == nil && addr.Unmap() == ip {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|