wire

package module
v0.0.3 Latest Latest
Warning

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

 Go to latest Published: Feb 10, 2025 License: MPL-2.0 Imports: 21 Imported by: 0

README

PSQL wire protocol 🔌

CI Go Reference Latest release Go Report Card

A pure Go PostgreSQL server wire protocol implementation. Build your own PostgreSQL server within a few lines of code. This project attempts to make it as straight forward as possible to set-up and configure your own PSQL server. Feel free to check out the examples directory for various ways on how to configure/set-up your own server.

🚧 This project does not include a PSQL parser. Please check out other projects such as auxten/postgresql-parser to parse PSQL SQL queries.

package main

import (
	"context"
	"fmt"

	wire "github.com/suprsend/psql-wire"
)

func main() {
	wire.ListenAndServe("127.0.0.1:5432", handler)
}

func handler(ctx context.Context, query string) (wire.PreparedStatements, error) {
	return wire.Prepared(wire.NewStatement(func(ctx context.Context, writer wire.DataWriter, parameters []wire.Parameter) error {
		fmt.Println(query)
		return writer.Complete("OK")
	})), nil
}

🚧 When wanting to debug issues and or inspect the PostgreSQL wire protocol please check out the psql-proxy cli

Support

Feel free to start a new discussion to discuss feature requests or issues.

Contributing

Thank you for your interest in contributing to psql-wire! Check out the open projects and/or issues and feel free to join any ongoing discussion. Feel free to checkout the open TODO's within the project.

Everyone is welcome to contribute, whether it's in the form of code, documentation, bug reports, feature requests, or anything else. We encourage you to experiment with the project and make contributions to help evolve it to meet your needs!

See the contributing guide for more details.

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var CopySignature = []byte("PGCOPY\n\377\r\n\000")

CopySignature is the signature that is used to identify the start of a copy-in stream. The signature is used to identify the start of a copy-in stream and is used to determine the start of the copy-in data. https://www.postgresql.org/docs/current/sql-copy.html

View Source 
var ErrClosedWriter = errors.New("closed writer")

ErrClosedWriter is returned when the data writer has been closed.

View Source 
var ErrDataWritten = errors.New("data has already been written")

ErrDataWritten is returned when an empty result is attempted to be sent to the client while data has already been written.

View Source 
var ErrUnknownOid = errors.New("unknown oid")
View Source 
var QueryParameters = regexp.MustCompile(`\$(\d+)|\?`)

QueryParameters represents a regex which could be used to identify and lookup parameters defined inside a given query. Parameters could be defined as positional parameters and non-positional parameters.

Functions

func AuthenticatedUsername 

func AuthenticatedUsername(ctx context.Context) string

AuthenticatedUsername returns the username of the authenticated user of the given connection context.

func ErrorCode 

func ErrorCode(writer *buffer.Writer, err error) error

ErrorCode writes an error message as response to a command with the given severity and error message. A ready for query message is written back to the client once the error has been written indicating the end of a command cycle. https://www.postgresql.org/docs/current/static/protocol-error-fields.html

func IsSuperUser 

func IsSuperUser(ctx context.Context) bool

IsSuperUser checks whether the given connection context is a super user.

func ListenAndServe 

func ListenAndServe(address string, handler ParseFn) error

ListenAndServe opens a new Postgres server using the given address and default configurations. The given handler function is used to handle simple queries. This method should be used to construct a simple Postgres server for testing purposes or simple use cases.

func NewErrMultipleCommandsStatements 

func NewErrMultipleCommandsStatements() error

NewErrMultipleCommandsStatements is returned whenever multiple statements have been given within a single query during the extended query protocol.

func NewErrUndefinedStatement 

func NewErrUndefinedStatement() error

NewErrUndefinedStatement is returned whenever no statement has been defined within the incoming query.

func NewErrUnimplementedMessageType 

func NewErrUnimplementedMessageType(t types.ClientMessage) error

NewErrUnimplementedMessageType is called whenever an unimplemented message type is sent. This error indicates to the client that the sent message cannot be processed at this moment in time.

func NewErrUnkownStatement 

func NewErrUnkownStatement(name string) error

NewErrUnkownStatement is returned whenever no executable has been found for the given name.

func ParseParameters 

func ParseParameters(query string) []oid.Oid

ParseParameters attempts to parse the parameters in the given string and returns the expected parameters. This is necessary for the query protocol where the parameter types are expected to be defined in the extended query protocol.

func RemoteAddress 

func RemoteAddress(ctx context.Context) net.Addr

RemoteAddress returns the Postgres remote address connection info if it has been set inside the given context.

func TypeMap 

func TypeMap(ctx context.Context) *pgtype.Map

TypeMap returns the Postgres type connection info if it has been set inside the given context.

Types

type AuthStrategy 

type AuthStrategy func(ctx context.Context, writer *buffer.Writer, reader *buffer.Reader) (_ context.Context, err error)

AuthStrategy represents an authentication strategy used to authenticate a user.

func ClearTextPassword 

func ClearTextPassword(validate func(ctx context.Context, database, username, password string) (context.Context, bool, error)) AuthStrategy

ClearTextPassword announces to the client to authenticate by sending a clear text password and validates if the provided username and password (received inside the client parameters) are valid. If the provided credentials are invalid or any unexpected error occurs, an error returned and the connection should be closed.

type BinaryCopyReader 

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

func NewBinaryColumnReader 

func NewBinaryColumnReader(ctx context.Context, copy *CopyReader) (_ *BinaryCopyReader, err error)

NewBinaryColumnReader creates a new column reader that reads rows from the given copy reader and returns the values as a slice of any values. The columns are used to determine the format of the data that is read from the reader. If the end of the copy-in stream is reached, an io.EOF error is returned.

func (*BinaryCopyReader) Read 

func (r *BinaryCopyReader) Read(ctx context.Context) (_ []any, err error)

Read reads a single row from the copy-in stream. The read row is returned as a slice of any values. If the end of the copy-in stream is reached, an io.EOF error is returned.

type CloseFn 

type CloseFn func(ctx context.Context) error

type Column 

type Column struct {
	Table        int32  // table id
	ID           int32  // column identifier
	Attr         int16  // column attribute number
	Name         string // column name
	AttrNo       int16  // column attribute no (optional)
	Oid          oid.Oid
	Width        int16
	TypeModifier int32
}

Column represents a table column and its attributes such as name, type and encode formatter.

func (Column) Define 

func (column Column) Define(ctx context.Context, writer *buffer.Writer, format FormatCode)

Define writes the column header values to the given writer. This method is used to define a column inside RowDescription message defining the column type, width, and name.

func (Column) Write 

func (column Column) Write(ctx context.Context, writer *buffer.Writer, format FormatCode, src any) (err error)

Write encodes the given source value using the column type definition and connection info. The encoded byte buffer is added to the given write buffer. This method Is used to encode values and return them inside a DataRow message.

type Columns 

type Columns []Column

Columns represent a collection of columns.

func (Columns) CopyIn 

func (columns Columns) CopyIn(ctx context.Context, writer *buffer.Writer, format FormatCode) error

CopyIn sends a [CopyInResponse] to the client, to initiate a CopyIn operation. Based on the given columns within the prepared statement. https://www.postgresql.org/docs/current/protocol-message-formats.html#PROTOCOL-MESSAGE-FORMATS-COPYINRESPONSE

func (Columns) Define 

func (columns Columns) Define(ctx context.Context, writer *buffer.Writer, formats []FormatCode) error

Define writes the table RowDescription headers for the given table and the containing columns. The headers have to be written before any data rows could be send back to the client. The given columns are encoded using the given format codes. Columns could be encoded as Text or Binary. If you provide a single format code, it will be applied to all columns.

func (Columns) Write 

func (columns Columns) Write(ctx context.Context, formats []FormatCode, writer *buffer.Writer, srcs []any) (err error)

Write writes the given column values back to the client. The given columns are encoded using the given format codes. Columns could be encoded as Text or Binary. If you provide a single format code, it will be applied to all columns.

type CopyReader 

type CopyReader struct {
	*buffer.Reader
	// contains filtered or unexported fields
}

func NewCopyReader 

func NewCopyReader(reader *buffer.Reader, writer *buffer.Writer, columns Columns) *CopyReader

NewCopyReader creates a new copy reader that reads copy-in data from the given reader and writes the data to the given writer. The columns are used to determine the format of the data that is read from the reader.

func (*CopyReader) Columns 

func (r *CopyReader) Columns() Columns

Columns returns the columns that are currently defined within the copy reader.

func (*CopyReader) Read 

func (r *CopyReader) Read() error

Read reads a single chunk from the copy-in stream. The read chunk is returned as a byte slice. If the end of the copy-in stream is reached, an io.EOF error is returned.

type DataWriter 

type DataWriter interface {
	// Row writes a single data row containing the values inside the given slice to
	// the underlaying Postgres client. The column headers have to be written before
	// sending rows. Each item inside the slice represents a single column value.
	// The slice length needs to be the same length as the defined columns. Nil
	// values are encoded as NULL values.
	Row([]any) error

	// Written returns the number of rows written to the client.
	Written() uint64

	// Empty announces to the client an empty response and that no data rows should
	// be expected.
	Empty() error

	// Columns returns the columns that are currently defined within the writer.
	Columns() Columns

	// Complete announces to the client that the command has been completed and
	// no further data should be expected.
	//
	// See [CommandComplete] for the expected format for different queries.
	//
	// [CommandComplete]: https://www.postgresql.org/docs/current/protocol-message-formats.html#PROTOCOL-MESSAGE-FORMATS-COMMANDCOMPLETE
	Complete(description string) error

	// CopyIn sends a [CopyInResponse] to the client, to initiate a CopyIn
	// operation. The copy operation can be used to send large amounts of data to
	// the server in a single transaction. A column reader has to be used to read
	// the data that is sent by the client to the CopyReader.
	CopyIn(format FormatCode) (*CopyReader, error)
}

DataWriter represents a writer interface for writing columns and data rows using the Postgres wire to the connected client.

func NewDataWriter 

func NewDataWriter(ctx context.Context, columns Columns, formats []FormatCode, reader *buffer.Reader, writer *buffer.Writer) DataWriter

NewDataWriter constructs a new data writer using the given context and buffer. The returned writer should be handled with caution as it is not safe for concurrent use. Concurrent access to the same data without proper synchronization can result in unexpected behavior and data corruption.

type DefaultPortalCache 

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

func (*DefaultPortalCache) Bind 

func (cache *DefaultPortalCache) Bind(ctx context.Context, name string, stmt *Statement, parameters []Parameter, formats []FormatCode) error

func (*DefaultPortalCache) Execute 

func (cache *DefaultPortalCache) Execute(ctx context.Context, name string, reader *buffer.Reader, writer *buffer.Writer) (err error)

func (*DefaultPortalCache) Get 

func (cache *DefaultPortalCache) Get(ctx context.Context, name string) (*Portal, error)

type DefaultStatementCache 

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

func (*DefaultStatementCache) Get 

func (cache *DefaultStatementCache) Get(ctx context.Context, name string) (*Statement, error)

Get attempts to get the prepared statement for the given name. An error is returned when no statement has been found.

func (*DefaultStatementCache) Set 

func (cache *DefaultStatementCache) Set(ctx context.Context, name string, stmt *PreparedStatement) error

Set attempts to bind the given statement to the given name. Any previously defined statement is overridden.

type FormatCode 

type FormatCode int16

FormatCode represents the encoding format of a given column 

const (
	// TextFormat is the default, text format.
	TextFormat FormatCode = 0
	// BinaryFormat is an alternative, binary, encoding.
	BinaryFormat FormatCode = 1
)

type OptionFn 

type OptionFn func(*Server) error

OptionFn options pattern used to define and set options for the given PostgreSQL server.

func CloseConn 

func CloseConn(fn CloseFn) OptionFn

CloseConn sets the close connection handle inside the given server instance.

func ExtendTypes 

func ExtendTypes(fn func(*pgtype.Map)) OptionFn

ExtendTypes provides the ability to extend the underlying connection types. Types registered inside the given github.com/jackc/pgx/v5/pgtype.Map are registered to all incoming connections.

func GlobalParameters 

func GlobalParameters(params Parameters) OptionFn

GlobalParameters sets the server parameters which are send back to the front-end (client) once a handshake has been established.

func Logger 

func Logger(logger *slog.Logger) OptionFn

Logger sets the given slog.Logger as the logger for the given server.

func MessageBufferSize 

func MessageBufferSize(size int) OptionFn

MessageBufferSize sets the message buffer size which is allocated once a new connection gets constructed. If a negative value or zero value is provided is the default message buffer size used.

func Portals 

func Portals(handler func() PortalCache) OptionFn

Portals sets the portals cache used to cache statements for later use. By default DefaultPortalCache is used.

func SessionAuthStrategy 

func SessionAuthStrategy(fn AuthStrategy) OptionFn

SessionAuthStrategy sets the given authentication strategy within the given server. The authentication strategy is called when a handshake is initiated.

func SessionMiddleware 

func SessionMiddleware(fn SessionHandler) OptionFn

SessionMiddleware sets the given session handler within the underlying server. The session handler is called when a new connection is opened and authenticated allowing for additional metadata to be wrapped around the connection context.

func Statements 

func Statements(handler func() StatementCache) OptionFn

Statements sets the statement cache used to cache statements for later use. By default DefaultStatementCache is used.

func TLSConfig 

func TLSConfig(config *tls.Config) OptionFn

TLSConfig sets the given TLS config to be used to initialize a secure connection between the front-end (client) and back-end (server).

func TerminateConn 

func TerminateConn(fn CloseFn) OptionFn

TerminateConn sets the terminate connection handle inside the given server instance.

func Version 

func Version(version string) OptionFn

Version sets the PostgreSQL version for the server which is send back to the front-end (client) once a handshake has been established.

type Parameter 

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

func NewParameter 

func NewParameter(types *pgtype.Map, format FormatCode, value []byte) Parameter

func (Parameter) Format 

func (p Parameter) Format() FormatCode

func (Parameter) Scan 

func (p Parameter) Scan(oid uint32) (any, error)

func (Parameter) Value 

func (p Parameter) Value() []byte

type ParameterStatus 

type ParameterStatus string

ParameterStatus represents a metadata key that could be defined inside a server/client metadata definition. 

const (
	ParamServerEncoding       ParameterStatus = "server_encoding"
	ParamClientEncoding       ParameterStatus = "client_encoding"
	ParamIsSuperuser          ParameterStatus = "is_superuser"
	ParamSessionAuthorization ParameterStatus = "session_authorization"
	ParamApplicationName      ParameterStatus = "application_name"
	ParamDatabase             ParameterStatus = "database"
	ParamUsername             ParameterStatus = "user"
	ParamServerVersion        ParameterStatus = "server_version"
)

At present there is a hard-wired set of parameters for which ParameterStatus will be generated. https://www.postgresql.org/docs/13/protocol-flow.html#PROTOCOL-ASYNC

type Parameters 

type Parameters map[ParameterStatus]string

Parameters represents a parameters collection of parameter status keys and their values.

func ClientParameters 

func ClientParameters(ctx context.Context) Parameters

ClientParameters returns the connection parameters if it has been set inside the given context.

func ServerParameters 

func ServerParameters(ctx context.Context) Parameters

ServerParameters returns the connection parameters if it has been set inside the given context.

type ParseFn 

type ParseFn func(ctx context.Context, query string) (PreparedStatements, error)

ParseFn parses the given query and returns a prepared statement which could be used to execute at a later point in time.

type Portal 

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

type PortalCache 

type PortalCache interface {
	Bind(ctx context.Context, name string, statement *Statement, parameters []Parameter, columns []FormatCode) error
	Get(ctx context.Context, name string) (*Portal, error)
	Execute(ctx context.Context, name string, reader *buffer.Reader, writer *buffer.Writer) error
}

PortalCache represents a cache which could be used to bind and execute prepared statements with parameters.

func DefaultPortalCacheFn 

func DefaultPortalCacheFn() PortalCache

type PreparedOptionFn 

type PreparedOptionFn func(*PreparedStatement)

PreparedOptionFn options pattern used to define options while preparing a new statement.

func WithColumns 

func WithColumns(columns Columns) PreparedOptionFn

WithColumns sets the given columns as the columns which are returned by the prepared statement.

func WithParameters 

func WithParameters(parameters []oid.Oid) PreparedOptionFn

WithParameters sets the given parameters as the parameters which are expected by the prepared statement.

type PreparedStatement 

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

func NewStatement 

func NewStatement(fn PreparedStatementFn, options ...PreparedOptionFn) *PreparedStatement

NewStatement constructs a new prepared statement for the given function.

type PreparedStatementFn 

type PreparedStatementFn func(ctx context.Context, writer DataWriter, parameters []Parameter) error

PreparedStatementFn represents a query of which a statement has been prepared. The statement could be executed at any point in time with the given arguments and data writer.

type PreparedStatements 

type PreparedStatements []*PreparedStatement

func Prepared 

func Prepared(stmts ...*PreparedStatement) PreparedStatements

Prepared is a small wrapper function returning a list of prepared statements. More then one prepared statement could be returned within the simple query protocol. An error is returned when more than one prepared statement is returned in the extended query protocol.

type Scanner 

type Scanner func(value []byte) (any, error)

Scanner is a function that scans a byte slice and returns the value as an any

func NewScanner 

func NewScanner(tm *pgtype.Map, column Column, format FormatCode) (Scanner, error)

NewScanner creates a new scanner that scans a byte slice and returns the value as an any. The scanner uses the given map to decode the value and the given type to determine the format of the data that is scanned.

type Server 

type Server struct {
	Auth            AuthStrategy
	BufferedMsgSize int
	Parameters      Parameters
	TLSConfig       *tls.Config

	Session       SessionHandler
	Statements    func() StatementCache
	Portals       func() PortalCache
	CloseConn     CloseFn
	TerminateConn CloseFn
	Version       string
	// contains filtered or unexported fields
}

Server contains options for listening to an address.

func NewServer 

func NewServer(parse ParseFn, options ...OptionFn) (*Server, error)

NewServer constructs a new Postgres server using the given address and server options.

func (*Server) Close 

func (srv *Server) Close() error

Close gracefully closes the underlaying Postgres server.

func (*Server) Handshake 

func (srv *Server) Handshake(conn net.Conn) (_ net.Conn, version types.Version, reader *buffer.Reader, err error)

Handshake performs the connection handshake and returns the connection version and a buffered reader to read incoming messages send by the client.

func (*Server) ListenAndServe 

func (srv *Server) ListenAndServe(address string) error

ListenAndServe opens a new Postgres server on the preconfigured address and starts accepting and serving incoming client connections.

func (*Server) Serve 

func (srv *Server) Serve(listener net.Listener) error

Serve accepts and serves incoming Postgres client connections using the preconfigured configurations. The given listener will be closed once the server is gracefully closed.

type Session 

type Session struct {
	*Server
	Statements StatementCache
	Portals    PortalCache
}

type SessionHandler 

type SessionHandler func(ctx context.Context) (context.Context, error)

SessionHandler represents a wrapper function defining the state of a single session. This function allows the user to wrap additional metadata around the shared context.

type Statement 

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

type StatementCache 

type StatementCache interface {
	// Set attempts to bind the given statement to the given name. Any
	// previously defined statement is overridden.
	Set(ctx context.Context, name string, fn *PreparedStatement) error
	// Get attempts to get the prepared statement for the given name. An error
	// is returned when no statement has been found.
	Get(ctx context.Context, name string) (*Statement, error)
}

StatementCache represents a cache which could be used to store and retrieve prepared statements bound to a name.

func DefaultStatementCacheFn 

func DefaultStatementCacheFn() StatementCache

Source Files

View all Source files

Directories

Path Synopsis
examples
copy
error
session
simple
tls
pkg
buffer
mock
types

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.