pipeline

package
v1.2.0 Latest Latest
Warning

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

 Go to latest Published: Oct 17, 2026 License: Apache-2.0 Imports: 25 Imported by: 0

Documentation

Overview

Package pipeline is the single orchestration entry point for protected SQL. It authenticates, loads policy, parses, evaluates two guard gates, executes only allowed statements, redacts result sets, and synchronously audits every outcome.

Index

Constants

View Source 
const (
	StageAuth         = "auth"
	StageLoad         = "load"
	StageParse        = "parse"
	StageGuardStatic  = "guard_static"
	StageGuardDynamic = "guard_dynamic"
	StageExecute      = "execute"
	StageRedact       = "redact"
	StageAudit        = "audit"
)

Variables

View Source 
var (
	ErrInvalidPorts            = errors.New("invalid pipeline ports")
	ErrInvalidSecret           = errors.New("invalid pipeline secret")
	ErrInvalidOption           = errors.New("invalid pipeline option")
	ErrInvalidRequest          = errors.New("invalid pipeline request")
	ErrAuditUnavailable        = errors.New("audit unavailable")
	ErrBusinessCommitUncertain = errors.New("business commit result uncertain")
	ErrTaskGrantUnavailable    = errors.New("task read grant validation unavailable")
)

Functions

func StaticAssess 

func StaticAssess(in StaticAssessInput) (model.Assessment, string, error)

StaticAssess parses one SQL string and runs only the production static rule gate. It has no store, executor, approval, or audit dependencies.

Types

type ApprovalWorkflow 

type ApprovalWorkflow interface {
	CreatePendingWithAudit(
		ctx context.Context,
		approval model.Approval,
		log model.AuditLog,
	) (model.Approval, model.AuditLog, error)
}

ApprovalWorkflow persists a pending approval and its approve audit event. A shared metadata/audit store provides all-or-nothing transaction semantics. With separate stores, implementations write audit first; metadata failure may therefore leave a valid immutable orphan audit and must return no approval.

type ApprovalWriter 

type ApprovalWriter interface {
	Create(ctx context.Context, approval model.Approval) (model.Approval, error)
}

type AuditRecorder 

type AuditRecorder interface {
	Record(ctx context.Context, log model.AuditLog) (model.AuditLog, error)
}

type ColumnAuthorizationConfiguration 

type ColumnAuthorizationConfiguration interface {
	ColumnAuthorizationConfigured() bool
}

ColumnAuthorizationConfiguration reports that B2 enforcement was requested even for a dialect that deliberately remains on the table-level route.

type ColumnAuthorizationController 

type ColumnAuthorizationController interface {
	Begin(context.Context, model.Agent, model.Datasource) (ColumnAuthorizationSnapshot, error)
}

type ColumnAuthorizationRouter 

type ColumnAuthorizationRouter interface {
	ColumnAuthorizationEnabled(model.Datasource) bool
}

ColumnAuthorizationRouter lets an activated controller keep unsupported dialects on the existing table-level path. Once it returns true, later runtime/fence failures are handled inside Begin and fail closed rather than silently falling back.

type ColumnAuthorizationSnapshot 

type ColumnAuthorizationSnapshot interface {
	Policies() []model.Policy
	Redactor() mask.Redactor
	RevisionDigest() string
	FinalCheck(context.Context) error
	Close() error
}

ColumnAuthorizationSnapshot owns one protocol-3 control snapshot. Its policies and mask plan source were read inside that snapshot; FinalCheck is called only after the business transaction and durable audit have ended.

type DatasourceReader 

type DatasourceReader interface {
	Get(ctx context.Context, id string) (model.Datasource, error)
}

type DecisionObserver 

type DecisionObserver interface {
	ObserveDecision(decision, dialect, stmtType string)
	ObserveRuleHit(ruleID, decision, risk string)
	ObserveStage(stage string, latencyMS int64)
}

DecisionObserver receives bounded, non-sensitive final pipeline signals.

type ExecutorProvider 

type ExecutorProvider interface {
	AuthorizedExecute(context.Context, model.Datasource, []byte, string, string) (executor.Statement, error)
}

type IdentityAuthenticator 

type IdentityAuthenticator interface {
	Authenticate(ctx context.Context, rawKey string) (model.Agent, error)
}

type Option 

type Option func(*pipelineOptions) error

Option configures immutable pipeline behavior at construction time.

func WithColumnAuthorization 

func WithColumnAuthorization(controller ColumnAuthorizationController) Option

WithColumnAuthorization enables the feature-gated S4 SELECT path. Normal construction omits this option, preserving the protocol-2 table pipeline.

func WithDemoConfig 

func WithDemoConfig(demo config.DemoConfig) Option

WithDemoConfig installs the server-validated demo boundary on the pipeline. A disabled or empty configuration makes every ProcessDemo call fail closed.

func WithObserver 

func WithObserver(observer DecisionObserver) Option

WithObserver installs an optional observer. A nil observer disables metrics.

func WithRuleLayers 

func WithRuleLayers(layers engine.RuleLayers) Option

WithRuleLayers supplies global, datasource, and Agent rule overrides.

type Pipeline 

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

Pipeline is the immutable coordinator for one protected SQL flow.

func New 

func New(ports Ports, secret []byte, options ...Option) (*Pipeline, error)

New validates all seven ports and constructs the shared guard components.

func (*Pipeline) ExecuteApproved added in v0.6.0 

func (pipeline *Pipeline) ExecuteApproved(ctx context.Context, apiKey, approvalID string) (Response, error)

ExecuteApproved atomically consumes a DBA-approved request before any business database access. A crash leaves it claimed for manual review.

func (*Pipeline) ExecuteApprovedDelegated added in v1.1.0 

func (pipeline *Pipeline) ExecuteApprovedDelegated(ctx context.Context, apiKey, approvalID, userID, tenantID string) (Response, error)

ExecuteApprovedDelegated binds an approved write to the same live human identity that created the request. The store checks this before claiming.

func (*Pipeline) Process 

func (pipeline *Pipeline) Process(ctx context.Context, request Request) (Response, error)

Process runs auth, load, parse, two guard gates, execute, redact, and audit.

func (*Pipeline) ProcessDemo 

func (pipeline *Pipeline) ProcessDemo(ctx context.Context, request Request) (Response, error)

ProcessDemo runs the production pipeline with immutable Live Demo safety constraints. Only trusted server-side code should map a demo profile to the API key carried by request.

type PolicyLoader 

type PolicyLoader interface {
	ListByAgentAndDatasource(
		ctx context.Context,
		agentID string,
		datasourceID string,
	) ([]model.Policy, error)
}

type Ports 

type Ports struct {
	Authenticator  IdentityAuthenticator
	Datasources    DatasourceReader
	Policies       PolicyLoader
	Executors      ExecutorProvider
	Approvals      ApprovalWriter
	Audit          AuditRecorder
	Redactors      RedactorBuilder
	RuleOverrides  RuleOverrideReader
	TaskReadGrants TaskReadGrantReader
}

Ports contains the seven required pipeline dependencies plus the optional runtime rule-override reader. A nil RuleOverrides preserves built-in rules.

type RedactorBuilder 

type RedactorBuilder interface {
	RedactorFor(ctx context.Context, datasourceID string) (mask.Redactor, error)
}

type Request 

type Request struct {
	APIKey            string
	DatasourceID      string
	SQL               string
	Reason            string
	SessionID         string
	ConversationID    *string
	MCPTool           string
	ClientIP          *string
	ModelName         *string
	DelegatedUserID   string
	DelegatedTenantID string
	ExplainOnly       bool
	RequireApproval   bool
}

Request is one protected SQL operation.

type Response 

type Response struct {
	Decision     model.Decision
	Assessment   model.Assessment
	Result       *model.QueryResult
	Redact       mask.RedactReport
	ApprovalID   string
	AuditID      int64
	ErrorCode    string                             `json:"error_code,omitempty"`
	ErrorStage   string                             `json:"error_stage,omitempty"`
	ErrorMessage string                             `json:"error_message,omitempty"`
	Suggestion   string                             `json:"suggestion,omitempty"`
	ColumnAuth   *executor.ColumnAuthorizationAudit `json:"column_auth,omitempty"`
	// contains filtered or unexported fields
}

Response is the complete guarded SQL outcome.

type RuleOverrideReader 

type RuleOverrideReader interface {
	List(ctx context.Context, dbType string) ([]model.Rule, error)
}

RuleOverrideReader reads administrator-maintained global rule overrides.

type StaticAssessInput 

type StaticAssessInput struct {
	SQL        string
	Dialect    model.DBDialect
	AgentLevel string
}

StaticAssessInput contains the isolated inputs accepted by the playground.

type TaskReadGrantReader added in v0.7.0 

type TaskReadGrantReader interface {
	Validate(context.Context, string, string, string, string, string, time.Time) (model.TaskReadGrant, error)
	Consume(context.Context, string, string, string, string, string, time.Time) error
}

TaskReadGrantReader checks a temporary grant against trusted identity and exact SQL digest at every admission, execution, and delivery gate.

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.