Documentation
¶
Overview ¶
Package tunnel implements secure access tunneling for plexd mesh nodes.
Index ¶
- Constants
- Variables
- func GenerateHostKey() (ssh.Signer, error)
- func HostKeyFingerprint(signer ssh.Signer) string
- func HostKeyPath(dataDir string) string
- func LoadOrGenerateHostKey(dataDir string, logger *slog.Logger) (ssh.Signer, error)
- func TerminatedByFromReason(reason string) string
- type ClosedSessionInfo
- type Config
- type Dispatcher
- type Ed25519JWTVerifier
- type JWTVerifier
- type K8sProxy
- type MeshServer
- type SSHServer
- type SSHServerConfig
- type Session
- type SessionActivityReporter
- type SessionManager
- func (m *SessionManager) ActiveCount() int
- func (m *SessionManager) ActiveSessions() map[string]time.Time
- func (m *SessionManager) CloseSession(sessionID, reason string) *ClosedSessionInfo
- func (m *SessionManager) CreateSession(ctx context.Context, sess api.NodeStateSession) (string, error)
- func (m *SessionManager) SetOnClosed(fn func(sessionID, reason string, info *ClosedSessionInfo))
- func (m *SessionManager) Shutdown()
Constants ¶
const DefaultMaxSSHSessions = 10
DefaultMaxSSHSessions is the default maximum number of concurrent SSH sessions.
const DefaultMaxSessions = 10
DefaultMaxSessions is the default maximum number of concurrent tunnel sessions.
const DefaultSSHIdleTimeout = 30 * time.Minute
DefaultSSHIdleTimeout is the default idle timeout for SSH connections.
const DefaultTimeout = 30 * time.Minute
DefaultTimeout is the default session timeout.
Variables ¶
var ErrTunnelingDisabled = errors.New("tunnel: tunneling is disabled")
ErrTunnelingDisabled is returned by CreateSession when tunneling is switched off in the node's configuration. It is a sentinel because that refusal is permanent for the life of the process, and a caller matching it with errors.Is can settle the session instead of retrying it on every pull.
Functions ¶
func GenerateHostKey ¶
GenerateHostKey generates a new Ed25519 keypair and returns it as an ssh.Signer.
func HostKeyFingerprint ¶ added in v0.4.0
HostKeyFingerprint renders the host key's public half as the canonical OpenSSH SHA-256 fingerprint, `SHA256:<base64>` with the padding omitted — the form `ssh-keygen -l` prints and the form the capability manifest's optional ssh_host_key_fingerprint field carries.
func HostKeyPath ¶ added in v0.4.0
HostKeyPath returns the path LoadOrGenerateHostKey reads and writes the node's SSH host key at. The integrity verifier watches the same file, so the name has one owner rather than two spellings that can drift apart.
func LoadOrGenerateHostKey ¶
LoadOrGenerateHostKey loads an existing Ed25519 host key from dataDir, or generates and persists a new one if none exists.
func TerminatedByFromReason ¶ added in v0.2.0
TerminatedByFromReason maps an internal session close reason to the wire terminated_by enum reported on a session_ended row: reasonExpired becomes ttl_expired, reasonIdle becomes idle_timeout, and every other reason becomes plexd_close.
api.TerminatedByOperatorRevoke is never produced here. It is a factual claim about a human action, and the node cannot make it: a revocation reaches the node as the absence of the entry, which is indistinguishable from a control plane that failed to serve the block. Asserting it would write a confidently wrong answer into the audit trail on every degraded pull.
Types ¶
type ClosedSessionInfo ¶
type ClosedSessionInfo struct {
Duration time.Duration
TargetHost string
TargetPort int
BytesIn int64
BytesOut int64
}
ClosedSessionInfo contains metadata about a session that was closed.
type Config ¶
type Config struct {
// Enabled controls whether tunneling is active.
// Default: true (set by ApplyDefaults).
Enabled bool `yaml:"enabled"`
// MaxSessions is the maximum number of concurrent tunnel sessions.
// Default: 10
MaxSessions int `yaml:"max_sessions"`
// DefaultTimeout is the default/maximum session timeout.
// Default: 30m
DefaultTimeout time.Duration `yaml:"default_timeout"`
// SSHListenAddr is the address for the SSH mesh server to listen on.
// If empty, the SSH server is not started.
SSHListenAddr string `yaml:"ssh_listen_addr"`
// HostKeyDir is the directory for storing the SSH host key.
// If empty, a transient key is generated (not persisted).
HostKeyDir string `yaml:"host_key_dir"`
}
Config holds the configuration for secure access tunneling.
func (*Config) ApplyDefaults ¶
func (c *Config) ApplyDefaults()
ApplyDefaults sets default values for zero-valued fields. On a zero-valued Config, Enabled defaults to true. To disable tunneling, set Enabled=false before or after calling ApplyDefaults.
type Dispatcher ¶ added in v0.3.0
type Dispatcher struct {
// contains filtered or unexported fields
}
Dispatcher consumes the sessions block of the reconciliation pull and holds the node's mediated-access listeners level with it. The block is desired state, not a delivery queue like the executions block: an entry stands for as long as the session is valid, and its disappearance is the teardown signal. Revocation and hard expiry both reach the node as that same absence — there is no revocation callback to answer and no terminal status to report.
Provisioning is tcp-kind only. An ssh or k8s entry is decoded and settled as unsupported: one warning, no listener, no activity row.
A Dispatcher is not safe for concurrent use. Handle is invoked only from the reconcile goroutine, one cycle at a time, so known and unreported need no mutex.
The pass carries no dispatch budget, unlike the executions block's: it is bounded by the live sessions the manager caps at MaxSessions plus the length of the block, and its only network call is a single bounded activity post per session whose started row is still outstanding.
func NewDispatcher ¶ added in v0.3.0
func NewDispatcher(manager *SessionManager, reporter SessionActivityReporter, logger *slog.Logger) *Dispatcher
NewDispatcher creates a Dispatcher that provisions the pull's sessions block through manager and reports each started listener through reporter.
func (*Dispatcher) Handle ¶ added in v0.3.0
func (d *Dispatcher) Handle(ctx context.Context, desired *api.NodeStateSnapshot)
Handle reconciles the snapshot's sessions block: it first tears down every live session the block no longer carries, then provisions the entries it does not yet serve, in block order, and finally reports the listeners it brought up. Its signature matches reconcile.DispatchHandler.
type Ed25519JWTVerifier ¶
type Ed25519JWTVerifier struct {
// contains filtered or unexported fields
}
Ed25519JWTVerifier verifies compact JWS tokens (alg=EdDSA) using an Ed25519 public key. It validates the signature and checks the "exp" claim without relying on any external JWT library.
func NewEd25519JWTVerifier ¶
func NewEd25519JWTVerifier(publicKey ed25519.PublicKey) *Ed25519JWTVerifier
NewEd25519JWTVerifier creates a new verifier with the given Ed25519 public key.
func (*Ed25519JWTVerifier) Verify ¶
func (v *Ed25519JWTVerifier) Verify(token string) error
Verify validates a compact JWS token (header.payload.signature). It checks the Ed25519 signature over the signing input and verifies the "exp" claim has not elapsed.
type JWTVerifier ¶
JWTVerifier validates JWT tokens for SSH authentication.
type K8sProxy ¶
type K8sProxy struct {
// contains filtered or unexported fields
}
K8sProxy is a reverse proxy that forwards HTTP requests to a Kubernetes API server.
func NewK8sProxy ¶
NewK8sProxy creates a K8sProxy targeting the given API server URL. If tlsConfig is non-nil, it is used for TLS connections to the API server.
type MeshServer ¶
type MeshServer struct {
// contains filtered or unexported fields
}
MeshServer composes the SSH mesh server with the existing SessionManager, managing their lifecycle together.
func NewMeshServer ¶
func NewMeshServer(cfg Config, meshIP string, hostKey ssh.Signer, verifier JWTVerifier, logger *slog.Logger) *MeshServer
NewMeshServer creates a MeshServer that manages the SSH server and session manager. meshIP is the node's registered mesh address, which every session listener binds — it is the reachability boundary of an otherwise unauthenticated forward, so it comes from the node identity rather than being guessed from the SSH listen address, which is empty in both documented configurations and would bind every interface on the host. If cfg.SSHListenAddr is empty, the SSH server is not created.
func (*MeshServer) SSHServer ¶
func (m *MeshServer) SSHServer() *SSHServer
SSHServer returns the underlying SSH server, or nil if not configured.
func (*MeshServer) SessionManager ¶
func (m *MeshServer) SessionManager() *SessionManager
SessionManager returns the underlying session manager.
func (*MeshServer) Shutdown ¶
func (m *MeshServer) Shutdown() error
Shutdown gracefully stops the mesh server. Ordering: close SSH listener first, then drain session manager.
type SSHServer ¶
type SSHServer struct {
// contains filtered or unexported fields
}
SSHServer is a mesh-facing SSH server that authenticates clients via JWT and provides direct-tcpip channel forwarding.
func NewSSHServer ¶
func NewSSHServer(cfg SSHServerConfig, hostKey ssh.Signer, verifier JWTVerifier, logger *slog.Logger) *SSHServer
NewSSHServer creates a new SSHServer with the given configuration.
type SSHServerConfig ¶
type SSHServerConfig struct {
// MaxSessions is the maximum number of concurrent SSH sessions.
// Default: 10
MaxSessions int
// IdleTimeout is the idle timeout for SSH connections.
// Default: 30m
IdleTimeout time.Duration
// ListenAddr is the address to listen on (mesh IP + port).
// Required.
ListenAddr string
}
SSHServerConfig holds the configuration for the SSH mesh server.
func (*SSHServerConfig) ApplyDefaults ¶
func (c *SSHServerConfig) ApplyDefaults()
ApplyDefaults sets default values for zero-valued fields.
func (*SSHServerConfig) Validate ¶
func (c *SSHServerConfig) Validate() error
Validate checks that configuration values are within acceptable ranges.
type Session ¶
type Session struct {
SessionID string
TargetHost string
TargetPort int
MeshIP string
// contains filtered or unexported fields
}
Session represents an active tunnel session with a local TCP listener that forwards connections to a target host through the mesh.
func NewSession ¶
func NewSession(sessionID, targetHost string, targetPort int, meshIP string, expiresAt time.Time, logger *slog.Logger) *Session
NewSession creates a Session with the given parameters.
func (*Session) Counters ¶ added in v0.2.0
Counters returns the bytes forwarded in each direction: in is client -> target (operator to target), out is target -> client (target to operator).
func (*Session) IdleFor ¶ added in v0.3.0
IdleFor returns how long the session has gone without observed byte flow. The listener bind counts as the first activity, so a started session never reports the whole time since the process began.
func (*Session) ListenAddr ¶
ListenAddr returns the listener address or empty string if not started.
func (*Session) Start ¶
Start opens a TCP listener bound to the mesh IP and begins accepting connections. Everything the session runs — the accept loop and, when an idle window is armed, the idle monitor — hangs off the child context Start derives here, which Close cancels; nothing outlives the session.
type SessionActivityReporter ¶ added in v0.2.0
type SessionActivityReporter interface {
ReportSessionStarted(ctx context.Context, sessionID, targetHost string, targetPort int, listenerEndpoint string) error
}
SessionActivityReporter reports the tcp-phase session_started row to the control plane. The session dispatcher posts it once the listener is up, so the row carries the bound listener address alongside the target. That address is the operator's only route to the listener, so the error is returned rather than swallowed: a dropped row leaves a listener nobody can reach. The matching session_ended row is emitted from the SessionManager's on-closed callback, not through here.
type SessionManager ¶
type SessionManager struct {
// contains filtered or unexported fields
}
SessionManager manages the lifecycle of tunnel sessions.
func NewSessionManager ¶
func NewSessionManager(cfg Config, meshIP string, logger *slog.Logger) *SessionManager
NewSessionManager creates a new SessionManager with default config applied.
func (*SessionManager) ActiveCount ¶
func (m *SessionManager) ActiveCount() int
ActiveCount returns the number of active sessions.
func (*SessionManager) ActiveSessions ¶ added in v0.3.0
func (m *SessionManager) ActiveSessions() map[string]time.Time
ActiveSessions returns the live sessions as session id to capped local expiry, in a fresh map taken under the lock. The session dispatcher's teardown pass consumes it: an id the pull's sessions block no longer carries is closed, and the expiry is what tells a revocation apart from a hard expiry — both reach the node as the same absence.
func (*SessionManager) CloseSession ¶
func (m *SessionManager) CloseSession(sessionID, reason string) *ClosedSessionInfo
CloseSession closes and removes a session by ID. Returns session metadata if the session existed, or nil if not found.
func (*SessionManager) CreateSession ¶
func (m *SessionManager) CreateSession(ctx context.Context, sess api.NodeStateSession) (string, error)
CreateSession creates and starts a tunnel session for one entry of the pull's sessions block and returns the bound listener address. Only tcp-kind entries are provisionable: the session dispatcher filters the block before calling in, and the kind guard here keeps the manager safe to call on its own.
func (*SessionManager) SetOnClosed ¶ added in v0.2.0
func (m *SessionManager) SetOnClosed(fn func(sessionID, reason string, info *ClosedSessionInfo))
SetOnClosed registers a callback invoked after a session is successfully closed and removed, for every close reason including "shutdown". The callback carries the close reason and the session's final metadata, and is the single path by which a session_ended activity row — TTL expiry, operator revoke, and node shutdown alike — reaches the control plane. Because it is the only carrier of the session's byte counters, skipping it on shutdown would leave the control plane's audit record for every live session without bytes_in, bytes_out, or terminated_by.
func (*SessionManager) Shutdown ¶
func (m *SessionManager) Shutdown()
Shutdown closes all active sessions, reporting each one through the on-closed callback with reason "shutdown" so its byte counters and a plexd_close terminated_by reach the control plane before the node goes offline.
The on-closed callback performs a blocking, bounded report, so the sessions are closed concurrently: total shutdown latency is then the single slowest report rather than their sum. Closing serially instead would let a slow or unreachable control plane stretch teardown to MaxSessions times the per-report bound, overrunning a typical orchestrator termination grace period.
Source Files