shared

package
v0.24.0 Latest Latest
Warning

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

 Go to latest Published: Aug 14, 2026 License: AGPL-3.0 Imports: 22 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	// MaxBatchRows is the largest number of rows sent in one INSERT.
	MaxBatchRows = 1000
	// MaxBatchBytes is the largest row payload sent in one INSERT (8 MiB).
	MaxBatchBytes = 8 << 20
	// QueueCapacityRows is the depth of the submit queue, in rows.
	QueueCapacityRows = 4096
	// MaxQueuedBytes bounds the row payload waiting in the queue (32 MiB).
	// A row cap alone does not bound memory — one wide row can be megabytes —
	// so the queue enforces both.
	MaxQueuedBytes = 32 << 20
)

Batch and queue sizing for the row writer.

The batch caps are deliberately large. Batching pays by amortizing the INSERT round-trip, and a bulk insert of 1000 small JSONB rows costs barely more than one of 50 — while at ~1 ms per insert a 50-row cap would ceiling throughput near 50k rows/s, over a minute of pure insert time on a multi-million-row capture.

The byte budget matters as much as the row count: 1000 rows is nothing for a single-column select and tens of megabytes for wide rows, so whichever cap trips first ends the batch.

View Source 
const (
	// ConnectionOpened marks a session that finished authenticating.
	ConnectionOpened = "opened"
	// ConnectionClosed marks a session that ended.
	ConnectionClosed = "closed"
)

Connection lifecycle states published on the connections topic.

View Source 
const (
	// ClientKeepAliveIdle is how long a connection may sit idle before the
	// first probe.
	ClientKeepAliveIdle = 30 * time.Second
	// ClientKeepAliveInterval is the gap between probes.
	ClientKeepAliveInterval = 10 * time.Second
	// ClientKeepAliveCount is how many unanswered probes declare the peer dead.
	ClientKeepAliveCount = 3
)

Client TCP keepalive settings. With no approval timeout, a hold ends only on approve, deny, or client disconnect — so "the client went away" has to be something the kernel actually tells us. A hard-killed client or a dead network path never sends a FIN, and without keepalive "until disconnect" silently means "forever", parking an upstream connection and any locks it holds indefinitely.

View Source 
const APIKeyUsageTimeout = 30 * time.Second

APIKeyUsageTimeout bounds one usage bump. It is generous because the write is a single-row UPDATE that nothing waits on: the number only has to be small enough that a wedged database cannot accumulate one parked goroutine per authenticated request for the life of the process.

View Source 
const DefaultLimitPollInterval = 250 * time.Millisecond

DefaultLimitPollInterval is how often the watchdog re-evaluates limits when no explicit interval is given. Small enough to cut a runaway stream promptly, large enough that the poll cost (two atomic loads + a time compare) is negligible.

View Source 
const GoroutineNameAPIKeyUsage = "api key usage bump"

GoroutineNameAPIKeyUsage is what a panic in the usage bump is logged under. It is one name for all five protocols and the REST API on purpose: the write is literally the same statement everywhere, so a second name would only make the log harder to grep.

View Source 
const GoroutineNameParkWatcher = "approval hold client read-watch"

GoroutineNameParkWatcher is what a panic in the park read-watch is logged under.

View Source 
const GoroutineNameRowWriterDrain = "captured row writer drain"

GoroutineNameRowWriterDrain is what a panic in the drain loop is logged under.

Variables

View Source 
var (
	// ErrApprovalAbandoned means the hold ended without a human decision —
	// the client disconnected, the query was canceled, the grant expired, or
	// the server drained. Nothing was ever forwarded upstream. It is rendered
	// distinctly from "denied" everywhere: to whoever finally looks at it, a
	// query nobody is waiting for anymore is a different thing from a query a
	// human rejected.
	ErrApprovalAbandoned = errors.New("approval hold abandoned: client is gone")
	// ErrApprovalUnavailable means the hold could not even be registered
	// (the pending row failed to persist). It fails **closed**: a statement
	// matching an approval pattern is never forwarded just because the
	// bookkeeping broke.
	ErrApprovalUnavailable = errors.New("approval required but the approval system is unavailable")
)

Approval hold errors. They travel back through each protocol's existing blocked-query path (the same one ErrDDLBlocked uses), so a denied statement reaches the client as a protocol-native error rather than a dropped socket.

View Source 
var (
	// ErrByteQuotaExceeded indicates the grant's max_bytes_transferred quota
	// was crossed while data was flowing.
	ErrByteQuotaExceeded = errors.New("bandwidth quota exceeded for this grant")
	// ErrGrantExpired indicates the grant's expiry time passed while the
	// session was still open.
	ErrGrantExpired = errors.New("grant expired")
	// ErrGrantRevoked indicates the grant backing the session was revoked
	// (by an admin, via the API) while the connection was still live.
	ErrGrantRevoked = errors.New("grant revoked")
)

Limit-enforcement errors shared across proxy implementations. They are surfaced both at command boundaries (a new query rejected because the grant is exhausted/expired) and mid-stream (a running query aborted the moment a limit is crossed).

View Source 
var (
	ErrReadOnlyViolation     = errors.New("write operations not permitted with read-only access")
	ErrDDLBlocked            = errors.New("DDL operations not permitted: your access grant blocks schema modifications")
	ErrPasswordChangeBlocked = errors.New("password modification is not allowed through the proxy")
	ErrOraclePatternBlocked  = errors.New("blocked: this Oracle operation is not permitted through the proxy")
	ErrMySQLPatternBlocked   = errors.New("blocked: this MySQL operation is not permitted through the proxy")
)

Validation errors shared across proxy implementations.

View Source 
var (
	ErrMongoReadOnly        = errors.New("dbbat: grant is read-only")
	ErrMongoDDLBlocked      = errors.New("dbbat: grant blocks DDL operations")
	ErrMongoCommandBlocked  = errors.New("dbbat: command not permitted through dbbat")
	ErrMongoUnknownCommand  = errors.New("dbbat: command not on the proxy allowlist")
	ErrMongoDatabaseBlocked = errors.New("dbbat: access to this database is not permitted")
	// ErrMongoPipelineNotCheckable — the aggregation pipeline could not be fully
	// inspected (too deeply nested, or a stage dbbat cannot parse), so the
	// databases it names cannot be established. Refused rather than forwarded:
	// an unreadable pipeline is not a pipeline that writes nowhere.
	ErrMongoPipelineNotCheckable = errors.New("dbbat: aggregation pipeline is too deeply nested " +
		"or malformed to be checked")
)

Mongo-specific validation errors (contract §7 surfaces these as the errmsg of an Unauthorized (13) reply).

View Source 
var ErrApprovalDenied = errors.New("query denied by approver")

ErrApprovalDenied is the sentinel every ApprovalDeniedError matches, so callers can branch with errors.Is without unwrapping.

View Source 
var ErrBastionNotSSH = errors.New("ssh: via_uid does not reference an ssh server")

ErrBastionNotSSH is returned when a via_uid resolves to a non-ssh row.

View Source 
var ErrConnParked = errors.New("connection is parked for an approval hold")

ErrConnParked is returned if something tries to read the connection while a hold owns it. It indicates a wiring bug, not a runtime condition.

View Source 
var ErrKubernetesViaUnsupported = errors.New("kubernetes: an ssh bastion cannot be reached through a kubernetes tunnel")

ErrKubernetesViaUnsupported is returned when an SSH bastion is itself placed behind a Kubernetes tunnel. The reverse (a cluster reached through a bastion) is supported; this direction would need a relay and is deliberately deferred.

View Source 
var ErrNoSSHAuthMethod = errors.New("ssh: bastion has no usable auth method (private key or password)")

ErrNoSSHAuthMethod is returned when a bastion row has neither a private key nor a password to authenticate with.

View Source 
var ErrSSHHostKeyMismatch = errors.New("ssh: host key mismatch with pinned known_host_key")

ErrSSHHostKeyMismatch is returned when a bastion presents a host key that differs from the TOFU-pinned one recorded on first connect.

View Source 
var ErrServerViaCycleDial = errors.New("ssh: via_uid chain forms a cycle")

ErrServerViaCycleDial mirrors store.ErrServerViaCycle for the dial path.

View Source 
var ErrViaNotTunnel = errors.New("via_uid does not reference a tunnel server (ssh or kubernetes)")

ErrViaNotTunnel is returned when a via_uid resolves to a row that is neither an SSH bastion nor a Kubernetes cluster — i.e. not a dial path at all.

Functions

func BuildUpstreamName added in v0.16.0 

func BuildUpstreamName(dbbatVersion, username, clientAppName string, maxLen int) string

BuildUpstreamName composes the canonical dbbat-branded application/program name sent to upstream databases, so a DBA looking at the target's session views (pg_stat_activity.application_name, V$SESSION.PROGRAM, MySQL's process list) can attribute a session to the dbbat user who initiated it.

Format: 

dbbat/$version @$username

and, when the client declared an application/program name dbbat was able to intercept: 

dbbat/$version @$username for $appName

The result is truncated to fit maxLen, preferring to truncate $appName first so the "dbbat/$version @$username" prefix survives intact. If even the bare prefix exceeds maxLen, the prefix itself is truncated as a last resort. maxLen <= 0 is treated as "no room at all" and returns "".

func BumpAPIKeyUsage added in v0.24.0 

func BumpAPIKeyUsage(ctx context.Context, logger *slog.Logger, st APIKeyUsageStore, keyID uuid.UUID)

BumpAPIKeyUsage records that an API key was just used, off the caller's latency path.

Every proxy and the REST API's bearer middleware had its own copy of this one-liner, each an unguarded `go func()`: a panic in any of them ended the *process*, taking every live session of every user on every database with it. One helper is one recover.

The write deliberately runs under context.WithoutCancel: it must outlive the login (or the HTTP request) that triggered it, which is canceled the moment that returns, while still carrying whatever tracing values the caller's context holds. Four of the five proxies used context.Background() before, which merely dropped those values; the REST middleware passed the *request* context, so its bump raced request completion — net/http cancels on ServeHTTP's return — and lost more often than not.

Detaching from cancellation removes a bound, so APIKeyUsageTimeout puts one back. Without it this would be a trade rather than a fix: the four proxies were already unbounded under context.Background(), but the REST middleware's bump inherited the server's own request timeout, and dropping that would leave a goroutine parked on a wedged database for as long as the process lives — one per authenticated request.

The store error stays swallowed — nothing has ever branched on it, and a failed usage bump is not worth a line at anything above debug.

func DialUpstream added in v0.17.0 

func DialUpstream(ctx context.Context, resolver ServerResolver, encryptionKey []byte, srv *store.Server) (net.Conn, error)

DialUpstream dials srv's host:port using the process-wide pooled dialer. resolver loads the via chain and persists TOFU host keys; encryptionKey decrypts bastion SSH secrets.

func EnableClientKeepAlive added in v0.20.0 

func EnableClientKeepAlive(conn net.Conn) error

EnableClientKeepAlive turns on TCP keepalive with dbbat's probe settings on the underlying TCP socket. Non-TCP conns (tests, unix sockets) are a no-op. Errors are returned for logging but are never fatal: a proxy that cannot set keepalive is degraded, not broken.

func HasNormalizedSQLPrefix added in v0.24.0 

func HasNormalizedSQLPrefix(sql, prefix string) bool

HasNormalizedSQLPrefix reports whether sql starts with prefix once comments are normalized away, trimmed and upper-cased — the prefix-shaped sibling of MatchesNormalizedSQL, for checks like PostgreSQL's `COPY `. prefix must already be upper case.

func IsAllowedAlterSession added in v0.24.0 

func IsAllowedAlterSession(sql string) bool

IsAllowedAlterSession reports whether sql is an `ALTER SESSION SET …` in which *every* parameter being set is allowed — see alterSessionAllowedParams for the list and for why it is a list.

It fails closed on everything else, deliberately, because "false" here just means the statement keeps the classification it had before the carve-out existed:

  • a statement that is not `ALTER SESSION SET` (`ALTER SESSION ENABLE …`, `ALTER SESSION CLOSE DATABASE LINK …`, `ALTER SYSTEM …`);
  • a statement setting one allowed parameter and one that is not — allowing it partially is not an option, so `ALTER SESSION SET CURRENT_SCHEMA=X CONTAINER=Y` is refused whole;
  • anything the scanner cannot read to the end with confidence: an unterminated quote, a missing `=`, a byte outside the value charset, a second statement stapled on behind a `;`.

The statement is still recorded either way. This is a classification change, not a visibility one: an allowed ALTER SESSION appears in /queries exactly like any other statement.

On the validation path it is reached from isWriteQuery/isDDLQuery and so sees the comment-stripped scratch copy; it does no stripping of its own, which is what keeps a statement from being normalised twice per validation call. Fed raw text with a comment in it, the scanner simply fails closed.

func IsDDLQuery 

func IsDDLQuery(sql string) bool

IsDDLQuery checks if a query is a DDL operation. Same comment normalisation as IsWriteQuery.

func IsMSSQLStatementParamName added in v0.24.0 

func IsMSSQLStatementParamName(name string) bool

IsMSSQLStatementParamName reports whether a parameter name could be the statement of a SQL-carrying system procedure.

func IsPasswordChangeQuery 

func IsPasswordChangeQuery(sql string) bool

IsPasswordChangeQuery checks if a query attempts to modify user/role passwords. Comment-normalised like the other two classifiers — `ALTER/**/USER bob PASSWORD 'x'` is an ALTER USER to the database.

func IsWriteQuery 

func IsWriteQuery(sql string) bool

IsWriteQuery checks if a query is a write operation.

The classification is prefix-shaped, so a leading comment changes what the statement looks like to dbbat but not to the database: `/*x*/INSERT …` is an INSERT either way. It therefore runs against the comment-stripped scratch copy (see sqlcomments.go); the caller's string is untouched.

func MSSQLDynamicSQL added in v0.24.0 

func MSSQLDynamicSQL(sql string) ([]string, bool)

MSSQLDynamicSQL returns the statement text of every dynamic-SQL form in a T-SQL batch whose text dbbat can read — `EXEC('…')`, `EXECUTE('…')` and `sp_executesql N'…'` — with the doubled quotes that escape a quote inside a literal undone, so the caller can run the same checks over it that the outer batch gets.

`false` means the batch carries dynamic SQL dbbat cannot vouch for and the caller must refuse it: either a quoted run was left open, or the extracted text nests *another* dynamic-SQL form. The nesting case is refused rather than unwrapped a second time — one level is where the recursion stops, and stopping silently would be a hole the shape of the one this file closes.

A form whose argument is *not* a literal — `EXEC(@sql)`, `EXEC('a' + @b)`, `EXEC dbo.some_proc` — yields no text and is deliberately **not** refused. The first two are undecidable; the third is an ordinary procedure call, which `describeRPCRequest` already fails closed on under a restrictive grant.

func MSSQLUseTargets added in v0.24.0 

func MSSQLUseTargets(sql string) ([]string, bool)

MSSQLUseTargets returns every database a T-SQL batch would switch to, and whether the batch could be read with confidence. `false` means a `USE` was found whose target dbbat could not parse (or that a quoted run was left open) — fail closed, the caller refuses.

It scans the whole batch rather than only its first statement, unlike MySQLUseTarget above, because a TDS SQLBatch is genuinely multi-statement and T-SQL needs no separator at all: `SELECT 1` followed by a newline and `USE otherdb` is one ordinary batch. An anchored check here would be a fig leaf.

Scanning for a keyword mid-statement is only safe because the scan skips string literals and quoted identifiers outright — `INSERT INTO t VALUES ('USE otherdb')` names no database — and because `USE` is a reserved word in T-SQL, so it cannot appear as a bare column or alias. The two constructs that do spell it without switching anything are the query hints `OPTION (USE PLAN …)` and `OPTION (USE HINT (…))`, which are skipped by name.

func MatchesAnyNormalizedSQL added in v0.24.0 

func MatchesAnyNormalizedSQL(sql string, patterns []*regexp.Regexp) bool

MatchesAnyNormalizedSQL is MatchesNormalizedSQL over a list, normalizing once rather than once per pattern.

func MatchesNormalizedSQL added in v0.24.0 

func MatchesNormalizedSQL(sql string, pattern *regexp.Regexp) bool

MatchesNormalizedSQL reports whether pattern matches sql's comment-normalized form. It exists for the protocol-local checks that sit alongside a ValidateQuery call — PostgreSQL's read-only-bypass list, SQL Server's bulk copy pattern — which are regex-shaped in exactly the same way and were evadable in exactly the same way.

Only the boolean escapes: the normalized string is never handed back, so no caller can relay it by accident. The syntax is the standard one (Oracle, PostgreSQL, SQL Server); MySQL statements are normalized inside this package by ValidateMySQLQuery, which knows the dialect's extra comment forms.

func MySQLPreparedText added in v0.24.0 

func MySQLPreparedText(sql string) (string, bool)

MySQLPreparedText returns the statement text a `PREPARE <name> FROM '<literal>'` would later execute, read from the comment-normalized scratch copy under the MySQL dialect.

The two returns are read together: 

("", true)       nothing to check — not a PREPARE, or one whose text is
                 built at runtime and so is not statically decidable
("<text>", true) the statement text, for the caller to check
("", false)      a PREPARE dbbat could not read all the way down — an
                 unterminated literal, a text that is not one single
                 literal, or a nested PREPARE. Fail closed: the caller
                 refuses.

Unwrapping stops at one level, and stops loudly. Recursion has to end somewhere, and ending silently would leave a hole the exact shape of the one this closes.

func MySQLUseTarget added in v0.24.0 

func MySQLUseTarget(sql string) (string, bool)

MySQLUseTarget reports whether sql is a MySQL `USE <database>` statement and, when it is, the database it names.

The two return values are read together: `isUse` true with an empty target means "this statement switches database and dbbat could not read where to", which the caller must refuse rather than forward — an unreadable switch is still a switch.

The match is anchored at the start of the statement, which is sound on this protocol rather than merely convenient: `USE` can only ever begin a statement, and the client leg does not negotiate `CLIENT_MULTI_STATEMENTS` (go-mysql's server capabilities offer `CLIENT_MULTI_RESULTS` only), so one COM_QUERY carries one statement. Anything trailing the target other than a `;` is therefore refused rather than parsed: `USE otherdb; SELECT 1` is not a shape dbbat can vouch for.

func NormalizeSQL added in v0.20.0 

func NormalizeSQL(sql string) string

NormalizeSQL is the canonical normalization applied before pattern matching: a trim, and nothing else.

Note the deliberate divergence from the static validators (IsWriteQuery, IsDDLQuery, IsPasswordChangeQuery), which upper-case before their keyword-prefix checks and are therefore case-insensitive for free. An approval pattern is a full regexp an operator writes and then reads back against the SQL shown in /queries, so rewriting the statement first would make patterns behave differently from what the UI displays.

The cost is that `^DELETE` does not match `delete from …`. Patterns should carry `(?i)` — which the definition form's placeholder and docs/approvals.md both teach — because a pattern that misses is a hold that never happens.

func SanitizeQueryError added in v0.22.0 

func SanitizeQueryError(ctx context.Context, logger *slog.Logger, queryError *string) *string

SanitizeQueryError is the last gate before a query error string reaches the store. It returns text a human could read, or nil when the value cannot be salvaged.

Query.Error is a human-facing column: everything legitimately written to it is a server diagnostic or a dbbat message. Two things can go wrong:

  • Control bytes. No diagnostic contains them; they only appear when a decoder misreads wire bytes — as the Oracle legacy fixed-offset Response layout did, copying column-compressed row data (0x15 descriptors, 0x07 separators) into the error field. Such a value is dropped outright.
  • Undecodable bytes. dbbat does not know the session charset, so a genuine diagnostic from a non-AL32UTF8 session is not valid UTF-8. Dropping those would silently lose real errors, so a mostly-decodable string is repaired instead: the bad bytes become U+FFFD and the text is kept. A mangled "contrainte unique viol<?>e" tells an operator what happened; nothing at all does not. Past maxUndecodableShare the string is not a sentence with accents in it — it is binary — and is dropped.

Both outcomes are logged at debug with the length only, never the bytes, so the misparse stays traceable without spilling row data into the logs.

func SanitizeStatementText added in v0.24.0 

func SanitizeStatementText(s string) (string, bool)

SanitizeStatementText applies the same judgement to a candidate *statement* run, and exists so the reasoning and the threshold live in one place rather than being reinvented per protocol.

The Oracle execute decoder needs exactly this question answered: a run of bytes is either the statement the header declared or it is misread binary, and "is it valid UTF-8" is the wrong test for the same reason it is the wrong test for a diagnostic — dbbat does not know the session charset, so a perfectly ordinary `INSERT INTO t VALUES ('café')` from a WE8ISO8859P1 session is not valid UTF-8. Refusing it there does not merely lose fidelity: the decoder falls back to a keyword scan that truncates the statement at the first accented byte, so a blocked pattern or an approval pattern in the tail stops matching.

The repaired text is what callers store, and that matters beyond readability: `queries.sql_text` is a Postgres `text` column, so a run kept verbatim would fail the insert and take the audit row with it.

func ValidateMongoCommand added in v0.16.0 

func ValidateMongoCommand(cmd, dbName string, body bson.Raw, db *store.Server, grant *store.Grant) error

ValidateMongoCommand enforces grant controls and the $db policy on a MongoDB command (contract §2). It operates on the command name and the kind-0 body. db is the session's resolved target database; grant carries the controls.

func ValidateMySQLQuery added in v0.7.0 

func ValidateMySQLQuery(sql string, grant *store.Grant) error

ValidateMySQLQuery runs shared validation plus MySQL-specific blocked patterns.

func ValidateOracleQuery 

func ValidateOracleQuery(sql string, grant *store.Grant) error

ValidateOracleQuery runs shared validation plus Oracle-specific blocked patterns.

func ValidatePatternsAgainstQueries added in v0.23.0 

func ValidatePatternsAgainstQueries(patterns, queries []string) (*CompiledApprovalPatterns, []QueryMatchResult)

ValidatePatternsAgainstQueries compiles the given approval-pattern sources and matches each query against them, applying exactly CompiledApprovalPatterns.Match's semantics — the same code NewApprovalGate/ApprovalGate.Match use on the proxy hot path.

It is a pure function: no store, no network, nothing but regexp compile and string matching. That is what makes it safe to expose directly as the grant-definition pattern-authoring endpoint (POST /grant-definitions/validate-patterns) — an author previews what a set of patterns will do against a bench of sample queries before saving anything.

Compile errors are returned alongside the match results, not as a hard failure: a pattern that fails to compile still lets the other patterns in the batch be tested, and it is up to the caller to decide compile errors block a save while non-matches don't (see docs on the API handler).

func ValidateQuery 

func ValidateQuery(sql string, grant *store.Grant) error

ValidateQuery checks SQL against grant controls. Used by the PostgreSQL, Oracle and SQL Server proxies (MySQL goes through ValidateMySQLQuery, which normalises with its own comment syntax).

sql is only ever read: the checks run against a comment-stripped scratch copy and the caller relays its own bytes, untouched.

Types

type APIKeyUsageStore added in v0.24.0 

type APIKeyUsageStore interface {
	IncrementAPIKeyUsage(ctx context.Context, id uuid.UUID) error
}

APIKeyUsageStore is the slice of the store the usage bump needs.

type ApprovalDeniedError added in v0.20.0 

type ApprovalDeniedError struct {
	Reason string
	By     string
}

ApprovalDeniedError is returned when a human explicitly denied the statement. It carries the approver's reason so each protocol can surface it verbatim to the client.

func (*ApprovalDeniedError) Error added in v0.20.0 

func (e *ApprovalDeniedError) Error() string

func (*ApprovalDeniedError) Is added in v0.20.0 

func (e *ApprovalDeniedError) Is(target error) bool

Is makes errors.Is(err, ErrApprovalDenied) work for the sentinel below.

type ApprovalDeps added in v0.20.0 

type ApprovalDeps struct {
	Enabled   bool
	Store     ApprovalStore
	Registry  *approval.Registry
	Broker    *events.Broker
	Escalator ApprovalEscalator
	Logger    *slog.Logger
	// PollInterval is how often quotas/expiry/revocation are re-evaluated
	// while a statement is parked. With no approval timeout, the LimitGuard
	// is the one remaining server-side bound, so it must keep running.
	PollInterval time.Duration
	// HoldRegistered, when set, is called with the pending row's uid *after*
	// the registry accepted the hold and *before* OnPending publishes that uid
	// to the protocol's cancellation path.
	//
	// It is a test seam and nothing else: that gap — the "arm window" — is a
	// handful of instructions wide, and the races that live in it (a cancel
	// landing there and then losing to a human approver) cannot be reached by
	// timing. Blocking here lets a test occupy the window deliberately. Nil in
	// production, where it costs one nil check per hold.
	HoldRegistered func(queryUID uuid.UUID)
}

ApprovalDeps are the collaborators every gate in the process shares.

type ApprovalEscalator added in v0.20.0 

type ApprovalEscalator interface {
	// Schedule arms the escalation timer for a hold.
	Schedule(ctx context.Context, hold ApprovalHoldInfo)
	// Resolved cancels a not-yet-fired escalation, or updates the posted
	// message in place if it already fired. Because there is no approval
	// timeout, a posted message stays actionable indefinitely — a stale
	// Approve button that silently no-ops is worse than no button.
	Resolved(ctx context.Context, queryUID uuid.UUID, status, byName, reason string)
}

ApprovalEscalator fires the delayed Slack notification for a hold and updates it in place once the hold resolves by any route. Optional: a nil escalator simply means no Slack.

type ApprovalGate added in v0.20.0 

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

ApprovalGate decides whether a statement needs a human, and parks the session while one is found.

It is constructed once per session: the grant's RE2 patterns are compiled here and reused for every statement, so the per-statement cost of the common (no-pattern) case is a nil check.

func NewApprovalGate added in v0.20.0 

func NewApprovalGate(deps ApprovalDeps, grant *store.Grant, connectionUID uuid.UUID, user *store.User, databaseName string) *ApprovalGate

NewApprovalGate compiles the grant's approval patterns. A gate is returned even when nothing is configured — callers check Active() (or just call Match, which reports false) rather than nil-checking.

Patterns that fail to compile are skipped with a loud log rather than failing the session: they are validated at definition-save time, so a bad one here means data predating validation, and refusing every connection over it would be a worse failure than gating one statement less.

func (*ApprovalGate) Active added in v0.20.0 

func (g *ApprovalGate) Active() bool

Active reports whether this gate can hold anything at all.

func (*ApprovalGate) Hold added in v0.20.0 

func (g *ApprovalGate) Hold(ctx context.Context, req HoldRequest) (uuid.UUID, error)

Hold persists the statement as pending, announces it, and blocks until a human resolves it or the hold dies. It returns the query uid it created (so the caller can complete that row instead of inserting a second one) and nil only on an explicit approval.

Every other exit — deny, disconnect, quota/expiry/revocation, shutdown — returns an error, and no path returns nil without an approval decision whose query uid equals the one this call created. That last clause is the TOCTOU guard: a client able to influence timing must not be able to have somebody else's approval released against its statement.

func (*ApprovalGate) Match added in v0.20.0 

func (g *ApprovalGate) Match(sql string) (string, bool)

Match reports the first pattern the statement matches. Matching runs on the same normalized SQL text the static validators use, so an operator writing a pattern does not have to reason about a second normalization.

Delegates to CompiledApprovalPatterns.Match, the same code the validate-patterns API endpoint calls — so a "test patterns" panel built on that endpoint reports exactly what the live gate will do, not a reimplementation that can drift from it.

func (*ApprovalGate) ResolutionFor added in v0.23.0 

func (g *ApprovalGate) ResolutionFor(queryUID uuid.UUID) *ApprovalResolution

ResolutionFor reports the outcome of a hold this session parked, but only for the statement it last resolved.

A session parks at most one statement at a time — the session goroutine is blocked for the whole hold — so remembering just the most recent resolution covers the completion event that immediately follows it, without the gate accumulating per-query state for the lifetime of a long-lived session. A miss simply means the completion event carries no approval fields, which is what it did before.

type ApprovalHoldInfo added in v0.20.0 

type ApprovalHoldInfo struct {
	QueryUID      uuid.UUID
	ConnectionUID uuid.UUID
	UserUID       uuid.UUID
	Username      string
	DatabaseName  string
	SQL           string
	Pattern       string
	StartedAt     time.Time
}

ApprovalHoldInfo describes a parked statement to the escalator.

type ApprovalResolution added in v0.23.0 

type ApprovalResolution struct {
	QueryUID uuid.UUID
	Status   string
	ByUID    *uuid.UUID
	ByName   string
	Reason   string
	At       time.Time
}

ApprovalResolution is the terminal outcome of a hold, kept on the gate for exactly as long as it takes the released statement to finish and publish.

type ApprovalStore added in v0.20.0 

type ApprovalStore interface {
	CreatePendingQuery(ctx context.Context, query *store.Query, pattern string) (*store.Query, error)
	ResolveQueryApproval(ctx context.Context, uid uuid.UUID, status string, resolvedBy *uuid.UUID, reason string) error
	NotifyEvent(ctx context.Context, channel string, payload store.EventNotification) error
}

ApprovalStore is the slice of the store the gate needs. An interface rather than *store.Store so protocol tests can drive a hold end-to-end without a database.

type CompiledApprovalPatterns added in v0.23.0 

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

CompiledApprovalPatterns is a set of RE2 approval-pattern sources compiled once, ready to match SQL with exactly ApprovalGate.Match's semantics: normalize with NormalizeSQL, then report the first pattern (in the order given) that matches — patterns that failed to compile never match anything.

This type is the single place that logic lives. NewApprovalGate uses it to build the live per-session gate, and the grant-definition validate-patterns API endpoint uses it to preview what that gate will do — so the "test patterns" panel in the UI can never drift from proxy behavior by reimplementing the match loop separately.

func CompileApprovalPatterns added in v0.23.0 

func CompileApprovalPatterns(sources []string) *CompiledApprovalPatterns

CompileApprovalPatterns compiles every pattern source. A pattern that fails to compile is recorded on its PatternCompileEntry rather than aborting the batch, so one typo does not hide the compile errors — or the matches — of every other pattern.

func (*CompiledApprovalPatterns) CompileErrors added in v0.23.0 

func (c *CompiledApprovalPatterns) CompileErrors() []PatternCompileError

CompileErrors reports every pattern that failed to compile, in source order. Empty when every pattern compiled.

func (*CompiledApprovalPatterns) Entries added in v0.23.0 

func (c *CompiledApprovalPatterns) Entries() []PatternCompileEntry

Entries reports the compile outcome of every source, in the order given to CompileApprovalPatterns — one entry per source, success or failure.

func (*CompiledApprovalPatterns) Len added in v0.23.0 

func (c *CompiledApprovalPatterns) Len() int

Len reports how many patterns compiled successfully.

func (*CompiledApprovalPatterns) Match added in v0.23.0 

func (c *CompiledApprovalPatterns) Match(sql string) (string, bool)

Match reports the first successfully-compiled pattern that matches sql after NormalizeSQL, or false if none does.

type CountingConn added in v0.10.0 

type CountingConn struct {
	net.Conn
	// contains filtered or unexported fields
}

CountingConn wraps a net.Conn and atomically tracks the number of bytes read from and written to it. The two counters live outside the wrapper so a session can share them across multiple wrapped conns (e.g. client and upstream): writes to one direction on one wrapper match reads from the same direction on the other.

Total() is safe to call concurrently with Read/Write — useful for taking per-query snapshots while the proxy is mid-stream.

func NewCountingConn added in v0.10.0 

func NewCountingConn(conn net.Conn, bytesRead, bytesWritten *atomic.Int64) *CountingConn

NewCountingConn wraps conn so Read accumulates into bytesRead and Write accumulates into bytesWritten. Either counter may be nil to disable that direction (rare; the typical caller passes both).

func (*CountingConn) Read added in v0.10.0 

func (c *CountingConn) Read(p []byte) (int, error)

Read implements net.Conn. Successful byte counts are added to the read counter even when the call returns an error (n > 0 with err is a valid outcome on a closing conn — those bytes did cross the wire).

func (*CountingConn) Unwrap added in v0.20.0 

func (c *CountingConn) Unwrap() net.Conn

Unwrap exposes the wrapped conn so helpers that need the raw socket (EnableClientKeepAlive) can walk down the wrapper chain.

func (*CountingConn) Write added in v0.10.0 

func (c *CountingConn) Write(p []byte) (int, error)

Write implements net.Conn with the same byte-counting semantics as Read.

type Dialer added in v0.17.0 

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

Dialer opens upstream connections, tunneling through a *via* row when a server row's ViaUID is set. Two kinds of via row exist and are pooled the same way, keyed by server UID: an SSH bastion yields a shared *ssh.Client carrying one channel per session, and a Kubernetes cluster yields a shared port-forward tunnel carrying one stream pair per session. A dead entry of either kind is transparently rebuilt.

func NewDialer added in v0.17.0 

func NewDialer() *Dialer

NewDialer builds an empty Dialer with its own via pools.

func (*Dialer) Close added in v0.18.0 

func (d *Dialer) Close()

Close tears down every pooled via entry. Used by short-lived dialers (connectivity checks) so a probe does not leak an SSH connection.

func (*Dialer) ConnectBastion added in v0.18.0 

func (d *Dialer) ConnectBastion(
	ctx context.Context,
	resolver ServerResolver,
	encryptionKey []byte,
	uid uuid.UUID,
) (*ssh.Client, error)

ConnectBastion dials (or reuses a pooled connection to) the SSH bastion row identified by uid, completing the handshake and — on first connect — pinning the presented host key via resolver.SetKnownHostKey.

It exists so a connectivity check can validate a `protocol: ssh` row on its own, with no database target behind it. Callers that want to force a real dial (rather than reuse a pooled client) must use a fresh Dialer.

func (*Dialer) ConnectKubernetes added in v0.24.0 

func (d *Dialer) ConnectKubernetes(
	ctx context.Context,
	resolver ServerResolver,
	encryptionKey []byte,
	uid uuid.UUID,
) (*upstream.KubernetesTunnel, error)

ConnectKubernetes returns a pooled (or freshly built) tunnel for the Kubernetes cluster row identified by uid.

It exists so a connectivity check can validate a `protocol: kubernetes` row on its own, with no database target behind it — the counterpart of ConnectBastion. Building the tunnel does no I/O; the caller is what probes.

func (*Dialer) DialUpstream added in v0.17.0 

func (d *Dialer) DialUpstream(ctx context.Context, resolver ServerResolver, encryptionKey []byte, srv *store.Server) (net.Conn, error)

DialUpstream dials srv's host:port directly, or through srv.ViaUID's via chain when set — an SSH bastion (recursing for multi-hop jump hosts) or a Kubernetes cluster.

type HoldRequest added in v0.20.0 

type HoldRequest struct {
	// SQL is the statement text, as it will be persisted and shown.
	SQL string
	// Params are the bind parameters, known at Execute time (not at Parse
	// time — which is exactly why the PostgreSQL gate hooks Execute).
	Params *store.QueryParameters
	// Pattern is the pattern that matched, recorded on the row.
	Pattern string
	// StartedAt is when the statement arrived.
	StartedAt time.Time
	// ClientGone fires when the parked client's socket dies. This is the
	// sole liveness bound on a hold: without it, "until disconnect" means
	// forever.
	ClientGone <-chan struct{}
	// Guard keeps quotas, expiry and revocation running while parked.
	Guard *LimitGuard
	// OnPending, when set, is called with the pending row's uid the instant
	// it exists — before the session blocks. Protocols use it to publish the
	// uid to their out-of-band cancellation path (PostgreSQL CancelRequest,
	// MySQL KILL QUERY, Mongo killOperations), which must be able to end a
	// hold that has not returned yet.
	OnPending func(queryUID uuid.UUID)
}

HoldRequest is one parked statement.

type LimitGuard added in v0.16.0 

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

LimitGuard evaluates a grant's time-window and bandwidth limits against the live wire-byte counters. It is designed to be called on the data path: Check() performs at most two atomic loads and a wall-clock comparison, with no allocation and no locking.

A guard built from a nil grant (or a grant with no limits) never trips, so callers can construct one unconditionally.

func NewLimitGuard added in v0.16.0 

func NewLimitGuard(grant *store.Grant, from, to *atomic.Int64) *LimitGuard

NewLimitGuard builds a guard for grant, reading live traffic from the two atomic counters (either may be nil). grant may be nil — the resulting guard enforces nothing.

func (*LimitGuard) Check added in v0.16.0 

func (g *LimitGuard) Check() error

Check reports the first limit that has been crossed, or nil if the grant is still within bounds. Bandwidth is checked before expiry so the "gigabytes in seconds" case is attributed to the byte quota, but either is a valid abort reason.

func (*LimitGuard) Watch added in v0.16.0 

func (g *LimitGuard) Watch(ctx context.Context, interval time.Duration, onViolation func(error))

Watch polls Check on a ticker until a limit is crossed or ctx is canceled. On the first violation it invokes onViolation with the offending error and returns; onViolation is never called more than once. interval <= 0 falls back to DefaultLimitPollInterval.

Watch is the guaranteed, protocol-agnostic enforcement path: it fires even when a query is blocked producing no traffic (idle expiry) and even for protocols whose client library owns the wire (MySQL). onViolation typically force-closes the client and upstream conns to tear the session down.

func (*LimitGuard) WithRevocation added in v0.16.0 

func (g *LimitGuard) WithRevocation(revoked *atomic.Bool) *LimitGuard

WithRevocation attaches the session's shared revocation flag to the guard so Check/Watch also trip when the grant is revoked mid-session. Returns the guard for fluent construction. A nil flag is a no-op (nothing to watch), keeping the plain NewLimitGuard signature stable for callers/tests that don't track revocation.

type PatternCompileEntry added in v0.23.0 

type PatternCompileEntry struct {
	Pattern string
	Regexp  *regexp.Regexp
	Err     error
}

PatternCompileEntry is the compile outcome for one approval pattern source, in the order the source was given to CompileApprovalPatterns. Exactly one of Regexp/Err is set: Err nil means the pattern compiled.

type PatternCompileError added in v0.23.0 

type PatternCompileError struct {
	Pattern string
	Err     error
}

PatternCompileError names one approval pattern source that failed to compile, and why. A narrower view of PatternCompileEntry for callers that only care about the failures.

type QueryMatchResult added in v0.23.0 

type QueryMatchResult struct {
	Query      string
	Normalized string
	Matched    bool
	Pattern    string
}

QueryMatchResult is the outcome of testing one candidate SQL statement against a set of compiled approval patterns: the NormalizeSQL form the patterns actually run against, and which pattern (if any) matched first.

type QuerySink added in v0.20.0 

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

QuerySink is one query's handle on the shared writer. It carries the parent query uid (once known) and records whether the capture lost rows.

A nil sink is inert: Add reports the row as not accepted, everything else is a no-op. That is what a session with no store gets.

func (*QuerySink) Add added in v0.20.0 

func (s *QuerySink) Add(row store.QueryRow) bool

Add queues one captured row. It never blocks: on a full queue (or an exhausted byte budget) the row is dropped, the capture is marked degraded, and false is returned. This is the entry point for capture paths that run inline with forwarding rows to the client.

func (*QuerySink) AddAll added in v0.20.0 

func (s *QuerySink) AddAll(ctx context.Context, rows []store.QueryRow)

AddAll queues a whole result set, waiting for queue space rather than dropping. It is only for callers that already hold every row in memory and run off the data path (MySQL, MongoDB, PostgreSQL COPY): those rows are already resident, so queueing them costs no extra memory, and dropping them would lose a capture that used to be stored whole.

The sink must already be resolved — the drain goroutine cannot make progress on rows whose parent query record is still unknown while the caller holds it.

func (*QuerySink) Dropped added in v0.20.0 

func (s *QuerySink) Dropped() bool

Dropped reports whether any row was lost — queue full, batch insert failed, or no parent query record. Read it after Flush so the answer covers the whole capture.

func (*QuerySink) Fail added in v0.20.0 

func (s *QuerySink) Fail()

Fail reports that the query record could not be created, so the sink's rows have no parent to hang from. Queued rows are discarded and the capture is marked as having dropped rows.

func (*QuerySink) Flush added in v0.20.0 

func (s *QuerySink) Flush(ctx context.Context)

Flush blocks until every row submitted so far has been inserted (or definitively lost). It is the barrier a query needs before it is marked complete: without it the UI would show a finished query with rows still arriving.

It is also where the capture's tamper-evident row chain is sealed, because the barrier is the first moment the chain head is final.

Call it from the completion goroutine, never from the capture path.

func (*QuerySink) QueryUID added in v0.20.0 

func (s *QuerySink) QueryUID(ctx context.Context) (uuid.UUID, bool)

QueryUID reports the uid of the query record this sink's rows hang from, blocking until it is known. It reports false when there is no sink, when the record could not be created, or when nothing resolved it in time — in which case the caller owns creating the record itself.

func (*QuerySink) Resolve added in v0.20.0 

func (s *QuerySink) Resolve(queryUID uuid.UUID)

Resolve publishes the uid of the query record the sink's rows belong to. Must be called exactly once, after the record has been inserted.

type RowStore added in v0.20.0 

type RowStore interface {
	StoreQueryRows(ctx context.Context, rows []store.PendingQueryRow) error

	// SealQueryRowChain stamps the final head of a capture's tamper-evident
	// row chain onto its query. It is called once per capture, at the flush
	// barrier, because that is the first moment the head is final — see
	// store.SealQueryRowChain and docs/audit-chain.md. Cheap for a query that
	// captured nothing.
	SealQueryRowChain(ctx context.Context, queryUID uuid.UUID) error
}

RowStore is the slice of the store the row writer needs.

type RowWriter added in v0.20.0 

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

RowWriter persists captured result rows in batches, off the proxy's data path. One writer is shared by the whole process: its batches span protocols, sessions and queries, so a busy proxy running many small result sets issues one INSERT per ~1000 rows overall rather than one per query.

The queue is a bounded channel drained opportunistically: the drain goroutine blocks for the first row, then takes whatever is already queued until a cap trips. Batches therefore size themselves to load — an idle producer gets batches of one and minimal latency, a fast producer gets full batches — with no timer to tune.

Submitting from a capture path is non-blocking and drops on a full queue (see QuerySink.Add). The capture path runs inline with forwarding rows to the client, so a slow moment in dbbat's own storage must never become a stall on the customer's query.

A nil *RowWriter is usable: every method is a no-op and NewSink returns a nil sink, which is likewise inert. That keeps the protocol call sites free of nil checks when there is no store to write to.

func NewRowWriter added in v0.20.0 

func NewRowWriter(st RowStore, logger *slog.Logger) *RowWriter

NewRowWriter starts a writer against st. A nil store yields a nil writer, which is inert — callers with no store keep working unchanged.

func (*RowWriter) Close added in v0.20.0 

func (w *RowWriter) Close(ctx context.Context)

Close drains what is queued and stops the writer. It is safe to call twice.

func (*RowWriter) NewSink added in v0.20.0 

func (w *RowWriter) NewSink() *QuerySink

NewSink returns a sink for one query whose record does not exist yet. Rows may be submitted immediately; they are held in the queue and only inserted once Resolve supplies the parent query uid, which is what keeps the query_rows -> queries foreign key satisfied without the producer ever blocking on the parent INSERT.

func (*RowWriter) NewSinkFor added in v0.20.0 

func (w *RowWriter) NewSinkFor(queryUID uuid.UUID) *QuerySink

NewSinkFor returns a sink for a query whose record already exists.

type ServerResolver added in v0.17.0 

type ServerResolver interface {
	GetServerByUID(ctx context.Context, uid uuid.UUID) (*store.Server, error)
	SetKnownHostKey(ctx context.Context, uid uuid.UUID, hostKey string) error
	SetKubernetesCACert(ctx context.Context, uid uuid.UUID, caCert string) error
}

ServerResolver resolves server rows and persists TOFU pins — an SSH bastion's host key, a Kubernetes API server's CA bundle. Satisfied by *store.Store; an interface so the dialer can be unit-tested with a fake.

type StreamPublisher added in v0.20.0 

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

StreamPublisher publishes a session's observable activity — every query on the connection, plus connection open/close — to the live event stream.

It is separate from ApprovalGate on purpose: streaming is unconditional and harmless, approval holds are opt-in and block a database connection. They share nothing but the connection identity.

Every method is nil-safe and non-blocking. A broken stream must never break a database connection, which is why nothing here returns an error.

func NewStreamPublisher added in v0.20.0 

func NewStreamPublisher(
	deps ApprovalDeps,
	connectionUID uuid.UUID,
	user *store.User,
	databaseName string,
) *StreamPublisher

NewStreamPublisher builds a publisher for one session. A nil broker yields a publisher whose methods are no-ops.

func (*StreamPublisher) Connection added in v0.20.0 

func (p *StreamPublisher) Connection(ctx context.Context, state string)

Connection announces a connection lifecycle change on the admin-only connections topic.

func (*StreamPublisher) Query added in v0.20.0 

func (p *StreamPublisher) Query(queryUID uuid.UUID, q *store.Query)

Query announces one executed statement on connection/<uid>/queries.

There is deliberately no global all-queries topic: a busy proxy issues thousands of statements a second, each carrying full SQL text, and firehosing that would be both a throughput problem and a data-exposure one. Watching is per-connection and opt-in.

func (*StreamPublisher) WithApprovals added in v0.23.0 

func (p *StreamPublisher) WithApprovals(gate *ApprovalGate) *StreamPublisher

WithApprovals links the session's approval gate so a released statement's completion event carries the decision that released it. Returns the receiver for chaining at construction.

type WatchedConn added in v0.20.0 

type WatchedConn struct {
	net.Conn
	// contains filtered or unexported fields
}

WatchedConn is a net.Conn wrapper that can be "parked": while parked, a background goroutine keeps reading the socket so a client FIN is noticed immediately instead of sitting unread until the session resumes.

This is load-bearing rather than decorative. During an approval hold the session goroutine is blocked on a human, so nothing is reading the client socket; without an active read-watch a disconnected client would leave the statement parked forever.

Bytes a pipelining client sent while parked are *not* dropped and *not* interpreted mid-hold: they are queued and replayed, in stream order, the moment the session resumes reading.

Placement matters: WatchedConn must sit *below* any TLS layer, so it parks on raw TLS records it never has to decrypt.

func NewWatchedConn added in v0.20.0 

func NewWatchedConn(conn net.Conn) *WatchedConn

NewWatchedConn wraps conn.

func (*WatchedConn) Buffered added in v0.20.0 

func (w *WatchedConn) Buffered() int

Buffered reports how many replay bytes are queued. Test/telemetry helper.

func (*WatchedConn) Park added in v0.20.0 

func (w *WatchedConn) Park() <-chan struct{}

Park starts the read-watch. The returned channel is closed when the client goes away (EOF, reset, or any other terminal read error). Unpark must be called before the session resumes reading.

func (*WatchedConn) Read added in v0.20.0 

func (w *WatchedConn) Read(p []byte) (int, error)

Read serves any bytes captured while parked before touching the socket, so the byte stream the session sees is exactly the byte stream the client sent.

func (*WatchedConn) Unpark added in v0.20.0 

func (w *WatchedConn) Unpark()

Unpark stops the read-watch and hands the socket back to the session. Any bytes the watcher captured stay queued for replay.

func (*WatchedConn) Unwrap added in v0.20.0 

func (w *WatchedConn) Unwrap() net.Conn

Unwrap exposes the wrapped conn so helpers like EnableClientKeepAlive can reach the TCP socket underneath.

Source Files

View all Source files

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.