inter

Documentation

Index

• Constants
• type AgentConnector
• type ConfigurationProvider
• type ConnectionManager
• type Connector
• type ConnectorInfo
• type ConnectorMessage
• type DDLResolver
• type Election
• type FlagApp
• type Framework
• type JsonLineOutput
• type Message
• type Pluggable
• type PluginType
  • func (t PluginType) String() string
• type ProtocolConstructor
• type RawNATSConnector
• type RequestSigner
• type RequestSignerConfig
• type SecurityProvider
• type SecurityTechnology

Constants

const (
	JsonLineErrorKind      = "error"
	JsonLineResultKind     = "result"
	JsonLineSummariesKind  = "summaries"
	JsonLineStatsKind      = "stats"
	JsonLineDiscoveredKind = "discovery"
)

const (
	MessageMessageType        string = "message"
	RequestMessageType        string = "request"
	DirectRequestMessageType  string = "direct_request"
	ServiceRequestMessageType string = "service_request"
	ReplyMessageType          string = "reply"
)

Types

type AgentConnector

type AgentConnector interface {
	ConnectedServer() string
	ConnectionOptions() nats.Options
	ConnectionStats() nats.Statistics
	QueueSubscribe(ctx context.Context, name string, subject string, group string, output chan ConnectorMessage) error
	Unsubscribe(name string) error
	AgentBroadcastTarget(collective string, agent string) string
	ServiceBroadcastTarget(collective string, agent string) string
}

AgentConnector provides the minimal Connector features for subscribing and unsubscribing agents

type ConfigurationProvider

type ConfigurationProvider interface {
	Configuration() *config.Config
}

ConfigurationProvider provides runtime Choria configuration

type ConnectionManager

type ConnectionManager interface {
	NewConnector(ctx context.Context, servers func() (srvcache.Servers, error), name string, logger *log.Entry) (conn Connector, err error)
}

ConnectionManager is capable of being a factory for connection, mcollective.Choria is one

type Connector

type Connector interface {
	AgentBroadcastTarget(collective string, agent string) string
	ChanQueueSubscribe(name string, subject string, group string, capacity int) (chan ConnectorMessage, error)
	Close()
	Connect(ctx context.Context) (err error)
	ConnectedServer() string
	ConnectionOptions() nats.Options
	ConnectionStats() nats.Statistics
	InboxPrefix() string
	IsConnected() bool
	Nats() *nats.Conn
	NodeDirectedTarget(collective string, identity string) string
	Publish(msg Message) error
	PublishRaw(target string, data []byte) error
	PublishRawMsg(msg *nats.Msg) error
	QueueSubscribe(ctx context.Context, name string, subject string, group string, output chan ConnectorMessage) error
	ReplyTarget(msg Message) (string, error)
	RequestRawMsgWithContext(ctx context.Context, msg *nats.Msg) (*nats.Msg, error)
	ServiceBroadcastTarget(collective string, agent string) string
	Unsubscribe(name string) error
}

Connector is the interface a connector must implement to be valid be it NATS, Stomp, Testing etc

type ConnectorInfo

type ConnectorInfo interface {
	ConnectedServer() string
	ConnectionOptions() nats.Options
	ConnectionStats() nats.Statistics
	InboxPrefix() string
}

ConnectorInfo provides information about the active connection without giving access to the connection

type ConnectorMessage

type ConnectorMessage interface {
	Subject() string
	Reply() string
	Data() []byte

	// Msg is the middleware specific message like *nats.Msg
	Msg() any
}

ConnectorMessage is received from middleware

type DDLResolver

type DDLResolver interface {
	// String indicates which resolver is in use
	String() string

	// DDL resolves a DDL and unmarshal it into target, which should be a pointer to a DDL type
	DDL(ctx context.Context, kind string, name string, target any, fw Framework) error

	// DDLBytes resolves a DDL and return the bytes
	DDLBytes(ctx context.Context, kind string, name string, fw Framework) ([]byte, error)

	// DDLNames returns all the DDL names this resolver knows about
	DDLNames(ctx context.Context, kind string, w Framework) ([]string, error)
}

DDLResolver allows DDLs to be found on the local system or via remote registries

type Election

type Election interface {
	// Start starts the election, interrupted by context. Blocks until stopped.
	Start(ctx context.Context) error
	// Stop stops the election
	Stop()
	// IsLeader determines if we are currently the leader
	IsLeader() bool
}

Election is a NATS Key-Value Store based Leader Election system

type FlagApp

type FlagApp interface {
	Flag(name, help string) *fisk.FlagClause
	Command(name, help string) *fisk.CmdClause
}

FlagApp is a fisk command or app

type Framework

type Framework interface {
	ProtocolConstructor
	ConfigurationProvider
	ConnectionManager

	BuildInfo() *build.Info
	CallerID() string
	Certname() string
	ClientTLSConfig() (*tls.Config, error)
	Colorize(c string, format string, a ...any) string
	ConfigureProvisioning(ctx context.Context)
	DDLResolvers() ([]DDLResolver, error)
	DisableTLSVerify() bool
	Enroll(ctx context.Context, wait time.Duration, cb func(digest string, try int)) error
	FacterCmd() string
	FacterDomain() (string, error)
	FacterFQDN() (string, error)
	FacterStringFact(fact string) (string, error)
	FederationCollectives() (collectives []string)
	FederationMiddlewareServers() (servers srvcache.Servers, err error)
	Getuid() int
	GovernorSubject(name string) string
	HTTPClient(secure bool) (*http.Client, error)
	HasCollective(collective string) bool
	IsFederated() (result bool)
	KV(ctx context.Context, conn Connector, bucket string, create bool, opts ...kv.Option) (nats.KeyValue, error)
	KVWithConn(ctx context.Context, conn Connector, bucket string, create bool, opts ...kv.Option) (nats.KeyValue, Connector, error)
	Logger(component string) *logrus.Entry
	MiddlewareServers() (servers srvcache.Servers, err error)
	NetworkBrokerPeers() (servers srvcache.Servers, err error)
	NewElection(ctx context.Context, conn Connector, name string, imported bool, opts ...election.Option) (Election, error)
	NewElectionWithConn(ctx context.Context, conn Connector, name string, imported bool, opts ...election.Option) (Election, Connector, error)
	NewGovernor(ctx context.Context, name string, conn Connector, opts ...governor.Option) (governor.Governor, Connector, error)
	NewGovernorManager(ctx context.Context, name string, limit uint64, maxAge time.Duration, replicas uint, update bool, conn Connector, opts ...governor.Option) (governor.Manager, Connector, error)
	OverrideCertname() string
	PQLQuery(query string) ([]byte, error)
	PQLQueryCertNames(query string) ([]string, error)
	ProgressWidth() int
	PrometheusTextFileDir() string
	ProvisionMode() bool
	ProvisioningServers(ctx context.Context) (srvcache.Servers, error)
	PublicCert() (*x509.Certificate, error)
	PuppetAIOCmd(command string, def string) string
	PuppetDBServers() (servers srvcache.Servers, err error)
	PuppetSetting(setting string) (string, error)
	QuerySrvRecords(records []string) (srvcache.Servers, error)
	SetLogWriter(out io.Writer)
	SetLogger(logger *logrus.Logger)
	SetupLogging(debug bool) (err error)
	SignerSeedFile() (f string, err error)
	SignerToken() (token string, exp time.Time, err error)
	SignerTokenFile() (f string, err error)
	SupportsProvisioning() bool
	TLSConfig() (*tls.Config, error)
	TrySrvLookup(names []string, defaultSrv srvcache.Server) (srvcache.Server, error)
	UniqueID() string
	UniqueIDFromUnverifiedToken() (id string, uid string, exp time.Time, token string, err error)
	ValidateSecurity() (errors []string, ok bool)
}

type JsonLineOutput

type JsonLineOutput struct {
	Kind             string          `json:"k"`
	ProtocolReply    json.RawMessage `json:"pr,omitempty"`
	RPCReply         json.RawMessage `json:"rr,omitempty"`
	Aggregates       json.RawMessage `json:"agg,omitempty"`
	Stats            json.RawMessage `json:"stat,omitempty"`
	Discovered       int             `json:"count,omitempty"`
	DiscoverySeconds float64         `json:"dt,omitempty"`
	DiscoveryMethod  string          `json:"dm,omitempty"`
	Error            string          `json:"err,omitempty"`
}

JsonLineOutput is a line format for json lines output from the cli

type Message

type Message interface {
	Agent() string
	Base64Payload() string
	CacheTransport()
	CallerID() string
	Collective() string
	CustomTarget() string
	DiscoveredHosts() []string
	ExpectedMessageID() string
	Filter() *protocol.Filter
	IsCachedTransport() bool
	NotifyPublish()
	OnPublish(func())
	Payload() []byte
	ProtocolVersion() protocol.ProtocolVersion
	ReplyTo() string
	Request() Message
	RequestID() string
	SenderID() string
	SetBase64Payload(payload []byte) error
	SetCollective(string) error
	SetCustomTarget(string)
	SetDiscoveredHosts(hosts []string)
	SetExpectedMsgID(id string) error
	SetFilter(*protocol.Filter)
	SetPayload([]byte)
	SetProtocolVersion(protocol.ProtocolVersion)
	SetReplyTo(string) error
	SetTTL(int)
	SetType(string) error
	String() string
	TTL() int
	TimeStamp() time.Time
	Transport(ctx context.Context) (protocol.TransportMessage, error)
	Type() string
	Validate() (bool, error)
	ValidateTTL() bool
	ReplyTarget() string
}

Message is a message that is transportable over the Choria Protocol

type Pluggable

type Pluggable interface {
	// PluginInstance is any structure that implements the plugin, should be right type for the kind of plugin
	PluginInstance() any

	// PluginName is a human friendly name for the plugin
	PluginName() string

	// PluginType is the type of the plugin, to match plugin.Type
	PluginType() PluginType

	// PluginVersion is the version of the plugin
	PluginVersion() string
}

Pluggable is a Choria Plugin

type PluginType

type PluginType int

PluginType are types of choria plugin

const (
	// UnknownPlugin is a unknown plugin type
	UnknownPlugin PluginType = iota

	// AgentProviderPlugin is a plugin that provide types of agents to Choria
	AgentProviderPlugin

	// AgentPlugin is a type of agent
	AgentPlugin

	// ProvisionTargetResolverPlugin is a plugin that helps provisioning mode Choria find its broker
	ProvisionTargetResolverPlugin

	// ConfigMutatorPlugin is a plugin that can dynamically adjust
	// configuration based on local site conditions
	ConfigMutatorPlugin

	// MachineWatcherPlugin is a plugin that adds a Autonomous Agent Watcher
	MachineWatcherPlugin

	// DataPlugin is a plugin that provides data to choria
	DataPlugin

	// MachinePlugin is an autonomous agent
	MachinePlugin
)

func (PluginType) String

func (t PluginType) String() string

type ProtocolConstructor

type ProtocolConstructor interface {
	NewMessage(payload []byte, agent string, collective string, msgType string, request Message) (msg Message, err error)
	NewMessageFromRequest(req protocol.Request, replyto string) (Message, error)
	NewReply(request protocol.Request) (reply protocol.Reply, err error)
	NewReplyFromMessage(version protocol.ProtocolVersion, msg Message) (rep protocol.Reply, err error)
	NewReplyFromSecureReply(sr protocol.SecureReply) (reply protocol.Reply, err error)
	NewReplyFromTransportJSON(payload []byte, skipvalidate bool) (msg protocol.Reply, err error)
	NewReplyTransportForMessage(msg Message, request protocol.Request) (protocol.TransportMessage, error)
	NewRequest(version protocol.ProtocolVersion, agent string, senderid string, callerid string, ttl int, requestid string, collective string) (request protocol.Request, err error)
	NewRequestFromMessage(version protocol.ProtocolVersion, msg Message) (req protocol.Request, err error)
	NewRequestFromSecureRequest(sr protocol.SecureRequest) (request protocol.Request, err error)
	NewRequestFromTransportJSON(payload []byte, skipvalidate bool) (msg protocol.Request, err error)
	NewRequestID() (string, error)
	NewRequestMessageFromTransportJSON(payload []byte) (Message, error)
	NewRequestTransportForMessage(ctx context.Context, msg Message, version protocol.ProtocolVersion) (protocol.TransportMessage, error)
	NewSecureReply(reply protocol.Reply) (secure protocol.SecureReply, err error)
	NewSecureReplyFromTransport(message protocol.TransportMessage, skipvalidate bool) (secure protocol.SecureReply, err error)
	NewSecureRequest(ctx context.Context, request protocol.Request) (secure protocol.SecureRequest, err error)
	NewSecureRequestFromTransport(message protocol.TransportMessage, skipvalidate bool) (secure protocol.SecureRequest, err error)
	NewTransportForSecureReply(reply protocol.SecureReply) (message protocol.TransportMessage, err error)
	NewTransportForSecureRequest(request protocol.SecureRequest) (message protocol.TransportMessage, err error)
	NewTransportFromJSON(data []byte) (message protocol.TransportMessage, err error)
	NewTransportMessage(version protocol.ProtocolVersion) (message protocol.TransportMessage, err error)
	RequestProtocol() protocol.ProtocolVersion
}

type RawNATSConnector

type RawNATSConnector interface {
	PublishRaw(target string, data []byte) error
	PublishRawMsg(msg *nats.Msg) error
	RequestRawMsgWithContext(ctx context.Context, msg *nats.Msg) (*nats.Msg, error)
}

RawNATSConnector sends NATS messages directly

type RequestSigner

type RequestSigner interface {
	// Sign signs request payload
	Sign(ctx context.Context, request []byte, cfg RequestSignerConfig) ([]byte, error)

	// Kind is the name of the provider
	Kind() string
}

RequestSigner signs request, typically remote signers over HTTP or Choria RPC

type RequestSignerConfig

type RequestSignerConfig interface {
	RemoteSignerURL() (*url.URL, error)
	RemoteSignerToken() ([]byte, error)
	RemoteSignerSeedFile() (string, error)
}

RequestSignerConfig configures RequestSigner

type SecurityProvider

type SecurityProvider interface {
	// Provider reports the name of the current security provider
	Provider() string

	// BackingTechnology indicates what the technology this provider is based on
	BackingTechnology() SecurityTechnology

	// Validate that the security provider is functional
	Validate() ([]string, bool)

	// Identity from the active certificates
	Identity() string

	// CallerName is a valid choria like foo=bar style caller name from the identity
	CallerName() string

	// CallerIdentity extracts the Identity from a caller name
	CallerIdentity(caller string) (string, error)

	// SignBytes signs bytes using the current active certificate
	SignBytes(b []byte) (signature []byte, err error)

	// VerifySignatureBytes verifies that dat signature was made using pubcert
	VerifySignatureBytes(dat []byte, sig []byte, public ...[]byte) (should bool, signer string)

	// RemoteSignRequest signs a choria request using a remote signer and returns a secure request
	RemoteSignRequest(ctx context.Context, str []byte) (signed []byte, err error)

	// IsRemoteSigning reports if the security provider is signing using a remote
	IsRemoteSigning() bool

	// ChecksumBytes produce a crypto checksum for data
	ChecksumBytes(data []byte) []byte

	// TLSConfig produce a tls.Config for the current identity using its certificates etc
	TLSConfig() (*tls.Config, error)

	// ClientTLSConfig produces a tls.Config specifically for clients
	ClientTLSConfig() (*tls.Config, error)

	// SSLContext produce a http.Transport for the current identity using its certificates etc
	SSLContext() (*http.Transport, error)

	// HTTPClient creates a standard HTTP client with optional security, it will
	// be set to use the CA and client certs for auth.
	HTTPClient(secure bool) (*http.Client, error)

	// PublicCert is the parsed public certificate
	PublicCert() (*x509.Certificate, error)

	// PublicCertBytes retrieves pem data in textual form for the public certificate of the current identity
	PublicCertBytes() ([]byte, error)

	// TokenBytes gives access to the users JWT token if any
	TokenBytes() ([]byte, error)

	// ShouldSignReplies indicates that the protocol should sign replies from servers, only supported in v2 protocol
	ShouldSignReplies() bool

	// ShouldAllowCaller validates the identity, the public data like certificate or JWT and checks
	// against allowed lists and is privileged user aware
	ShouldAllowCaller(name string, callers ...[]byte) (privileged bool, err error)

	// Enroll creates a new cert with the active identity and attempt to enroll it with the security system
	// if there's a process of waiting for the certificate to be signed for example this should wait
	// no more than wait.  cb gets called on every attempt to download a cert with the attempt number
	// as argument
	Enroll(ctx context.Context, wait time.Duration, cb func(digest string, try int)) error
}

SecurityProvider provides a security plugin for the choria subsystem

type SecurityTechnology

type SecurityTechnology int

SecurityTechnology indicates the technology a security provider is based on

const (
	SecurityTechnologyUnknown    SecurityTechnology = 0
	SecurityTechnologyX509       SecurityTechnology = 1
	SecurityTechnologyED25519JWT SecurityTechnology = 2
)