peer

package
v2.0.0-rc.2 Latest Latest
Warning

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

 Go to latest Published: Aug 30, 2026 License: Apache-2.0 Imports: 28 Imported by: 0

Documentation

Overview

Package peer runs the bounded single-host AGTP discovery product profile. Three Go nodes exchange Presence and ANS deltas over real loopback TLS ports, authenticate each peer action with ASB, route DHT lookups by XOR distance, and persist state needed for restart convergence.

It is not a general AGTP wire implementation or a cross-host DHT. Peer trust, Manager-issued grants, certificates, and binding keys are verifier-local deployment configuration.

Index

Constants

View Source 
const (
	DefaultMaxRecords        = 100
	DefaultMaxTombstones     = 200
	DefaultMaxPeers          = 2
	DefaultMaxRequestBytes   = 1 << 20
	DefaultAuditMaxBytes     = 10 << 20
	DefaultRequestsPerSecond = 20
	DefaultRequestBurst      = 40
)
View Source 
const (
	ProtocolVersion = 1
	NoncePath       = "/v1/peer/nonce"
	ReplicatePath   = "/v1/peer/replicate"
	FindNodePath    = "/v1/peer/find-node"
	HealthPath      = "/healthz"
	MetricsPath     = "/metrics"

	IdentityGrantHeader  = "ASB-Identity-Grant"
	SessionBindingHeader = "ASB-Session-Binding"
	VerifierNonceHeader  = "ASB-Verifier-Nonce"
)

Variables

View Source 
var (
	ErrInvalidProtocol = errors.New("agtp discovery peer: invalid protocol message")
	ErrUnauthorized    = errors.New("agtp discovery peer: unauthorized peer")
	ErrRateLimited     = errors.New("agtp discovery peer: rate limited")
)
View Source 
var (
	ErrCorruptState     = errors.New("agtp discovery peer: corrupt persistent state")
	ErrUnsupportedState = errors.New("agtp discovery peer: unsupported persistent state version")
)

Functions

This section is empty.

Types

type Action 

type Action string

Action is a separately authorized peer operation. 

const (
	ActionReplicate Action = "replicate"
	ActionFindNode  Action = "find-node"
)

type AuditEvent 

type AuditEvent struct {
	Time   time.Time `json:"time"`
	NodeID string    `json:"node_id"`
	PeerID string    `json:"peer_id,omitempty"`
	Action string    `json:"action"`
	Result string    `json:"result"`
	Reason string    `json:"reason,omitempty"`
}

AuditEvent contains metadata only; tokens and Presence payloads are never written to the audit log.

type AuditLog 

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

AuditLog is a synchronous JSONL audit sink.

func NewAuditLog 

func NewAuditLog(path string, maxBytes int64) (*AuditLog, error)

NewAuditLog opens a mode-0600 append-only audit file.

func (*AuditLog) Close 

func (l *AuditLog) Close() error

Close flushes the audit file.

func (*AuditLog) Write 

func (l *AuditLog) Write(event AuditEvent) error

Write appends and syncs one event.

type Client 

type Client struct {
	AgentID          string
	Issuer           string
	KeyID            string
	PrivateKey       ed25519.PrivateKey
	TLSConfig        *tls.Config
	Remotes          map[string]RemoteAuthorization
	Timeout          time.Duration
	MaxResponseBytes int64
	Now              func() time.Time
}

Client performs one mTLS+ASB peer action per TLS connection.

func (*Client) FindNode 

func (c *Client) FindNode(ctx context.Context, peer discovery.NodeInfo, request FindNodeRequest) (FindNodeResponse, error)

FindNode performs one authenticated DHT lookup hop.

func (*Client) Replicate 

func (c *Client) Replicate(ctx context.Context, peer discovery.NodeInfo, request ReplicateRequest) (ReplicateResponse, error)

Replicate exchanges Presence and ANS deltas with one peer.

type Config 

type Config struct {
	Info               discovery.NodeInfo
	ListenAddress      string
	TLSConfig          *tls.Config
	Directory          *PeerDirectory
	Client             *Client
	StatePath          string
	AuditPath          string
	AuditMaxBytes      int64
	GossipInterval     time.Duration
	RequestTimeout     time.Duration
	TombstoneRetention time.Duration
	MaxRecords         int
	MaxTombstones      int
	MaxPeers           int
	MaxRequestBytes    int64
	RequestsPerSecond  float64
	RequestBurst       int
	Now                func() time.Time
	ErrorLog           *log.Logger
}

Config fixes the bounded single-host product profile.

type FileReplayCache 

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

FileReplayCache is a single-host durable replay cache. It is intentionally small and synchronous for the three-node product scope.

func NewFileReplayCache 

func NewFileReplayCache(path string, now func() time.Time) (*FileReplayCache, error)

NewFileReplayCache loads or creates a durable replay cache.

func (*FileReplayCache) MarkUsed 

func (c *FileReplayCache) MarkUsed(key string, expiresAt time.Time) error

MarkUsed commits replay state before returning success.

type FindNodeRequest 

type FindNodeRequest struct {
	Protocol int                `json:"protocol"`
	Sender   discovery.NodeInfo `json:"sender"`
	Target   string             `json:"target"`
	Count    int                `json:"count"`
}

FindNodeRequest performs one authenticated DHT lookup hop.

type FindNodeResponse 

type FindNodeResponse struct {
	Protocol int                  `json:"protocol"`
	Peers    []discovery.NodeInfo `json:"peers"`
}

FindNodeResponse contains the receiver's nearest trusted peers.

type Metrics 

type Metrics struct {
	PeerRequests      atomic.Uint64
	AuthRejected      atomic.Uint64
	RateLimited       atomic.Uint64
	GossipSuccess     atomic.Uint64
	GossipFailure     atomic.Uint64
	PersistenceErrors atomic.Uint64
	AuditErrors       atomic.Uint64
}

Metrics contains dependency-free counters for the bounded local product.

type Node 

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

Node is one persistent Presence, DHT, ANS, and gossip peer.

func NewNode 

func NewNode(config Config) (*Node, error)

NewNode loads durable state and constructs one peer. It does not open a port until Start succeeds.

func (*Node) AddPeer 

func (n *Node) AddPeer(peer discovery.NodeInfo) error

AddPeer adds a verifier-configured peer and commits routing state.

func (*Node) Announce 

func (n *Node) Announce(record discovery.Record) (bool, error)

Announce applies and durably commits one Presence update.

func (*Node) Counts 

func (n *Node) Counts() (records, tombstones, peers int)

Counts reports bounded live state.

func (*Node) Deregister 

func (n *Node) Deregister(name string, version uint64) (bool, error)

Deregister removes an ANS binding and commits its Presence withdrawal.

func (*Node) Discover 

func (n *Node) Discover(ctx context.Context, query discovery.Query, requester discovery.Requester) (discovery.Response, error)

Discover queries local live Presence state.

func (*Node) Errors 

func (n *Node) Errors() <-chan error

Errors reports asynchronous serving failures.

func (*Node) GossipOnce 

func (n *Node) GossipOnce(ctx context.Context) error

GossipOnce exchanges bounded Presence and ANS deltas with all reachable configured peers.

func (*Node) Info 

func (n *Node) Info() discovery.NodeInfo

Info returns the current Node-ID and bound endpoint.

func (*Node) Locate 

func (n *Node) Locate(ctx context.Context, target string, count int) ([]discovery.NodeInfo, error)

Locate performs a real multi-peer authenticated DHT lookup.

func (*Node) Register 

func (n *Node) Register(binding discovery.NameBinding) (bool, error)

Register applies and durably commits one ANS binding.

func (*Node) Resolve 

func (n *Node) Resolve(name string) (discovery.NameBinding, bool)

Resolve resolves one live ANS name.

func (*Node) SetPeerBlocked 

func (n *Node) SetPeerBlocked(peerID string, blocked bool)

SetPeerBlocked is an operator/test partition control. It does not alter durable routing state.

func (*Node) Start 

func (n *Node) Start() error

Start opens the configured real TCP port and starts periodic gossip.

func (*Node) Stop 

func (n *Node) Stop(ctx context.Context) error

Stop stops gossip, drains HTTP requests, and commits a final snapshot.

func (*Node) Withdraw 

func (n *Node) Withdraw(agentID string, version uint64) (bool, error)

Withdraw applies and durably commits a retained deletion marker.

type PeerDirectory 

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

PeerDirectory is verifier-local policy. Network input cannot add entries.

func NewPeerDirectory 

func NewPeerDirectory() *PeerDirectory

NewPeerDirectory creates an empty trust directory.

func (*PeerDirectory) Add 

func (d *PeerDirectory) Add(certificate *x509.Certificate, identity PeerIdentity) error

Add binds a peer certificate to its expected Node-ID.

func (*PeerDirectory) LookupCertificate 

func (d *PeerDirectory) LookupCertificate(certificate *x509.Certificate) (PeerIdentity, bool)

LookupCertificate returns local policy for an authenticated TLS leaf.

func (*PeerDirectory) LookupNode 

func (d *PeerDirectory) LookupNode(nodeID string) (PeerIdentity, bool)

LookupNode returns local policy for one configured Node-ID.

type PeerIdentity 

type PeerIdentity struct {
	Node    discovery.NodeInfo
	Profile production.SoftwareOnlyProfile
}

PeerIdentity binds a configured DHT Node-ID and endpoint to one mTLS client certificate and one ASB verification profile.

type PersistentState 

type PersistentState struct {
	Version  int                     `json:"version"`
	SavedAt  time.Time               `json:"saved_at"`
	Presence discovery.Delta         `json:"presence"`
	Names    []discovery.NameBinding `json:"names"`
	Peers    []discovery.NodeInfo    `json:"peers"`
}

PersistentState is the durable state owned by one discovery node.

type RemoteAuthorization 

type RemoteAuthorization struct {
	Audience                string
	ServerName              string
	ServerCertificateSHA256 string
	Grants                  map[Action]string
}

RemoteAuthorization contains verifier-specific values for one destination. Identity Grants are issued out of band by the Manager and may be reused until expiry; each request receives a fresh verifier nonce and binding proof.

type ReplicateRequest 

type ReplicateRequest struct {
	Protocol   int                     `json:"protocol"`
	Sender     discovery.NodeInfo      `json:"sender"`
	Digest     discovery.Digest        `json:"digest"`
	Delta      discovery.Delta         `json:"delta"`
	NameDigest map[string]uint64       `json:"name_digest"`
	Names      []discovery.NameBinding `json:"names"`
}

ReplicateRequest exchanges Presence and ANS digests and deltas.

type ReplicateResponse 

type ReplicateResponse struct {
	Protocol   int                     `json:"protocol"`
	Digest     discovery.Digest        `json:"digest"`
	Delta      discovery.Delta         `json:"delta"`
	NameDigest map[string]uint64       `json:"name_digest"`
	Names      []discovery.NameBinding `json:"names"`
}

ReplicateResponse returns state newer than the requester's digest.

type StateStore 

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

StateStore persists a checksummed JSON snapshot with atomic replacement.

func NewStateStore 

func NewStateStore(path string) (*StateStore, error)

NewStateStore creates a snapshot store at path.

func (*StateStore) Load 

func (s *StateStore) Load() (PersistentState, bool, error)

Load returns false when no snapshot exists yet.

func (*StateStore) Save 

func (s *StateStore) Save(state PersistentState) error

Save atomically replaces the durable snapshot.

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.