replica

package
v0.0.74 Latest Latest
Warning

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

 Go to latest Published: May 12, 2026 License: AGPL-3.0 Imports: 57 Imported by: 0

README

Pigeonhole Storage Replica

This section of code implements the storage replica server from our unpublished paper, Echomix: a Strong Anonymity System with Messaging which you can read here: https://arxiv.org/abs/2501.02933

This storage server operates "outside the mixnet". The "courier" services which operates on the service nodes in the mix network, is responsible for proxying queries and replies to and from the storage replicas. The replicas perform their replication and communication with the courier services outside of the mixnet, however they do make use of our PQ Noise based transport protocol for all of this communication. Detailed design docs forthcoming.

dependencies

Note that this component might be slightly more tricky to build than the rest of Katzenpost because of the dependency on a slightly older version of RocksDB in order to maintain compatibility with the golang bindings.

building / running

Install the RocksDB dependencies on your host system. Run these commands as root:

apt install cmake

cd /tmp && \
    git clone https://github.com/gflags/gflags.git && \
    cd gflags && \
    mkdir build && \
    cd build && \
    cmake -DBUILD_SHARED_LIBS=1 -DGFLAGS_INSTALL_SHARED_LIBS=1 .. && \
    make install && \
    cd /tmp && \
    rm -R /tmp/gflags/

cd /tmp && \
    git clone https://github.com/facebook/rocksdb.git && \
    cd rocksdb && \
    git checkout v10.2.1 && \
    make shared_lib && \
    mkdir -p /usr/local/rocksdb/lib && \
    mkdir -p /usr/local/rocksdb/include && \
    cp librocksdb.so* /usr/local/rocksdb/lib && \
    cp /usr/local/rocksdb/lib/librocksdb.so* /usr/lib/ && \
    cp -r include /usr/local/rocksdb/ && \
    cp -r include/* /usr/include/ && \
    rm -R /tmp/rocksdb/

and then you can run go build as usual:

cd cmd/replica
go build

Debugging (as replica operator)

apt install rocksdb-tools/testing

for i in {1..5}; do echo "Replica $i"; ldb --db=voting_mixnet/replica${i}/replica.db --hex --ignore_unknown_options dump | awk -F '==' '/==/ {print "\t"$1}'; done

Replica 1
        0x3B0B39B05170202479198C5D63A7D8A2A30FF3D17A0C20EBCF4FF124A7D56DBC 
        0xA9F171DCEE449B661CFDDCDAAF82E9808628BF105E2144AA779CFDA30C6B180A 
Replica 2
        0xA9F171DCEE449B661CFDDCDAAF82E9808628BF105E2144AA779CFDA30C6B180A 
Replica 3
Replica 4
        0xA9F171DCEE449B661CFDDCDAAF82E9808628BF105E2144AA779CFDA30C6B180A 
Replica 5
        0x3B0B39B05170202479198C5D63A7D8A2A30FF3D17A0C20EBCF4FF124A7D56DBC 
        0xA9F171DCEE449B661CFDDCDAAF82E9808628BF105E2144AA779CFDA30C6B180A

Documentation

Index

Constants

View Source 
const (
	// GracePeriod is the duration after key expirey that we keep the keys.
	GracePeriod = 3 * time.Hour
)
View Source 
const (
	NumPKIDocsToFetch = 3
)
View Source 
const PKIDocNum = 3
View Source 
const ValidEnvelopeEpochWindow uint64 = 1

ValidEnvelopeEpochWindow is the symmetric tolerance the replica applies when decapsulating an inbound MKEM ciphertext. It MUST match the courier's ValidCourierEnvelopeEpochWindow, so any envelope the courier decided to forward has a matching private key to try here.

A window of 1 means three candidate epochs at any moment: {current-1, current, current+1}. See the Pigeonhole specification section "Epoch tolerance for CourierEnvelope" for the reasoning.

Variables

View Source 
var (
	ErrBoxIDNotFound       = errors.New("Box ID not found")
	ErrBoxAlreadyExists    = errors.New("BoxID already exists, writes are immutable")
	ErrFailedDBRead        = errors.New("Failed to read from database")
	ErrFailedToDeserialize = errors.New("Failed to deserialize data from DB")
	ErrDBClosed            = errors.New("DB is closed")
)
View Source 
var ErrGenerateOnly = errors.New("server: GenerateOnly set")

ErrGenerateOnly is the error returned when the server initialization terminates due to the `GenerateOnly` debug config option.

View Source 
var GitCommit = "unknown"

GitCommit is the git commit hash, set at build time via -ldflags

View Source 
var (
	PublishDeadline = vServer.MixPublishDeadline
)

Functions

func CreateTestConfig 

func CreateTestConfig(t *testing.T, schemes *TestSchemes, geometry *geo.Geometry, dataDir, identifier string, addresses []string) *config.Config

CreateTestConfig creates a standard test configuration

func CreateTestGeometry 

func CreateTestGeometry(schemes *TestSchemes) *geo.Geometry

CreateTestGeometry creates a standard test geometry

func CreateTestGeometryCustom 

func CreateTestGeometryCustom(schemes *TestSchemes, payloadSize, nrHops int) *geo.Geometry

CreateTestGeometryCustom creates a custom test geometry

func CreateTestGeometryECDH 

func CreateTestGeometryECDH(forwardPayloadLength, nrHops int) *geo.Geometry

CreateTestGeometryECDH creates a test geometry using ECDH

func CreateTestPKIDocument 

func CreateTestPKIDocument(t *testing.T, replicas []*pki.ReplicaDescriptor, serviceNodes []*pki.MixDescriptor) *pki.Document

CreateTestPKIDocument creates a test PKI document

func CreateTestTempDir 

func CreateTestTempDir(t *testing.T, prefix string) string

CreateTestTempDir creates a temporary directory for tests

func GenerateTestReplica 

func GenerateTestReplica(t *testing.T, schemes *TestSchemes, index int) *pki.ReplicaDescriptor

GenerateTestReplica creates a test replica descriptor

func StoreTestDocument 

func StoreTestDocument(t *testing.T, pkiWorker *PKIWorker, doc *pki.Document)

StoreTestDocument stores a PKI document in a PKI worker

Types

type Connector 

type Connector struct {
	sync.RWMutex
	worker.Worker
	// contains filtered or unexported fields
}

func (*Connector) CloseAllCh 

func (co *Connector) CloseAllCh() chan interface{}

func (*Connector) ConnectionCount added in v0.0.73 

func (co *Connector) ConnectionCount() int

ConnectionCount returns the number of active outgoing connections.

func (*Connector) DispatchCommand 

func (co *Connector) DispatchCommand(cmd commands.Command, idHash *[32]byte)

func (*Connector) DispatchReplication 

func (co *Connector) DispatchReplication(cmd *commands.ReplicaWrite)

func (*Connector) ForceUpdate 

func (co *Connector) ForceUpdate()

func (*Connector) Halt 

func (co *Connector) Halt()

func (*Connector) OnClosedConn 

func (co *Connector) OnClosedConn(c *outgoingConn)

func (*Connector) QueueForRetry added in v0.0.73 

func (co *Connector) QueueForRetry(cmd commands.Command, idHash [32]byte)

QueueForRetry adds a command to the retry queue when no connection is available. Dedup is by (destination replica, BoxID): re-queuing the same write collapses and bumps attempts; writes for different boxes to the same peer accumulate as distinct entries. Evicts expired entries and, if at capacity, the oldest entry.

func (*Connector) Server 

func (co *Connector) Server() *Server

type EnvelopeKeys 

type EnvelopeKeys struct {
	worker.Worker
	// contains filtered or unexported fields
}

func NewEnvelopeKeys 

func NewEnvelopeKeys(scheme nike.Scheme, log *logging.Logger, datadir string, epoch uint64) (*EnvelopeKeys, error)

func (*EnvelopeKeys) EnsureKey 

func (k *EnvelopeKeys) EnsureKey(replicaEpoch uint64) (*replicaCommon.EnvelopeKey, error)

EnsureKey returns the keypair for replicaEpoch, generating it on demand for the current or a future replica epoch (the PKI publisher's use case). For any replicaEpoch strictly in the past it returns whatever is already cached in memory, or an error — it never fabricates a fresh random keypair for a past epoch, because such a key cannot decrypt ciphertexts that were encrypted to the previously PKI-published public key for that epoch.

func (*EnvelopeKeys) Generate 

func (k *EnvelopeKeys) Generate(replicaEpoch uint64) error

func (*EnvelopeKeys) GetKeypair 

func (k *EnvelopeKeys) GetKeypair(replicaEpoch uint64) (*replicaCommon.EnvelopeKey, error)

func (*EnvelopeKeys) Prune 

func (k *EnvelopeKeys) Prune() bool

type GenericConnector 

type GenericConnector interface {
	Halt()
	Server() *Server
	OnClosedConn(conn *outgoingConn)
	CloseAllCh() chan interface{}
	ForceUpdate()
	DispatchCommand(cmd commands.Command, idHash *[32]byte)
	DispatchReplication(cmd *commands.ReplicaWrite)
	QueueForRetry(cmd commands.Command, idHash [32]byte)
	ConnectionCount() int
}

type GenericListener 

type GenericListener interface {
	Halt()
	CloseOldConns(interface{}) error
	GetConnIdentities() (map[[constants.RecipientIDLength]byte]interface{}, error)
}

type Listener 

type Listener struct {
	sync.Mutex
	worker.Worker
	// contains filtered or unexported fields
}

func (*Listener) CloseOldConns 

func (l *Listener) CloseOldConns(ptr interface{}) error

func (*Listener) GetConnIdentities 

func (l *Listener) GetConnIdentities() (map[[sConstants.RecipientIDLength]byte]interface{}, error)

GetConnIdentities returns a slice of byte slices each corresponding to a currently connected client identity.

func (*Listener) Halt 

func (l *Listener) Halt()

type PKIWorker 

type PKIWorker struct {
	worker.Worker

	*pki.WorkerBase
	// contains filtered or unexported fields
}

func (*PKIWorker) ForceFetchPKI 

func (p *PKIWorker) ForceFetchPKI() error

ForceFetchPKI forces the PKI worker to fetch a new PKI document for the current epoch. This is useful for integration tests where you want to ensure the replica has the latest PKI document without waiting for the normal fetch cycle.

func (*PKIWorker) HasCurrentPKIDocument 

func (p *PKIWorker) HasCurrentPKIDocument() bool

HasCurrentPKIDocument returns true if the replica has a PKI document for the current epoch. This is useful for integration tests to check if the replica is ready.

func (*PKIWorker) PKIDocument 

func (p *PKIWorker) PKIDocument() *cpki.Document

PKIDocument returns the PKI document for the current epoch

func (*PKIWorker) ReplicasCopy 

func (p *PKIWorker) ReplicasCopy() map[[32]byte]*pki.ReplicaDescriptor

ReplicasCopy returns a copy of the replicas map

type ProxyRequest added in v0.0.54 

type ProxyRequest struct {
	ResponseCh      chan *commands.ReplicaMessageReply
	MKEMPrivateKey  nike.PrivateKey
	TargetPublicKey nike.PublicKey
	OriginalRequest *commands.ReplicaMessage
	Timestamp       time.Time
}

ProxyRequest represents a pending proxy request

type ProxyRequestManager added in v0.0.54 

type ProxyRequestManager struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

ProxyRequestManager manages pending proxy requests

func NewProxyRequestManager added in v0.0.54 

func NewProxyRequestManager(log *logging.Logger, requestTimeout time.Duration) *ProxyRequestManager

NewProxyRequestManager creates a new proxy request manager

func (*ProxyRequestManager) CleanupExpiredRequests added in v0.0.54 

func (p *ProxyRequestManager) CleanupExpiredRequests(timeout time.Duration)

CleanupExpiredRequests removes requests that have been waiting too long

func (*ProxyRequestManager) HandleReply added in v0.0.54 

func (p *ProxyRequestManager) HandleReply(reply *commands.ReplicaMessageReply) bool

HandleReply processes an incoming reply and routes it to the waiting request

func (*ProxyRequestManager) RegisterProxyRequest added in v0.0.54 

func (p *ProxyRequestManager) RegisterProxyRequest(envelopeHash [32]byte, mkemPrivateKey nike.PrivateKey, targetPublicKey nike.PublicKey, originalRequest *commands.ReplicaMessage) chan *commands.ReplicaMessageReply

RegisterProxyRequest registers a new proxy request and returns a response channel

func (*ProxyRequestManager) Shutdown added in v0.0.54 

func (p *ProxyRequestManager) Shutdown()

Shutdown gracefully shuts down the proxy request manager

type Server 

type Server struct {
	sync.WaitGroup

	PKIWorker *PKIWorker
	// contains filtered or unexported fields
}

func CreateTestServer 

func CreateTestServer(t *testing.T, cfg *config.Config, keys *TestKeys, logBackend *log.Backend) *Server

CreateTestServer creates a test server with PKI worker

func New 

func New(cfg *config.Config) (*Server, error)

New returns a new Server instance parameterized with the specific configuration.

func NewWithPKI 

func NewWithPKI(cfg *config.Config, pkiClient pki.Client) (*Server, error)

NewWithPKI returns a new Server instance with a custom PKI implementation. If pkiFactory is nil, the default PKI worker is used.

func (*Server) ConnectionCount added in v0.0.73 

func (s *Server) ConnectionCount() int

ConnectionCount returns the number of active outgoing connections to other replicas. This is useful for testing to verify inter-replica connections are established.

func (*Server) ForceConnectorUpdate added in v0.0.73 

func (s *Server) ForceConnectorUpdate()

ForceConnectorUpdate triggers the connector to rescan PKI and spawn new connections.

func (*Server) LogBackend 

func (s *Server) LogBackend() *log.Backend

func (*Server) ProxyManager added in v0.0.54 

func (s *Server) ProxyManager() *ProxyRequestManager

ProxyManager returns the proxy request manager

func (*Server) RotateLog 

func (s *Server) RotateLog()

RotateLog rotates the log file if logging to a file is enabled.

func (*Server) Shutdown 

func (s *Server) Shutdown()

Shutdown cleanly shuts down a given Server instance.

func (*Server) Wait 

func (s *Server) Wait()

Wait waits till the server is terminated for any reason.

type TestKeys 

type TestKeys struct {
	IdentityPubKey  sign.PublicKey
	IdentityPrivKey sign.PrivateKey
	IdentityKeyBlob []byte
	LinkPubKey      kem.PublicKey
	LinkPrivKey     kem.PrivateKey
	LinkKeyBlob     []byte
	ReplicaPubKey   nike.PublicKey
	ReplicaPrivKey  nike.PrivateKey
	ReplicaKeyBlob  []byte
}

GenerateTestKeys generates a complete set of test keys

func GenerateTestKeys 

func GenerateTestKeys(t *testing.T, schemes *TestSchemes) *TestKeys

GenerateTestKeys creates a complete set of test keys

type TestSchemes 

type TestSchemes struct {
	PKI     sign.Scheme
	Link    kem.Scheme
	Replica nike.Scheme
	Sphinx  nike.Scheme
}

TestSchemes holds the cryptographic schemes used in tests

func NewTestSchemes 

func NewTestSchemes() *TestSchemes

NewTestSchemes creates standard test schemes

func NewTestSchemesAdvanced 

func NewTestSchemesAdvanced() *TestSchemes

NewTestSchemesAdvanced creates advanced test schemes

Source Files

View all Source files

Directories

Path Synopsis
Package common contains code that is useful to replicas and couriers.
 Package common contains code that is useful to replicas and couriers.

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.