agent

Documentation

Overview

Package agent runs the egress-side Skybridge data plane in either deployment mode:

• listener: the agent listens locally for native clients (psql/mysql/mongosh) and proxies to the upstream DB, masking result rows. Clients reach the agent directly.
• tunnel: the agent dials OUT to the relay gateway (egress-only), registers the targets it can reach, and serves the gateway's logical streams by running the same wire engines + masker against the upstream DB. Raw data never leaves the egress network.

Both modes share the engine selection and masking pipeline, so masking behaviour is identical no matter how the client arrives.

Index

• Constants
• func BuildMasker(cfg config.Agent) mask.Masker
• func BuildMaskerWithPathLabelSync(ctx context.Context, cfg config.Agent, logger *slog.Logger) (masker mask.Masker, detector *mask.Remote, store *remotestore.Store)
• func ContextWithWireClientIP(ctx context.Context, clientAddr string) context.Context
• func EngineFor(dbType string) (wire.Engine, error)
• func MaskingMode(cfg config.Agent) string
• func NewHTTPCredentialResolver(cfg config.Agent) wire.CredentialResolver
• func NewHTTPK8sCredentialResolver(cfg config.Agent) k8sapi.CredentialResolver
• func RunListener(ctx context.Context, cfg config.Agent, logger *slog.Logger) error
• func RunTunnel(ctx context.Context, cfg config.Agent, deps Deps, logger *slog.Logger) error
• func ServeTunnelConn(ctx context.Context, conn net.Conn, cfg config.Agent, deps Deps, ...) error
• type Deps

Constants

const DefaultOrgHeader = "X-Organization-Id"

DefaultOrgHeader is the request header carrying the agent's organization id when fetching the dynamic PII overlay, used unless SKYBRIDGE_PII_OVERLAY_ORG_HEADER overrides it.

Functions

func BuildMasker

func BuildMasker(cfg config.Agent) mask.Masker

BuildMasker assembles the masking chain (remote masker + your column overlay) from config. It does not start the dynamic overlay/path-label sync loops (no ctx/logger available here) — use BuildMaskerWithPathLabelSync when you have a lifecycle context to run those against.

func BuildMaskerWithPathLabelSync

func BuildMaskerWithPathLabelSync(ctx context.Context, cfg config.Agent, logger *slog.Logger) (masker mask.Masker, detector *mask.Remote, store *remotestore.Store)

BuildMaskerWithPathLabelSync assembles the masking chain and starts its dynamic sources (the flat column overlay poller and, when cfg.PathLabelURL is set, the pathlabel remotestore's pull/push loops) against ctx. Use this instead of BuildMasker wherever a lifecycle ctx is available — currently RunListener/RunTunnel (below) and cmd/skybridge/edge.go's shared studio/dbexec masker, since dbquery's detector hook (internal/edge/dbquery/mask.go) needs the store's push loop actually running for its Put calls to ever reach the control plane.

The returned detector/store are for wiring dbquery.Options.Detector/ProposeStore (the "propose" half of the pathlabel workflow) at the same call sites that already receive masker — they ride on the same Remote analyze call and the same synced Store, rather than a second detection pass. Either return value is nil when its respective prerequisite isn't configured (no MaskAnalyzeURL/MaskAnonymizeURL, or no PathLabelURL).

func ContextWithWireClientIP

func ContextWithWireClientIP(ctx context.Context, clientAddr string) context.Context

ContextWithWireClientIP annotates ctx with the native client's IP for credential exchange.

func EngineFor

func EngineFor(dbType string) (wire.Engine, error)

EngineFor selects a wire engine by database type (no client-TLS termination). The agent uses the TLS-aware engineFactory at runtime; this stays for callers/tests that want the plaintext default.

func MaskingMode

func MaskingMode(cfg config.Agent) string

MaskingMode returns a short label describing the active masking layers.

func NewHTTPCredentialResolver

func NewHTTPCredentialResolver(cfg config.Agent) wire.CredentialResolver

NewHTTPCredentialResolver builds a wire.CredentialResolver that exchanges the client-presented session token for an upstream credential via the control plane. Returns nil when injection is not configured (the caller then uses the verbatim Proxy path).

func NewHTTPK8sCredentialResolver

func NewHTTPK8sCredentialResolver(cfg config.Agent) k8sapi.CredentialResolver

NewHTTPK8sCredentialResolver builds a k8sapi.CredentialResolver that exchanges a client-presented session token for the real cluster bearer token via the control plane, mirroring NewHTTPCredentialResolver's shape (credsource.go) for the DB engines. Returns nil when the exchange URL is not configured — the caller then cannot serve Kubernetes targets at all (unlike DB engines there is no verbatim-passthrough fallback; see k8sapi's package doc).

func RunListener

func RunListener(ctx context.Context, cfg config.Agent, logger *slog.Logger) error

RunListener serves native clients directly (listener mode).

func RunTunnel

func RunTunnel(ctx context.Context, cfg config.Agent, deps Deps, logger *slog.Logger) error

RunTunnel dials the gateway and serves its streams (tunnel mode), reconnecting on failure.

func ServeTunnelConn

func ServeTunnelConn(ctx context.Context, conn net.Conn, cfg config.Agent, deps Deps, logger *slog.Logger) error

ServeTunnelConn registers over an established gateway connection and serves inbound streams. It is separated from RunTunnel so tests can drive it over an in-memory pipe.

Types

type Deps

type Deps struct {
	Dial        func(ctx context.Context, network, addr string) (net.Conn, error)
	Engine      func(dbType string) (wire.Engine, error)
	Masker      mask.Masker
	Resolver    wire.CredentialResolver // non-nil enables credential injection (handoff)
	UpstreamTLS *upstreamTLSPolicy      // non-nil enables agent→database TLS (Postgres)
}

Deps are injectable collaborators (overridable in tests); zero values fall back to real defaults.